"""
多源知识路由器示例
此示例演示了用于多智能体系统的路由器模式。
路由器对查询进行分类，将它们并行路由到专门的智能体，
并将结果综合成一个组合响应。
"""
import operator
from typing import Annotated, Literal, TypedDict
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from pydantic import BaseModel, Field
# 状态定义
class AgentInput(TypedDict):
"""每个子智能体的简单输入状态。"""
query: str
class AgentOutput(TypedDict):
"""每个子智能体的输出。"""
source: str
result: str
class Classification(TypedDict):
"""单个路由决策：使用什么查询调用哪个智能体。"""
source: Literal["github", "notion", "slack"]
query: str
class RouterState(TypedDict):
query: str
classifications: list[Classification]
results: Annotated[list[AgentOutput], operator.add]
final_answer: str
# 分类器的结构化输出模式
class ClassificationResult(BaseModel):
"""将用户查询分类为特定于智能体的子问题的结果。"""
classifications: list[Classification] = Field(
description="要调用的智能体列表及其针对性的子问题"
)
# 工具
@tool
def search_code(query: str, repo: str = "main") -> str:
"""在 GitHub 仓库中搜索代码。"""
return f"Found code matching '{query}' in {repo}: authentication middleware in src/auth.py"
@tool
def search_issues(query: str) -> str:
"""搜索 GitHub 议题和拉取请求。"""
return f"Found 3 issues matching '{query}': #142 (API auth docs), #89 (OAuth flow), #203 (token refresh)"
@tool
def search_prs(query: str) -> str:
"""搜索拉取请求以获取实现细节。"""
return f"PR #156 added JWT authentication, PR #178 updated OAuth scopes"
@tool
def search_notion(query: str) -> str:
"""在 Notion 工作区中搜索文档。"""
return f"Found documentation: 'API Authentication Guide' - covers OAuth2 flow, API keys, and JWT tokens"
@tool
def get_page(page_id: str) -> str:
"""通过 ID 获取特定的 Notion 页面。"""
return f"Page content: Step-by-step authentication setup instructions"
@tool
def search_slack(query: str) -> str:
"""搜索 Slack 消息和讨论串。"""
return f"Found discussion in #engineering: 'Use Bearer tokens for API auth, see docs for refresh flow'"
@tool
def get_thread(thread_id: str) -> str:
"""获取特定的 Slack 讨论串。"""
return f"Thread discusses best practices for API key rotation"
# 模型和智能体
model = init_chat_model("openai:gpt-5.4")
router_llm = init_chat_model("openai:gpt-5.4-mini")
github_agent = create_agent(
model,
tools=[search_code, search_issues, search_prs],
system_prompt=(
"You are a GitHub expert. Answer questions about code, "
"API references, and implementation details by searching "
"repositories, issues, and pull requests."
),
)
notion_agent = create_agent(
model,
tools=[search_notion, get_page],
system_prompt=(
"You are a Notion expert. Answer questions about internal "
"processes, policies, and team documentation by searching "
"the organization's Notion workspace."
),
)
slack_agent = create_agent(
model,
tools=[search_slack, get_thread],
system_prompt=(
"You are a Slack expert. Answer questions by searching "
"relevant threads and discussions where team members have "
"shared knowledge and solutions."
),
)
# 工作流节点
def classify_query(state: RouterState) -> dict:
"""对查询进行分类并确定要调用哪些智能体。"""
structured_llm = router_llm.with_structured_output(ClassificationResult)
result = structured_llm.invoke([
{
"role": "system",
"content": """分析此查询并确定要咨询哪些知识库。
对于每个相关来源，生成一个针对该来源优化的针对性子问题。
可用来源：
- github：代码、API 参考、实现细节、议题、拉取请求
- notion：内部文档、流程、政策、团队维基
- slack：团队讨论、非正式知识共享、最近的对话
仅返回与查询相关的来源。"""
},
{"role": "user", "content": state["query"]}
])
return {"classifications": result.classifications}
def route_to_agents(state: RouterState) -> list[Send]:
"""根据分类扇出到智能体。"""
return [
Send(c["source"], {"query": c["query"]})
for c in state["classifications"]
]
def query_github(state: AgentInput) -> dict:
"""查询 GitHub 智能体。"""
result = github_agent.invoke({
"messages": [{"role": "user", "content": state["query"]}]
})
return {"results": [{"source": "github", "result": result["messages"][-1].content}]}
def query_notion(state: AgentInput) -> dict:
"""查询 Notion 智能体。"""
result = notion_agent.invoke({
"messages": [{"role": "user", "content": state["query"]}]
})
return {"results": [{"source": "notion", "result": result["messages"][-1].content}]}
def query_slack(state: AgentInput) -> dict:
"""查询 Slack 智能体。"""
result = slack_agent.invoke({
"messages": [{"role": "user", "content": state["query"]}]
})
return {"results": [{"source": "slack", "result": result["messages"][-1].content}]}
def synthesize_results(state: RouterState) -> dict:
"""将所有智能体的结果组合成一个连贯的答案。"""
if not state["results"]:
return {"final_answer": "No results found from any knowledge source."}
formatted = [
f"**From {r['source'].title()}:**\n{r['result']}"
for r in state["results"]
]
synthesis_response = router_llm.invoke([
{
"role": "system",
"content": f"""综合这些搜索结果以回答原始问题："{state['query']}"
- 组合来自多个来源的信息，避免冗余
- 突出最相关和可操作的信息
- 注意来源之间的任何差异
- 保持响应简洁且组织良好"""
},
{"role": "user", "content": "\n\n".join(formatted)}
])
return {"final_answer": synthesis_response.content}
# 构建工作流
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()
)
if __name__ == "__main__":
result = workflow.invoke({
"query": "How do I authenticate API requests?"
})
print("Original query:", result["query"])
print("\nClassifications:")
for c in result["classifications"]:
print(f" {c['source']}: {c['query']}")
print("\n" + "=" * 60 + "\n")
print("Final Answer:")
print(result["final_answer"])
