构建一个多Agent系统,在当下可以说是AI工程化落地的一个关键方向。尤其当你面对的是那些任务复杂度高、需要明确专业分工、同时还得兼顾可维护性和扩展性的场景时,多Agent架构几乎就成了一个绕不开的选择。下面这套方法论,结合了LangGraph、CrewAI、smolagents等主流框架的实践经验,从步骤到细节,希望能帮你构建一个真正能打的多Agent系统。

一、明确系统目标与架构模式
1.1 典型应用场景
在实际落地中,下面这几个场景是目前多Agent系统用得最多、效果也最明显的地方:
- 智能客服:从意图识别到知识检索,再到回答生成,一套流程下来,体验提升不少。
- 自动化报告:数据采集、分析、内容撰写,一条龙服务。
- 工业运维助手:设备状态查询、故障诊断、工单生成,一气呵成。
- 医疗问诊:症状收集、知识库匹配、给出建议,整个过程可以更规范。
1.2 选择协作架构(关键!)
选对了协作模式,系统的复杂度和可控性就有了基础。这里直接上干货:
| 架构类型 | 特点 | 适用场景 |
|---|---|---|
| Supervisor(主管模式) | 中央协调者决定任务路由 | 大多数企业级应用,可控性强 |
| Handoffs(交接模式) | Agent 主动移交控制权 | 需灵活跳转、支持人工干预 |
| Team(团队模式) | 类似人类团队分工协作 | 内容创作、研究分析等流程化任务 |
| Network(全互联) | 任意Agent可调用其他 | 小规模实验,生产慎用 |
二、选择技术栈与框架
画个重点,技术选型直接影响到开发的效率和系统未来的可维护性。
| 框架 | 特点 | 适合人群 |
|---|---|---|
| LangGraph | 基于状态图,支持条件边、Human-in-the-loop、可观测性 | 企业级开发,需精细控制流 |
| CrewAI | 面向“角色-任务-团队”抽象,代码简洁 | 快速原型、内容生成类任务 |
| smolagents / Dify | 轻量、支持 CodeAgent(可生成并执行代码) | 需要动态编程能力的场景 |
| AutoGen(微软) | 支持多Agent对话、GroupChat | 学术研究、对话模拟 |
三、核心组件设计(以 LangGraph 为例)
3.1 定义共享状态(State)
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage
class AgentState(TypedDict):
messages: Annotated[List[AnyMessage], add_messages] # 对话历史
task_type: str # 任务类型(如 "diagnosis", "report")
device_id: str # 当前操作对象(可选)
remaining_steps: int # 防止无限循环
3.2 创建专业化 Agent 节点
def knowledge_agent(state: AgentState):
query = state["messages"][-1].content
# 调用RAG或API
result = rag_retriever.invoke(query)
return {"messages": [AIMessage(content=result)]}
def reasoning_agent(state: AgentState):
context = "n".join([m.content for m in state["messages"]])
response = llm.invoke(f"基于以下信息推理:{context}")
return {"messages": [AIMessage(content=response)]}
3.3 设计主管(Supervisor)决策逻辑
def supervisor(state: AgentState):
prompt = f"""用户问题:{state['messages'][-1].content}
请选择下一步执行的Agent:
- 若需查资料 → 返回 'knowledge'
- 若需推理总结 → 返回 'reasoning'
- 若已完成 → 返回 'END'"""
decision = llm.with_structured_output(SupervisorSchema).invoke(prompt)
return {"next": decision.agent_name}
3.4 编排图结构
from langgraph.graph import StateGraph, START, END
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("knowledge", knowledge_agent)
builder.add_node("reasoning", reasoning_agent)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
"supervisor",
lambda s: s["next"],
{"knowledge": "knowledge", "reasoning": "reasoning", "END": END}
)
builder.add_edge("knowledge", "supervisor")
builder.add_edge("reasoning", "supervisor")
graph = builder.compile()
四、增强系统可靠性(高级技巧)
✅ 加入 Human-in-the-Loop
# 在关键节点插入人工审批
def human_review(state):
print("请审核以下内容:", state["messages"][-1].content)
approval = input("是否通过?(y/n): ")
return {"approved": approval == "y"}
✅ 支持任务回滚与修正
靠谱的系统从来不是一把梭。可以借用全局监控Agent监听用户中途修改的意图(比如“我改时间了”),然后触发状态重置或者直接跳转到指定节点。
✅ 引入长期记忆
from langgraph.store.memory import PostgresStore
long_term_store = PostgresStore(conn_string="postgresql://...")
# 在Agent中读写用户偏好、历史记录
五、部署与可观测性
系统上线只是开始,真正的挑战在于持续运维。这几个点值得花精力去打磨:
- 日志追踪:集成LangFuse或LangSmith,记录每一步Agent的决策逻辑,出了问题才能追根溯源。
- 性能监控:统计各Agent的调用频次、耗时、失败率,用数据说话。
- 容器化部署:用Docker加FastAPI把Agent封装成微服务,方便扩展和管理。
- 安全控制:如果用了CodeAgent,一定得做好import权限限制,比如只允许
datetime、json这类基础模块,避免安全风险。
六、快速上手示例(CrewAI 版)
如果只是想快速原型的,CrewAI这套极简组合拳足够用了:
from crewai import Agent, Task, Crew
researcher = Agent(role="数据研究员", goal="收集2025年AI趋势", verbose=True)
writer = Agent(role="技术作家", goal="撰写清晰报告", verbose=True)
task1 = Task(description="调研多Agent框架进展", agent=researcher)
task2 = Task(description="基于调研写一篇综述", agent=writer)
crew = Crew(agents=[researcher, writer], tasks=[task1, task2])
result = crew.kickoff()
print(result)
总结:构建多Agent系统的 Checklist
最后,附上一张实用Checklist,照着这个框架走,基本能避开大多数常见的坑:
- 明确任务是否需要多Agent(工具 >3 个?任务 >2 步?)
- 选择合适架构(Supervisor / Team / Handoffs)
- 定义清晰的 State 和消息协议
- 每个 Agent 保持单一职责
- 加入错误处理与防循环机制
- 集成可观测性工具(LangFuse 等)
- 考虑 Human-in-the-Loop 场景
