大模型实战：专门为 RAG 设计一套检索、解释、评估的Agent 组合
前面已经写了很多关于：
- RAG 知识库构建与实时更新；
- 多智能体协作框架；
- 工作流编排、监控与自我调优；
- 以及 Agent 的内部结构、自我反思、服务封装。
但真正上线一段时间后，你会发现一个很现实的问题：
“RAG 整体还行，但有时候答得莫名其妙。
明明检索到了对的文档，回答却没抓住重点；
或者回答看着挺像回事，其实跟文档不太相关。”
这通常是因为：
- 检索逻辑、解释逻辑、质量评估都揉在一个大 Agent 里；
- 调一侧会连带影响另一侧，很难针对性优化；
- 也很难在监控层清楚地看到：
“是检索错了？”还是“解释错了？”还是“检索没错但胡编了？”
这篇就专门给 RAG 做一个专用 Agent 组合设计，把职责拆清楚：
- 检索 Agent（Retriever Agent）：专职“找文档”的；
- 解释 Agent（Explainer Agent）：专职“读文档、构造答案”的；
- 评估 Agent（Evaluator Agent）：专职“事后打分与纠偏”的。
目标是让你：
- 能清楚定位 RAG 问题到底出在哪一层；
- 能针对性优化检索或回答策略；
- 能渐进式把“低质量回答”转换成可学习的样本。
一、把 RAG 拆成三个独立 Agent 有什么好处？
先从结果倒推：
为什么不搞一个“大 RAG Agent”，而要拆？
1.1 可观测性：问题定位更准确
拆开后，你可以在日志和监控里单独看到：
retriever_agent：召回的文档相关性如何？是否经常“0 命中”？explainer_agent：在“文档相关”前提下，答案质量如何？evaluator_agent：用户反馈差的样本，具体是文档问题还是解释问题？
这样当你看到：
- 文档 top-k 看起来就不对 → 优先调检索 Agent；
- 文档对但回答偏了 → 优先调解释 Agent 的 Prompt / 模型；
- 都勉强对，但表述不清/过度自信 → 优先调评估 Agent + 自我反思策略。
1.2 策略可插拔：不同 Agent 可以用不同模型/参数
比如：
- 检索 Agent：用参数小一点、速度快的模型做 Query 重写；
- 解释 Agent：用质量更好的模型负责“读文档 + 写答案”；
- 评估 Agent：用便宜的模型先粗评一遍，只对“风险高”的再用好模型复评。
1.3 更利于“学习闭环”
当你开始做“自动收集错误样本→后处理/微调”的时候，很重要的一点是知道错的是哪一步。
拆成三个 Agent 后，你可以分别写：
- “检索失败样本池”；
- “解释偏离样本池”；
- “过度自信样本池”。
后面做专项优化会轻松很多。
二、检索 Agent：专职把“问题 → 最合适的文档块”
2.1 职责边界
检索 Agent 只做三件事：
- 解析问题 → 生成适合检索的查询（可以包含重写）；
- 调用底层向量库 / 混合检索接口；
- 返回干净的文档片段列表 + 检索日志，不负责回答。
接口形态建议固定为：
class RetrieverAgent(BaseAgent):
def handle(self, query: str, context: dict) -> dict:
"""
输入：原始用户 query + 少量上下文（语言、业务线等）
输出：
{
"docs": [ { "id": str, "content": str, "score": float, "meta": {...} }, ... ],
"used_query": str, # 最终用于检索的 query（可能重写过）
"retrieval_logs": {...}
}
"""
2.2 一个简单实现骨架
class RetrieverAgent:
def __init__(self, llm, vector_store, top_k: int = 5):
self.llm = llm
self.vs = vector_store
self.top_k = top_k
def _rewrite_query(self, query: str, context: dict) -> str:
# 可选：用小模型做 Query 重写
prompt = (
"请把下面的问题改写成更适合知识检索的简短查询语句，保留关键信息：
"
f"原问题：{query}
"
"只输出改写后的内容。"
)
rewritten = self.llm(prompt)
return rewritten.strip() if rewritten else query
def _search(self, q: str) -> list[dict]:
hits = self.vs.search(q, k=self.top_k) # 自行封装向量库
docs = []
for h in hits:
docs.append({
"id": h["id"],
"content": h["text"],
"score": h["score"],
"meta": h.get("metadata", {})
})
return docs
def handle(self, query: str, context: dict | None = None) -> dict:
context = context or {}
used_query = self._rewrite_query(query, context)
docs = self._search(used_query)
return {
"docs": docs,
"used_query": used_query,
"retrieval_logs": {
"original_query": query,
"used_query": used_query,
"doc_count": len(docs)
}
}
注意：
- 检索 Agent 不回答问题，只输出“干净的检索结果”；
- 方便后面解释 Agent 和评估 Agent 复用。
三、解释 Agent：专职“读文档、写答案”
3.1 职责边界
解释 Agent 只做：
- 接收用户问题 + 文档片段；
- 组织成合适的 Prompt 结构（可以带模板）；
- 用 LLM 生成面向用户的自然语言答案；
- 结构化地返回答案 + 引用信息（如用到哪些文档片段）。
接口建议：
class ExplainerAgent(BaseAgent):
def handle(self, question: str, docs: list[dict], context: dict) -> dict:
"""
输出：
{
"answer": str,
"used_doc_ids": [...],
"reasoning": str # 可选：隐藏或仅用于调试
}
"""
3.2 一个实现骨架
class ExplainerAgent:
def __init__(self, llm, max_docs: int = 5):
self.llm = llm
self.max_docs = max_docs
def _build_context(self, docs: list[dict]) -> str:
ctx_parts = []
for i, d in enumerate(docs[:self.max_docs], start=1):
ctx_parts.append(f"[文档片段{i} - {d['id']}]
{d['content']}")
return "
".join(ctx_parts)
def handle(self, question: str, docs: list[dict], context: dict | None = None) -> dict:
context = context or {}
ctx_text = self._build_context(docs)
prompt = (
"你是一个基于文档的问答助手，请严格根据【文档片段】回答用户问题。
"
"要求：
"
"1. 优先使用文档中的原始表述，避免自己编造。
"
"2. 如果文档没有相关信息，请明确说明“资料中未提及”。
"
"3. 回答尽量结构化、有条理。
"
f"【文档片段】
{ctx_text}
"
f"【问题】
{question}
"
)
answer = self.llm(prompt)
used_ids = [d["id"] for d in docs[:self.max_docs]]
return {
"answer": answer.strip(),
"used_doc_ids": used_ids
}
你可以在这里额外接入前面提到的：
- 自我反思逻辑（让解释 Agent 先审查自己的答案）；
- 模型切换策略（不同场景用不同模型）。
四、评估 Agent：专职“挑毛病 + 打分 + 标记风险”
4.1 职责边界
评估 Agent 不参与生成，只用来：
- 根据问题 + 文档 + 答案，做一次质量打分；
- 标出可能的问题类型：
“文档不相关 / 答案偏离 / 过度自信 / 格式不佳”等； - 输出结构化结果，方便后续统计与自动化处理。
接口建议：
class EvaluatorAgent(BaseAgent):
def handle(self, question: str, docs: list[dict], answer: str, context: dict) -> dict:
"""
输出：
{
"score": float, # 综合质量评分(0~1)
"labels": [ ... ], # 问题标签
"suggestion": str, # 可选：用来反馈给系统或人工
"raw_review": str # 审稿 LLM 的原始文字（可选）
}
"""
4.2 一个评估实现骨架（结构化 JSON 输出）
import json
class EvaluatorAgent:
def __init__(self, llm, max_docs: int = 5):
self.llm = llm
self.max_docs = max_docs
def _build_docs_text(self, docs: list[dict]) -> str:
parts = []
for i, d in enumerate(docs[:self.max_docs], start=1):
parts.append(f"[文档{i} - {d['id']}]
{d['content'][:400]}...")
return "
".join(parts)
def handle(self, question: str, docs: list[dict], answer: str, context: dict | None = None) -> dict:
docs_text = self._build_docs_text(docs)
prompt = f"""
你是一个严格的答案评审助手，请根据【文档】审查【答案】。
【问题】
{question}
【文档】
{docs_text}
【答案】
{answer}
请从以下角度打分并标注问题：
1. 相关性：答案是否紧扣问题？
2. 精确性：事实、数字、名称是否与文档一致？
3. 完整性：有没有遗漏文档中重要信息？
4. 过度自信：在没有文档支持时是否使用“肯定、必须、一定”等表述？
请仅输出一段 JSON，格式如下：
{{
"score": 0.0-1.0,
"labels": ["相关性良好", "轻微遗漏", "可能过度自信", ...],
"suggestion": "简要说明如何改进",
"checks": {{
"relevance": 0.0-1.0,
"factuality": 0.0-1.0,
"completeness": 0.0-1.0,
"overconfidence": 0.0-1.0
}}
}}
"""
review = self.llm(prompt)
try:
data = json.loads(review)
except json.JSONDecodeError:
data = {
"score": 0.0,
"labels": ["评审解析失败"],
"suggestion": "评审 LLM 输出格式错误，需调整 Prompt。",
"checks": {}
}
return data
评估 Agent 最重要的是返回结构化信息，方便后续统计、触发重试/自我反思、写入训练样本库。
五、三类 Agent 如何在一个请求里协作？
把三个 Agent 接到现有协调器（Coordinator）里，流程大概是：
用户问题
│
▼
[RetrieverAgent] → docs + 检索日志
│
▼
[ExplainerAgent] → 答案
│
▼
[EvaluatorAgent] → 质量评分 + 标签 + 建议
│
▼
返回给用户（必要时触发自我反思 / 降级策略）
5.1 一个最小协作示例
class RAGOrchestrator:
def __init__(self, retriever: RetrieverAgent, explainer: ExplainerAgent, evaluator: EvaluatorAgent):
self.retriever = retriever
self.explainer = explainer
self.evaluator = evaluator
def run(self, question: str, context: dict | None = None) -> dict:
context = context or {}
# Step 1: 检索
r_res = self.retriever.handle(question, context)
docs = r_res["docs"]
# Step 2: 解释
e_res = self.explainer.handle(question, docs, context)
answer = e_res["answer"]
# Step 3: 评估
v_res = self.evaluator.handle(question, docs, answer, context)
# 可以在这里追加简单策略：低分时触发重试 / 自我反思
if v_res["score"] < 0.6:
# 例如：让 ExplainerAgent 用不同 Prompt 再答一遍，或提醒用户“本回答可能不够准确”
warning = "本回答依据的文档有限，可能不够准确，仅供参考。"
else:
warning = ""
return {
"answer": answer,
"warning": warning,
"meta": {
"retrieval": r_res["retrieval_logs"],
"used_doc_ids": e_res["used_doc_ids"],
"evaluation": v_res
}
}
六、如何逐步把你现有 RAG 改造成这套结构？
不用一口气重构完，可以按下面顺序来：
-
先抽出 RetrieverAgent
- 把原来混在一起的“检索逻辑”单独封装成一个类；
- 先只在内部使用，不改对外接口。
-
再抽出 ExplainerAgent
- 把“拼 Prompt + 调 LLM + 生成回答”的逻辑集中到一个类里；
- 在现有代码里先改调用路径（RAG 入口 → RetrieverAgent → ExplainerAgent）。
-
最后加 EvaluatorAgent
- 一开始可以只在日志/离线评估管道里用；
- 等效果稳定后，再把“低分触发重试/降级”的逻辑放进线上流程。
-
配合监控与看板
- 在日志中分开记录三类 Agent 的调用：检索耗时、解释耗时、评估耗时；
- 在看板上按 Agent 维度画图，很容易发现瓶颈和异常。
七、小结
围绕 RAG 场景，设计了一套**“检索 / 解释 / 评估”三类专用 Agent**：
- 检索 Agent：只关心“把合适的文档找出来”；
- 解释 Agent：只关心“在文档约束下写出好答案”；
- 评估 Agent：只关心“给答案打分、挑毛病、结构化输出质量信息”。
通过这种拆分，你可以：
- 在监控和日志层面清晰地看到：问题究竟出在检索、解释还是质量控制；
- 针对不同职责选用不同模型和参数，做到性价比更优；
- 为后续“自动收集错误样本、做数据闭环和微调”打下干净的基础。
