Xiaohongshu KOL Screening Agent (Tutorial)
Build an agent that evaluates 200 Xiaohongshu influencers in minutes: engagement scoring, content classification, audience quality signals, all for under $6.
TL;DR — A Python agent that screens 200 Xiaohongshu KOLs for a brand campaign in under 10 minutes, scoring each on engagement quality, content fit, and growth trajectory. Total data cost: $4-$6. Uses PGY (蒲公英) influencer marketplace data + note-level engagement metrics + Anthropic Claude for scoring.
Screening KOLs for a Xiaohongshu campaign manually means: open 200 profiles, scroll through posts, mentally estimate engagement, compare across candidates, and somehow rank them. Takes a marketing team 2-3 days.
An agent does it in 10 minutes for $6. Here is how.
What the agent produces
For each of 200 KOL candidates:
| Output | Source | Cost |
|---|---|---|
| PGY engagement score | xiaohongshu/pgy/blogger-detail | $0.02 |
| Follower growth trend (30 days) | xiaohongshu/pgy/blogger-fans-history | $0.02 |
| Content category distribution | xiaohongshu/app-v2/user-notes (sample 10) | $0.01 |
| Average engagement rate | Computed from note samples | — |
| LLM fitness score (1-10) | Claude Sonnet 4 | ~$0.01/batch |
Final output: a ranked CSV with scores, reasoning, and red flags.
The scoring model
The agent evaluates each KOL on four dimensions:
Total Score = 0.3 × Engagement Quality
+ 0.25 × Content Fit
+ 0.25 × Growth Trajectory
+ 0.2 × Audience Quality SignalsEngagement Quality (from note samples):
- Engagement rate = (likes + comments + saves) / estimated impressions
- Comment-to-like ratio (high ratio = genuine discussion, not just passive likes)
- Save rate (high saves = utility content, strong purchase intent signal on XHS)
Content Fit (LLM-evaluated):
- Does the creator’s recent content align with the brand’s category?
- Visual style consistency
- Tone match (educational vs entertainment vs lifestyle)
Growth Trajectory (from PGY fans history):
- 30-day follower growth rate
- Growth stability (steady vs spike-and-decline)
- Red flag: >50% growth in a week without viral content (possible fake followers)
Audience Quality Signals (inferred):
- Fan count vs engagement ratio (low engagement + high fans = suspicious)
- Geographic distribution if available
- PGY platform score (Xiaohongshu’s own internal ranking)
Complete code
#!/usr/bin/env python3
"""Xiaohongshu KOL screening agent.
Evaluates KOL candidates using PGY data + content sampling + LLM scoring.
Output: ranked CSV ready for brand team review.
"""
import csv
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
import requests
from openai import OpenAI
SANDBASE_API_KEY = os.environ["SANDBASE_API_KEY"]
HEADERS = {
"Authorization": f"Bearer {SANDBASE_API_KEY}",
"Content-Type": "application/json",
}
# OpenAI client for LLM calls only
client = OpenAI(
base_url="https://api.sandbase.ai/v1",
api_key=SANDBASE_API_KEY,
)
@dataclass
class KOLScore:
blogger_id: str
nickname: str
follower_count: int
engagement_rate: float
save_rate: float
growth_30d: float
pgy_score: float
content_fit: float
total_score: float
reasoning: str
red_flags: list[str]
def fetch_blogger_detail(blogger_id: str) -> dict:
"""Get PGY blogger overview with engagement metrics."""
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={"model": "xiaohongshu/pgy/blogger-detail", "blogger_id": blogger_id},
)
return resp.json()
def fetch_fans_history(blogger_id: str) -> dict:
"""Get 30-day follower growth data."""
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={"model": "xiaohongshu/pgy/blogger-fans-history", "blogger_id": blogger_id},
)
return resp.json()
def fetch_recent_notes(blogger_id: str, count: int = 10) -> list:
"""Sample recent notes for engagement analysis."""
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={"model": "xiaohongshu/app-v2/user-faved-notes", "user_id": blogger_id, "count": count},
)
return resp.json()
def compute_engagement_metrics(notes: list) -> dict:
"""Calculate engagement rates from note samples."""
if not notes:
return {"engagement_rate": 0, "save_rate": 0, "comment_ratio": 0}
total_likes = sum(n.get("likes", 0) for n in notes)
total_comments = sum(n.get("comments", 0) for n in notes)
total_saves = sum(n.get("collects", 0) for n in notes)
total_interactions = total_likes + total_comments + total_saves
# Engagement rate approximation (without impression data, use followers as proxy)
return {
"total_interactions": total_interactions,
"avg_interactions_per_note": total_interactions / len(notes) if notes else 0,
"save_rate": total_saves / max(total_interactions, 1),
"comment_ratio": total_comments / max(total_likes, 1),
}
def compute_growth(fans_history: dict) -> dict:
"""Analyze follower growth trajectory."""
data_points = fans_history.get("history", [])
if len(data_points) < 2:
return {"growth_30d": 0, "stable": True, "spike_detected": False}
first = data_points[0].get("count", 0)
last = data_points[-1].get("count", 0)
growth_rate = (last - first) / max(first, 1)
# Detect suspicious spikes
max_daily_change = 0
for i in range(1, len(data_points)):
daily_change = abs(data_points[i].get("count", 0) - data_points[i-1].get("count", 0))
max_daily_change = max(max_daily_change, daily_change)
spike_threshold = first * 0.1 # >10% in a single day is suspicious
return {
"growth_30d": growth_rate,
"stable": max_daily_change < spike_threshold,
"spike_detected": max_daily_change >= spike_threshold,
}
def score_content_fit(notes: list, brand_category: str) -> dict:
"""LLM evaluates content fit with the brand."""
if not notes:
return {"score": 0, "reasoning": "No notes available"}
note_descriptions = [n.get("desc", n.get("title", ""))[:100] for n in notes[:10]]
resp = client.chat.completions.create(
model="anthropic/claude-sonnet-4",
messages=[{
"role": "user",
"content": f"""Rate this KOL's content fit for a brand in the "{brand_category}" category.
Their recent 10 note titles/descriptions:
{json.dumps(note_descriptions, ensure_ascii=False)}
Score 1-10 where:
- 1-3: Completely unrelated content
- 4-6: Some overlap but not core focus
- 7-9: Strong category alignment
- 10: Perfect fit, exact target audience
Respond in JSON: {{"score": N, "reasoning": "one sentence"}}"""
}],
max_tokens=200,
)
try:
return json.loads(resp.choices[0].message.content)
except json.JSONDecodeError:
return {"score": 5, "reasoning": "Could not parse LLM response"}
def screen_kol(blogger_id: str, brand_category: str) -> KOLScore:
"""Full screening of a single KOL candidate."""
# Fetch data (4-5 API calls per KOL)
detail = fetch_blogger_detail(blogger_id)
fans = fetch_fans_history(blogger_id)
notes = fetch_recent_notes(blogger_id, count=10)
# Compute metrics
engagement = compute_engagement_metrics(notes)
growth = compute_growth(fans)
content = score_content_fit(notes, brand_category)
# Detect red flags
red_flags = []
follower_count = detail.get("follower_count", 0)
if follower_count > 100000 and engagement["avg_interactions_per_note"] < 100:
red_flags.append("Very low engagement for follower count")
if growth.get("spike_detected"):
red_flags.append("Suspicious follower spike detected")
if engagement["save_rate"] < 0.05:
red_flags.append("Low save rate (weak purchase intent)")
# Calculate total score
engagement_score = min(engagement["avg_interactions_per_note"] / 500, 10) # Normalize to 0-10
growth_score = min(growth["growth_30d"] * 100, 10) # 10% monthly growth = score 10
pgy_score = detail.get("pgy_score", 5)
content_score = content.get("score", 5)
total = (
0.3 * engagement_score
+ 0.25 * content_score
+ 0.25 * growth_score
+ 0.2 * pgy_score
)
return KOLScore(
blogger_id=blogger_id,
nickname=detail.get("nickname", "Unknown"),
follower_count=follower_count,
engagement_rate=engagement["avg_interactions_per_note"],
save_rate=engagement["save_rate"],
growth_30d=growth["growth_30d"],
pgy_score=pgy_score,
content_fit=content_score,
total_score=round(total, 2),
reasoning=content.get("reasoning", ""),
red_flags=red_flags,
)
def run_screening(blogger_ids: list[str], brand_category: str, output_path: str):
"""Screen all candidates and output ranked CSV."""
results: list[KOLScore] = []
for i, blogger_id in enumerate(blogger_ids):
print(f"[{i+1}/{len(blogger_ids)}] Screening {blogger_id}...")
try:
score = screen_kol(blogger_id, brand_category)
results.append(score)
except Exception as e:
print(f" Error: {e}")
time.sleep(0.5) # Gentle rate limiting
# Sort by total score descending
results.sort(key=lambda x: x.total_score, reverse=True)
# Write CSV
with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow([
"Rank", "Nickname", "Blogger ID", "Followers", "Engagement/Note",
"Save Rate", "30d Growth", "PGY Score", "Content Fit",
"Total Score", "Reasoning", "Red Flags"
])
for rank, r in enumerate(results, 1):
writer.writerow([
rank, r.nickname, r.blogger_id, r.follower_count,
f"{r.engagement_rate:.0f}", f"{r.save_rate:.2%}",
f"{r.growth_30d:.1%}", r.pgy_score, r.content_fit,
r.total_score, r.reasoning, "; ".join(r.red_flags) or "None"
])
print(f"\nScreening complete. {len(results)} KOLs scored.")
print(f"Output: {output_path}")
print(f"Top 5:")
for r in results[:5]:
print(f" {r.nickname}: {r.total_score} ({r.reasoning})")
if __name__ == "__main__":
# Example: screen 200 beauty KOLs
BLOGGER_IDS = ["id_1", "id_2", "..."] # Your 200 candidate IDs
run_screening(BLOGGER_IDS, brand_category="beauty & skincare", output_path="kol_ranking.csv")Cost calculation for 200 KOLs
| Operation | Per KOL | × 200 | Unit price | Total |
|---|---|---|---|---|
| PGY blogger detail | 1 call | 200 | $0.02 | $4.00 |
| PGY fans history | 1 call | 200 | $0.02 | $4.00 |
| Recent notes | 1 call | 200 | $0.001 | $0.20 |
| LLM content scoring | 1 call | 200 | ~$0.005 | $1.00 |
| Total | $9.20 |
Wait — that is higher than I initially estimated. The $0.02 PGY calls add up. Here is how to optimize:
Optimization 1: Skip fans history for first pass. Screen on engagement + content fit first (cost: $5.20 for 200). Only fetch fans history for the top 50 finalists ($1.00). New total: $6.20.
Optimization 2: Batch LLM scoring. Instead of 200 individual LLM calls, batch 10 KOLs per prompt. 20 LLM calls × $0.02 = $0.40 instead of $1.00. New total: $5.60.
Limitations
- PGY data availability: Not all Xiaohongshu creators are on PGY. Smaller creators (<10K followers) may not have PGY profiles. Fall back to note-level data only.
- Engagement approximation: Without impression data, engagement rate uses interactions/followers as a proxy. This disadvantages large accounts with old followers.
- Content fit scoring: LLM scoring is subjective. Run calibration with 10 known-good and 10 known-bad KOLs to tune the prompt before full batch.
- No fake-follower detection: The spike detection is a heuristic, not definitive. Combine with third-party audit tools for high-stakes campaigns.
FAQ
How do I get 200 candidate blogger IDs?
Use xiaohongshu/pgy/blogger-list with category and follower range filters. One call returns a paginated list. Example: {"category": "beauty", "follower_min": 10000, "follower_max": 500000, "page": 1}.
Can I run this for categories other than beauty?
Yes. Change brand_category to anything: “fitness”, “food”, “tech”, “parenting”, “fashion”. The LLM scoring adapts to whatever category you specify.
How long does screening 200 KOLs take?
About 8-12 minutes with 0.5s delays between candidates. The bottleneck is the LLM scoring calls, not the data fetches.
What is a “good” total score?
In practice: >7.0 is strong fit, 5.0-7.0 is worth considering, <5.0 is likely not aligned. But calibrate against your specific brand — run 10 known-good KOLs first.
For more on Xiaohongshu data coverage, see Weibo & Xiaohongshu APIs on SandBase. For the broader social data landscape, see our social media data API guide.
