多 Agent 协作实战：基于 LangGraph Swarm 架构实现智能任务路由、动态交接与企业级工程化落地

前言

随着大模型应用逐渐从简单问答发展到复杂业务流程，单个 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 发现用户需要查询实时订单数据，可以交给 Data Agent；
• Data Agent 查询完数据后，可以交给 Quality Agent 总结；
• Action Agent 发现操作需要审批时，可以暂停任务并等待用户确认。

Swarm 的核心不是“Agent 越多越好”，而是：

每个 Agent 只负责自己擅长的事情，并且能够在明确规则下完成任务交接。

二、项目场景设计

本文以企业售后服务系统为例。

用户输入：

查询一下华东区最近三个月的客户投诉情况，并找出投诉最多的产品。如果问题严重，创建一张售后工单。

系统可以拆分为以下 Agent：

完整流程如下：

用户提出问题
↓
Triage Agent 判断任务
↓
Data Agent 查询投诉数据
↓
Knowledge Agent 查询售后标准
↓
Quality Agent 生成分析结果
↓
Action Agent 根据用户确认创建工单

三、项目初始化

安装主要依赖：

pip install -U langgraph langchain langchain-openai pydantic

如果需要接入向量数据库，可以额外安装：

pip install chromadb 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.py

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict, total=False):
messages: Annotated[list[AnyMessage], add_messages]

# 当前正在处理任务的 Agent
active_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.py

from typing import Literal
from pydantic import BaseModel, Field

class AgentDecision(BaseModel):
next_agent: Literal[
"knowledge_agent",
"data_agent",
"action_agent",
"quality_agent"
] = Field(description="下一步负责处理任务的 Agent")

reason: str = Field(description="任务交接原因")

response: str = Field(
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.py

from langchain_core.messages import AIMessage
from langgraph.types import Command

from app.state import AgentState
from app.agents.protocol import AgentDecision
from app.agents.rules import ALLOWED_HANDOFFS

def create_agent_node(agent_name, system_prompt, llm):
decision_model = llm.with_structured_output(AgentDecision)

def node(state: AgentState):
messages = [
{
"role": "system",
"content": system_prompt
}
]

messages.extend(state.get("messages", [])[–12:])

decision = decision_model.invoke(messages)

allowed_targets = ALLOWED_HANDOFFS.get(agent_name, set())

if decision.next_agent not in allowed_targets:
next_agent = "quality_agent"
else:
next_agent = decision.next_agent

history = list(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(
name=agent_name,
content=decision.response
)
]
}

return Command(
goto=next_agent,
update=updates
)

return node

这个节点做了几件事：

这就是 Swarm 架构中的核心交接逻辑。

七、创建不同职责的 Agent

1. Triage Agent

triage_prompt = """
你是任务分流 Agent。

你的职责是判断用户问题属于哪一类：

– 需要检索企业文档，交给 knowledge_agent；
– 需要查询数据库或实时业务数据，交给 data_agent；
– 需要执行写操作，交给 action_agent；
– 信息已经足够，交给 quality_agent。

不要自己执行数据库查询和写操作。
"""

2. Knowledge Agent

knowledge_prompt = """
你是企业知识库 Agent。

你的职责是：

1. 查询产品手册、售后规范和内部制度；
2. 提取与当前问题最相关的内容；
3. 保存文档来源和页码；
4. 如果问题需要实时数据，交给 data_agent；
5. 如果需要创建工单，交给 action_agent。

回答必须基于检索结果，不要编造企业规则。
"""

3. Data Agent

data_prompt = """
你是业务数据 Agent。

你的职责是：

1. 查询投诉、订单、客户和库存数据；
2. 只执行只读 SQL；
3. 对查询结果进行简单统计；
4. 不得修改数据库；
5. 数据查询完成后交给 quality_agent。

如果用户要求修改数据，必须交给 action_agent。
"""

4. Action Agent

action_prompt = """
你是业务操作 Agent。

你可以创建售后工单、发送通知和更新业务状态。

所有写操作都必须：

1. 明确操作对象；
2. 展示操作摘要；
3. 请求用户确认；
4. 用户确认后才可以执行；
5. 记录操作审计日志。
"""

5. Quality Agent

quality_prompt = """
你是最终质量检查 Agent。

你的职责是：

1. 汇总其他 Agent 的结果；
2. 检查数字是否来自数据查询；
3. 检查知识库引用是否存在；
4. 对无法确认的信息明确说明；
5. 生成简洁、结构化的最终答案。
"""

八、构建 LangGraph Swarm

# app/graph.py

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from app.state import AgentState
from app.agents.base import create_agent_node

def build_graph(llm):
builder = StateGraph(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)

checkpointer = MemorySaver()

return builder.compile(checkpointer=checkpointer)

这里的路由由各个 Agent 通过 Command(goto=…) 动态完成。

在生产环境中，不建议使用 MemorySaver 作为唯一持久化方案。可以替换成数据库 Checkpointer：

开发环境：MemorySaver
单机生产：SQLite Checkpointer
企业生产：PostgreSQL Checkpointer
高并发场景：Redis + PostgreSQL

九、执行一次多 Agent 任务

graph = build_graph(llm)

config = {
"configurable": {
"thread_id": "conversation_10001"
}
}

result = graph.invoke(
{
"messages": [
{
"role": "user",
"content": "查询华东区最近三个月投诉最多的产品，并分析原因"
}
],
"active_agent": "triage_agent",
"handoff_history": [],
"status": "running"
},
config=config
)

thread_id 非常重要。

它用于关联：

• 当前会话；
• Checkpoint；
• 中断任务；
• 人工审批；
• 后续恢复执行。

同一个 thread_id 可以让系统继续之前没有完成的任务。

十、加入 RAG 工具

知识库 Agent 不应该直接读取所有文档，而应该通过受控工具访问知识库。

from langchain_core.tools import tool

@tool
def search_knowledge(query: str) –> str:
"""
查询企业知识库。
"""
results = vector_store.similarity_search(
query=query,
k=5
)

if not results:
return "没有找到相关资料。"

output = []

for item in results:
output.append(
f"来源：{item.metadata.get('source')}\\n"
f"页码：{item.metadata.get('page')}\\n"
f"内容：{item.page_content}"
)

return "\\n\\n".join(output)

工具返回结果时，必须带上来源信息：

来源：售后服务规范.pdf
页码：12
内容：重大质量问题需要在 24 小时内创建售后工单。

这样最终回答才能做到可追溯。

十一、加入数据库查询工具

数据库工具必须限制权限。

from langchain_core.tools import tool

@tool
def query_complaints(region: str, months: int = 3) –> str:
"""
查询指定区域最近几个月的客户投诉数据。
该工具只读，不允许修改数据库。
"""
if months < 1 or months > 12:
raise ValueError("months 参数必须在 1 到 12 之间")

if region not in {"华东", "华南", "华北", "西南"}:
raise ValueError("region 参数不在允许范围内")

rows = database.fetch_complaints(
region=region,
months=months
)

return format_complaint_result(rows)

不要让模型直接生成任意 SQL：

# 不推荐
sql = llm.invoke("请生成 SQL")
database.execute(sql)

更安全的方式是：

• 只暴露固定查询工具；
• 参数使用 Schema 校验；
• 数据库账号只读；
• 限制查询时间和返回行数；
• 禁止访问系统表；
• 记录完整审计日志。

十二、高风险操作与人工确认

创建工单、发送邮件、修改订单等操作，不能让 Agent 自动执行。

LangGraph 提供了 interrupt 机制，可以在执行前暂停流程。

from langgraph.types import interrupt

def action_agent_node(state: AgentState):
action = {
"type": "create_ticket",
"title": "华东区产品投诉异常",
"priority": "high",
"description": "最近三个月投诉数量明显上升"
}

approval = interrupt({
"type": "approval_required",
"message": "是否创建高优先级售后工单？",
"action": action
})

if not approval.get("approved"):
return {
"status": "cancelled",
"messages": [
{
"role": "assistant",
"content": "用户拒绝创建售后工单。"
}
]
}

ticket_id = ticket_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_HANDOFFS = 6

history = state.get("handoff_history", [])

if len(history) >= MAX_HANDOFFS:
return Command(
goto="quality_agent",
update={
"status": "handoff_limit_reached"
}
)

2. 检测重复路径

def has_loop(history: list[dict]) –> bool:
if len(history) < 4:
return False

recent = [
(item["from"], item["to"])
for item in history[–4:]
]

return len(set(recent)) <= 2

3. 设置超时和取消

每次任务都应该设置：

• 最大执行时间；
• 最大模型调用次数；
• 最大工具调用次数；
• 最大上下文长度；
• 用户取消入口。

4. 记录交接原因

不要只记录：

A → B

还要记录：

A → B
原因：用户问题需要实时投诉数据

没有原因的路由，很难调试。

十四、上下文管理

多 Agent 系统不能把完整历史消息无限传给每个 Agent。

建议采用三种方式：

最近消息窗口

recent_messages = state["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. 路由测试

def test_complaint_question_goes_to_data_agent():
decision = triage_agent.invoke(
"查询华东区最近三个月投诉数量"
)

assert decision.next_agent == "data_agent"

2. 知识库测试

def test_knowledge_agent_must_return_source():
result = knowledge_agent.invoke(
"重大质量问题如何处理？"
)

assert result.sources
assert result.sources[0]["page"] is not None

3. 高风险操作测试

def test_action_requires_approval():
result = run_task(
"删除所有客户数据"
)

assert result.status == "approval_required"

4. 循环测试

def test_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"}
]
}

assert has_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 才能从简单聊天机器人，逐步演进成真正可以落地到企业业务中的智能协作系统。