04 May 2026 / AI Agent 多智能体 LangGraph AutoGen CrewAI LLM 人工智能

AI Agent 多智能体协作框架实战：从原理到生产部署

什么是多智能体协作系统？

随着大语言模型（LLM）能力的快速提升，单一 Agent 已经无法满足复杂业务场景的需求。多智能体协作系统（Multi-Agent System）通过将复杂任务拆分给多个具有不同能力的 Agent 并行或串行处理，显著提升了任务完成的质量和效率。

核心思想是：每个 Agent 专注于自己的"专业领域"，通过明确的通信协议相互协作，就像一个高效的团队分工合作。典型的场景包括：

代码研究员负责搜索和阅读文档，代码编写者负责生成代码，测试 Agent 负责验证
数据分析师 Agent 处理数据，报告撰写 Agent 生成报告，审核 Agent 做质量把控
客服路由 Agent 判断意图，专业 Agent 处理具体业务，汇总 Agent 整合回复

主流框架选型对比

目前市面上有多个成熟的多智能体框架，各有侧重。选择合适的框架是项目成功的第一步。

LangGraph：基于有向图的状态机模型，适合有明确工作流程和条件分支的场景，对 LangChain 生态兼容好，上手门槛适中
AutoGen（Microsoft）：对话驱动型框架，Agent 之间通过自然语言消息交互，适合需要多轮人机协作的场景，灵活但不确定性较高
CrewAI：角色导向框架，用"角色（Role）+目标（Goal）+背景（Backstory）"定义 Agent，适合模拟团队协作，配置简洁，适合快速原型

综合考量：生产环境推荐 LangGraph（流程可控、可审计）；快速验证推荐 CrewAI；研究场景推荐 AutoGen。

用 LangGraph 构建一个代码审查多 Agent

下面我们用 LangGraph 实现一个包含"代码分析 Agent + 安全检查 Agent + 汇总 Agent"的协作流程。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定义共享状态
class ReviewState(TypedDict):
    code: str
    analysis: str
    security_issues: list[str]
    summary: str

# 代码分析 Agent
def code_analyzer(state: ReviewState) -> ReviewState:
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")
    result = llm.invoke(f"分析以下代码的逻辑和质量问题：\n{state['code']}")
    return {"analysis": result.content}

# 安全检查 Agent
def security_checker(state: ReviewState) -> ReviewState:
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")
    result = llm.invoke(f"检查以下代码的安全漏洞（SQL注入、XSS、敏感信息泄露等）：\n{state['code']}")
    issues = result.content.split('\n') if result.content else []
    return {"security_issues": [i for i in issues if i.strip()]}

# 汇总 Agent
def summarizer(state: ReviewState) -> ReviewState:
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"代码质量分析：{state['analysis']}\n安全问题：{chr(10).join(state['security_issues'])}\n请生成简洁的审查报告。"
    result = llm.invoke(prompt)
    return {"summary": result.content}

# 构建图
builder = StateGraph(ReviewState)
builder.add_node("analyzer", code_analyzer)
builder.add_node("security", security_checker)
builder.add_node("summarizer", summarizer)

# 并行执行分析和安全检查，最后汇总
builder.set_entry_point("analyzer")
builder.add_edge("analyzer", "security")
builder.add_edge("security", "summarizer")
builder.add_edge("summarizer", END)

graph = builder.compile()

# 运行
result = graph.invoke({"code": "SELECT * FROM users WHERE id = " + user_input})
print(result["summary"])

关键设计原则：让多 Agent 系统更可靠

在生产环境中，多 Agent 系统面临的最大挑战是不确定性和错误传播。以下是经过实践验证的设计原则：

状态不可变原则：每个 Agent 只追加或更新自己负责的状态字段，不修改其他 Agent 的输出，避免相互覆盖
幂等性设计：每个 Agent 的处理逻辑应该是幂等的，相同输入始终产生相同输出，方便重试和调试
超时与熔断：为每个 Agent 设置执行超时（建议 30s），并实现熔断机制，防止一个 Agent 的失败拖垮整个流水线
人工审核节点：对于高风险操作（如发送邮件、写数据库），在流程图中插入 human-in-the-loop 节点，保留人工干预能力
完整的日志追踪：记录每个 Agent 的输入输出，使用 LangSmith 或自定义日志系统，便于事后审计和优化

生产部署注意事项

将多 Agent 系统从本地部署到生产环境，需要关注以下几个关键问题：

并发控制：使用 asyncio 实现 Agent 并行执行，但要注意 LLM API 的 Rate Limit，建议使用令牌桶算法控制并发数
成本优化：为不同复杂度的任务选择不同的模型，简单分类任务用 gpt-4o-mini，复杂推理任务才用 claude-3-5-sonnet
状态持久化：使用 Redis 或 PostgreSQL 持久化 Agent 状态，支持长时间运行任务的断点续传
监控告警：监控 Agent 的成功率、平均耗时、Token 消耗，及时发现性能退化

import asyncio
from langchain_openai import ChatOpenAI

async def run_agent_async(agent_func, state, semaphore):
    """带并发控制的异步 Agent 执行器"""
    async with semaphore:
        try:
            result = await asyncio.wait_for(
                asyncio.to_thread(agent_func, state),
                timeout=30.0
            )
            return result
        except asyncio.TimeoutError:
            return {"error": f"Agent {agent_func.__name__} timed out"}
        except Exception as e:
            return {"error": str(e)}

# 最多同时运行 3 个 Agent
semaphore = asyncio.Semaphore(3)

async def run_parallel_agents(state):
    tasks = [
        run_agent_async(code_analyzer, state, semaphore),
        run_agent_async(security_checker, state, semaphore),
    ]
    results = await asyncio.gather(*tasks)
    return results

多智能体系统代表了 AI 工程化的重要方向。从单一 LLM 调用到精心设计的 Agent 协作网络，需要开发者同时具备 AI 能力和系统工程思维。建议从简单的两 Agent 场景入手，逐步迭代到复杂系统。