【从0搭建AI智能体·9】多智能体协作实战:让几个 Agent 分工合作 1+1>2

📚 本文是《从 0 搭建你的 AI 智能体》专栏第 9 篇。
上一篇:《ReAct vs Plan-and-Execute:编排模式对比》

标签:多智能体 Multi-Agent AI Agent LangGraph 协作 大模型

📌 前言:一个人干不完,就组个团队

前面我们做的都是「单个 Agent」。但有些任务,让一个 Agent 从头包到尾,效果并不好:

  • 让它「写一篇高质量技术文」——它既要调研、又要写作、还要审校,样样都做,样样平庸;
  • 让它处理「客户咨询」——技术问题、售后问题、销售问题混在一起,一个万能 Agent 反而哪个都不精。

现实世界怎么解决?分工。一个团队里有调研员、写手、审校,各司其职,产出远胜一个「全能选手」单打独斗。多智能体(Multi-Agent)就是把这套搬进 AI:让多个各有专长、各有专属 System Prompt 和工具的 Agent 分工协作,共同完成单个 Agent 搞不定的复杂任务。

这一篇,我们从最简单的「流水线协作」讲起,逐步到「主管调度」「辩论评审」等模式,每种都给可跑代码,让你真正搭出一个 AI 团队。

💡 本文适合谁:已掌握单 Agent 构建,想挑战复杂任务的开发者。本文用纯 Python 手搓,不依赖框架,理解了再看 LangGraph 等就秒懂。阅读约 15 分钟。

目录

  1. 为什么多个智能体比一个强
  2. 核心:智能体之间怎么「传话」
  3. 模式一:流水线(Pipeline)
  4. 模式二:主管调度(Supervisor / Router)
  5. 模式三:辩论 / 评审(Debate / Review)
  6. 完整可跑 Demo:AI 写作团队
  7. 进阶实战:会拆解任务的编排器(Orchestrator)
  8. 多智能体的成本与失控风险
  9. 什么时候「别」用多智能体
  10. FAQ
  11. 总结

① 为什么多个 Agent 比一个强

同一个模型,拆成多个专职 Agent,效果往往显著更好,原因有三:

优势 说明
专注度 每个 Agent 只干一件事,System Prompt 高度聚焦,产出更专业
可组合 像搭积木,不同 Agent 自由组合成不同工作流
可维护 某环节效果差,只改那一个 Agent,不影响其他
可并行 相互独立的 Agent 能同时跑,提速

一句话:与其让一个 Agent「样样通样样松」,不如让多个 Agent「各精一样」再协作。


② 核心:Agent 之间怎么「传话」

多 Agent 系统的本质,是把上一个 Agent 的输出,作为下一个 Agent 的输入。所以每个 Agent 可以抽象成一个函数:

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("AGENT_KEY"), base_url=os.getenv("API_BASE_URL"))
MODEL = "standard-agent-v1"


def agent(role_prompt, user_input, temperature=0.7):
    """一个通用 Agent:给它一个角色设定和输入,返回它的输出"""
    resp = client.chat.completions.create(
        model=MODEL, temperature=temperature,
        messages=[
            {"role": "system", "content": role_prompt},   # 这个 Agent 的专属人设
            {"role": "user", "content": user_input},
        ],
    )
    return resp.choices[0].message.content

有了这个原子函数(配好 .env 即可跑),「多 Agent 协作」就变成了怎么把这些 agent() 调用组织(编排)起来——串行、并行、还是循环。下面三种模式就是三种组织方式。本文所有代码都基于这一个 agent() 函数,复制到同一个文件里即可连续运行。


③ 模式一:流水线(Pipeline)

最简单直接:几个 Agent 排成一条流水线,前一个的输出喂给后一个。像工厂装配线。

要点

初稿

主题输入

调研 Agent
收集要点

写作 Agent
成文

审校 Agent
润色纠错

成品输出

数据单向流动,一环接一环——上一个 Agent 的输出,就是下一个的输入。

def pipeline_demo(topic):
    # Agent 1:调研员,负责收集要点
    research = agent(
        "你是资深调研员。针对给定主题,列出 5 个最重要的核心要点,简明扼要。",
        f"主题:{topic}")

    # Agent 2:写手,基于要点成文
    draft = agent(
        "你是专业科技写手。根据提供的要点,写一篇 300 字左右、通俗流畅的短文。",
        f"要点:\n{research}")

    # Agent 3:审校,润色纠错
    final = agent(
        "你是严格的审校编辑。修正下文的语法、逻辑和表达问题,让它更专业流畅,直接输出修改后的完整版本。",
        f"待审校:\n{draft}")

    return final


if __name__ == "__main__":
    print(pipeline_demo("边缘计算的应用前景"))

特点:结构清晰、易调试。适合步骤固定、线性推进的任务(内容生产、数据处理管道)。

💡 流水线的每个 Agent 都可以有自己的工具(如调研员配联网搜索,见第7篇),让每一环更强。


④ 模式二:主管调度(Supervisor / Router)

流水线是「写死」的顺序。但很多场景需要动态决定「派给谁」——比如客服系统,来了个问题得先判断是技术、售后还是销售,再转给对应专家。

这就是主管模式:一个「主管 Agent」负责判断意图、分发任务给合适的「专家 Agent」。

技术类

售后类

销售类

用户提问

主管 Agent
判断意图

技术专家

售后专家

销售专家

回复

关键是那个菱形的"判断"节点——路由是动态的,同一个问题运行时才决定走哪条分支。

def supervisor_demo(user_question):
    # 主管:只负责分类,输出路由决策
    route = agent(
        "你是客服调度主管。判断用户问题属于哪类,只回答一个词:"
        "「技术」「售后」或「销售」,不要其他内容。",
        user_question, temperature=0).strip()

    # 根据路由,交给对应专家(各有专属人设)
    experts = {
        "技术": "你是技术支持专家,专业解答产品使用和故障问题。",
        "售后": "你是售后专家,负责退换货、维修、投诉处理,态度耐心。",
        "销售": "你是销售顾问,介绍产品优势、报价和优惠,热情专业。",
    }
    role = experts.get(route, experts["技术"])   # 兜底给技术
    print(f"[主管路由到] {route}")
    return agent(role, user_question)


if __name__ == "__main__":
    print(supervisor_demo("我买的机器开不了机了怎么办?"))
    print(supervisor_demo("你们企业版多少钱?"))

运行效果(同一套代码,主管把两个问题分给了不同专家):

[主管路由到] 技术
您好,机器无法开机请先按住电源键 10 秒强制重启,若仍无反应,请检查……
[主管路由到] 销售
您好!我们企业版提供多用户协作、专属客服和高级安全功能,标准报价……

特点:能动态路由,各专家高度专业。适合多领域、需要分诊的场景(智能客服、多技能助手)。

🎯 主管模式是生产级客服/助手的常见架构。主管也可以更复杂——不只分类,还能拆任务、调度多个专家协同、汇总结果。


⑤ 模式三:辩论 / 评审(Debate / Review)

对于需要高质量、高可靠的输出,可以让 Agent「互相挑刺」——一个生成、一个批判、再改进。类似「作者 + 评审」反复打磨。

有问题

无问题 / 用尽轮数

问题

生成智能体
产出答案

评审智能体
审阅并挑刺

判断
有问题吗?

生成智能体
按评审意见改进

输出终稿

对照代码看这张图:生成answer = agent(...))→ 评审critique = agent(...))→ 判断if "无问题" in critique)→ 有问题就改进answer = agent(...改进...))并回到评审,无问题(或达到 rounds 上限)就输出。这个回环正是"生成↔批判"反复打磨的精髓。

def review_demo(question, rounds=2):
    answer = agent("你是问题解答专家,尽力给出准确的答案。", question)

    for i in range(rounds):
        # 评审 Agent:专门找茬
        critique = agent(
            "你是挑剔的评审专家。指出下面答案中的错误、遗漏或不严谨之处,"
            "如果没有问题就只回答「无问题」。",
            f"问题:{question}\n答案:{answer}")

        if "无问题" in critique:
            break

        # 生成 Agent:根据批评改进
        answer = agent(
            "你是解答专家。根据评审意见改进你的答案,输出改进后的完整答案。",
            f"问题:{question}\n原答案:{answer}\n评审意见:{critique}")
        print(f"[第{i+1}轮改进完成]")

    return answer

运行效果(评审揪出问题 → 改进 → 再审通过):

[第1轮改进完成]          ← 首稿有疏漏,评审提了意见,改进了一版
(第2轮评审返回"无问题",循环提前结束)
最终答案:……(比首稿更严谨、更完整)

特点:通过「对抗」提升质量,能显著减少错误和遗漏。适合对准确性要求高的场景(代码审查、方案评估、事实核查)。代价是多轮调用、成本较高。

💡 这就是很多「AI 自我反思(Self-Reflection)」技术的核心——让模型批判自己的输出再改进,往往比一次生成质量高得多。


⑥ 完整可跑 Demo:AI 写作团队

把流水线 + 评审结合,搭一个「AI 写作团队」,还用上并行提速:

import os
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("AGENT_KEY"), base_url=os.getenv("API_BASE_URL"))
MODEL = "standard-agent-v1"


def agent(role, text, temperature=0.7):
    return client.chat.completions.create(
        model=MODEL, temperature=temperature,
        messages=[{"role": "system", "content": role}, {"role": "user", "content": text}],
    ).choices[0].message.content


def writing_team(topic):
    # 1. 两个调研 Agent 从不同角度并行调研
    #    这两次调研互不依赖,没必要一个等一个跑完(那样要等两次 API 往返)。
    #    ThreadPoolExecutor 是 Python 的"线程池",能让多个任务同时进行——
    #    调用大模型时程序大部分时间在"等网络返回"(IO 密集),
    #    用线程并发就能让两次调研的等待时间重叠,总耗时≈慢的那一次,而非两次之和。
    angles = ["技术原理角度", "商业应用角度"]
    with ThreadPoolExecutor() as pool:
        # pool.map(函数, 列表):把 angles 里每个元素丢给函数并发执行,
        # 按输入顺序收集返回值(这里即两份调研结果)
        researches = list(pool.map(
            lambda a: agent(f"你是调研员,从【{a}】列出该主题 3 个关键点。", topic),
            angles))
    material = "\n".join(researches)
    print("① 调研完成(2 个角度并行)")

    # 2. 写手综合成文
    draft = agent(
        "你是科技专栏作者,根据调研材料写一篇 400 字、有观点、通俗的短文。",
        f"主题:{topic}\n调研材料:\n{material}")
    print("② 初稿完成")

    # 3. 评审改进一轮
    critique = agent(
        "你是主编,指出文章的问题(逻辑、准确性、可读性),没问题回答「无问题」。",
        f"文章:{draft}")
    if "无问题" not in critique:
        draft = agent("你是作者,根据主编意见修改,输出完整终稿。",
                      f"原文:{draft}\n主编意见:{critique}")
        print("③ 主编提了意见,已改进 → 终稿")
    else:
        print("③ 主编审核通过 → 终稿")

    return draft


if __name__ == "__main__":
    print(writing_team("大模型如何改变软件开发"))

运行轨迹(能清楚看到"团队"在流水线上依次接力):

① 调研完成(2 个角度并行)
② 初稿完成
③ 主编提了意见,已改进 → 终稿
大模型正在重塑软件开发的每一个环节……(400 字成稿,略)

这个「团队」里:2 个调研员(并行)+ 1 个写手 + 1 个主编,分工协作产出一篇文章——比单个 Agent「写篇文章」质量高得多。

6.1 眼见为实:单 Agent vs 团队,差在哪

「团队更强」不能只靠嘴说。跑个对比就知道——同一个主题,一个 Agent 一把梭 vs 上面的团队分工:

def single_agent(topic):
    """对照组:一个 Agent 独自完成全部"""
    return agent("你是科技作者,就给定主题写一篇 400 字短文。", topic)


if __name__ == "__main__":
    topic = "大模型如何改变软件开发"
    print("===== 单 Agent =====\n", single_agent(topic))
    print("\n===== 团队协作 =====\n", writing_team(topic))

差异在哪(把两篇并排读,你会明显感到):

维度 单 Agent 一把梭 团队协作
内容广度 容易只谈一个角度(如只说效率) 调研阶段已覆盖技术+商业双视角,更全面
观点深度 泛泛而谈、平铺直叙 写手在充分材料上提炼,更有观点
硬伤 无人复核,错了就错了 主编环节能揪出逻辑/事实问题

🎯 1+1>2 的本质:不是"人多力量大",而是每个环节都在专注做好一件事——调研只管全、写作只管顺、审校只管挑错。关注点分离(Separation of Concerns)在 AI 协作里同样成立。 这也解释了为什么"能单智能体搞定的简单任务,拆开反而更差"(见第 ⑨ 节)——没有值得分离的关注点时,分工只剩开销。


⑦ 进阶实战:会拆解任务的编排器(Orchestrator)

前面第 ④ 节的主管只会「分类路由」——把一个问题丢给一个专家。但真实的复杂问题,往往一个专家答不全。比如:

「帮我评估一下’在小城市开一家咖啡馆’这个想法:市场怎么样、大概要花多少钱、有哪些风险?」

这一个问题里其实藏着三个不同领域的子问题(市场、财务、风险),该分给三个不同专家并行处理,最后再汇总成一份完整报告。这就是编排器(Orchestrator)模式——主管的进阶形态:拆解 → 并行调度 → 汇总

子任务1

子任务2

子任务3

复杂问题

规划器
拆解成子任务

市场专家

财务专家

风险专家

汇总器
综合成报告

完整报告

它把本专栏学过的三样东西合成了一体:主管的拆解判断(第④节)+ 流水线的多角色(第③节)+ ThreadPoolExecutor 的并行(第⑥节)。完整代码如下(仍基于第 ② 节的 agent() 函数,配好 .env 即可跑):

import json
from concurrent.futures import ThreadPoolExecutor
# 复用第②节的 agent(role_prompt, user_input, temperature) 函数

# 可调度的专家库:每个专家一个专属人设
EXPERTS = {
    "市场分析": "你是市场分析师,擅长评估需求、竞争和目标客群。",
    "财务分析": "你是财务顾问,擅长估算成本、收入和回本周期。",
    "风险评估": "你是风险管理专家,擅长识别潜在风险并给出应对建议。",
    "法律合规": "你是法务顾问,擅长解答证照、合规、合同相关问题。",
}


def plan_subtasks(question):
    """① 规划器:把复杂问题拆成【子任务 + 指派专家】的清单"""
    expert_list = "、".join(EXPERTS.keys())
    raw = agent(
        f"你是任务规划器。把用户的复杂问题拆解成 2~4 个独立子任务,"
        f"每个子任务指派给最合适的一位专家(可选专家:{expert_list})。"
        f'只输出 JSON 数组,格式:[{{"expert": "市场分析", "subtask": "..."}}],'
        f"不要输出任何其他文字。",
        question, temperature=0)
    raw = raw.replace("```json", "").replace("```", "").strip()   # 容错剥壳
    return json.loads(raw)


def run_one(item):
    """② 单个子任务:交给指派的专家处理(会被并行调用)"""
    role = EXPERTS.get(item["expert"], "你是通用助手。")
    answer = agent(role, item["subtask"])
    return {"expert": item["expert"], "subtask": item["subtask"], "answer": answer}


def aggregate(question, results):
    """③ 汇总器:把各专家的结果综合成一份连贯报告"""
    joined = "\n\n".join(
        f"【{r['expert']}】针对「{r['subtask']}」:\n{r['answer']}" for r in results)
    return agent(
        "你是首席顾问。把下面各领域专家的分析,整合成一份结构清晰、"
        "有总体结论的完整报告,不要简单罗列,要融会贯通。",
        f"用户问题:{question}\n\n各专家分析:\n{joined}")


def orchestrate(question):
    """编排主流程:拆解 → 并行执行 → 汇总"""
    plan = plan_subtasks(question)
    print(f"① 规划完成,拆出 {len(plan)} 个子任务:")
    for p in plan:
        print(f"   - [{p['expert']}] {p['subtask']}")

    # 关键:多个子任务互不依赖,用线程池并行跑,总耗时≈最慢的一个而非累加
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(run_one, plan))
    print(f"② {len(results)} 位专家并行处理完毕")

    report = aggregate(question, results)
    print("③ 汇总完成\n")
    return report


if __name__ == "__main__":
    q = "帮我评估'在小城市开一家咖啡馆':市场怎么样、大概要花多少钱、有哪些风险?"
    print(orchestrate(q))

运行轨迹(拆解 → 并行 → 汇总,全程可见):

① 规划完成,拆出 3 个子任务:
   - [市场分析] 评估小城市咖啡馆的市场需求与竞争情况
   - [财务分析] 估算开一家小城市咖啡馆的启动成本与回本周期
   - [风险评估] 识别小城市开咖啡馆的主要风险及应对
② 3 位专家并行处理完毕
③ 汇总完成

【综合评估报告】总体来看,小城市咖啡馆是一门"低天花板、需精细化运营"的生意……
(市场、财务、风险三部分融会贯通的完整报告)

7.1 这段代码的三个关键设计

  • 规划器输出结构化 JSON{"expert": ..., "subtask": ...} 的格式让程序能直接解析并派发。这里复用了第 3 篇 Function Calling 的思路——让模型产出机器可读的结果,而非自由文本(生产中更稳的做法是直接用 Function Calling / JSON mode 强制结构化)。
  • 专家库可扩展:想加能力,只需往 EXPERTS 字典里加一条,规划器会自动把它纳入调度范围——新增专家零改动主流程,这正是多智能体"可组合"的体现。
  • 并行是性能关键:三个子任务互不依赖,ThreadPoolExecutor 让它们同时进行。若串行,总耗时是三次调用相加;并行后≈最慢的那一次,3 个专家几乎和 1 个专家一样快(详见下一节的量化)。

⚠️ 两个必须处理的坑:① 规划器偶尔输出不合法 JSON,json.loads 要包 try/except,失败就降级为"单专家直接回答";② 拆出的子任务数要设上限(这里靠 Prompt 约束 2~4 个),否则一个问题被拆成十几个子任务,成本和延迟都会失控。

🎯 认知升级:到这里,你已经把主管模式从「一个问题 → 一个专家」进化到了「一个问题 → 拆成多个 → 多专家并行 → 汇总」。这套「编排器」正是 AutoGPT、MetaGPT 这类复杂 Agent 系统的核心骨架——它们只是在拆解、调度、汇总上做得更精细而已。


⑧ 多智能体的成本与失控风险

多 Agent 强大,但代价和风险也真实存在,上生产前必须清楚:

风险 说明 应对
成本翻倍 N 个 Agent = N 倍(甚至更多)模型调用 算清账,非必要不堆 Agent;能并行的并行
延迟叠加 串行流水线,总耗时是各环节之和 能并行的环节并行;减少不必要的轮次
错误传播 上游 Agent 出错,下游照单全收、越错越远 关键节点加校验;评审模式兜底
失控循环 Agent 互相调用不收敛 设全局步数/轮数上限、总 Token 预算
调试困难 出问题不知是哪个 Agent 的锅 每个 Agent 的输入输出都记日志(接第10篇)

🚨 最容易翻车的是「错误传播」:流水线第一环调研错了,后面写手、审校再努力也是在错误地基上盖楼。越靠前的 Agent,越要保证质量,或在早期节点加校验。

8.1 算笔账:多智能体到底贵多少、慢多少

别只说"翻倍",我们拿具体数字估一估。假设单次大模型调用 ≈ 消耗 1 份 Token 成本、耗时约 2 秒(实际因模型和长度而异,这里只看倍率关系):

方案 模型调用次数 成本(相对单次) 延迟(估算)
单 Agent 一把梭 1 ~2 秒
3 步流水线(串行) 3 ~3× ~6 秒(2+2+2,逐环等待)
本文写作团队 5(2调研+1写+1审+1改) ~5× ~8 秒(调研并行省了 2 秒,否则要 ~10 秒)
编排器(第⑦节,3 专家并行) 5(1规划+3专家+1汇总) ~5× ~6 秒(3 专家并行只算 1 次,规划+并行+汇总≈2+2+2)
3 轮辩论评审 生成1 +(评审+改进)×3 ≈ 7 ~7× ~14 秒

从这张表能读出三条实用结论:

  • 成本几乎等于调用次数:拆几个 Agent,账单大致就乘几。5 个 Agent 的团队,成本约是单 Agent 的 5 倍——质量提升不是免费的
  • 串行是延迟的主要来源:流水线每多一环,就多约一次 API 往返(~2 秒)。用户能明显感到"变慢";
  • 并行是最划算的优化:写作团队把两次调研并行,直接省下约 2 秒(20% 延迟)。凡是互不依赖的环节,都该并行(第 ⑥ 节的 ThreadPoolExecutor 就是干这个的)。

🎯 决策心法:多智能体的成本和延迟是"看得见"的(乘以调用数),质量提升是"感觉到"的。上线前务必用真实数据实测一次每步的 Token 和耗时(接第 10 篇可观测性),再判断这笔"质量 vs 成本"的买卖划不划算。


⑨ 什么时候「别」用多智能体

多智能体不是银弹。以下情况单智能体更好,别为了炫技硬上:

  • 任务简单:一个智能体 + 好的 System Prompt 就能搞定,多智能体纯属增加成本和延迟;
  • 实时性要求高:多智能体串行延迟大,不适合要求秒回的场景;
  • 预算敏感:调用次数翻几倍,成本吃不消。

🎯 判断标准:只有当任务明显能拆成几个「不同专长」的子任务、且单智能体做效果确实差时,多智能体才划算。先用单智能体试,不够好再拆——别一上来就上复杂架构


⑩ FAQ

Q1:多个 Agent 必须用不同模型吗?
不必。通常同一个模型,靠不同 System Prompt 扮演不同角色即可。也可给关键角色配更强的模型、简单角色配便宜模型来省钱。

Q2:需要用 LangGraph / AutoGen / CrewAI 这些框架吗?
不是必须。本文纯 Python 就实现了三种模式。框架的价值在于状态管理、可视化、复杂图编排——任务复杂到手搓吃力时再上。理解本文,用框架会非常轻松。

Q3:Agent 之间怎么共享上下文/记忆?
简单场景直接传文本;复杂场景用一个共享的「黑板(Blackboard)」或数据库存中间结果,各 Agent 读写(可结合第2篇 RAG、第6篇 Redis)。

Q4:主管路由分错了怎么办?
给主管更清晰的分类标准和示例;加兜底类别;必要时让专家 Agent 发现「不该归我」时能「退回」重新路由。

Q5:怎么控制多智能体的总成本?
设全局预算:统计所有智能体的 Token 消耗,超过阈值就终止(详见下一篇可观测性)。

Q6:智能体之间怎么传递复杂数据结构(列表、字典)甚至文件?
这是多智能体系统的高频实操问题。核心原则:智能体之间交换的「消息」本质是字符串,所以复杂数据要序列化成 JSON 文本来传,收到后再反序列化回对象。文件则不塞进对话(会爆 Token),改为传"引用"(路径/URL/ID),真实内容放在共享存储里按需读取

import json

# —— 场景:分析智能体产出结构化数据,交给报告智能体使用 ——

def analyst_agent(topic):
    """分析智能体:要求模型输出结构化 JSON,而非自由文本"""
    raw = agent(
        "你是数据分析师。分析给定主题,严格按此 JSON 输出,不要多余文字:"
        '{"score": 0-100的整数, "pros": [字符串数组], "cons": [字符串数组]}',
        topic, temperature=0)
    raw = raw.replace("```json", "").replace("```", "").strip()
    return json.loads(raw)          # 字符串 → Python 字典,拿到结构化对象


def report_agent(topic, data: dict):
    """报告智能体:接收上游的字典,序列化后放进 prompt 供其参考"""
    # 把字典转回 JSON 文本喂给模型(ensure_ascii=False 保留中文可读)
    data_str = json.dumps(data, ensure_ascii=False, indent=2)
    return agent(
        "你是报告撰写员。根据下面的结构化分析数据,写一段通俗的评价。",
        f"主题:{topic}\n分析数据:\n{data_str}")


if __name__ == "__main__":
    topic = "远程办公模式"
    data = analyst_agent(topic)                     # {'score': 78, 'pros': [...], 'cons': [...]}
    print("传递的结构化数据:", data)
    print(report_agent(topic, data))                # 报告智能体基于字典生成文字

要点小结:

要传的东西 怎么传 说明
列表 / 字典 json.dumps 序列化成文本传,收方 json.loads 还原 让模型输出结构化 JSON是关键(配合第3篇 Function Calling / JSON mode 更稳)
大文本 / 中间结果 存进共享存储(Redis / DB / 向量库),只传 key 别把大块内容塞进对话,省 Token(结合第2篇 RAG、第6篇 Redis)
文件(图片/PDF等) 文件路径或 URL,处理方按需读取 对话里传引用而非内容;多模态文件走对应的文件接口

🎯 一句话原则智能体之间传"结构化的数据引用",而不是"一大坨原始内容"。 JSON 负责结构、共享存储负责体量——两者配合,既清晰又省钱。


⑪ 总结

这一篇把单个智能体扩展成了「AI 团队」,讲了三种核心协作模式,外加一个进阶编排器:

模式 结构 适用
流水线 串行,前输出→后输入 步骤固定的内容生产
主管调度 主管分诊 → 专家处理 多领域、需分诊(客服)
辩论/评审 生成 ↔ 批判 ↔ 改进 高质量、高可靠要求
编排器(进阶) 拆解 → 并行 → 汇总 一个问题跨多领域、要综合报告

核心心法:把复杂任务拆成「不同专长」的子任务,让专职智能体各司其职。但切记多智能体有成本、延迟、失控风险——能单智能体搞定就别硬拆

到这里,你已经掌握了从单个到多个智能体的完整构建能力。但「能跑」不等于「跑得好」——线上到底花了多少钱、慢在哪、错在哪?下一篇讲 Agent 可观测性,给你的智能体装上「仪表盘」。


🔜 下一篇预告:《Agent 可观测性:日志、追踪、成本监控》——给智能体装上仪表盘,把黑盒变白盒。
👍 如果本文帮到你,点赞 / 收藏 / 关注,追更不迷路。

Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐