知识库 + ReAct + 多 Agent：故障诊断的智能解答系统

问题场景

用户报告：“订单提交失败，报错 500”。

传统 RAG：检索”订单 500 错误”，返回相关文档片段。

但这不够。真实的故障诊断需要：

1. 分析错误日志，定位具体模块
2. 查询相关配置和依赖状态
3. 对比历史相似案例
4. 给出可执行的修复步骤

这需要推理 + 检索 + 行动的组合——ReAct 模式。

---

ReAct 模式回顾

ReAct（Reasoning + Acting）的核心循环：

Thought: 分析当前情况，决定下一步
Action:  执行动作（检索、调用工具、查询数据库）
Observation: 获取结果
→ 循环直到问题解决

与知识库结合的关键点：

环节	知识库的作用
Thought	基于领域知识生成推理路径
Action	检索相关文档、案例、配置
Observation	验证检索结果是否相关

---

多 Agent 架构

单一 Agent 处理复杂故障诊断时容易：

• 推理路径单一
• 检索策略固定
• 缺乏专业分工

多 Agent 分工：

┌─────────────────────────────────────────────────────┐
│                   Orchestrator                       │
│              (协调器：分解任务、汇总结果)              │
└──────────────┬──────────────────────────────────────┘
               │
       ┌───────┼───────┐
       ▼       ▼       ▼
   ┌──────┐ ┌──────┐ ┌──────┐
   │诊断   │ │检索   │ │方案   │
   │Agent  │ │Agent  │ │Agent  │
   └──────┘ └──────┘ └──────┘
       │       │       │
       └───────┴───────┘
               │
         知识库系统

Orchestrator（协调器）

职责：

• 接收用户问题
• 分解为子任务
• 分配给专业 Agent
• 汇总结果，生成最终答案

诊断 Agent

职责：分析问题，生成假设

Thought: 用户报告订单 500 错误，可能原因有：
         1. 数据库连接问题
         2. 支付网关超时
         3. 库存服务异常
Action: query_logs(service="order", error="500", time_range="1h")
Observation: 发现 order-service 日志中有 3 条超时错误，
             指向 inventory-service

检索 Agent

职责：从知识库中检索相关信息

Action: search_kb(query="inventory-service timeout troubleshooting")
Observation: 返回 3 篇相关文档：
             1. 库存服务超时排查指南
             2. 服务间调用超时配置
             3. 历史案例 #2847

方案 Agent

职责：基于诊断结果和检索到的知识，生成解决方案

Thought: 结合日志分析和知识库文档，问题是 inventory-service
         响应超时。历史案例 #2847 中通过调整超时配置解决。
Action: generate_solution(
  diagnosis="inventory-service timeout",
  references=[doc1, doc2, case_2847]
)

---

ReAct + 知识库的完整流程

示例：订单服务故障诊断

用户：订单提交失败，报错 500，昨天还正常

=== Orchestrator 分解任务 ===
Task 1: 分析日志，定位错误源 → 诊断 Agent
Task 2: 检索相关故障案例 → 检索 Agent
Task 3: 生成修复方案 → 方案 Agent

=== 诊断 Agent 执行 ===
Thought: 需要先看日志，确认是哪个服务报 500
Action: query_logs(service="order", level="error", time_range="24h")
Observation: order-service 调用 payment-gateway 返回 504 Gateway Timeout

Thought: 支付网关超时，可能是网关本身问题或配置变更
Action: check_config_changes(service="payment-gateway", time_range="48h")
Observation: 昨晚 23:00 更新了超时配置，从 30s 改为 10s

=== 检索 Agent 执行 ===
Action: search_kb(query="payment gateway timeout configuration")
Observation: 找到文档《支付网关最佳实践》：
             "建议超时设置不低于 20s，高并发场景建议 30s"

Action: search_cases(query="payment timeout after config change")
Observation: 找到历史案例 #1923：
             "超时配置过短导致高并发时段超时，恢复为 30s 后解决"

=== 方案 Agent 执行 ===
Thought: 根本原因是昨晚的超时配置变更。
         知识库建议 30s，当前配置 10s。
         历史案例 #1923 有相同问题和解决方案。

Action: generate_solution()
Result:
  ## 诊断结果
  - 根因：支付网关超时配置从 30s 改为 10s
  - 影响：高并发时段订单提交超时

  ## 修复步骤
  1. 立即恢复配置：`kubectl set env deployment/payment-gateway TIMEOUT=30s`
  2. 验证：提交测试订单确认正常
  3. 监控：观察后续 1 小时错误率

  ## 参考
  - 文档：《支付网关最佳实践》
  - 历史案例：#1923

=== Orchestrator 汇总 ===
返回完整诊断报告给用户

---

Self-Reflective RAG：让 Agent 自我纠错

传统 RAG 的检索结果可能不相关。Self-Reflective RAG 加入反思环节：

检索 → 评估相关性 → 不相关则重写查询 → 重新检索

文档评估（Document Grading）

def grade_document(query: str, doc: str) -> str:
    """评估文档与查询的相关性"""
    prompt = f"""
    问题：{query}
    文档：{doc}

    该文档是否与问题相关？回答：relevant / irrelevant
    """
    return llm.invoke(prompt)

查询重写（Query Rewriting）

def rewrite_query(original: str, context: str) -> str:
    """基于上下文重写查询"""
    prompt = f"""
    原始查询：{original}
    检索失败原因：{context}

    生成更好的检索查询：
    """
    return llm.invoke(prompt)

完整的反思循环

1. 检索：search_kb("订单失败")
2. 评估：文档 A 相关，文档 B 不相关
3. 决策：有相关文档 → 进入生成
4. 生成：基于文档 A 生成答案
5. 评估答案：是否解决了问题？
6. 如果不：重写查询，回到步骤 1

---

知识库设计要点

内容组织

内容类型	说明	示例
故障案例	历史问题的完整记录	案例 #2847：库存服务超时
排查指南	标准化的排查流程	《数据库连接问题排查》
配置参考	系统配置说明	《超时配置最佳实践》
架构文档	系统架构和依赖关系	《订单服务依赖图》

检索优化

1. 混合检索：向量检索 + 关键词检索
2. 元数据过滤：按服务、时间、严重程度过滤
3. 重排序：用 LLM 对检索结果重新排序

知识更新

• 故障解决后自动生成案例记录
• 定期评审和更新文档
• 从排查过程中提取新模式

---

实现框架

LangGraph 状态图

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 定义状态
class AgentState(TypedDict):
    query: str           # 用户问题
    diagnosis: str       # 诊断结果
    documents: list      # 检索到的文档
    generation: str      # 生成的方案
    iterations: int      # 迭代次数

# 节点函数
def diagnose(state: AgentState) -> AgentState:
    """诊断 Agent：分析问题，生成假设"""
    query = state["query"]
    # 调用诊断 Agent 进行分析
    diagnosis = diagnosis_agent.invoke(query)
    return {"diagnosis": diagnosis}

def retrieve(state: AgentState) -> AgentState:
    """检索 Agent：从知识库检索相关文档"""
    query = state["query"]
    diagnosis = state["diagnosis"]
    # 组合查询
    search_query = f"{query}\n诊断: {diagnosis}"
    docs = knowledge_base.search(search_query)
    return {"documents": docs}

def grade_documents(state: AgentState) -> AgentState:
    """评估文档相关性"""
    docs = state["documents"]
    query = state["query"]
    # 过滤不相关的文档
    relevant_docs = [d for d in docs if is_relevant(query, d)]
    return {"documents": relevant_docs}

def generate(state: AgentState) -> AgentState:
    """方案 Agent：生成解决方案"""
    diagnosis = state["diagnosis"]
    docs = state["documents"]
    solution = solution_agent.invoke(diagnosis, docs)
    return {"generation": solution}

def should_retry(state: AgentState) -> str:
    """决策：是否需要重新检索"""
    if not state["documents"]:
        return "rewrite"  # 没有相关文档，重写查询
    if state["iterations"] >= 3:
        return "end"      # 超过最大迭代次数
    return "generate"     # 有相关文档，生成方案

def rewrite_query(state: AgentState) -> AgentState:
    """重写查询"""
    original = state["query"]
    new_query = query_rewriter.invoke(original)
    return {"query": new_query, "iterations": state["iterations"] + 1}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("diagnose", diagnose)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite_query)

# 设置入口
workflow.set_entry_point("diagnose")

# 添加边
workflow.add_edge("diagnose", "retrieve")
workflow.add_edge("retrieve", "grade")

# 条件边
workflow.add_conditional_edges(
    "grade",
    should_retry,
    {
        "rewrite": "rewrite",
        "generate": "generate",
        "end": END,
    }
)

workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)

# 编译并运行
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 执行
result = app.invoke({
    "query": "订单提交失败，报错 500",
    "iterations": 0
})

---

小结

要点	说明
ReAct 循环	Thought → Action → Observation
多 Agent 分工	诊断、检索、方案各司其职
Self-Reflective	评估检索结果，自我纠错
知识库	案例库 + 文档库 + 配置库

核心思想：不是简单检索，而是推理驱动的检索——Agent 知道什么时候需要什么信息，去哪里找，找到后如何使用。

---

参考资料

• ReAct: Synergizing Reasoning and Acting in Language Models
• Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection
• Corrective Retrieval Augmented Generation (CRAG)
• LLM Powered Autonomous Agents - Lilian Weng
• Self-Reflective RAG with LangGraph