12 实战：用 Router 搭多源知识库（先分诊，再并行查，最后汇总）

1
import getpass, os
2

3
os.environ["LANGSMITH_TRACING"] = "true"
4
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

1
from typing import Annotated, Literal, TypedDict
2
import operator
3

4

5
class AgentInput(TypedDict):
6
    """每个子 Agent 的输入状态：只要一个问题。"""
7
    query: str
8

9

10
class AgentOutput(TypedDict):
11
    """每个子 Agent 的输出。"""
12
    source: str
13
    result: str
14

15

16
class Classification(TypedDict):
17
    """一条路由决策：调哪个 Agent、问它什么。"""
18
    source: Literal["github", "notion", "slack"]
19
    query: str
20

21

22
class RouterState(TypedDict):
23
    query: str
24
    classifications: list[Classification]
25
    results: Annotated[list[AgentOutput], operator.add]  # 归并函数：并行结果累加
26
    final_answer: str

1
from langchain.tools import tool
2

3

4
@tool
5
def search_code(query: str, repo: str = "main") -> str:
6
    """在 GitHub 仓库里搜代码。"""
7
    return f"Found code matching '{query}' in {repo}: authentication middleware in src/auth.py"
8

9

10
@tool
11
def search_issues(query: str) -> str:
12
    """搜 GitHub 的 issue 和 pull request。"""
13
    return f"Found 3 issues matching '{query}': #142 (API auth docs), #89 (OAuth flow), #203 (token refresh)"
14

15

16
# ...（略）另有 search_prs / search_notion / get_page / search_slack / get_thread，共 7 个

1
from langchain.agents import create_agent
2
from langchain.chat_models import init_chat_model
3

4
model = init_chat_model("openai:gpt-5.5")
5

6
github_agent = create_agent(
7
    model,
8
    tools=[search_code, search_issues, search_prs],
9
    system_prompt=(
10
        "You are a GitHub expert. Answer questions about code, "
11
        "API references, and implementation details by searching "
12
        "repositories, issues, and pull requests."
13
    ),
14
)
15
# 中文大意：GitHub 专家，只从仓库 / issue / PR 里找代码与实现细节的答案
16

17
# notion_agent / slack_agent 写法完全一样，只是 tools 与 system_prompt 不同
18
# ...（略）

1
from pydantic import BaseModel, Field
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.types import Send
4

5
router_llm = init_chat_model("openai:gpt-5.4-mini")
6

7

8
class ClassificationResult(BaseModel):
9
    """把用户问题拆成若干「面向某个知识源的子问题」。"""
10
    classifications: list[Classification] = Field(
11
        description="List of agents to invoke with their targeted sub-questions"
12
    )
13

14

15
def classify_query(state: RouterState) -> dict:
16
    """分类：决定叫哪几个 Agent，并给每个 Agent 派一个定制子问题。"""
17
    structured_llm = router_llm.with_structured_output(ClassificationResult)
18

19
    result = structured_llm.invoke([
20
        {"role": "system", "content": SYSTEM_PROMPT},
21
        {"role": "user", "content": state["query"]},
22
    ])
23

24
    return {"classifications": result.classifications}

1
def route_to_agents(state: RouterState) -> list[Send]:
2
    """扇出：把每条分类变成一次 Send，谁被点到谁就跑。"""
3
    return [
4
        Send(c["source"], {"query": c["query"]})
5
        for c in state["classifications"]
6
    ]

1
def query_github(state: AgentInput) -> dict:
2
    """查 GitHub Agent。"""
3
    result = github_agent.invoke({
4
        "messages": [{"role": "user", "content": state["query"]}]
5
    })
6
    return {"results": [{"source": "github", "result": result["messages"][-1].content}]}
7

8

9
# ...（略）query_notion / query_slack 写法一模一样，只换 agent 变量与 source 名

1
def synthesize_results(state: RouterState) -> dict:
2
    """汇总：等所有并行分支跑完，把结果揉成一个答案。"""
3
    if not state["results"]:
4
        return {"final_answer": "No results found from any knowledge source."}
5

6
    formatted = [
7
        f"**From {r['source'].title()}:**\n{r['result']}"
8
        for r in state["results"]
9
    ]
10

11
    synthesis_response = router_llm.invoke([
12
        {"role": "system", "content": (
13
            f'把下面的检索结果汇总成对原始问题的回答："{state["query"]}"\n'
14
            "- 合并多源信息，不要重复\n"
15
            "- 突出最相关、最可执行的内容\n"
16
            "- 标出来源之间的分歧\n"
17
            "- 保持简洁、有条理"
18
        )},
19
        {"role": "user", "content": "\n\n".join(formatted)},
20
    ])
21

22
    return {"final_answer": synthesis_response.content}

1
workflow = (
2
    StateGraph(RouterState)
3
    .add_node("classify", classify_query)
4
    .add_node("github", query_github)
5
    .add_node("notion", query_notion)
6
    .add_node("slack", query_slack)
7
    .add_node("synthesize", synthesize_results)
8
    .add_edge(START, "classify")
9
    .add_conditional_edges("classify", route_to_agents, ["github", "notion", "slack"])
10
    .add_edge("github", "synthesize")
11
    .add_edge("notion", "synthesize")
12
    .add_edge("slack", "synthesize")
13
    .add_edge("synthesize", END)
14
    .compile()
15
)

1
result = workflow.invoke({
2
    "query": "How do I authenticate API requests?"
3
})
4

5
print("Original query:", result["query"])
6
print("\nClassifications:")
7
for c in result["classifications"]:
8
    print(f"  {c['source']}: {c['query']}")
9
print("\nFinal Answer:")
10
print(result["final_answer"])

1
return [Send(c["source"], {"query": c["query"]}) for c in state["classifications"]]