Agent架构解析与实战(五)--Multi Agent
告别单体智能的瓶颈,构建你的‘AI 专家特种部队’——Multi-Agent 架构通过模拟人类团队的专业分工与协作,让复杂任务的解决能力实现从‘泛泛而谈’到‘深度洞察’的质的飞跃。

参考来源:all-agentic-architectures[1]
目录
- 架构定义
- 宏观工作流
- 常见架构模式
- 主流框架对比
- 应用场景
- 优缺点分析
- 代码实现详解
- 总结与核心要点
1. 架构定义
Multi-Agent System (多智能体系统) 是一种超越单一 Agent 概念的架构,它模拟一个由专业化智能体组成的团队,通过协作来解决复杂问题。
核心理念

| 特性 | 描述 |
|---|---|
| 分工协作 | 每个 Agent 拥有独特的角色、人设和技能集,模拟人类专家团队的工作方式 |
| 任务分解 | 复杂问题被分解为子任务,分配给最适合的 Agent 处理 |
| 中央协调 | 通过中央控制器或定义好的工作流协议来管理 Agent 间的通信和任务路由 |
2. 宏观工作流
Multi-Agent System 的工作流程遵循 "分解-分配-协作-综合" 的模式:

详细步骤说明
| 步骤 | 名称 | 描述 |
|---|---|---|
| 1 | 分解任务 | 主控制器或用户提供复杂任务 |
| 2 | 角色分配 | 系统根据定义的角色(如"研究员"、"编码员"、"评论家"、"写手")将子任务分配给专业 Agent |
| 3 | 协作执行 | Agent 并行或串行执行任务,将输出传递给彼此或传递到中央"黑板" |
| 4 | 综合输出 | 最终的"Manager"或"Synthesizer" Agent 收集专家 Agent 的输出,组装最终的综合响应 |
3. 常见架构模式
3.1 Supervisor (监督者) 架构
在监督者架构中,一个中央"监督者"Agent 负责编排多个"子 Agent"或"工作者 Agent"。

工作流程
- 接收查询:监督者接收用户查询或初始输入
- 推理与委派:利用 LLM 进行推理,决定将任务委派给哪个专业 Worker Agent
- 任务执行:Worker Agent 使用特定工具执行分配的任务
- 反馈与进一步行动:监督者接收输出,决定是否需要进一步委派
3.2 Hierarchical (层次化) 架构
层次化架构扩展了监督者模型,适用于更大、更复杂的系统。

关键特征
| 特征 | 描述 |
|---|---|
| 分层控制 | 顶层编排器 → 中层监督者 → 底层专业 Agent |
| 任务分解 | 高层管理战略规划,低层专注具体执行 |
| 专业化 | 每一层和每个 Agent 可以专注于不同的抽象层次或领域 |
3.3 高级协作模式
随着深度学习与 Agent 研究的深入,出现了更复杂的协作模式以提升系统的可靠性与推理能力:
| 模式 | 描述 | 适用场景 |
|---|---|---|
| Multi-Agent Debate (多智能体辩论) | 让持不同人设或观点的 Agent 针对同一问题进行辩论。通过多轮“提出-反驳-修正”循环,消除幻觉和偏见。 | 事实核查、复杂逻辑推理、减少模型幻觉 |
| Reflection (反思/自我修正) | 引入“批判者 (Critic)”角色,专门负责检查执行 Agent 的输出,并提出修改建议,形成“执行-评估-修正”闭环。 | 代码生成、长文本写作、数学解题 |
| Voting / Ensembling (投票共识) | 多个 Agent 独立生成答案,通过多数投票 (Majority Vote) 或加权方式决定最终结果,利用群体智慧提升准确率。 | 高风险决策、开放式问答、提高系统鲁棒性 |
4. 主流框架对比
Multi-Agent 生态日益成熟,三大主流框架各有千秋:
| 特性 | LangGraph | AutoGen | CrewAI |
|---|---|---|---|
| 核心机制 | 图论:基于状态机和有向图,将 Agent 视为图中的节点,Edge 定义流转逻辑。 | 对话:Agent 之间通过自然语言对话驱动任务进展,模拟人类群聊。 | 角色:模拟企业团队结构,强调明确的角色定义、层级和任务流。 |
| 控制粒度 | 极高:支持循环、条件分支、中断与状态持久化,适合构建确定性强的逻辑。 | 中等:侧重于交互模式,流程相对灵活,适合涌现式协作。 | 中/高:结构化强,封装了大量协同逻辑,适合按部就班的任务链。 |
| 适用场景 | 企业级生产系统、复杂业务流自动化、需要 Human-in-the-loop 的应用。 | 快速原型验证、开放式问题探索、代码解释器交互、多角色扮演游戏。 | 内容创作团队、市场分析流水线、无需复杂状态管理的标准作业程序。 |
5. 应用场景
| 场景 | 描述 | 示例 |
|---|---|---|
| 复杂报告生成 | 需要多领域专业知识的详细报告 | 财务分析报告、科学研究报告 |
| 软件开发流水线 | 模拟开发团队 | 程序员 + 代码审查员 + 测试员 + 项目经理 |
| 创意头脑风暴 | 不同"个性"的 Agent 团队 | 乐观者 + 谨慎者 + 狂野创意者 |
6. 优缺点分析
✅ 优点
| 优点 | 说明 |
|---|---|
| 专业化与深度 | 每个 Agent 可以针对特定人设和工具进行调优,在其领域产出更高质量的工作 |
| 模块化与可扩展性 | 可以轻松添加、移除或升级单个 Agent,无需重新设计整个系统 |
| 并行性 | 多个 Agent 可以同时处理其子任务,可能减少总体任务时间 |
| 健壮性与错误恢复 | 层次化分解可以提高对错误和不确定性的健壮性 |
❌ 缺点
| 缺点 | 说明 |
|---|---|
| 协调开销 | 管理 Agent 之间的通信和工作流增加了系统设计的复杂性 |
| 成本与延迟增加 | 运行多个 Agent 涉及更多 LLM 调用,可能比单 Agent 方法更昂贵和缓慢 |
7. 代码实现详解
6.1 环境配置
依赖库安装
pip install -q -U langchain-openai langchain langgraph python-dotenv rich tavily-python pydantic langchain-core
导入依赖并设置密钥
import os
import re
from typing import List, Annotated, TypedDict, Optional
from dotenv import load_dotenv
from tavily import TavilyClient
# LangChain组件
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
# LangGraph组件
from langgraph.graph import StateGraph, END
from langgraph.graph.message import AnyMessage, add_messages
from langgraph.prebuilt import ToolNode, tools_condition
# 美化输出
from rich.console import Console
from rich.markdown import Markdown
# API密钥和追踪配置
load_dotenv()
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Agentic Architecture - Multi-Agent (OpenAI)"
for key in ["OPENAI_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:
if not os.environ.get(key):
print(f"{key} 未找到. 请创建.env文件并配置对应密钥.")
print("环境变量和LangChain追踪配置完成.")
运行结果:
环境变量和LangChain追踪配置完成.
6.2 基础对照——单一"通才型" Agent
首先测试单个 Agent 在复杂任务上的表现。
console = Console()
# 定义Agent间通用的状态
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
# 定义工具和LLM
client = TavilyClient(os.environ["TAVILY_API_KEY"])
@tool
def web_search(query: str) -> str:
"""根据传入的query在网上搜索相关信息"""
result = client.search(query, max_results=3)
return str(result["results"])
api_key=os.environ["OPENAI_API_KEY"]
llm = ChatOpenAI(
model="gpt-5-mini",
temperature=0.2,
api_key=api_key,
base_url="https://api.openai.com/v1"
)
llm_with_tools = llm.bind_tools([web_search])
# 定义单一通才型Agent
def monolithic_agent_node(state: AgentState):
console.print("--- MONOLITHIC AGENT: Thinking... ---")
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
tool_node = ToolNode([web_search])
# 创建为通才型Agent创建ReAct图
mono_graph_builder = StateGraph(AgentState)
mono_graph_builder.add_node("agent", monolithic_agent_node)
mono_graph_builder.add_node("tools", tool_node)
mono_graph_builder.set_entry_point("agent")
# 条件函数
def should_continue(state: AgentState):
"""检查是否需要继续执行工具"""
messages = state["messages"]
last_message = messages[-1]
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return "tools"
return END
mono_graph_builder.add_conditional_edges("agent", should_continue)
mono_graph_builder.add_edge("tools", "agent")
monolithic_agent_app = mono_graph_builder.compile()
print("单一通才型Agent创建成功.")
运行结果:
单一通才型Agent创建成功.
单一 Agent 架构图

6.3 多 Agent 专家组
接下来构建专家团队:新闻分析师、技术分析师和财务分析师。每个都是独立的 Agent 节点,具有特定的人设。最终由报告撰写者担任 Manager,汇编他们的工作。
定义多 Agent 状态
# 多Agent系统的状态会包含每一个专家的输出
class MultiAgentState(TypedDict):
user_request: str
news_report: Optional[str]
technical_report: Optional[str]
financial_report: Optional[str]
final_report: Optional[str]
创建专家节点工厂函数
def create_specialist_node(persona: str, output_key: str):
"""Factory function to create a specialist agent node."""
system_prompt = persona + "\n\nYou have access to a web search tool. Your output MUST be a concise report section in Chinese, formatted in markdown, focusing only on your area of expertise."
# 构建 ChatPromptTemplate
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{user_request}")
])
# 使用 prompt_template 格式化输入
agent = prompt_template | llm_with_tools
def specialist_node(state: MultiAgentState):
console.print(f"--- CALLING {output_key.replace('_report','').upper()} ANALYST ---")
result = agent.invoke({"user_request": state["user_request"]})
content = result.content if result.content else f"No direct content, tool calls: {result.tool_calls}"
return {output_key: content}
return specialist_node
创建专家节点
# 创建专家节点
news_analyst_node = create_specialist_node(
"You are an expert News Analyst. Your specialty is scouring the web for the latest news, articles, and social media sentiment about a company.",
"news_report"
)
technical_analyst_node = create_specialist_node(
"You are an expert Technical Analyst. You specialize in analyzing stock price charts, trends, and technical indicators.",
"technical_report"
)
financial_analyst_node = create_specialist_node(
"You are an expert Financial Analyst. You specialize in interpreting financial statements and performance metrics.",
"financial_report"
)
创建报告撰写节点
def report_writer_node(state: MultiAgentState):
"""The manager agent that synthesizes the specialist reports."""
console.print("--- CALLING REPORT WRITER ---")
prompt = f"""You are an expert financial editor. Your task is to combine the following specialist reports into a single, professional, and cohesive market analysis report in Chinese. Add a brief introductory and concluding paragraph.
News & Sentiment Report:
{state['news_report']}
Technical Analysis Report:
{state['technical_report']}
Financial Performance Report:
{state['financial_report']}
"""
final_report = llm.invoke(prompt).content
return {"final_report": final_report}
print("专家节点以及报告撰写节点已定义...")
运行结果:
专家节点以及报告撰写节点已定义...
6.4 构建多 Agent 图
multi_agent_graph_builder = StateGraph(MultiAgentState)
# 加入所有节点
multi_agent_graph_builder.add_node("news_analyst", news_analyst_node)
multi_agent_graph_builder.add_node("technical_analyst", technical_analyst_node)
multi_agent_graph_builder.add_node("financial_analyst", financial_analyst_node)
multi_agent_graph_builder.add_node("report_writer", report_writer_node)
# 定义工作流
multi_agent_graph_builder.set_entry_point("news_analyst")
multi_agent_graph_builder.add_edge("news_analyst", "technical_analyst")
multi_agent_graph_builder.add_edge("technical_analyst", "financial_analyst")
multi_agent_graph_builder.add_edge("financial_analyst", "report_writer")
multi_agent_graph_builder.add_edge("report_writer", END)
multi_agent_app = multi_agent_graph_builder.compile()
print("多Agent专家组创建成功.")
运行结果:
多Agent专家组创建成功.
多 Agent 架构图

[!NOTE]
专家 Agent 按顺序工作(在实际应用中可以并行运行),最终步骤始终是报告撰写者进行综合。
7.1 测试通才型 Agent
company = "Zhipu AI"
monolithic_query = f"Create a brief but comprehensive market analysis report for {company} in Chinese. The report should include three sections: 1. A summary of recent news and market sentiment. 2. A basic technical analysis of the stock's price trend. 3. A look at the company's recent financial performance."
console.print(f"[bold yellow]Testing MONOLITHIC agent on a multi-faceted task:[/bold yellow]\n'{monolithic_query}'\n")
final_mono_output = monolithic_agent_app.invoke({
"messages": [
SystemMessage(content="You are a single, expert financial analyst. You must create a comprehensive report covering all aspects of the user's request in Chinese."),
HumanMessage(content=monolithic_query)
]
})
console.print("\n--- [bold red]Final Report from Monolithic Agent[/bold red] ---")
console.print(Markdown(final_mono_output['messages'][-1].content))
运行过程:
Testing MONOLITHIC agent on a multi-faceted task:
'Create a brief but comprehensive market analysis report for Zhipu AI in Chinese...'
--- MONOLITHIC AGENT: Thinking... ---
--- MONOLITHIC AGENT: Thinking... ---
--- MONOLITHIC AGENT: Thinking... ---
--- Final Report from Monolithic Agent ---
7.2 测试多 Agent 专家组
multi_agent_query = f"Create a brief but comprehensive market analysis report for {company}."
initial_multi_agent_input = {"user_request": multi_agent_query}
console.print(f"[bold green]Testing MULTI-AGENT TEAM on the same task:[/bold green]\n'{multi_agent_query}'\n")
final_multi_agent_output = multi_agent_app.invoke(initial_multi_agent_input)
console.print("\n--- [bold green]Final Report from Multi-Agent Team[/bold green] ---")
console.print(Markdown(final_multi_agent_output['final_report']))
运行过程:
Testing MULTI-AGENT TEAM on the same task:
'Create a brief but comprehensive market analysis report for Zhipu AI.'
--- CALLING NEWS ANALYST ---
--- CALLING TECHNICAL ANALYST ---
--- CALLING FINANCIAL ANALYST ---
--- CALLING REPORT WRITER ---
--- Final Report from Multi-Agent Team ---
7.3 输出对比分析
通才型 Agent 输出特点
| 问题 | 描述 |
|---|---|
| 结构模糊 | 各部分可能混合在一起,缺乏清晰的标题或专业格式 |
| 分析肤浅 | 试图同时成为三个领域的专家,每个领域只能提供概述 |
| 语言通用 | 语言可能过于泛化,缺乏每个领域真正专家的特定术语和焦点 |
多 Agent 专家组输出特点
| 优势 | 描述 |
|---|---|
| 高度结构化 | 每个分析领域都有清晰、独立的部分,因为每个部分都是由具有特定格式指令的专家生成 |
| 更深入分析 | 每个部分包含更详细的、特定领域的语言和见解 |
| 更专业 | 由报告撰写者组装的最终报告读起来像专业文档,有清晰的引言、正文和结论 |
7.4 量化评估
class ReportEvaluation(BaseModel):
"""Schema for evaluating a financial report."""
clarity_and_structure_score: int = Field(description="Score 1-10 on the report's organization, structure, and clarity in Chinese.")
analytical_depth_score: int = Field(description="Score 1-10 on the depth and quality of the analysis in each section in Chinese.")
completeness_score: int = Field(description="Score 1-10 on how well the report addressed all parts of the user's request in Chinese.")
justification: str = Field(description="A brief justification for the scores in Chinese.")
judge_llm = llm.with_structured_output(ReportEvaluation)
def evaluate_report(query: str, report: str):
prompt = f"""You are an expert judge of financial analysis reports. Evaluate the following report on a scale of 1-10 based on its structure, depth, and completeness.
**Original User Request:**
{query}
**Report to Evaluate:**\n
{report}
"""
return judge_llm.invoke(prompt)
评估结果对比
| 评估维度 | 通才型 Agent | 多 Agent 专家组 |
|---|---|---|
| 结构清晰度 | 9 | 9 |
| 分析深度 | 6 | 8 |
| 完整性 | 7 | 8 |
通才型 Agent 评估反馈:
优点:报告结构清晰,按用户要求分为三大部分并附结论与监控要点,语言简洁、要点明确。
不足:分析深度偏概括性,技术面缺乏具体量化指标(如 RSI、均线数值、成交量变化的具体数据或图示)、缺乏时间序列或回测支撑;财务部分依赖媒体汇总,缺少经审计的关键财务数据。
多 Agent 专家组评估反馈:
报告整体结构清晰、条理分明(引言、舆情、技术面、财务与商业模式、风险、催化剂、结论与免责声明),语言简练,便于不同类型投资者快速抓取要点。
分析深度较好:既覆盖了舆情与技术面(含具体可套用的技术指标),又讨论了商业化路径、成本结构、现金/融资敏感性与关键风险,提供了可操作的观察清单与投资者分层建议。
[!IMPORTANT]
评估结果证明了假设:多 Agent 团队的报告在分析深度和完整性方面得分明显更高。这证实了对于可分解为多个专业领域的复杂任务,Multi-Agent 架构是生成高质量、结构化和可靠结果的更优方法。
8. 总结与核心要点
🎓 核心成果
通过创建一个由专业化 Agent 组成的团队(每个都有专注的人设和角色)以及一个 Manager 来综合他们的工作,产生了一个质量明显更高的最终输出。
专业化的力量

| 关键点 | 说明 |
|---|---|
| 专业化分工 | 正如人类组织一样,将大问题分解并将各部分分配给专家会产生更好的结果 |
| 协调复杂性 | 虽然这种架构在编排方面引入了更多复杂性,但最终输出的结构、深度和专业性的显著改进使其成为不可或缺的模式 |
| 适用场景 | 任何需要跨多个领域提供专家级表现的严肃 Agentic 应用 |
关键代码速查表
| 组件 | 代码 |
|---|---|
| 多 Agent 状态 | class MultiAgentState(TypedDict) |
| 专家节点工厂 | create_specialist_node(persona, output_key) |
| 创建图 | StateGraph(MultiAgentState) |
| 添加节点 | graph_builder.add_node("name", node_func) |
| 定义流程 | graph_builder.add_edge("from", "to") |
| 编译运行 | graph_builder.compile() |
架构选择指南
| 场景 | 推荐架构 |
|---|---|
| 简单任务,单一领域 | 单一 Agent (ReAct) |
| 中等复杂度,需要协调 | Supervisor 架构 |
| 高复杂度,多团队协作 | Hierarchical 架构 |
| 需要独立并行处理的多领域任务 | 并行 Multi-Agent |
参考链接
[1] all-agentic-architectures: https://github.com/FareedKhan-dev/all-agentic-architectures/tree/main
更多推荐


所有评论(0)