AI Agent 多智能体协作框架实战：从零构建你的自动化工作流
什么是 AI Agent 多智能体系统？
随着大语言模型（LLM）能力的持续突破，单一 AI 助手已经难以胜任复杂、多步骤的业务场景。多智能体系统（Multi-Agent System）应运而生——它将复杂任务拆解为多个子任务，分配给专门化的 Agent 分工协作，最终汇聚结果，极大提升了自动化程度与任务完成质量。
多智能体系统的核心价值在于：
分工明确：每个 Agent 专注于自身擅长的领域，如信息检索 Agent、代码生成 Agent、质量评审 Agent 等；
并行加速：多个 Agent 可以并发执行互不依赖的子任务，显著缩短整体完成时间；
容错鲁棒：单个 Agent 失败可由其他 Agent 补偿或重试，整体流程更加稳健；
可扩展性强：新增业务只需插入新的 Agent，无需重构整体架构。
三大主流框架横向对比
目前社区最活跃的多智能体框架主要有三个：LangGraph、AutoGen 和 CrewAI。它们各有侧重，适合不同的使用场景。
LangGraph：由 LangChain 团队开发，以有向图（DAG/循环图）为核心抽象，强调状态管理与流程控制，适合需要复杂分支与回退逻辑的工作流；
AutoGen：微软开源，以"对话驱动"为核心，Agent 之间通过自然语言消息协作，上手门槛低，适合快速原型验证；
CrewAI：角色扮演风格，强调 Agent 的角色（Role）与目标（Goal）定义，适合模拟团队协作场景，代码简洁优雅。
以下是三个框架的对比总结：
框架 核心抽象 学习曲线 适合场景
LangGraph 状态图 中等 复杂工作流、需回退
AutoGen 对话消息 低 快速原型、对话任务
CrewAI 角色/任务 低 团队协作模拟
LangGraph 实战：构建代码审查 Agent 流水线
下面以 LangGraph 为例，构建一个包含「需求解析 → 代码生成 → 质量审查 → 输出报告」四个节点的 Agent 流水线。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 1. 定义全局状态
class PipelineState(TypedDict):
requirement: str
code: str
review_result: str
report: str
# 2. 定义各 Agent 节点
def parse_requirement(state: PipelineState):
"""需求解析 Agent"""
# 调用 LLM 将自然语言需求转化为结构化描述
structured = llm.invoke(f"请将以下需求结构化：{state['requirement']}")
return {"requirement": structured.content}
def generate_code(state: PipelineState):
"""代码生成 Agent"""
code = llm.invoke(f"根据需求生成 Python 代码：{state['requirement']}")
return {"code": code.content}
def review_code(state: PipelineState):
"""质量审查 Agent"""
review = llm.invoke(f"请对以下代码进行安全和质量审查：\n{state['code']}")
return {"review_result": review.content}
def write_report(state: PipelineState):
"""报告生成 Agent"""
report = llm.invoke(
f"根据代码审查结果生成最终报告：\n{state['review_result']}"
)
return {"report": report.content}
# 3. 构建图
graph = StateGraph(PipelineState)
graph.add_node("parse", parse_requirement)
graph.add_node("codegen", generate_code)
graph.add_node("review", review_code)
graph.add_node("report", write_report)
# 4. 连接节点
graph.set_entry_point("parse")
graph.add_edge("parse", "codegen")
graph.add_edge("codegen", "review")
graph.add_edge("review", "report")
graph.add_edge("report", END)
app = graph.compile()
result = app.invoke({"requirement": "实现一个线程安全的单例模式"})
print(result["report"])
上述示例展示了 LangGraph 最核心的使用方式：定义状态类型、编写节点函数、连接有向边，最终 compile() 生成可执行的流水线。
AutoGen 实战：多 Agent 对话协作
AutoGen 的多 Agent 协作以消息传递为核心，非常适合需要多轮对话的场景，比如「产品经理 → 开发工程师 → 测试工程师」的模拟协作。
import autogen
config_list = [{"model": "gpt-4o", "api_key": "YOUR_API_KEY"}]
llm_config = {"config_list": config_list, "temperature": 0.3}
# 创建各角色 Agent
pm = autogen.AssistantAgent(
name="ProductManager",
system_message="你是产品经理，负责分析需求、拆解任务，输出清晰的功能规格书。",
llm_config=llm_config,
)
dev = autogen.AssistantAgent(
name="Developer",
system_message="你是资深 Python 开发工程师，负责根据规格书编写高质量代码。",
llm_config=llm_config,
)
tester = autogen.AssistantAgent(
name="Tester",
system_message="你是测试工程师，负责为代码编写单元测试，找出潜在 Bug。",
llm_config=llm_config,
)
user_proxy = autogen.UserProxyAgent(
name="UserProxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
code_execution_config={"work_dir": "coding", "use_docker": False},
)
# 启动群聊
groupchat = autogen.GroupChat(
agents=[user_proxy, pm, dev, tester],
messages=[],
max_round=12,
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)
user_proxy.initiate_chat(
manager,
message="请协作完成：实现一个支持过期时间的 LRU 缓存类（Python）",
)
AutoGen 的 GroupChat 模式允许多个 Agent 根据上下文自动决策下一个发言者，整个协作过程自然流畅，非常接近真实团队的工作方式。
CrewAI 实战：角色化团队任务
CrewAI 采用「船员（Crew）」隐喻，每个 Agent 拥有明确的角色、目标和背景故事，任务（Task）与 Agent 绑定，整体代码极为简洁。
from crewai import Agent, Task, Crew, Process
# 定义 Agent
researcher = Agent(
role="AI 研究员",
goal="调研最新的 Transformer 优化技术并整理摘要",
backstory="你是一位专注于深度学习的研究员，擅长阅读论文和总结技术趋势。",
verbose=True,
allow_delegation=False,
)
writer = Agent(
role="技术博客作者",
goal="将研究摘要转化为面向开发者的高质量技术博客文章",
backstory="你是有 10 年经验的技术写作专家，擅长将复杂概念用通俗语言表达。",
verbose=True,
allow_delegation=False,
)
# 定义任务
research_task = Task(
description="调研 2024-2025 年 Transformer 注意力机制的最新优化方案（如 FlashAttention3、MLA 等），输出 500 字技术摘要。",
expected_output="包含技术要点、性能对比和应用场景的结构化摘要",
agent=researcher,
)
write_task = Task(
description="基于研究摘要，撰写一篇 1500 字面向 Python 开发者的技术博客文章，需包含代码示例。",
expected_output="完整的 Markdown 格式博客文章",
agent=writer,
)
# 组建 Crew 并执行
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential, # 顺序执行
verbose=True,
)
result = crew.kickoff()
print(result)
工程化落地：关键设计原则
将多智能体系统从 Demo 推向生产环境，需要关注以下几个核心工程问题：
状态持久化：使用 Redis 或数据库存储 Agent 的中间状态，支持任务中断后恢复，避免重复消耗 Token；
工具调用安全：所有 Agent 的 Tool 调用都应设置超时与重试策略，防止单个外部 API 超时拖垮整个流水线；
Token 预算管理：为每个 Agent 设置最大 Token 上限，防止无限循环的对话消耗过多成本；
可观测性：集成 LangSmith、Arize Phoenix 等 LLM 可观测平台，记录每次 Agent 调用的输入输出、耗时和费用；
幂等性设计：对于有副作用的工具（如发邮件、写数据库），确保 Agent 调用具有幂等性，避免重试导致重复执行。
# 示例：带重试和超时的 Tool 调用包装器
import functools
import time
def with_retry(max_retries=3, timeout=30, delay=1.0):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
import signal
def handler(signum, frame):
raise TimeoutError(f"Tool call timed out after {timeout}s")
signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout)
result = func(*args, **kwargs)
signal.alarm(0)
return result
except TimeoutError as e:
print(f"Attempt {attempt+1} timed out: {e}")
except Exception as e:
print(f"Attempt {attempt+1} failed: {e}")
if attempt < max_retries - 1:
time.sleep(delay * (2 ** attempt)) # 指数退避
raise RuntimeError(f"Tool {func.__name__} failed after {max_retries} retries")
return wrapper
return decorator
@with_retry(max_retries=3, timeout=30)
def call_external_api(query: str) -> str:
# 你的外部 API 调用逻辑
pass
性能调优与成本控制
多智能体系统的运行成本不可忽视，以下是几个行之有效的优化策略：
分级模型策略：简单子任务（分类、摘要）使用轻量模型（如 GPT-4o-mini），复杂推理任务才调用旗舰模型，综合成本可降低 60%-80%；
提示词缓存（Prompt Caching）：Anthropic Claude 和 OpenAI 均支持系统提示缓存，重复调用时 Token 成本大幅降低；
结果缓存：对相同输入的 Agent 调用结果做语义哈希缓存，命中率高的场景可直接返回缓存结果；
并行化执行：使用 asyncio 并发执行互不依赖的 Agent 子任务，将串行时间复杂度降为并行。
import asyncio
from langchain_openai import ChatOpenAI
llm_fast = ChatOpenAI(model="gpt-4o-mini") # 轻量模型
llm_smart = ChatOpenAI(model="gpt-4o") # 旗舰模型
async def run_parallel_agents(tasks: list[str]) -> list[str]:
"""并发执行多个独立子任务"""
coroutines = [llm_fast.ainvoke(task) for task in tasks]
results = await asyncio.gather(*coroutines, return_exceptions=True)
return [r.content if not isinstance(r, Exception) else str(r) for r in results]
# 并发执行 5 个独立子任务
tasks = [
"总结第1章要点",
"总结第2章要点",
"总结第3章要点",
"总结第4章要点",
"总结第5章要点",
]
results = asyncio.run(run_parallel_agents(tasks))
总结与选型建议
多智能体系统已经从实验室走向生产落地，选择合适的框架至关重要：
需要精确流程控制、复杂分支判断→ 优先选 LangGraph；
需要快速验证想法、对话式协作 → 优先选 AutoGen；
需要角色化团队、代码简洁 → 优先选 CrewAI；
生产环境建议搭配 LangSmith 做可观测，Redis 做状态持久化，分级模型策略控制成本。
AI Agent 时代的核心竞争力不再只是"会用大模型"，而是能否设计出高效、稳健、可维护的多智能体协作架构。希望本文的框架对比与实战代码能帮助你快速上手，在项目中落地属于自己的 AI 自动化工作流。
发布评论
热门评论区：