Tutorials

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.

By SandBase Team
#xiaohongshu #kol-marketing #ai-agent #tutorial #python

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:

OutputSourceCost
PGY engagement scorexiaohongshu/pgy/blogger-detail$0.02
Follower growth trend (30 days)xiaohongshu/pgy/blogger-fans-history$0.02
Content category distributionxiaohongshu/app-v2/user-notes (sample 10)$0.01
Average engagement rateComputed 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 Signals

Engagement 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

from openai import OpenAI

client = OpenAI(
    base_url="https://api.sandbase.ai/v1",
    api_key=os.environ["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 = client.chat.completions.create(
        model="xiaohongshu/pgy/blogger-detail",
        messages=[],
        extra_body={"blogger_id": blogger_id},
    )
    return json.loads(resp.choices[0].message.content)


def fetch_fans_history(blogger_id: str) -> dict:
    """Get 30-day follower growth data."""
    resp = client.chat.completions.create(
        model="xiaohongshu/pgy/blogger-fans-history",
        messages=[],
        extra_body={"blogger_id": blogger_id},
    )
    return json.loads(resp.choices[0].message.content)


def fetch_recent_notes(blogger_id: str, count: int = 10) -> list:
    """Sample recent notes for engagement analysis."""
    resp = client.chat.completions.create(
        model="xiaohongshu/app-v2/user-faved-notes",
        messages=[],
        extra_body={"user_id": blogger_id, "count": count},
    )
    return json.loads(resp.choices[0].message.content)


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

OperationPer KOL× 200Unit priceTotal
PGY blogger detail1 call200$0.02$4.00
PGY fans history1 call200$0.02$4.00
Recent notes1 call200$0.001$0.20
LLM content scoring1 call200~$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.

You May Also Like

Cover image for Build a Douyin Competitor Monitor Agent
Tutorials

Build a Douyin Competitor Monitor Agent

Step-by-step tutorial: build an agent that tracks competitor Douyin accounts, detects new videos, and generates daily briefings for under $1/month.

Cover image for Build a Social Listening Agent: Weibo + Douyin
Tutorials

Build a Social Listening Agent: Weibo + Douyin

Tutorial: build an agent that monitors Weibo hot search and Douyin trending every 15 minutes, flags brand mentions, and sends alerts via webhook.

Cover image for Weibo & Xiaohongshu Data APIs Now on SandBase
Product Updates

Weibo & Xiaohongshu Data APIs Now on SandBase

64 Weibo and 36 Xiaohongshu data operations are now available through SandBase, covering hot search, user posts, influencer analytics, and commerce data.