前言随着大模型应用逐渐从简单问答发展到复杂业务流程单个 Agent 已经很难覆盖所有场景。例如一个企业智能客服系统可能同时需要查询企业知识库查询订单、客户和库存数据分析销售报表创建售后工单发送通知或邮件对高风险操作进行人工确认。如果把所有能力都放到一个 Agent 中Prompt 会越来越长工具数量会越来越多模型也更容易出现误判。更合理的方式是把复杂任务拆分给多个专业 Agent用户请求 ↓ 路由 Agent ├─ 知识库 Agent ├─ 数据分析 Agent ├─ 售后处理 Agent └─ 通知 Agent本文将使用 LangGraph 构建一个简化的企业智能服务系统重点实现多 Agent 协作Swarm 架构Agent 间动态任务交接共享状态管理RAG 与工具调用高风险操作人工确认任务恢复与断点续跑路由循环检测和可观测性。一、什么是 Swarm 架构1. Supervisor 架构传统多 Agent 系统通常采用 Supervisor 架构。用户 ↓ Supervisor ├─ Agent A ├─ Agent B └─ Agent C所有任务都必须先经过 Supervisor再由 Supervisor 决定交给哪个 Agent。这种架构比较容易理解适合固定流程但也存在几个问题Supervisor 容易变成性能瓶颈所有路由逻辑都集中在一个节点Agent 之间不能直接协作业务扩展后 Supervisor Prompt 会变得复杂。2. Swarm 架构Swarm 架构强调多个 Agent 之间的动态交接。┌──────────────┐ │ Knowledge │ │ Agent │ └──────┬───────┘ │ ┌──────────┐ ┌──────▼───────┐ ┌──────────────┐ │ Triage │ ──→ │ Data Agent │ ──→ │ Action Agent │ │ Agent │ └──────────────┘ └──────────────┘ └──────────┘ │ ▼ ┌──────────────┐ │ Quality │ │ Agent │ └──────────────┘在 Swarm 中每个 Agent 都具备两类能力处理自己擅长的任务判断是否应该把任务交给其他 Agent。例如知识库 Agent 发现用户需要查询实时订单数据可以交给 Data AgentData Agent 查询完数据后可以交给 Quality Agent 总结Action Agent 发现操作需要审批时可以暂停任务并等待用户确认。Swarm 的核心不是“Agent 越多越好”而是每个 Agent 只负责自己擅长的事情并且能够在明确规则下完成任务交接。二、项目场景设计本文以企业售后服务系统为例。用户输入查询一下华东区最近三个月的客户投诉情况并找出投诉最多的产品。如果问题严重创建一张售后工单。系统可以拆分为以下 AgentAgent主要职责Triage Agent判断任务类型和初始路由Knowledge Agent检索产品手册、售后规范和企业知识库Data Agent查询订单、客户和投诉数据库Action Agent创建工单、发送通知等写操作Quality Agent汇总结果、检查引用和生成最终答案完整流程如下用户提出问题 ↓ Triage Agent 判断任务 ↓ Data Agent 查询投诉数据 ↓ Knowledge Agent 查询售后标准 ↓ Quality Agent 生成分析结果 ↓ Action Agent 根据用户确认创建工单三、项目初始化安装主要依赖pipinstall-Ulanggraph langchain langchain-openai pydantic如果需要接入向量数据库可以额外安装pipinstallchromadb sentence-transformers推荐目录结构multi-agent-swarm/ ├─ app/ │ ├─ graph.py │ ├─ state.py │ ├─ agents/ │ │ ├─ triage.py │ │ ├─ knowledge.py │ │ ├─ data.py │ │ ├─ action.py │ │ └─ quality.py │ ├─ tools/ │ │ ├─ knowledge_search.py │ │ ├─ database.py │ │ └─ ticket.py │ └─ observability.py ├─ evals/ │ └─ cases.jsonl ├─ .env └─ requirements.txt四、定义共享状态多个 Agent 能够协作关键在于共享状态。# app/state.pyfromtypingimportAnnotated,TypedDictfromlangchain_core.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesclassAgentState(TypedDict,totalFalse):messages:Annotated[list[AnyMessage],add_messages]# 当前正在处理任务的 Agentactive_agent:str# Agent 交接记录handoff_history:list[dict]# 检索结果retrieved_documents:list[dict]# 数据查询结果query_result:dict|None# 待审批动作pending_action:dict|None# 最终答案final_answer:str|None# 任务状态status:str这里的messages使用add_messages作为 Reducer新的消息会追加到历史消息中。其他字段则可以保存当前任务的中间结果。例如{active_agent:data_agent,retrieved_documents:[],query_result:{region:华东,total_complaints:126},handoff_history:[{from:triage_agent,to:data_agent,reason:需要查询实时投诉数据}]}五、定义 Agent 路由协议为了避免 Agent 自由发挥建议使用结构化输出描述下一步动作。# app/agents/protocol.pyfromtypingimportLiteralfrompydanticimportBaseModel,FieldclassAgentDecision(BaseModel):next_agent:Literal[knowledge_agent,data_agent,action_agent,quality_agent]Field(description下一步负责处理任务的 Agent)reason:strField(description任务交接原因)response:strField(default,description当前 Agent 产生的阶段性结果)使用结构化输出有三个好处避免从自然语言中猜测路由可以限制 Agent 只能跳转到合法节点方便记录和统计 Agent 的决策结果。路由白名单ALLOWED_HANDOFFS{triage_agent:{knowledge_agent,data_agent,quality_agent},knowledge_agent:{data_agent,action_agent,quality_agent},data_agent:{knowledge_agent,action_agent,quality_agent},action_agent:{quality_agent},quality_agent:set()}生产环境中不要允许模型返回任意节点名称所有目标 Agent 都必须经过白名单校验。六、实现 Agent 节点下面封装一个通用的 Agent 节点。# app/agents/base.pyfromlangchain_core.messagesimportAIMessagefromlanggraph.typesimportCommandfromapp.stateimportAgentStatefromapp.agents.protocolimportAgentDecisionfromapp.agents.rulesimportALLOWED_HANDOFFSdefcreate_agent_node(agent_name,system_prompt,llm):decision_modelllm.with_structured_output(AgentDecision)defnode(state:AgentState):messages[{role:system,content:system_prompt}]messages.extend(state.get(messages,[])[-12:])decisiondecision_model.invoke(messages)allowed_targetsALLOWED_HANDOFFS.get(agent_name,set())ifdecision.next_agentnotinallowed_targets:next_agentquality_agentelse:next_agentdecision.next_agent historylist(state.get(handoff_history,[]))history.append({from:agent_name,to:next_agent,reason:decision.reason})updates{active_agent:next_agent,handoff_history:history,messages:[AIMessage(nameagent_name,contentdecision.response)]}returnCommand(gotonext_agent,updateupdates)returnnode这个节点做了几件事给当前 Agent 注入系统提示词只保留最近一部分上下文使用结构化输出得到下一步路由校验目标 Agent写入交接记录使用Command跳转到下一个节点。这就是 Swarm 架构中的核心交接逻辑。七、创建不同职责的 Agent1. Triage Agenttriage_prompt 你是任务分流 Agent。 你的职责是判断用户问题属于哪一类 - 需要检索企业文档交给 knowledge_agent - 需要查询数据库或实时业务数据交给 data_agent - 需要执行写操作交给 action_agent - 信息已经足够交给 quality_agent。 不要自己执行数据库查询和写操作。 2. Knowledge Agentknowledge_prompt 你是企业知识库 Agent。 你的职责是 1. 查询产品手册、售后规范和内部制度 2. 提取与当前问题最相关的内容 3. 保存文档来源和页码 4. 如果问题需要实时数据交给 data_agent 5. 如果需要创建工单交给 action_agent。 回答必须基于检索结果不要编造企业规则。 3. Data Agentdata_prompt 你是业务数据 Agent。 你的职责是 1. 查询投诉、订单、客户和库存数据 2. 只执行只读 SQL 3. 对查询结果进行简单统计 4. 不得修改数据库 5. 数据查询完成后交给 quality_agent。 如果用户要求修改数据必须交给 action_agent。 4. Action Agentaction_prompt 你是业务操作 Agent。 你可以创建售后工单、发送通知和更新业务状态。 所有写操作都必须 1. 明确操作对象 2. 展示操作摘要 3. 请求用户确认 4. 用户确认后才可以执行 5. 记录操作审计日志。 5. Quality Agentquality_prompt 你是最终质量检查 Agent。 你的职责是 1. 汇总其他 Agent 的结果 2. 检查数字是否来自数据查询 3. 检查知识库引用是否存在 4. 对无法确认的信息明确说明 5. 生成简洁、结构化的最终答案。 八、构建 LangGraph Swarm# app/graph.pyfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromapp.stateimportAgentStatefromapp.agents.baseimportcreate_agent_nodedefbuild_graph(llm):builderStateGraph(AgentState)builder.add_node(triage_agent,create_agent_node(triage_agent,triage_prompt,llm))builder.add_node(knowledge_agent,create_agent_node(knowledge_agent,knowledge_prompt,llm))builder.add_node(data_agent,create_agent_node(data_agent,data_prompt,llm))builder.add_node(action_agent,create_agent_node(action_agent,action_prompt,llm))builder.add_node(quality_agent,create_agent_node(quality_agent,quality_prompt,llm))builder.add_edge(START,triage_agent)builder.add_edge(quality_agent,END)checkpointerMemorySaver()returnbuilder.compile(checkpointercheckpointer)这里的路由由各个 Agent 通过Command(goto...)动态完成。在生产环境中不建议使用MemorySaver作为唯一持久化方案。可以替换成数据库 Checkpointer开发环境MemorySaver 单机生产SQLite Checkpointer 企业生产PostgreSQL Checkpointer 高并发场景Redis PostgreSQL九、执行一次多 Agent 任务graphbuild_graph(llm)config{configurable:{thread_id:conversation_10001}}resultgraph.invoke({messages:[{role:user,content:查询华东区最近三个月投诉最多的产品并分析原因}],active_agent:triage_agent,handoff_history:[],status:running},configconfig)thread_id非常重要。它用于关联当前会话Checkpoint中断任务人工审批后续恢复执行。同一个thread_id可以让系统继续之前没有完成的任务。十、加入 RAG 工具知识库 Agent 不应该直接读取所有文档而应该通过受控工具访问知识库。fromlangchain_core.toolsimporttooltooldefsearch_knowledge(query:str)-str: 查询企业知识库。 resultsvector_store.similarity_search(queryquery,k5)ifnotresults:return没有找到相关资料。output[]foriteminresults:output.append(f来源{item.metadata.get(source)}\nf页码{item.metadata.get(page)}\nf内容{item.page_content})return\n\n.join(output)工具返回结果时必须带上来源信息来源售后服务规范.pdf 页码12 内容重大质量问题需要在 24 小时内创建售后工单。这样最终回答才能做到可追溯。十一、加入数据库查询工具数据库工具必须限制权限。fromlangchain_core.toolsimporttooltooldefquery_complaints(region:str,months:int3)-str: 查询指定区域最近几个月的客户投诉数据。 该工具只读不允许修改数据库。 ifmonths1ormonths12:raiseValueError(months 参数必须在 1 到 12 之间)ifregionnotin{华东,华南,华北,西南}:raiseValueError(region 参数不在允许范围内)rowsdatabase.fetch_complaints(regionregion,monthsmonths)returnformat_complaint_result(rows)不要让模型直接生成任意 SQL# 不推荐sqlllm.invoke(请生成 SQL)database.execute(sql)更安全的方式是只暴露固定查询工具参数使用 Schema 校验数据库账号只读限制查询时间和返回行数禁止访问系统表记录完整审计日志。十二、高风险操作与人工确认创建工单、发送邮件、修改订单等操作不能让 Agent 自动执行。LangGraph 提供了interrupt机制可以在执行前暂停流程。fromlanggraph.typesimportinterruptdefaction_agent_node(state:AgentState):action{type:create_ticket,title:华东区产品投诉异常,priority:high,description:最近三个月投诉数量明显上升}approvalinterrupt({type:approval_required,message:是否创建高优先级售后工单,action:action})ifnotapproval.get(approved):return{status:cancelled,messages:[{role:assistant,content:用户拒绝创建售后工单。}]}ticket_idticket_service.create(action)return{status:completed,messages:[{role:assistant,content:f售后工单已创建编号{ticket_id}}]}前端收到approval_required事件后展示确认弹窗。用户确认后使用相同的thread_id恢复任务。graph.invoke(Command(resume{approved:True}),config{configurable:{thread_id:conversation_10001}})这类机制比单纯在 Prompt 中写“请先确认”可靠得多因为审批逻辑由工作流引擎控制而不是依赖模型自觉。十三、如何避免 Agent 无限循环多 Agent 协作最容易出现的问题是knowledge_agent → data_agent data_agent → knowledge_agent knowledge_agent → data_agent如果没有限制任务可能一直循环。1. 限制最大交接次数MAX_HANDOFFS6historystate.get(handoff_history,[])iflen(history)MAX_HANDOFFS:returnCommand(gotoquality_agent,update{status:handoff_limit_reached})2. 检测重复路径defhas_loop(history:list[dict])-bool:iflen(history)4:returnFalserecent[(item[from],item[to])foriteminhistory[-4:]]returnlen(set(recent))23. 设置超时和取消每次任务都应该设置最大执行时间最大模型调用次数最大工具调用次数最大上下文长度用户取消入口。4. 记录交接原因不要只记录A → B还要记录A → B 原因用户问题需要实时投诉数据没有原因的路由很难调试。十四、上下文管理多 Agent 系统不能把完整历史消息无限传给每个 Agent。建议采用三种方式最近消息窗口recent_messagesstate[messages][-12:]阶段性摘要summary 用户关注华东区最近三个月客户投诉情况。 Data Agent 已查询到 126 条投诉记录。 Knowledge Agent 找到售后处理规范第 12 页。 Agent 专属上下文不同 Agent 只读取与自己相关的信息Knowledge Agent读取问题、文档和检索结果 Data Agent读取问题、筛选条件和数据库查询结果 Action Agent读取操作摘要和审批状态 Quality Agent读取所有最终结果这样可以减少 token 消耗也能降低上下文污染。十五、Swarm 架构的可观测性多 Agent 系统必须记录每次路由和工具调用。推荐事件结构{trace_id:trace_001,thread_id:conversation_10001,agent:data_agent,event:handoff,target:quality_agent,reason:数据查询已完成,latency_ms:820,status:success}需要重点统计每个 Agent 的调用次数平均处理耗时Agent 之间的交接次数最常见的路由路径循环路由数量工具调用失败率人工审批通过率不同 Agent 的 token 消耗最终任务完成率。可以把一次任务画成链路trace_001 ├─ triage_agent 120ms ├─ data_agent 820ms │ └─ query_complaints 230ms ├─ knowledge_agent 640ms │ └─ search_knowledge 410ms └─ quality_agent 950ms当任务失败时可以快速判断是哪个节点出现问题。十六、测试策略多 Agent 系统不能只测试最终答案还要测试过程。1. 路由测试deftest_complaint_question_goes_to_data_agent():decisiontriage_agent.invoke(查询华东区最近三个月投诉数量)assertdecision.next_agentdata_agent2. 知识库测试deftest_knowledge_agent_must_return_source():resultknowledge_agent.invoke(重大质量问题如何处理)assertresult.sourcesassertresult.sources[0][page]isnotNone3. 高风险操作测试deftest_action_requires_approval():resultrun_task(删除所有客户数据)assertresult.statusapproval_required4. 循环测试deftest_handoff_loop_is_blocked():state{handoff_history:[{from:knowledge_agent,to:data_agent},{from:data_agent,to:knowledge_agent},{from:knowledge_agent,to:data_agent},{from:data_agent,to:knowledge_agent}]}asserthas_loop(state[handoff_history])5. 恢复测试重点测试Agent 执行中服务重启用户关闭页面后重新打开人工审批后继续执行工具调用失败后重试模型超时后降级。十七、Swarm 架构的优缺点优点Agent 职责清晰支持动态任务交接复杂任务更容易拆解新增 Agent 时不需要修改所有逻辑适合企业客服、数据分析和自动化流程。缺点路由调试难度更高更容易出现循环上下文管理复杂每次交接都会增加模型调用成本Agent 权限边界必须设计清楚。因此不是所有项目都需要 Swarm。如果业务流程是固定的解析问题 → 查询数据 → 生成报告 → 发送邮件使用普通 StateGraph 或 Supervisor 可能更简单。如果任务经常变化且不同专家之间需要动态协作Swarm 才更有价值。十八、上线前检查清单每个 Agent 都有明确职责Agent 之间的交接目标有白名单共享状态字段有明确含义所有工具都有参数校验数据库工具默认只读高风险操作必须人工确认Agent 有最大交接次数支持循环检测支持超时和任务取消使用 Checkpointer 保存任务状态每个任务都有thread_id记录 Agent 路由和工具调用链路Prompt 具有版本号建立路由、工具和最终答案评测集支持失败重试和版本回滚总结多 Agent 协作的关键不是简单地把多个模型调用放在一起而是建立一套稳定的任务协作机制。LangGraph 为我们提供了图结构工作流状态管理动态路由Command 跳转interrupt 人工确认Checkpoint 任务恢复节点级可观测性。Swarm 架构则进一步让多个 Agent 从“被动执行节点”变成“可以自主交接任务的协作单元”。一套可靠的企业级多 Agent 系统应该同时具备专业化 Agent 清晰的任务路由 受控的工具调用 可恢复的任务状态 高风险操作审批 完整的链路追踪 完善的评测与回归当这些能力组合起来之后AI Agent 才能从简单聊天机器人逐步演进成真正可以落地到企业业务中的智能协作系统。