Solo Unicorn2,820 字
我怎么设计了一个 8-Agent 社群管理系统
AI AgentMulti-Agent社群管理一人独角兽俱乐部实战案例
我怎么设计了一个 8-Agent 社群管理系统
开场
一人独角兽俱乐部现在有 2000 多名成员，横跨金融、科技、设计、法律等行业。一个人管这么大的社群，靠人力根本不现实。我花了三周时间搭了一套 8-Agent 系统，现在日均自动处理 200+ 条消息，每月 API 成本不到 $45。这篇文章完整拆解这个系统的设计思路、架构细节和踩过的坑。
问题背景
社群管理的痛点很具体：
- 消息量大：每天 200-400 条消息，内容从技术问题到活动报名到闲聊
- 响应时效：成员期望几分钟内得到回复，而不是几小时
- 知识分散：历史讨论、活动信息、资源链接散落在各个角落
- 重复劳动：60% 以上的问题是重复的，但每次都要手动回答
我试过几种方案：
- 纯人工：一天花 3-4 小时回消息，不可持续
- 关键词机器人：太傻，匹配不准，成员体验差
- 单个 AI Agent：prompt 塞了 8000 token，什么都能做但什么都做不好
最后选择了 Multi-Agent 架构，核心原因：社群管理的子任务差异太大——迎新、问答、内容推荐、活动管理，每个任务需要完全不同的上下文和行为模式。
核心架构
设计原则
三条原则，搭之前就定好的：
- Router-first：所有消息先过 Router Agent，它只做分类不做回复，保证分流准确
- 专才优于通才：每个 Worker Agent 只做一件事，prompt 精简到 800 token 以内
- Human escalation：任何 Agent 不确定的问题（confidence < 0.7）自动 escalate 到我
系统架构图
用户消息 → Router Agent (分类)
│
┌───────────┼───────────────────────┐
│ │ │ │
▼ ▼ ▼ ▼
Greeter Q&A Agent Content Event
Agent (问答) Curator Scheduler
(迎新) (内容推荐) (活动管理)
│ │ │ │
│ ┌─────┼─────┐ │ │
│ ▼ ▼ │ │
│ Knowledge Search │ │
│ Base Agent Agent │ │
│ (知识库) (搜索) │ │
│ │ │
└───────────┬───────────┘ │
▼ │
Digest Agent ←────────────────┘
(日报汇总)
8 个 Agent 的职责
| Agent | 职责 | 模型 | 平均延迟 | 日均调用 |
|---|---|---|---|---|
| Router | 消息分类和分流 | GPT-4.1-mini | 0.4s | 220 |
| Greeter | 新成员欢迎和引导 | GPT-4.1-mini | 0.8s | 8 |
| Q&A Agent | 回答社群问题 | Claude Sonnet 4.5 | 2.1s | 85 |
| Content Curator | 推荐相关内容和资源 | GPT-4.1 | 1.5s | 35 |
| Event Scheduler | 活动创建和报名管理 | GPT-4.1-mini | 0.6s | 15 |
| Knowledge Base | 从历史讨论中检索信息 | text-embedding-3-small | 0.3s | 85 |
| Search Agent | 搜索外部信息 | GPT-4.1-mini | 1.8s | 25 |
| Digest Agent | 生成每日/每周社群摘要 | Claude Sonnet 4.5 | 4.2s | 1 |
实现细节
Step 1：Router Agent — 最关键的一环
Router 的准确率直接决定整个系统的表现。我用 GPT-4.1-mini 做分类，因为这个任务不需要强推理，需要的是速度和稳定性。
from openai import OpenAI
from pydantic import BaseModel
from enum import Enum
client = OpenAI()
class MessageCategory(str, Enum):
GREETING = "greeting" # 新人打招呼
QUESTION = "question" # 提问
CONTENT = "content" # 内容分享/讨论
EVENT = "event" # 活动相关
CHITCHAT = "chitchat" # 闲聊（不处理）
ESCALATE = "escalate" # 需要人工介入
class RouterResult(BaseModel):
category: MessageCategory
confidence: float # 0-1 的置信度
sub_topic: str # 细分主题，给下游 Agent 用
needs_context: bool # 是否需要查历史消息
ROUTER_PROMPT = """你是社群消息分类器。对每条消息做分类。
规则：
- greeting: 新成员自我介绍或打招呼
- question: 任何形式的提问（技术、工具、职业、商业）
- content: 分享文章、工具、观点
- event: 活动报名、时间查询、线下聚会
- chitchat: 日常闲聊、表情包、无实质内容
- escalate: 投诉、争议、敏感话题、你不确定的内容
置信度低于 0.7 时，category 设为 escalate。"""
def route_message(message: str, user_info: dict) -> RouterResult:
"""消息分类：快速、准确、低成本"""
response = client.beta.chat.completions.parse(
model="gpt-4.1-mini",
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": f"用户信息：{user_info}\n消息：{message}"}
],
response_format=RouterResult,
temperature=0.1, # 分类任务用低温度
)
return response.choices[0].message.parsed
踩坑经验：最初 Router 的 prompt 里没有 escalate 分类，导致遇到模糊消息时随机分到其他类别。加了 escalate + confidence 阈值后，分流准确率从 82% 提升到 94%。
Step 2：Q&A Agent — 核心工作马
Q&A Agent 处理量最大，也是最需要质量的。它的工作流程：先查 Knowledge Base，有历史回答就直接引用；没有就调 Search Agent 搜外部信息。
async def qa_agent(question: str, context: dict) -> str:
"""Q&A Agent：先查内部知识库，再查外部"""
# Step 1: 检索内部知识库
kb_results = await knowledge_base.search(
query=question,
top_k=3,
min_score=0.78 # 相似度阈值
)
if kb_results and kb_results[0].score > 0.85:
# 高相似度：直接基于历史回答生成
source = "knowledge_base"
reference = kb_results[0].content
else:
# 低相似度：搜索外部信息
source = "web_search"
reference = await search_agent.search(question)
# Step 2: 生成回答
response = await client.chat.completions.create(
model="claude-sonnet-4-5", # 回答质量要高，用好模型
messages=[
{"role": "system", "content": QA_PROMPT},
{"role": "user", "content": f"""问题：{question}
参考资料（来源：{source}）：{reference}
用户背景：{context.get('user_profile', '未知')}"""}
],
max_tokens=500, # 社群回答不宜太长
)
answer = response.choices[0].message.content
# Step 3: 新回答写入知识库（供下次检索）
await knowledge_base.upsert(
content=f"Q: {question}\nA: {answer}",
metadata={"source": source, "timestamp": datetime.now().isoformat()}
)
return answer
Step 3：Greeter Agent — 新人体验
GREETER_PROMPT = """你是一人独角兽俱乐部的迎新助手。风格：热情但不油腻，专业但不冷淡。
任务：
1. 欢迎新成员
2. 根据他们的自我介绍，推荐 2-3 个可能感兴趣的讨论话题
3. 告知社群规则（简洁版）
4. 引导他们做一个简短的自我介绍
保持在 150 字以内。不要用表情符号。"""
async def greet_new_member(intro_message: str, member_name: str) -> str:
"""新成员迎新：个性化欢迎 + 话题推荐"""
# 检索最近的热门讨论，推荐给新人
hot_topics = await get_recent_hot_topics(limit=5)
response = await client.chat.completions.create(
model="gpt-4.1-mini",
messages=[
{"role": "system", "content": GREETER_PROMPT},
{"role": "user", "content": f"""新成员：{member_name}
自我介绍：{intro_message}
最近热门话题：{hot_topics}"""}
],
max_tokens=200,
)
return response.choices[0].message.content
Step 4：Knowledge Base — RAG 的核心
知识库用 Qdrant 做 vector store，存所有历史问答和讨论内容。
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
class CommunityKnowledgeBase:
def __init__(self):
self.client = QdrantClient(url="http://localhost:6333")
self.collection = "community_knowledge"
self.embedding_model = "text-embedding-3-small"
async def search(self, query: str, top_k: int = 3,
min_score: float = 0.75) -> list:
"""语义检索：找到最相关的历史内容"""
# 生成 query embedding
query_vector = await self._embed(query)
results = self.client.query_points(
collection_name=self.collection,
query=query_vector,
limit=top_k,
score_threshold=min_score,
)
return results.points
async def upsert(self, content: str, metadata: dict):
"""新内容写入知识库"""
vector = await self._embed(content)
point = PointStruct(
id=str(uuid4()),
vector=vector,
payload={"content": content, **metadata}
)
self.client.upsert(
collection_name=self.collection,
points=[point]
)
关键参数：embedding 用 text-embedding-3-small（$0.02/1M tokens），约 15,000 条历史内容的完整索引成本不到 $0.30。检索延迟 p95 约 45ms。
Step 5：Digest Agent — 日报自动生成
每天凌晨自动跑一次，汇总当天的讨论精华。
async def generate_daily_digest(date: str) -> str:
"""生成每日社群精华摘要"""
# 拉取当天所有非闲聊消息
messages = await get_messages_by_date(date, exclude=["chitchat"])
response = await client.chat.completions.create(
model="claude-sonnet-4-5", # 摘要需要强归纳能力
messages=[
{"role": "system", "content": """生成社群日报。格式：
## 今日精华（日期）
### 热门讨论（3-5 条）
### 实用资源（附链接）
### 新成员（欢迎 XX 位新朋友）
### 明日活动预告
保持简洁，每条不超过两句话。"""},
{"role": "user", "content": f"今日消息（共 {len(messages)} 条）：\n{format_messages(messages)}"}
],
max_tokens=800,
)
return response.choices[0].message.content
实战经验
生产数据（过去 30 天平均）
| 指标 | 数据 |
|---|---|
| 日均消息处理量 | 215 条 |
| Router 分流准确率 | 94.2% |
| Q&A 回答满意度（成员反馈） | 87% |
| 平均响应延迟 | 2.3 秒 |
| Human escalation 比例 | 8.5% |
| 月均 API 成本 | $42.80 |
成本拆解
| Agent | 模型 | 月均 token | 月成本 |
|---|---|---|---|
| Router | GPT-4.1-mini | 1.2M | $0.72 |
| Greeter | GPT-4.1-mini | 180K | $0.11 |
| Q&A | Claude Sonnet 4.5 | 1.8M | $18.90 |
| Content Curator | GPT-4.1 | 650K | $2.60 |
| Event Scheduler | GPT-4.1-mini | 280K | $0.17 |
| Knowledge Base | embedding-3-small | 900K | $0.02 |
| Search Agent | GPT-4.1-mini | 480K | $0.29 |
| Digest Agent | Claude Sonnet 4.5 | 350K | $3.68 |
| Qdrant hosting | - | - | $16.31 |
| 合计 | $42.80 |
Q&A Agent 吃掉了 44% 的 API 成本，因为它用了 Claude Sonnet 4.5 且调用量最大。如果换成 GPT-4.1，成本能降 40%，但回答质量会明显下降——这是我测试过的。
踩过的坑
坑 1：Router 的 prompt 太复杂。最初 Router 的 prompt 有 2000 token，不仅分类还分析情绪、提取关键词。结果延迟高且不稳定。后来砍到 500 token，只做分类，准确率反而提高了。
坑 2：Knowledge Base 的冷启动。系统上线第一周，知识库几乎是空的，Q&A Agent 动不动就要调 Search Agent，延迟和成本都高。解决方案：上线前先导入了 500 条 FAQ，覆盖 80% 的常见问题。
坑 3：Digest Agent 的幻觉。有一次日报里出现了不存在的讨论话题，因为模型"补脑"了一些内容。解决方案：在 prompt 里明确写"只汇总提供的消息，不要添加任何原文中没有的信息"，并加了一个简单的事实核查步骤——把生成的摘要和原始消息做关键词交叉验证。
坑 4：晚间高峰期的并发。晚 8-10 点消息量是平时的 3 倍，API rate limit 成了瓶颈。解决方案：对非紧急消息（内容推荐、日报）做异步处理，高峰期只保证 Router 和 Q&A 的实时响应。
总结
三条 takeaway：
- Router Agent 是 Multi-Agent 系统的命门——投入 50% 的调试时间在 Router 上不为过。分流准确率每提高 1%，下游所有 Agent 的有效输出都在提升
- Knowledge Base 决定系统的长期价值——冷启动要准备好种子数据，运行后持续积累。三个月后，我的知识库已经覆盖了 90%+ 的常见问题，Q&A Agent 的搜索调用降了 60%
- 成本控制的关键是模型分层——不是每个 Agent 都需要最强的模型。Router 和 Greeter 用 mini 模型，Q&A 和 Digest 用强模型，整体成本降 60% 而不影响体验
如果你也在管社群，先从两个 Agent 开始：一个 Router + 一个 Q&A。跑通了再加其他的。想看具体的 prompt 和配置，来一人独角兽俱乐部，我可以直接分享。