一个 AI 不够用？我用 Python 搭了 3 个智能体，让它们分工协作解决复杂任务

很多人刚接触 AI Agent 的时候，第一反应都是：

“给大模型一个强一点的提示词，不就行了吗？”

一开始，这种想法没什么问题。
让一个模型回答问题、写文章、总结资料，很多时候确实已经够用了。

但只要任务一复杂，问题就会立刻暴露出来：

• 既要理解需求，又要规划步骤
• 还要查资料、执行动作、整合信息
• 最后最好还能自己检查一遍结果靠不靠谱

这时候，一个 Agent 往往会开始“身兼数职”，最后变成：

• 该规划的时候在瞎执行
• 该执行的时候又在重新理解任务
• 该审查结果的时候却继续生成新内容

于是输出越来越乱，上下文越来越长，结果越来越不稳定。

所以，多智能体系统开始变得重要。

这篇文章，我不打算只讲概念。
我会直接带你搭一个最小可运行的三智能体协作 Demo，让你真正理解：

• 多智能体到底解决了什么问题
• 它和“多个提示词”有什么区别
• 一个三智能体系统应该怎么拆
• 用 Python 怎么把它真正跑起来
• 工程上最容易踩哪些坑

---

一、为什么一个 AI 往往不够

单 Agent 最大的问题，不是它不聪明，而是它职责太杂。

在简单任务里，一个 Agent 可以同时做这些事：

1. 理解用户意图
2. 制定解决方案
3. 执行具体步骤
4. 组织最终输出

但一旦任务复杂，这种“全包式”设计就容易失控。

比如你给它一个任务：

“帮我整理一份关于多智能体系统的博客提纲，补充一些代码示例，并对内容质量做一次自检。”

看起来只是一个任务，实际上里面至少混了三类能力：

• 任务理解与拆解
• 信息生成与执行
• 结果评估与修正

如果都交给同一个 Agent，它就会出现一种典型现象：

什么都能做一点，但没有一个环节做得特别稳。

这背后的原因很简单：
复杂任务不是“一个大问题”，而是多个不同类型的小问题串在一起。

而多智能体的本质，就是把这种复杂度拆开。

---

二、多智能体到底是什么，不是什么

先说一个常见误区：

不是把三个对话框摆在一起，就叫多智能体。

真正的多智能体系统，不是“模型数量变多了”，而是系统内部出现了明确的角色分工、状态流转和协作机制。

也就是说，它至少要回答清楚这几个问题：

• 每个 Agent 的职责是什么？
• 它们的输入和输出是什么？
• 上一个 Agent 的结果，怎么传给下一个？
• 谁来做最终决策？
• 如果某一步失败了，系统怎么处理？

所以，多智能体更像一个小型智能协作系统，而不是几个模型一起聊天。

你可以把它类比成一个项目小组：

• 有人负责拆任务
• 有人负责干活
• 有人负责审核

如果三个人都在做同样的事，那不叫协作，那叫混乱。

---

三、最常见的多智能体协作模式

多智能体不是只有一种形态，工程里常见的有下面几类。

1）规划者 + 执行者

这是最基础的一种。

• 规划者（Planner）：理解目标，拆解步骤
• 执行者（Executor）：按步骤完成具体任务

适合那种“先想清楚，再逐步做”的问题。

---

2）主控 Agent + 专家 Agent

• 主控 Agent：决定当前该调用谁
• 专家 Agent：只处理自己擅长的领域任务

比如：

• 数学题给数学 Agent
• 法律文本给法律 Agent
• 数据分析给数据 Agent

这种模式的核心不在“专家多厉害”，而在主控能不能路由对。

---

3）检索 Agent + 分析 Agent + 写作 Agent

这类很适合知识密集型任务。

• 检索 Agent：查资料
• 分析 Agent：理解信息、提炼观点
• 写作 Agent：整理成结构化输出

这其实已经很接近很多 RAG 系统和 Research Agent 的雏形了。

---

4）审核 Agent / 反思 Agent / 裁判 Agent

这类 Agent 不负责“生产”，而负责“纠偏”。

它常见的职责包括：

• 检查事实错误
• 检查逻辑冲突
• 检查是否跑题
• 给出修改建议

很多系统不是败在“不会生成”，而是败在“生成之后没人检查”。

---

四、一个三智能体系统该怎么设计

如果你是第一次做多智能体，我建议不要一上来就设计五六个角色。
三智能体其实已经足够说明问题了。

一个非常经典、也非常实用的拆法是：

Agent 1：任务理解与拆解

它负责：

• 理解用户目标
• 把复杂任务拆成步骤
• 明确每一步要产出什么

它不直接做内容，只负责“想清楚”。

---

Agent 2：信息搜集与执行

它负责：

• 根据任务计划执行具体动作
• 生成中间结果
• 补充必要内容

它是系统里真正“干活”的角色。

---

Agent 3：结果审查与优化

它负责：

• 检查结果是否完整
• 看有没有逻辑问题
• 给出修正建议
• 生成最终优化版输出

它像一个“质检员”。

---

它们之间怎么传递信息？

这一步非常关键。

最差的做法是：
把所有历史对话、所有中间结果、所有提示词一股脑塞给每个 Agent。

这样会导致：

• 上下文越来越长
• 噪声越来越多
• 每个 Agent 开始做不属于自己的事

更合理的方式是：
只传必要结构化状态。

例如：

{
  "task": "写一篇多智能体系统入门博客",
  "plan": ["解释单Agent问题", "介绍多Agent模式", "给出Python Demo"],
  "draft": "......",
  "review": "结构基本完整，但代码部分解释不够"
}

也就是说，Agent 之间传的不是“整段大脑”，而是有限、清晰、可控的状态。

---

五、用 Python 实现一个最小多智能体 Demo

下面我们直接上代码。

这个 Demo 的目标不是做一个工业级 Agent 平台，而是让你快速看懂：

• 多智能体到底怎么协作
• 每个 Agent 如何只做自己的事
• 状态如何在系统中流动

为了让代码更容易迁移，我这里采用一种非常轻量的实现方式：

• 一个统一的 llm_call() 负责调用模型
• 三个 Agent 各自封装自己的职责
• 一个 MultiAgentSystem 负责调度流程
• 用一个共享状态 state 串起整个过程

---

1）先定义一个最小 LLM 调用层

你可以把它接到 OpenAI 兼容接口、本地模型、Ollama，或者任何你自己的推理服务。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

def llm_call(system_prompt: str, user_prompt: str, model: str = "gpt-4o-mini") -> str:
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.7,
    )
    return response.choices[0].message.content.strip()

如果你暂时不想接真实模型，也可以先写一个 mock 版本：

def llm_call(system_prompt: str, user_prompt: str, model: str = "mock-model") -> str:
    return f"[Mocked Response]\\nSYSTEM: {system_prompt[:60]}\\nUSER: {user_prompt[:120]}"

---

2）定义共享状态

多智能体系统一定要有“状态”这个概念。
否则 Agent 之间就只是各说各话。

from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class AgentState:
    user_task: str
    plan: List[str] = field(default_factory=list)
    execution_result: str = ""
    review_result: str = ""
    final_output: str = ""
    logs: List[Dict[str, Any]] = field(default_factory=list)

    def add_log(self, agent_name: str, content: str):
        self.logs.append({
            "agent": agent_name,
            "content": content
        })

这个 AgentState 很重要，因为它体现了多智能体系统和“多段提示词”的区别：

• 这里有显式状态
• 状态在不同 Agent 之间流转
• 每个 Agent 只读取自己需要的部分
• 最终系统可以回溯每一步发生了什么

---

3）定义三个 Agent

Planner：负责任务拆解

class PlannerAgent:
    name = "PlannerAgent"

    def run(self, state: AgentState) -> AgentState:
        system_prompt = """
你是一个任务规划智能体。
你的职责只有一个：理解用户目标，并将其拆解为3到5个清晰步骤。
不要直接完成任务，不要写最终答案。
输出要求：
1. 使用编号列表
2. 每一步尽量简洁
3. 只输出计划内容
"""
        user_prompt = f"用户任务：{state.user_task}"

        result = llm_call(system_prompt, user_prompt)
        state.add_log(self.name, result)

        # 简单解析：按行提取步骤
        plan_lines = [
            line.strip()
            for line in result.splitlines()
            if line.strip() and (line.strip()[0].isdigit() or line.strip().startswith("-"))
        ]

        if not plan_lines:
            plan_lines = [result]

        state.plan = plan_lines
        return state

---

Executor：根据计划生成执行结果

class ExecutorAgent:
    name = "ExecutorAgent"

    def run(self, state: AgentState) -> AgentState:
        plan_text = "\n".join(state.plan)

        system_prompt = """
你是一个执行智能体。
你的职责是根据已有计划完成任务内容。
要求：
1. 严格围绕计划执行
2. 产出结构化、完整的初稿
3. 不要解释你自己是谁
4. 不要重复计划内容
"""
        user_prompt = f"""
用户原始任务：
{state.user_task}

任务计划：
{plan_text}

请根据以上计划，完成初稿。
"""

        result = llm_call(system_prompt, user_prompt)
        state.execution_result = result
        state.add_log(self.name, result)
        return state

---

Reviewer：审查并优化结果

class ReviewerAgent:
    name = "ReviewerAgent"

    def run(self, state: AgentState) -> AgentState:
        system_prompt = """
你是一个审查智能体。
你的职责是检查初稿是否存在以下问题：
1. 是否偏题
2. 是否结构不清晰
3. 是否有明显重复
4. 是否缺少关键点
然后给出一个优化后的最终版本。

输出格式：
【审查意见】
...
【优化后版本】
...
"""
        user_prompt = f"""
用户任务：
{state.user_task}

任务计划：
{chr(10).join(state.plan)}

执行初稿：
{state.execution_result}
"""

        result = llm_call(system_prompt, user_prompt)
        state.review_result = result
        state.add_log(self.name, result)

        if "【优化后版本】" in result:
            state.final_output = result.split("【优化后版本】", 1)[1].strip()
        else:
            state.final_output = result

        return state

---

4）定义调度器，把三个 Agent 串起来

多智能体系统最核心的，不是 Agent 本身，而是编排器（Orchestrator）。

class MultiAgentSystem:
    def __init__(self):
        self.planner = PlannerAgent()
        self.executor = ExecutorAgent()
        self.reviewer = ReviewerAgent()

    def run(self, user_task: str) -> AgentState:
        state = AgentState(user_task=user_task)

        state = self.planner.run(state)
        state = self.executor.run(state)
        state = self.reviewer.run(state)

        return state

---

5）运行一个完整示例

if __name__ == "__main__":
    task = "写一篇面向初学者的短文，解释什么是多智能体系统，并给出一个最小 Python Demo 的思路。"

    system = MultiAgentSystem()
    result_state = system.run(task)

    print("========== PLAN ==========")
    for step in result_state.plan:
        print(step)

    print("\n========== EXECUTION RESULT ==========")
    print(result_state.execution_result)

    print("\n========== REVIEW RESULT ==========")
    print(result_state.review_result)

    print("\n========== FINAL OUTPUT ==========")
    print(result_state.final_output)

---

6）这个 Demo 的运行逻辑到底是什么？

整个流程其实非常清晰：

用户任务
   ↓
PlannerAgent：拆解任务
   ↓
ExecutorAgent：根据计划执行
   ↓
ReviewerAgent：审查并优化
   ↓
最终输出

如果你把它画成更工程化一点的流程图，大概是这样：

+------------------+
|   User Task      |
+------------------+
         |
         v
+------------------+
|  Planner Agent   |
|  任务理解与拆解   |
+------------------+
         |
         v
+------------------+
| Executor Agent   |
|  信息生成与执行   |
+------------------+
         |
         v
+------------------+
| Reviewer Agent   |
|  结果审查与优化   |
+------------------+
         |
         v
+------------------+
|   Final Output   |
+------------------+

这就是一个最小多智能体系统的骨架。

---

六、这个 Demo 为什么比单 Agent 更稳？

很多人会问：

“这不就是把三次调用串起来吗？真的有必要吗？”

有必要。因为这里不是简单串联，而是职责切分。

单 Agent 的问题

如果你只用一个 Agent，它会在一个 prompt 里同时承担：

• 理解任务
• 制定计划
• 执行内容
• 检查质量

这时候它会反复在多个认知角色里来回切换。

而且最麻烦的是：
你很难知道它到底是在哪一步做错了。

---

三智能体的优势

用了多智能体之后：

• Planner 出问题：你看 plan 就知道
• Executor 出问题：你看执行结果就知道
• Reviewer 出问题：你看审查意见就知道

也就是说，系统开始具备一种很重要的工程属性：

可调试性

真正能落地的 Agent 系统，不是“偶尔答对”，而是“出了问题能定位”。

---

七、多智能体最容易踩的坑

多智能体不是角色越多越高级。
很多系统做坏，恰恰是因为 Agent 加太多了。

下面这几个坑非常常见。

---

坑 1：分工不清，多个 Agent 做重复劳动

比如：

• Planner 在写内容
• Executor 又重新规划
• Reviewer 又开始补充新知识

最后三个 Agent 都在做“通用大模型”的事，没有真正分工。

解决方式很简单：

• 给每个 Agent 明确职责边界
• 明确禁止它做什么
• 输入输出尽量结构化

一句话说就是：

角色定义越模糊，协作成本越高。

---

坑 2：上下文传递过多，系统越来越乱

这是非常典型的工程问题。

很多人设计多智能体时，喜欢把完整历史、所有草稿、所有日志都传给后面的 Agent。
结果就是：

• token 开销越来越大
• 噪声越来越多
• 后面的 Agent 开始抓不住重点

正确做法是：

• 只传当前阶段需要的字段
• 用结构化状态代替大段自然语言
• 中间结果尽量摘要化，而不是无限堆历史

---

坑 3：每个 Agent 都太“聪明”，反而不受控

很多人给每个 Agent 写提示词的时候，都会忍不住加很多能力描述：

• 你是一个顶级专家
• 你擅长规划、分析、写作、审查、决策
• 你可以灵活发挥

结果就是每个 Agent 都变成“全能选手”。

这会直接破坏协作边界。

多智能体里，很多时候不是要 Agent 更强，而是要 Agent 更专。

---

坑 4：协作成本大于收益

这点非常重要。

如果你的任务只是：

• 改一段文案
• 写一个函数
• 总结一篇文章

那一个 Agent 往往就够了。

不要为了“高级感”强行多 Agent。
系统越复杂，维护成本越高，延迟也越高。

---

八、什么时候该用多智能体，什么时候不该用

这是最容易被讲偏的地方。

很多文章会让你觉得：

“只要任务高级，就应该上多智能体。”

其实不是。

适合单 Agent 的任务

• 目标明确
• 步骤不多
• 不依赖外部工具
• 不需要多轮审查

比如：

• 改写一段介绍
• 写个简单 SQL
• 总结一篇短文

这种时候，多智能体纯属加戏。

---

适合工作流的任务

有些任务虽然复杂，但流程是固定的。

比如：

1. 读取文件
2. 清洗数据
3. 调用模型总结
4. 输出报告

这时候更适合的是工作流，而不是多智能体自主协作。

因为重点在“固定流程执行”，不是“角色智能协商”。

---

真正适合多智能体的任务

高价值多智能体任务通常具备几个特征：

• 任务复杂，能明显拆成多个子任务
• 子任务之间能力类型不同
• 中间结果需要被审查或修正
• 存在一定程度的动态决策

例如：

• 自动研究与报告生成
• 多步骤工具调用任务
• 复杂分析与写作
• 带审核机制的执行系统
• 多角色协同的客服/运营/知识问答系统

一句话判断：

如果问题本身已经具有“组织协作结构”，多智能体才真正有价值。

---

九、多智能体系统的工程关键，不是提示词，而是编排

很多初学者在做 Agent 时，最爱研究的是提示词。
但真正上线以后你会发现，决定系统上限的往往不是提示词，而是下面这些东西：

1）流程设计

谁先做、谁后做、失败了怎么办、能不能重试。
这是系统层问题，不是模型层问题。

---

2）状态管理

状态是多智能体的“血液”。

如果没有清晰状态：

• Agent 之间就无法稳定协作
• 结果不可追踪
• 错误无法定位

---

3）可观测性

你至少要知道：

• 哪个 Agent 被调用了
• 它拿到了什么输入
• 它输出了什么
• 哪一步出了问题

所以真实系统里，日志、trace、调试面板都非常重要。

---

4）失败兜底

多智能体不是永远成功的。

你要考虑：

• Planner 计划太差怎么办？
• Executor 结果为空怎么办？
• Reviewer 误判怎么办？
• 某一步超时怎么办？

没有兜底的多智能体系统，本质上只是一个“更复杂的随机系统”。

---

十、把这个 Demo 再往前走一步，可以怎么升级？

如果你已经理解了上面的最小版本，下一步可以从这几个方向继续演进。

---

升级 1：让 Agent 输出 JSON，而不是自由文本

这样更容易解析，也更容易做状态控制。

比如 Planner 输出：

{
  "goal": "写多智能体博客",
  "steps": [
    "解释单 Agent 问题",
    "介绍多 Agent 协作模式",
    "实现 Python Demo",
    "总结适用场景"
  ]
}

这样系统会更稳。

---

升级 2：给 Executor 增加工具能力

比如：

• 搜索工具
• 文件读取工具
• 数据分析工具
• 代码执行工具

这时候 Executor 就不只是“写内容”，而是真正能干活了。

---

升级 3：给 Reviewer 增加评分机制

比如让它输出：

• 完整性评分
• 逻辑性评分
• 可读性评分
• 是否需要重试

然后由主控决定要不要让 Executor 再跑一轮。

---

升级 4：引入路由器或者主控 Agent

让系统根据任务类型动态决定：

• 要不要先检索
• 要不要走审核链路
• 要不要调用某个专家 Agent

这一步开始，系统就从“三段式流程”进化成“有策略的协作系统”了。

---

十一、一个更接近真实项目的改进版思路

如果你想把上面的 Demo 再做得工程化一点，可以把系统拆成下面这几个模块：

multi_agent/
├── llm.py          # 模型调用封装
├── state.py        # 状态定义
├── agents/
│   ├── planner.py
│   ├── executor.py
│   └── reviewer.py
├── orchestrator.py # 协作调度
├── tools.py        # 搜索、文件、数据库等工具
└── main.py         # 启动入口

这样做的好处是：

• 角色逻辑更清晰
• 更容易替换模型
• 更容易接工具
• 更容易做日志和测试

这也是为什么说：

多智能体系统的重点从来不是“角色名”，而是“工程结构”。

---

十二、给初学者的一个实用建议

如果你现在正准备做自己的第一个多智能体项目，我建议你按这个顺序来：

第一步：先做单 Agent，确认任务本身能跑通

不要一上来就搞多角色。
先确认任务目标、输入输出、评估标准是清楚的。

第二步：找到单 Agent 最不稳定的那个环节

比如：

• 规划不稳
• 检索不稳
• 输出质量不稳

第三步：只把最不稳定的环节拆出来

这才是多智能体真正有价值的地方。
不是“为了复杂而复杂”，而是“为了稳定而拆分”。

---

十三、总结

多智能体不是噱头，但也绝对不是银弹。

它真正的价值，不在于你能起多少个炫酷角色名，
而在于你能不能让复杂任务具备下面三种能力：

• 可拆解
• 可协作
• 可治理

这也是为什么，真正能落地的多智能体系统，重点永远不是提示词，而是：

• 架构设计
• 状态管理
• 流程编排
• 可观测性
• 失败兜底

最后再回到文章开头那句话：

一个 AI 不够用，并不是因为模型不够强，
而是因为复杂任务本来就不该只靠一个角色硬扛。

当任务开始变复杂，
你要思考的就不再是“提示词怎么写更厉害”，
而是：

这个系统，应该怎么分工。

---

附：完整可运行示例代码

为了方便你直接复制测试，我把完整代码放在一起。

import os
from dataclasses import dataclass, field
from typing import List, Dict, Any

from openai import OpenAI


# =========================
# 1. LLM Client
# =========================
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

def llm_call(system_prompt: str, user_prompt: str, model: str = "gpt-4o-mini") -> str:
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.7,
    )
    return response.choices[0].message.content.strip()


# =========================
# 2. Shared State
# =========================
@dataclass
class AgentState:
    user_task: str
    plan: List[str] = field(default_factory=list)
    execution_result: str = ""
    review_result: str = ""
    final_output: str = ""
    logs: List[Dict[str, Any]] = field(default_factory=list)

    def add_log(self, agent_name: str, content: str):
        self.logs.append({
            "agent": agent_name,
            "content": content
        })


# =========================
# 3. Agents
# =========================
class PlannerAgent:
    name = "PlannerAgent"

    def run(self, state: AgentState) -> AgentState:
        system_prompt = """
你是一个任务规划智能体。
你的职责只有一个：理解用户目标，并将其拆解为3到5个清晰步骤。
不要直接完成任务，不要写最终答案。
输出要求：
1. 使用编号列表
2. 每一步尽量简洁
3. 只输出计划内容
"""
        user_prompt = f"用户任务：{state.user_task}"
        result = llm_call(system_prompt, user_prompt)
        state.add_log(self.name, result)

        plan_lines = [
            line.strip()
            for line in result.splitlines()
            if line.strip() and (line.strip()[0].isdigit() or line.strip().startswith("-"))
        ]
        if not plan_lines:
            plan_lines = [result]

        state.plan = plan_lines
        return state


class ExecutorAgent:
    name = "ExecutorAgent"

    def run(self, state: AgentState) -> AgentState:
        plan_text = "\n".join(state.plan)

        system_prompt = """
你是一个执行智能体。
你的职责是根据已有计划完成任务内容。
要求：
1. 严格围绕计划执行
2. 产出结构化、完整的初稿
3. 不要解释你自己是谁
4. 不要重复计划内容
"""
        user_prompt = f"""
用户原始任务：
{state.user_task}

任务计划：
{plan_text}

请根据以上计划，完成初稿。
"""
        result = llm_call(system_prompt, user_prompt)
        state.execution_result = result
        state.add_log(self.name, result)
        return state


class ReviewerAgent:
    name = "ReviewerAgent"

    def run(self, state: AgentState) -> AgentState:
        system_prompt = """
你是一个审查智能体。
你的职责是检查初稿是否存在以下问题：
1. 是否偏题
2. 是否结构不清晰
3. 是否有明显重复
4. 是否缺少关键点
然后给出一个优化后的最终版本。

输出格式：
【审查意见】
...
【优化后版本】
...
"""
        user_prompt = f"""
用户任务：
{state.user_task}

任务计划：
{chr(10).join(state.plan)}

执行初稿：
{state.execution_result}
"""
        result = llm_call(system_prompt, user_prompt)
        state.review_result = result
        state.add_log(self.name, result)

        if "【优化后版本】" in result:
            state.final_output = result.split("【优化后版本】", 1)[1].strip()
        else:
            state.final_output = result

        return state


# =========================
# 4. Orchestrator
# =========================
class MultiAgentSystem:
    def __init__(self):
        self.planner = PlannerAgent()
        self.executor = ExecutorAgent()
        self.reviewer = ReviewerAgent()

    def run(self, user_task: str) -> AgentState:
        state = AgentState(user_task=user_task)

        state = self.planner.run(state)
        state = self.executor.run(state)
        state = self.reviewer.run(state)

        return state


# =========================
# 5. Main
# =========================
if __name__ == "__main__":
    task = "写一篇面向初学者的短文，解释什么是多智能体系统，并给出一个最小 Python Demo 的思路。"

    system = MultiAgentSystem()
    result_state = system.run(task)

    print("========== PLAN ==========")
    for step in result_state.plan:
        print(step)

    print("\n========== EXECUTION RESULT ==========")
    print(result_state.execution_result)

    print("\n========== REVIEW RESULT ==========")
    print(result_state.review_result)

    print("\n========== FINAL OUTPUT ==========")
    print(result_state.final_output)

---

结尾一句话

多智能体系统真正有价值的地方，不是“让多个 AI 同时说话”，而是：

让复杂任务第一次拥有了像团队一样的分工结构。