LangGraph构建企业级多源知识问答系统

1. 项目概述

在构建企业级知识问答系统时，我们经常面临一个核心挑战：如何高效整合分散在不同平台的专业知识？传统的单一智能体方案往往难以兼顾多领域知识的深度与广度。本文将详细介绍基于LangGraph的Router架构，通过智能路由机制实现GitHub代码库、Notion文档和Slack讨论的多源知识整合。

这个系统的独特之处在于其三层架构设计：

• 路由层（Router）：作为系统的"交通指挥中心"，负责分析用户问题并分发给合适的专业智能体
• 专业智能体层（Specialized Agents）：包括GitHub、Notion和Slack三个垂直领域的专家
• 汇总层（Synthesize）：将多个来源的结果整合成连贯的最终答案

2. 核心架构解析

2.1 Router层设计原理

Router层的本质是一个智能分流器，其工作流程遵循"分类-分发-汇总"的范式。与常规智能体不同，Router不直接处理问题，而是专注于：

1. 问题拆解：将复杂问题分解为适合各专业领域处理的子问题
2. 智能路由：根据问题类型决定调用哪些专业智能体
3. 结果协调：确保最终答案呈现统一的视角

这种设计的优势在于：

• 降低单个智能体的认知负荷
• 提高系统可维护性（各模块职责清晰）
• 便于扩展新的知识领域

2.2 专业智能体配置

系统包含三个核心专业智能体，每个都针对特定平台优化：

每个智能体都配备：

• 专用提示词（限定回答范围和风格）
• 平台特定工具集
• 独立的大模型实例

2.3 结果合成机制

汇总层采用两阶段处理：

1. 去重与冲突检测：识别不同来源间的矛盾信息
2. 视角统一：以用户友好的方式组织答案，标注信息来源

关键技术点：

• 使用单独的LLM实例进行结果合成
• 保留原始数据引用（便于追溯）
• 动态调整摘要长度基于问题复杂度

3. 技术实现细节

3.1 状态管理设计

系统采用TypedDict定义严格的状态结构：

python复制class RouterState(TypedDict):
    query: str  # 原始问题
    classifications: list[Classification]  # 路由决策
    results: Annotated[list[AgentOutput], operator.add]  # 并行结果收集
    final_answer: str  # 最终答案

这种设计确保了：

• 类型安全：每个字段都有明确定义的类型
• 可追溯性：保留完整的处理链路
• 并行友好：results字段支持自动合并并行任务结果

3.2 工具集实现

每个专业智能体配备量身定制的工具：

GitHub工具集：

python复制@tool
def search_code(query: str, repo: str = "main") -> str:
    """在指定仓库搜索代码"""
    return f"在{repo}中找到匹配'{query}'的代码..."

@tool 
def search_issues(query: str) -> str:
    """搜索GitHub问题"""
    return f"找到3个相关issue: #142, #89, #203"

Notion工具集：

python复制@tool
def search_notion(query: str) -> str:
    """搜索Notion工作区"""
    return "找到文档：'API认证指南'"

@tool
def get_page(page_id: str) -> str:
    """获取特定页面内容"""
    return "页面内容：分步认证设置说明"

工具设计遵循"搜索-获取"两段式模式，既保证召回率又确保结果精准。

3.3 智能体提示工程

每个专业智能体的提示词都经过精心设计：

GitHub Agent提示词：

code复制你是一个GitHub专家。通过搜索仓库、issue和PR来回答有关代码实现、API参考的问题。
- 只基于GitHub中的事实回答
- 避免理论解释和最佳实践建议
- 优先使用工具获取信息

Notion Agent提示词：

code复制你是一个Notion文档专家。回答有关内部流程、政策和团队文档的问题。
- 代表官方口径回答
- 保持回答严谨规范
- 必须通过搜索验证信息

提示词通过：

1. 角色定位（明确专家身份）
2. 边界限定（规定回答范围）
3. 行为约束（强制工具优先）
   确保各智能体保持专业性和一致性。

4. LangGraph工作流构建

4.1 路由决策实现

分类器使用结构化输出确保决策可预测：

python复制class ClassificationResult(BaseModel):
    classifications: list[Classification] = Field(
        description="待调用的智能体及其专属子问题"
    )

def classify_query(state: RouterState) -> dict:
    structured_llm = router_llm.with_structured_output(ClassificationResult)
    result = structured_llm.invoke([
        {"role": "system", "content": "分析问题并决定咨询哪些知识源..."},
        {"role": "user", "content": state["query"]}
    ])
    return {"classifications": result.classifications}

关键设计：

• 强制模型输出结构化结果
• 为每个目标智能体生成优化后的子问题
• 支持多智能体并行调用

4.2 并行执行机制

路由到专业智能体采用LangGraph的Send机制：

python复制def route_to_agents(state: RouterState) -> list[Send]:
    return [
        Send(c["source"], {"query": c["query"]})
        for c in state["classifications"] 
    ]

这实现了：

• 动态决定调用的智能体
• 每个智能体接收定制化的查询
• 真正的并行执行（非顺序）

4.3 结果合成策略

汇总阶段采用"对比-整合"方法：

python复制def synthesize_results(state: RouterState) -> dict:
    formatted = [
        f"来自{r['source']}:/n{r['result']}" 
        for r in state["results"]
    ]
    
    response = router_llm.invoke([
        {"role": "system", "content": "综合以下结果回答原始问题..."},
        {"role": "user", "content": "/n/n".join(formatted)}
    ])
    return {"final_answer": response.content}

合成过程强调：

• 消除冗余信息
• 标注知识来源
• 处理冲突陈述
• 保持回答简洁

5. 完整工作流集成

5.1 图形化流程定义

使用LangGraph的声明式API构建完整流程：

python复制workflow = (
    StateGraph(RouterState)
    .add_node("classify", classify_query)
    .add_node("github", query_github)
    .add_node("notion", query_notion)
    .add_node("slack", query_slack)
    .add_node("synthesize", synthesize_results)
    .add_edge(START, "classify")
    .add_conditional_edges("classify", route_to_agents, ["github", "notion", "slack"])
    .add_edge("github", "synthesize")
    .add_edge("notion", "synthesize")
    .add_edge("slack", "synthesize")
    .add_edge("synthesize", END)
    .compile()
)

流程特点：

• 清晰的阶段划分
• 条件路由支持
• 自动并行处理
• 可视化调试友好

5.2 记忆增强实现

通过封装工作流为工具实现对话记忆：

python复制@tool
def search_knowledge_base(query: str) -> str:
    """跨知识源搜索"""
    result = workflow.invoke({"query": query})
    return result["final_answer"]

conversational_agent = create_agent(
    model,
    tools=[search_knowledge_base],
    checkpointer=InMemorySaver(),
)

实现效果：

• 保留完整对话历史
• 支持多轮追问
• 上下文感知的回答

6. 实战应用示例

6.1 典型查询处理

示例问题："我们的API认证机制是什么？"

系统处理流程：

1. Router识别需要查询GitHub(实现细节)���Notion(官方文档)
2. 并行调用两个智能体：
   • GitHub Agent返回JWT中间件代码位置
   • Notion Agent提供OAuth2流程文档
3. 汇总层生成包含代码引用和文档链接的完整回答

6.2 结果对比展示

原始结果：

code复制GitHub: 在src/auth.py找到JWT认证中间件...
Notion: 文档显示支持OAuth2和API Key两种方式...

合成后回答：

code复制我们支持三种认证方式：
1. JWT令牌（推荐）：实现见src/auth.py
2. OAuth2流程：详见《API认证指南》
3. API密钥：用于服务间通信

7. 性能优化建议

7.1 路由优化策略

1. 缓存常见问题模式：对高频问题缓存路由决策
2. 元分类器：简单问题直接回答，复杂问题再路由
3. 动态权重：根据各平台数据新鲜度调整查询优先级

7.2 并行执行优化

1. 超时控制：为每个智能体设置合理超时
2. 结果缓存：相同子问题复用之前结果
3. 负载均衡：监控各智能体响应时间动态调整流量

7.3 合成阶段改进

1. 分块合成：对超长结果先分块总结再整体合成
2. 可信度标注：根据来源可靠性标注信息可信度
3. 差异提示：显式标注不同来源间的矛盾点

8. 扩展应用场景

8.1 企业内部知识中枢

整合更多数据源：

• Confluence文档
• Jira工单系统
• 内部数据库
• CRM系统

8.2 客户支持系统

应用模式：

1. 路由识别问题类型（技术、账单、使用指导）
2. 分发给对应部门的专业知识库
3. 生成符合客户认知水平的回答

8.3 教育领域应用

构建多模态学习助手：

• 教材内容（结构化知识）
• 实验指导（操作步骤）
• 讨论记录（常见疑问）
• 作业范例（实践参考）

9. 常见问题排查

9.1 路由决策不准确

症状：问题被分配给不相关的智能体
排查步骤：

1. 检查分类器的few-shot示例是否覆盖足够场景
2. 验证结构化输出模式是否约束足够严格
3. 分析错误案例中的query改写是否合理

解决方案：

• 增加更多路由示例
• 强化输出schema的description字段
• 添加query改写校验规则

9.2 结果合成质量差

症状：最终回答遗漏重要信息或存在冲突
排查步骤：

1. 检查各智能体原始输出是否完整
2. 验证合成提示词是否明确要求对比来源
3. 分析结果合并算法是否合理

解决方案：

• 添加信息重要性评估步骤
• 强化合成阶段的冲突检测逻辑
• 实现分阶段合成（先提取要点再组织）

9.3 系统响应延迟

症状：简单问题处理时间过长
排查步骤：

1. 监控各智能体响应时间
2. 检查是否有智能体阻塞整体流程
3. 分析网络延迟情况

解决方案：

• 设置智能体调用超时
• 实现缓存机制
• 优化LangGraph执行计划

10. 架构演进方向

10.1 混合路由模式

结合Router与Subagents的优势：

• 第一层：粗粒度路由到领域
• 第二层：领域内使用Subagents深度处理
• 统一协调最终输出

10.2 动态智能体注册

实现热插拔架构：

1. 智能体注册中心管理可用实例
2. Router动态获取当前可用的智能体列表
3. 运行时加载所需的智能体

10.3 自适应学习机制

持续优化系统表现：

1. 收集用户对答案的反馈
2. 分析路由决策的正确性
3. 自动调整分类器few-shot示例
4. 优化各智能体的提示词

这种架构特别适合知识不断更新的企业环境，能够随着组织知识库的增长而持续进化。