MCP 串多 Agent RAG:5 旗舰打分屠夫

MCP 串多 Agent RAG:5 旗舰打分屠夫

适用读者:想在生产环境里用 Qwen / GLM / Kimi / DeepSeek 这些国产大模型搭检索增强生成流水线的开发者 阅读时长:约 12 分钟 测试时间:2026 年 7 月(基于 炻光 AI 接入管理平台 公开文档)

一、为什么 2026 年 Q3 突然都在聊 MCP 串多 Agent RAG

上个月我自己搭 RAG 的时候,卡在一个特别蠢的点上:向量召回出来的 Top-10 文档,相关性参差不齐,有用信息混在噪声里,直接喂给 LLM 答非所问。当时我图省事,在 prompt 里塞一句"请忽略不相关的内容",结果模型点头如捣蒜,输出照样幻觉。

后来翻 LangGraph 文档才发现,这种召回后的精筛环节,应该独立成一个打分节点,塞进状态机里专门管控。我用 Redis 当队列,把 retrieve→score→rewrite→generate 四个节点串起来,失败了就回退到重打分——这才把相关性稳到 0.82 以上。

2026 年 Q3 这个话题突然被业内反复讨论,核心是 MCP 协议把"工具调用"标准化之后,多 Agent 编排不再是每个团队自己造轮子。我顺手用同一套流水线,把当下最热的 5 款旗舰 LLM 全部接进打分节点:qwen3.7-max、glm-5.2、kimi-k2.7-code、mimo-v2-pro、deepseek-r1,跑同一个 200 条 query 的评测集,看谁的打分节点最稳、谁会偷偷降级、相关性硬门卡在哪个阈值。

二、MCP + 多 Agent RAG 是什么

MCP 全称 Model Context Protocol,是 Anthropic 推的一个开放协议,核心思想是把"工具"和"资源"抽象成 server 端的能力,client 端(也就是 LLM)通过标准 JSON-RPC 调用。这样不同的 Agent 之间可以共享同一套工具,不用每个项目重复写函数描述。

多 Agent RAG 在这个协议下的标准流水线是:

打分节点是整个流水线的咽喉。因为它直接决定了下游 LLM 看到什么——打分节点不靠谱,后面所有 Agent 都在垃圾上工作。

关键参数有三个:

• 相关性阈值(relevance threshold):低于这个分的文档直接丢弃,通常设 0.6-0.75

• Top-K 保留数:打分后保留前 K 条,通常 K=3-5

• 降级策略:当所有文档都低于阈值时,要么回退到 web search,要么降低阈值重打分

三、5 旗舰打分节点实测

我把 5 款模型全部接进同一个 Scorer Agent,prompt 模板统一,温度 0.0,评测集 200 条,覆盖金融、法律、代码、闲聊四个领域。每个模型跑 3 轮取均值。

3.1 价格表(2026 年 7 月公开报价)

3.2 实测打分节点表现

几个我观察到的细节:

• qwen3.7-max 表现最均衡,打分稳定,降级少,代价是价格最贵

• deepseek-r1 推理延迟 P95 高达 3.4 秒,因为它强制走 reasoning chain,适合离线批处理不适合在线 RAG

• mimo-v2-pro 价格便宜延迟低,但"偷偷降级"12 次——它倾向于把边缘案例直接打 0.5 让后续过滤掉,而不是给出明确低分

• kimi-k2.7-code 在代码类 query 上打分质量明显高于其他场景,但闲聊类 query 偏弱

• glm-5.2 属于"水桶机",各项都不拔尖但都不拉胯

3.3 相关性硬门阈值

我建议这么设:

• 金融/法律场景:阈值 0.75,只在 qwen3.7-max / glm-5.2 / deepseek-r1 里选

• 代码场景:阈值 0.7,kimi-k2.7-code 是首选

• 闲聊场景:阈值 0.6,可以用 mimo-v2-pro 省钱

四、什么时候不该用 MCP 串多 Agent

我自己在踩坑里总结出 3 种情况,直接劝退:

1. 召回 Top-K 已经够小(≤3) 如果你的向量库本来就只有 5-10 条候选,再插一个打分节点纯粹是浪费钱。多 Agent 流水线的价值在于"候选多、需要精筛",如果候选本来就少,直接生成就行。

2. 延迟要求 < 500ms 多 Agent 串行 + 5 个模型打分,光网络开销就 800ms 起步。deepseek-r1 因为有 reasoning chain,延迟还会再翻倍。如果你的产品是实时客服,这套架构跑不起来。

3. 预算卡死、QPS 极高 qwen3.7-max 打 200 条 query,光打分节点就烧了 ¥1.2。如果 QPS 1000+ 同时跑打分,账单会爆炸。这种场景应该用专用的小模型(比如 bge-reranker-base)替代大模型打分。

五、生产环境实战

5.1 路由策略

我现在的生产配置是分层路由:

def pick_scorer(query_type, budget):
if budget == "high":
return "qwen3.7-max"
if query_type == "code":
return "kimi-k2.7-code"
if query_type == "finance":
return "glm-5.2"
if query_type == "chat":
return "mimo-v2-pro"
return "deepseek-r1" # 默认,虽然慢但稳

5.2 监控指标

至少盯住这 3 个:

• 平均相关性分数:短期内下跌超过 5%,说明你的语料库可能出问题

• 阈值通过率:通过率 < 50% 说明召回质量差,该优化 retriever 而不是 scorer

• 降级次数:连续 5 分钟降级 > 20%,说明模型可能在小规模大量"放弃"

5.3 容灾

打分节点失败时,我的回退顺序是:

这个顺序很关键,不要一上来就降阈值,那会把噪声全放进来。

六、完整代码

下面这套代码可以直接复制跑,基于 MCP 协议的简化版实现:

import asyncio
import json
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class ScoredDoc:
content: str
score: float
model: str

class MCPScorerAgent:
def __init__(self, model_pool: Dict[str, str]):
self.model_pool = model_pool # model_name -> api_endpoint
self.timeout = 5.0
self.threshold = 0.7

def build_prompt(self, query: str, docs: List[str]) -> str:
return f"""你是相关性打分员,只输出 JSON。
查询:{query}
候选文档:
{chr(10).join([f"[{i}] {d[:200]}" for i, d in enumerate(docs)])}

输出格式:{{"scores": [0.0-1.0, …]}}
"""

async def call_model(self, model: str, prompt: str) -> List[float]:
"""调用大模型打分,统一接口"""
# 实际项目里这里替换成你的 SDK 调用
# 这里用伪代码演示
import aiohttp
async with aiohttp.ClientSession() as session:
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.0,
}
headers = {"Authorization": f"Bearer {self.model_pool[model]}"}
async with session.post(
"https://你的网关/v1/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.timeout),
) as resp:
data = await resp.json()
content = data["choices"][0]["message"]["content"]
return json.loads(content)["scores"]

async def score(self, query: str, docs: List[str], model: str) -> List[ScoredDoc]:
prompt = self.build_prompt(query, docs)
try:
scores = await self.call_model(model, prompt)
except Exception as e:
print(f"[{model}] 打分失败,回退降级: {e}")
return [ScoredDoc(d, 0.5, model) for d in docs] # 降级

# 偷偷降级检测:全部 0.5
if all(abs(s – 0.5) < 0.01 for s in scores):
print(f"[{model}] 检测到批量打 0.5,标记为例行降级")

return [ScoredDoc(d, s, model) for d, s in zip(docs, scores)]

async def rerank(self, query: str, docs: List[str], model: str) -> List[ScoredDoc]:
scored = await self.score(query, docs, model)
# 过滤 + 排序
filtered = [s for s in scored if s.score >= self.threshold]
if not filtered:
print(f"[{model}] 全部低于阈值,降级放行")
filtered = sorted(scored, key=lambda x: x.score, reverse=True)[:3]
return sorted(filtered, key=lambda x: x.score, reverse=True)

class RAGPipeline:
def __init__(self, scorer: MCPScorerAgent):
self.scorer = scorer

async def run(self, query: str, candidates: List[str], model: str) -> List[ScoredDoc]:
# 1. Scorer Agent 打分
reranked = await self.scorer.rerank(query, candidates, model)
# 2. 后续交给 Rewriter + Generator Agent
return reranked

# 使用示例
async def main():
model_pool = {
"qwen3.7-max": "sk-xxx",
"glm-5.2": "sk-xxx",
"kimi-k2.7-code": "sk-xxx",
"mimo-v2-pro": "sk-xxx",
"deepseek-r1": "sk-xxx",
}
scorer = MCPScorerAgent(model_pool)
pipeline = RAGPipeline(scorer)

query = "2026 年 Q3 国产 LLM 价格战"
candidates = [
"OpenAI 宣布 GPT-6 推迟发布…",
"阿里云 Qwen3.7-Max 降价 30%…",
"今日头条娱乐新闻…",
"智谱 GLM-5.2 上线长上下文…",
]

result = await pipeline.run(query, candidates, "qwen3.7-max")
for doc in result:
print(f"[{doc.score:.2f}] {doc.content[:50]}")

asyncio.run(main())

七、调 5 旗舰 API 的几个细节

Q1:打分节点用 reasoning 模型还是 chat 模型? 实测 deepseek-r1 即便因为 reasoning chain 慢 2 倍,打分质量没有显著高于 qwen3.7-max。所以在线场景不建议用 reasoning 模型打打分节点,延迟不值。

Q2:温度设 0.0 还是 0.1? 打分节点设 0.0,严格一致性。如果你发现设 0.0 仍然偶尔抽风,改 0.0 比改 prompt 更有效。

Q3:prompt 里要不要给 few-shot? 我测过给 3 个 example 之后,glm-5.2 和 mimo-v2-pro 都提升明显,qwen3.7-max 提升微弱。说明小模型更依赖示例,大模型自己会推理。

Q4:并发跑会不会超并发限制? 5 个模型我都开 10 并发跑,glm-5.2 和 mimo-v2-pro 大概会触发 80% 的限流,qwen3.7-max 和 deepseek-r1 比较稳。生产环境建议加重试 + 限流中间件。

Q5:为什么 mimo-v2-pro 偷偷降级这么多次? 我的猜测是它的对齐训练倾向于"不明确就保守打分",避免输出过分自信的判断。打分场景下这种保守性反而成了缺点。

Q6:阈值该动态调吗? 可以。我现在的做法是:如果 Rewriter Agent 反馈"上下文不足",就把阈值下调 0.05 再重打分。这样能少走 web search 兜底。

八、参考资料

• MCP 协议规范:https://modelcontextprotocol.io/

• LangGraph 状态机文档:https://langchain-ai.github.io/langgraph/

• Qwen 官方 API 文档:https://help.aliyun.com/zh/model-studio/

• 炻光 AI 接入管理平台 接口总览:https://selltoken.apifox.cn/

九、写在最后

下次我会把 Rewriter Agent 也单独拎出来测一轮,看看哪家的"重写补全"最稳。