/**
* 多源知识 router 示例
*
* 此示例演示多智能体系统的 router pattern。
* router 对查询进行分类，将查询并行路由到专用智能体，
* 并将结果合成为组合响应。
*/
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}' 匹配的 issues：#142（API 身份验证文档）、#89（OAuth 流程）、#203（token 刷新）`;
},
{
name: "search_issues",
description: "搜索 GitHub issues 和 pull requests。",
schema: z.object({
query: z.string(),
}),
}
);
const searchPrs = tool(
async ({ query }) => {
return `PR #156 添加了 JWT 身份验证，PR #178 更新了 OAuth scopes`;
},
{
name: "search_prs",
description: "搜索 pull requests 以查找实现细节。",
schema: z.object({
query: z.string(),
}),
}
);
const searchNotion = tool(
async ({ query }) => {
return `找到文档：'API Authentication Guide'，涵盖 OAuth2 流程、API keys 和 JWT tokens`;
},
{
name: "search_notion",
description: "在 Notion workspace 中搜索文档。",
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 中找到讨论：'使用 Bearer tokens 进行 API 身份验证，刷新流程请参阅文档'`;
},
{
name: "search_slack",
description: "搜索 Slack 消息和线程。",
schema: z.object({
query: z.string(),
}),
}
);
const getThread = tool(
async ({ threadId }) => {
return `线程讨论了 API key 轮换的最佳实践`;
},
{
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.4" });
const githubAgent = createAgent({
model: llm,
tools: [searchCode, searchIssues, searchPrs],
systemPrompt: `
你是 GitHub 专家。通过搜索仓库、issues 和 pull requests，
回答有关代码、API references 和实现细节的问题。
`.trim(),
});
const notionAgent = createAgent({
model: llm,
tools: [searchNotion, getPage],
systemPrompt: `
你是 Notion 专家。通过搜索组织的 Notion workspace，
回答有关内部流程、政策和团队文档的问题。
`.trim(),
});
const slackAgent = createAgent({
model: llm,
tools: [searchSlack, getThread],
systemPrompt: `
你是 Slack 专家。通过搜索团队成员分享知识和解决方案的
相关线程和讨论来回答问题。
`.trim(),
});
const routerLlm = new ChatOpenAI({ model: "gpt-5.4-mini" });
// 为分类器定义结构化输出 schema
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 references、实现细节、issues、pull requests
- notion：内部文档、流程、政策、团队 wiki
- slack：团队讨论、非正式知识分享、近期对话
仅返回与查询相关的来源。每个来源都应包含
针对该特定知识领域优化的定向子问题。
以“如何对 API 请求进行身份验证？”为例：
- github："存在哪些身份验证代码？搜索 auth middleware 和 JWT handling"
- notion："存在哪些身份验证文档？查找 API auth guides"
（省略 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);
