用 OpenAI SDK 搭社媒监控 Agent
用 OpenAI SDK 做 LLM 分析 + requests 直接调 SandBase 社媒数据 API 搭建社交监控 Agent。完整代码教程。
结论先行 — 约 100 行 Python 核心代码实现一个完整可用的社媒监控 Agent。使用
requests直接调 SandBase 社媒数据 API（POST /v1/run），使用 OpenAI SDK 做 LLM 推理。监控抖音和微博品牌提及、分析情绪、产生告警。数据获取走直接 HTTP，LLM 分析走 OpenAI 兼容接口。
我们在构建什么
一个社媒监控 Agent：
- 定期检查抖音和微博上指定关键词的提及
- 用 LLM 分析情绪和紧急程度
- 检测到负面情绪或量级飙升时产生结构化告警
- 数据获取用 requests 直接 HTTP，LLM 推理用 OpenAI SDK
这是最小生产可用模式。可以扩展更多平台、更复杂分析或自动回复。
为什么这样分
SandBase 的社媒数据 API 通过 POST /v1/run 直接返回结构化数据，LLM 模型通过 OpenAI 兼容的 chat completions 接口。这意味着：
requests库做数据获取（社媒 API 直接返回 JSON）- OpenAI SDK做 LLM 推理（chat completions + tool-calling）
- 一把 API key 同时认证两者
- 清晰分离：数据调用同步立即返回；LLM 调用走熟悉的 chat 接口
前置要求
pip install openai requests pydantic你需要一个 SandBase API key，有权访问：
- LLM 模型（我们用 GPT-4.1-mini 兼顾成本）
- 抖音数据 API
- 微博数据 API
完整 Agent 代码
"""
社媒监控 Agent — OpenAI SDK + 直接 HTTP 模式
监控抖音和微博品牌提及，分析情绪。
数据获取用 requests，LLM 推理用 OpenAI SDK。
"""
import json
import time
from datetime import datetime
import requests
from openai import OpenAI
from pydantic import BaseModel
# --- 配置 ---
SANDBASE_API_KEY = "your-sandbase-api-key"
MONITOR_KEYWORDS = ["你的品牌", "品牌产品名", "竞品名"]
CHECK_INTERVAL_SECONDS = 900 # 15 分钟
ALERT_THRESHOLD_NEGATIVE = 0.3 # 负面 > 30% 触发告警
# --- 客户端设置 ---
# 直接 HTTP 请求头，用于社媒数据 API
HEADERS = {
"Authorization": f"Bearer {SANDBASE_API_KEY}",
"Content-Type": "application/json",
}
# OpenAI SDK 客户端，仅用于 LLM 推理
client = OpenAI(
base_url="https://api.sandbase.ai/v1",
api_key=SANDBASE_API_KEY,
)
# --- 数据模型 ---
class Alert(BaseModel):
timestamp: str
severity: str # "low", "medium", "high", "critical"
keyword: str
platform: str
reason: str
details: str
# --- 工具定义 ---
tools = [
{
"type": "function",
"function": {
"name": "search_douyin_content",
"description": "搜索抖音匹配关键词的视频内容，返回近期视频及互动指标。",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "搜索关键词"},
"count": {"type": "integer", "description": "返回结果数（最多 20）", "default": 10}
},
"required": ["keyword"]
}
}
},
{
"type": "function",
"function": {
"name": "search_weibo_topics",
"description": "搜索微博匹配关键词的帖子和讨论，返回近期帖子及互动数据。",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "搜索关键词"},
"count": {"type": "integer", "description": "返回结果数（最多 20）", "default": 10}
},
"required": ["keyword"]
}
}
},
{
"type": "function",
"function": {
"name": "get_douyin_video_comments",
"description": "获取特定抖音视频的评论用于情绪分析。",
"parameters": {
"type": "object",
"properties": {
"video_id": {"type": "string", "description": "抖音视频 ID"},
"count": {"type": "integer", "description": "获取评论数", "default": 20}
},
"required": ["video_id"]
}
}
}
]
# --- 工具执行（通过直接 HTTP 调 SandBase 数据 API） ---
def execute_tool(tool_name: str, arguments: dict) -> str:
"""路由工具调用到对应的 SandBase 数据 API（直接 HTTP）。"""
if tool_name == "search_douyin_content":
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={
"model": "douyin/search/general",
"keyword": arguments["keyword"],
"count": arguments.get("count", 10),
"sort_type": "0",
},
)
return json.dumps(resp.json(), ensure_ascii=False)
elif tool_name == "search_weibo_topics":
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={
"model": "weibo/web-v2/search",
"keyword": arguments["keyword"],
"count": arguments.get("count", 10),
},
)
return json.dumps(resp.json(), ensure_ascii=False)
elif tool_name == "get_douyin_video_comments":
resp = requests.post(
"https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={
"model": "douyin/web/fetch-video-comments",
"video_id": arguments["video_id"],
"count": arguments.get("count", 20),
},
)
return json.dumps(resp.json(), ensure_ascii=False)
return json.dumps({"error": f"未知工具: {tool_name}"})
# --- Agent 循环 ---
def run_monitoring_cycle(keywords: list[str]) -> list[Alert]:
"""运行一个监控周期：搜索、分析、告警。"""
alerts = []
for keyword in keywords:
# 步骤 1：让 Agent 监控此关键词
messages = [
{
"role": "system",
"content": """你是社媒监控 Agent。对每个关键词：
1. 搜索抖音和微博的近期提及
2. 分析结果的情绪
3. 识别紧急项（投诉、危机、负面病毒内容）
4. 产出结构化评估
简洁明了。聚焦可操作信号，不要总结中性内容。
如果负面内容互动异常高，立即标记。"""
},
{
"role": "user",
"content": f"监控关键词：'{keyword}'。搜索两个平台并提供情绪分析。"
}
]
# 步骤 2：Agent 推理并调用工具
max_iterations = 5
for _ in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4.1-mini",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
if not message.tool_calls:
break
for tool_call in message.tool_calls:
arguments = json.loads(tool_call.function.arguments)
result = execute_tool(tool_call.function.name, arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 步骤 3：提取结构化评估
assessment_prompt = messages + [{
"role": "user",
"content": """基于你的分析，输出 JSON：
{
"keyword": "关键词",
"total_mentions": <数量>,
"sentiment": {"positive": <百分比>, "neutral": <百分比>, "negative": <百分比>},
"alert_needed": true/false,
"alert_severity": "low|medium|high|critical",
"alert_reason": "告警原因",
"summary": "一段话总结"
}
只输出 JSON。"""
}]
assessment_response = client.chat.completions.create(
model="gpt-4.1-mini",
messages=assessment_prompt
)
try:
assessment = json.loads(assessment_response.choices[0].message.content)
except json.JSONDecodeError:
continue
# 步骤 4：需要时产生告警
if assessment.get("alert_needed"):
alert = Alert(
timestamp=datetime.now().isoformat(),
severity=assessment.get("alert_severity", "medium"),
keyword=keyword,
platform="douyin, weibo",
reason=assessment.get("alert_reason", "负面情绪升高"),
details=assessment.get("summary", "")
)
alerts.append(alert)
print(f"🚨 告警 [{alert.severity.upper()}] — {alert.keyword}: {alert.reason}")
else:
neg_pct = assessment.get('sentiment', {}).get('negative', 0)
print(f"✓ {keyword}: 正常（{assessment.get('total_mentions', 0)} 条提及，"
f"{neg_pct}% 负面）")
return alerts
# --- 主循环 ---
def main():
"""按计划运行监控 Agent。"""
print(f"🔍 社媒监控 Agent 启动")
print(f" 关键词：{MONITOR_KEYWORDS}")
print(f" 间隔：{CHECK_INTERVAL_SECONDS} 秒")
print()
while True:
print(f"--- 周期开始：{datetime.now().isoformat()} ---")
alerts = run_monitoring_cycle(MONITOR_KEYWORDS)
if alerts:
for alert in alerts:
if alert.severity in ("high", "critical"):
print(f"\n{'='*60}")
print(f"🔴 {alert.severity.upper()} — {alert.keyword}")
print(f" 原因：{alert.reason}")
print(f" 详情：{alert.details}")
print(f"{'='*60}\n")
else:
print("本周期无告警。")
print(f"--- 下一周期 {CHECK_INTERVAL_SECONDS}s 后 ---\n")
time.sleep(CHECK_INTERVAL_SECONDS)
if __name__ == "__main__":
main()模式工作原理
核心洞察：一切通过同一个 OpenAI 客户端：
# LLM 推理 — OpenAI SDK（chat completions 正确）
client.chat.completions.create(model="gpt-4.1-mini", messages=...)
# 数据获取 — 直接 HTTP 到 /v1/run（社媒数据 API 正确方式）
requests.post("https://api.sandbase.ai/v1/run",
headers=HEADERS,
json={"model": "douyin/search/general", "keyword": "...", "count": 10})SandBase 的社媒数据 API 不是 LLM——它们不接受 messages，也不返回 chat completions。它们接受结构化参数，通过 /v1/run 端点直接返回数据。OpenAI SDK 仅用于 LLM 推理层（GPT-4.1-mini 做分析和 tool-calling）。
成本拆解
每个监控周期（3 个关键词，每个查 2 个平台）：
数据 API 调用：
6 次搜索（3 关键词 × 2 平台）：6 × $0.001 = $0.006
~3 次评论获取（标记的视频）：3 × $0.001 = $0.003
LLM 调用：
6 次推理（tool-use）：~2K 输入 + 500 输出 tokens 每次
= 6 × $0.0019 = $0.0114
3 次评估提取：~1K 输入 + 200 输出每次
= 3 × $0.001 = $0.003
每周期合计：~$0.023每 15 分钟运行：
日：96 周期 × $0.023 = $2.21
月：~$66对比人工监控：一名分析师每天花 2 小时做品牌监控，月成本 ¥15,000–30,000。这个 Agent 以 2% 的成本提供 7×24 覆盖。
扩展 Agent
添加更多平台
# 添加小红书监控
{
"type": "function",
"function": {
"name": "search_xiaohongshu_notes",
"description": "搜索小红书匹配关键词的笔记。",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string"},
"count": {"type": "integer", "default": 10}
},
"required": ["keyword"]
}
}
}添加趋势对比
class TrendTracker:
def __init__(self):
self.history = {}
def record(self, keyword: str, mentions: int, negative_pct: float):
if keyword not in self.history:
self.history[keyword] = []
self.history[keyword].append((time.time(), mentions, negative_pct))
def detect_spike(self, keyword: str, window_hours: int = 24) -> bool:
"""检测当前量级是否为 24 小时均值的 3 倍。"""
if keyword not in self.history or len(self.history[keyword]) < 10:
return False
cutoff = time.time() - (window_hours * 3600)
recent = [m for t, m, _ in self.history[keyword] if t > cutoff]
if not recent:
return False
avg = sum(recent) / len(recent)
return recent[-1] > avg * 3添加自动回复草拟
def draft_response(alert: Alert) -> str:
"""用 LLM 为高严重度告警草拟回复。"""
response = client.chat.completions.create(
model="gpt-4.1-mini",
messages=[
{"role": "system", "content": "你是品牌公关专家。"
"为以下社媒问题草拟简短、共情的回复。"
"保持专业，承认关切，提供下一步。"},
{"role": "user", "content": f"问题：{alert.reason}\n详情：{alert.details}\n"
"草拟一条适合在社媒发布的回复。"}
]
)
return response.choices[0].message.content生产考量
限流
from collections import deque
class RateLimiter:
def __init__(self, max_calls: int, window_seconds: int):
self.max_calls = max_calls
self.window = window_seconds
self.calls = deque()
async def acquire(self):
now = time.time()
while self.calls and self.calls[0] < now - self.window:
self.calls.popleft()
if len(self.calls) >= self.max_calls:
sleep_time = self.calls[0] + self.window - now
await asyncio.sleep(sleep_time)
self.calls.append(time.time())
limiter = RateLimiter(max_calls=60, window_seconds=60)错误恢复
def execute_tool_with_retry(tool_name: str, arguments: dict, max_retries: int = 3) -> str:
"""带指数退避的工具执行。"""
for attempt in range(max_retries):
try:
return execute_tool(tool_name, arguments)
except Exception as e:
if attempt == max_retries - 1:
return json.dumps({"error": str(e), "tool": tool_name})
time.sleep(2 ** attempt)
return json.dumps({"error": "超出最大重试"})与完整版对比
| 特性 | 本教程 | 完整版 Agent |
|---|---|---|
| 平台 | 2（抖音、微博） | 3+（+ 小红书） |
| 分析深度 | 关键词级情绪 | 线程级、KOL 追踪 |
| 状态管理 | 每周期无状态 | 历史趋势对比 |
| 输出 | 控制台告警 | 多渠道（Slack、邮件、仪表盘） |
| 代码复杂度 | ~100 行核心 | ~400 行 |
| 月成本 | ~$66 | ~$200–400 |
用这个模式作起点。只在简单版的限制真正阻碍你的用例时才增加复杂度。
要点总结
- 一个 SDK，一个客户端。 OpenAI SDK 模式让你不需要为数据访问和 LLM 推理准备不同库。
- Tool-calling 是集成模式。 把社媒数据端点定义为工具，让 LLM 决定何时调用。
- 成本由 LLM 调用主导。 $0.001 的数据 API 调用可忽略。先优化 LLM 层。
- 先简单，后扩展。 100 行的能用 Agent 胜过永远发不出去的完美 Agent。
- Sync API 让循环简洁。 因为社媒数据 API 是 sync-only，Agent 循环保持干净——无回调、无轮询、无任务管理。
