/**
* 多源知识路由器示例
*
* 本示例演示了多智能体系统的路由器模式。
* 路由器对查询进行分类，并行路由到专业化的智能体，
* 并将结果综合成一个组合响应。
*/
import { z } from "zod/v4";
import { tool } from "langchain";
import { StateGraph, START, END, Send, StateSchema, ReducedValue } from "@langchain/langgraph";
const AgentOutput = z.object({
source: z.string(),
result: z.string(),
});
const RouterState = new StateSchema({
query: z.string(),
classifications: z.array(
z.object({
source: z.enum(["github", "notion", "slack"]),
query: z.string(),
})
),
results: new ReducedValue(
z.array(AgentOutput).default(() => []),
{ reducer: (current, update) => current.concat(update) }
),
finalAnswer: z.string(),
});
const searchCode = tool(
async ({ query, repo }) => {
return `在 ${repo || "main"} 中找到匹配 '${query}' 的代码：src/auth.py 中的认证中间件`;
},
{
name: "search_code",
description: "在 GitHub 仓库中搜索代码。",
schema: z.object({
query: z.string(),
repo: z.string().optional().default("main"),
}),
}
);
const searchIssues = tool(
async ({ query }) => {
return `找到 3 个匹配 '${query}' 的议题：#142（API 认证文档），#89（OAuth 流程），#203（令牌刷新）`;
},
{
name: "search_issues",
description: "搜索 GitHub 议题和拉取请求。",
schema: z.object({
query: z.string(),
}),
}
);
const searchPrs = tool(
async ({ query }) => {
return `PR #156 添加了 JWT 认证，PR #178 更新了 OAuth 范围`;
},
{
name: "search_prs",
description: "搜索拉取请求以获取实现细节。",
schema: z.object({
query: z.string(),
}),
}
);
const searchNotion = tool(
async ({ query }) => {
return `找到文档：'API 认证指南' - 涵盖 OAuth2 流程、API 密钥和 JWT 令牌`;
},
{
name: "search_notion",
description: "在 Notion 工作区中搜索文档。",
schema: z.object({
query: z.string(),
}),
}
);
const getPage = tool(
async ({ pageId }) => {
return `页面内容：分步认证设置说明`;
},
{
name: "get_page",
description: "通过 ID 获取特定的 Notion 页面。",
schema: z.object({
pageId: z.string(),
}),
}
);
const searchSlack = tool(
async ({ query }) => {
return `在 #engineering 中找到讨论：'为 API 认证使用 Bearer 令牌，请参阅文档了解刷新流程'`;
},
{
name: "search_slack",
description: "搜索 Slack 消息和线程。",
schema: z.object({
query: z.string(),
}),
}
);
const getThread = tool(
async ({ threadId }) => {
return `该线程讨论了 API 密钥轮换的最佳实践`;
},
{
name: "get_thread",
description: "获取特定的 Slack 线程。",
schema: z.object({
threadId: z.string(),
}),
}
);
import { createAgent } from "langchain";
import { ChatOpenAI } from "@langchain/openai";
const llm = new ChatOpenAI({ model: "gpt-5.5" });
const githubAgent = createAgent({
model: llm,
tools: [searchCode, searchIssues, searchPrs],
systemPrompt: `
你是一个 GitHub 专家。通过搜索仓库、议题和拉取请求来回答关于代码、
API 参考和实现细节的问题。
`.trim(),
});
const notionAgent = createAgent({
model: llm,
tools: [searchNotion, getPage],
systemPrompt: `
你是一个 Notion 专家。通过搜索组织的 Notion 工作区来回答关于内部
流程、政策和团队文档的问题。
`.trim(),
});
const slackAgent = createAgent({
model: llm,
tools: [searchSlack, getThread],
systemPrompt: `
你是一个 Slack 专家。通过搜索团队成员分享知识和解决方案的相关
线程和讨论来回答问题。
`.trim(),
});
const routerLlm = new ChatOpenAI({ model: "gpt-5.4-mini" });
// 为分类器定义结构化输出模式
const ClassificationResultSchema = z.object({
classifications: z
.array(
z.object({
source: z.enum(["github", "notion", "slack"]),
query: z.string(),
})
)
.describe("要调用的智能体列表及其针对性的子问题"),
});
async function classifyQuery(state: typeof RouterState.State) {
const structuredLlm = routerLlm.withStructuredOutput(
ClassificationResultSchema
);
const result = await structuredLlm.invoke([
{
role: "system",
content: `分析此查询并确定要咨询哪些知识库。
为每个相关的来源，生成一个针对该来源优化的子问题。
可用来源：
- github: 代码、API 参考、实现细节、议题、拉取请求
- notion: 内部文档、流程、政策、团队维基
- slack: 团队讨论、非正式知识共享、近期对话
仅返回与查询相关的来源。每个来源应有一个针对该特定知识领域优化的子问题。
例如对于"如何认证 API 请求？":
- github: "存在什么认证代码？搜索认证中间件、JWT 处理"
- notion: "存在什么认证文档？查找 API 认证指南"
（省略 slack，因为它与这个技术问题无关）`,
},
{ role: "user", content: state.query },
]);
return { classifications: result.classifications };
}
function routeToAgents(state: typeof RouterState.State): Send[] {
return state.classifications.map(
(c) => new Send(c.source, { query: c.query })
);
}
async function queryGithub(state: typeof RouterState.State) {
const result = await githubAgent.invoke({
messages: [{ role: "user", content: state.query }],
});
return {
results: [{ source: "github", result: result.messages.at(-1)?.content }],
};
}
async function queryNotion(state: typeof RouterState.State) {
const result = await notionAgent.invoke({
messages: [{ role: "user", content: state.query }],
});
return {
results: [{ source: "notion", result: result.messages.at(-1)?.content }],
};
}
async function querySlack(state: typeof RouterState.State) {
const result = await slackAgent.invoke({
messages: [{ role: "user", content: state.query }],
});
return {
results: [{ source: "slack", result: result.messages.at(-1)?.content }],
};
}
async function synthesizeResults(state: typeof RouterState.State) {
if (state.results.length === 0) {
return { finalAnswer: "未从任何知识源找到结果。" };
}
// 格式化结果以便综合
const formatted = state.results.map(
(r) =>
`**来自 ${r.source.charAt(0).toUpperCase() + r.source.slice(1)}:**\n${r.result}`
);
const synthesisResponse = await routerLlm.invoke([
{
role: "system",
content: `综合这些搜索结果来回答原始问题："${state.query}"
- 组合来自多个来源的信息，避免冗余
- 突出最相关和可操作的信息
- 注意来源之间的任何差异
- 保持响应简洁且组织良好`,
},
{ role: "user", content: formatted.join("\n\n") },
]);
return { finalAnswer: synthesisResponse.content };
}
const workflow = new StateGraph(RouterState)
.addNode("classify", classifyQuery)
.addNode("github", queryGithub)
.addNode("notion", queryNotion)
.addNode("slack", querySlack)
.addNode("synthesize", synthesizeResults)
.addEdge(START, "classify")
.addConditionalEdges("classify", routeToAgents, ["github", "notion", "slack"])
.addEdge("github", "synthesize")
.addEdge("notion", "synthesize")
.addEdge("slack", "synthesize")
.addEdge("synthesize", END)
.compile();
const result = await workflow.invoke({
query: "如何认证 API 请求？",
});
console.log("原始查询:", result.query);
console.log("\n分类:");
for (const c of result.classifications) {
console.log(` ${c.source}: ${c.query}`);
}
console.log(`\n${"=".repeat(60)}\n`);
console.log("最终答案:");
console.log(result.finalAnswer);
