参考来源:all-agentic-architectures[1]


目录

  1. 架构定义
  2. 宏观工作流
  3. 常见架构模式
  4. 主流框架对比
  5. 应用场景
  6. 优缺点分析
  7. 代码实现详解
  8. 总结与核心要点

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"。

工作流程
  1. 接收查询:监督者接收用户查询或初始输入
  2. 推理与委派:利用 LLM 进行推理,决定将任务委派给哪个专业 Worker Agent
  3. 任务执行:Worker Agent 使用特定工具执行分配的任务
  4. 反馈与进一步行动:监督者接收输出,决定是否需要进一步委派

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

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐