把两个大模型智能体放进同一条业务流里跑,和让它们各自写一份报告,是完全不同的两件事。前阵子我把团队里的一套信息分析链路整体搬到 AgentScope 上,跑了将近两个月,回头整理时最想记下来的不是某个 Agent 的 Prompt 技巧,而是智能体之间那套被标准化下来的交互方式。这篇文章会从核心机制讲到具体代码,再讲我在一条标记为 24.4 的实验分支里踩过的坑,希望给正在做多智能体编排的人留下一些能直接拿去用的经验。
1. 多智能体开发真正稀缺的是交互契约,不是单个Agent的聪明程度
先讲一个我反复踩过的场景。假设你已经有一个很擅长做风险识别的 Agent,每次给一段材料,它都能输出很漂亮的风险清单。问题出在当它和另一个负责写行动建议的 Agent 对接时,前者的输出是 Markdown 风格的“风险项:xxx;等级:高”,后者期待的是 JSON 数组[{"risk": "...", "level": "high"}]。于是你不得不在中间写一层解析函数,把字符串还原成结构化数据。如果链条里有四个 Agent、六个环节,这种“私货格式”就会像野草一样长满整个项目。
1.1 缺一份双方都能读的公共消息体
上面这个场景的本质,是两个 Agent 之间没有公共协议。传统的软件架构解决这类问题靠的是接口定义和 DTO,但大模型场景下有一个特殊之处:模型接收、产出的主体是自然语言,强行把它压成 RPC 请求结构,反而会让 Agent 丢掉上下文。
AgentScope 的做法比较实在:不管你是给模型发指令,还是让 A 智能体把分析结果传给 B 智能体,统一走Msg这种消息体。Msg里既放得下content这种正文内容,也能带name、role、metadata之类的元信息。正文可以是纯文本,也可以是结构化 JSON。下游 Agent 拿到消息后,先读元信息,再决定怎么解析正文,这一下就把“双方各说各话”的问题压到了最小。
在我 24.4 这条分支上,我最终把团队内部所有 Agent 的出参约定收敛成两种:一种是给模型读的自然语言结果,一种是给规则逻辑读的 JSON 结果。都装在Msg.content里,靠metadata["format"]区分。这样既保留了模型的自然表达,又让链路里的非模型组件能稳定解析。
1.2 面向大模型原生设计,而不是把传统服务编排生搬硬套
为什么我特别强调“原生适配大模型”?因为很多编排框架其实是把传统微服务那套搬了过来:消息要序列化、接口要强类型、调用要低延迟。但大模型智能体最核心的资源是上下文窗口和对话记忆,不是网络吞吐。AgentScope 在设计上把“记忆”和“消息流转”放在平级位置,Agent 收到消息后,不是简单转发,而是会把这轮对话内容合并进自己的 Memory,再由模型读取完整上下文后输出。
这个差异在写多智能体业务时非常关键。拿我刚才的信息分析链路举例:事实整理员刚输出完摘要,紧接着风险识别员要的不是“你有一条新消息”这种通知,而是希望把摘要直接拼接进它的提示词里,让它基于完整上下文做判断。AgentScope 的 Agent 生命周期天然支持这件事,你不用手动维护一套“消息中心”,再自己去拼 Prompt。
1.3 为什么我把这条探索线标成 24.4
说实话,24.4 并不是官方版本号,而是我自己在 2024 年 4 月整理代码时给这条实验分支起的代号。当时我集中重构了一层消息交互逻辑,核心目标就是让多智能体之间的传递更标准化。AgentScope 本身版本迭代也快,所以下面聊到的部分 API 字段可能会跟官方最新版有差异,等你动手时如果发现对不上,先翻一下安装包里的源码,思路是通的。
2. 拆开看核心部件:模型适配、消息单元与流程编排怎么长在一起
很多人第一次看 AgentScope 类图时容易被概念绕晕,觉得又是 Agent 又是 Pipeline 又是 Memory,代码还没写就开始焦虑。其实它的核心只有三块:模型配置、消息单元、执行流程。这三块拼在一起,就是一套能跑起来的多智能体系统。
2.1 模型适配层:换模型不换 Agent 代码
AgentScope 里每个 Agent 都通过model_config来绑定自己要调用的模型服务。这个配置通常包括模型服务地址、密钥、模型名、温度这类采样参数。好处是 Agent 的业务代码和数据流完全不用关心底层到底是哪个模型,只要配置里声明好了,同一个ReActAgent既能跑一个大参数模型,也能切换到一个本地小模型做快速实验。
我在这条分支上的标准做法是:所有 Agent 共用一份model_config,只把temperature按角色差异化。需要强事实输出的 Agent 用 0.2,需要发散建议的 Agent 用 0.7。这样配置文件里能一眼看出每个环节的“性格”,而不是散落在各处代码里。
2.2 Msg 的字段与语义:不只是消息,是带身份的对白
Msg是 AgentScope 里最基础,也最容易被低估的对象。一个典型的Msg长这样:
from agentscope.message import Msg msg = Msg( name="事实整理员", role="assistant", content="原料价格在过去一个月上涨了 12%,主要原因是物流成本上升。", metadata={"source": "模拟数据源", "confidence": 0.92}, )name标明消息是谁发出的,role告诉拼 Prompt 的逻辑这条消息该放在用户侧还是助手侧,metadata则是给程序逻辑看的附加信息。你完全可以把它理解成一封带规范信封的信:信件正文是给模型读的,信封上的地址和标签是给程序读的。
这里面有一点很容易被忽略:Msg的content并不仅限字符串。当你需要让 Agent 输出严格 JSON 时,可以在上游先把结构算好,把dict塞进content也行。只不过如果你把dict塞进去,一定要确保下游 Agent 的提示词里明确写清楚“正文是 JSON 对象”,否则模型可能会把它当自然语言重新解释一遍。
2.3 Agent 的执行循环:收到消息、合并记忆、调用模型、回写结果
以ReActAgent为例,它的执行逻辑可以概括成四个步骤:
- 接收一条
Msg。 - 从 Memory 里取出相关历史,和当前消息一起拼装成系统提示词。
- 调用模型,让模型决定是直接回答还是调用工具。
- 把模型输出的内容包成新的
Msg,同时写回 Memory。
这四步听着简单,但正是因为有这样固定的生命周期,你才不用在每个 Agent 里重复写“先取历史、再调模型、再存结果”的样板代码。多个 Agent 组装成系统时,每个 Agent 对输入的期待和对输出的承诺都是一致的,排错的时候就能沿着链路逐个环节查。
2.4 Pipeline:串行、并行、分支,三件套足够覆盖大多数协作场景
在 AgentScope 里,多个 Agent 的协作通常通过 Pipeline 串起来。最基础的是串行:前一个 Agent 的输出是后一个 Agent 的输入。这个模式适合“整理事实 → 识别风险 → 给出建议”这种有明显先后依赖的流程。
真正让多智能体发挥价值的是并行和分支。同一个原始材料,可以让一个 Agent 提取事实,另一个 Agent 直接跳出来识别风险,两个结果再汇合给总编 Agent。这样总编 Agent 拿到的是来自不同视角的素材,而不是一个被串行链路层层“污染”过的结果。分支的触发条件可以直接读刚才那条Msg的metadata。比如消息里标注了risk_level == "high",就走深度审查分支,否则走快速通道。这个判断逻辑放在 Pipeline 的解析函数里,干净利落。
3. 手写一个研报协作流程:让三个Agent各司其职
理论讲再多,不如一段能跑起来的代码。下面这个例子我不追求面面俱到,重点是让你看到 Agent、Msg、Pipeline 是怎么串起来的。
3.1 场景设定
我模拟的场景是:系统收到一份市场原始材料,需要依次完成三件事:
- 事实整理:把材料里的客观信息抽出来,过滤掉主观判断。
- 风险识别:基于事实清单,输出结构化风险项。
- 行动建议:针对风险给出可执行的建议。
如果用手写方式,你可能要维护 3 段 Prompt、3 段模型调用代码、3 个结果变量,再手动把中间结果传给下一段。用 AgentScope,重点就从“传递变量”变成了“定义 Agent 行为”。
3.2 完整代码结构
下面的代码基于我的 24.4 实验分支,具体 API 在不同版本里可能略有出入,但结构是一致的:
from agentscope.agents import ReActAgent from agentscope.message import Msg from agentscope.pipelines import SequentialPipeline model_config = { "model_type": "chat", "base_url": "http://127.0.0.1:8000/v1", "api_key": "local-key", "model_name": "local-llm", } fact_agent = ReActAgent( name="事实整理员", sys_prompt="你只提取输入材料中的客观事实,按条目输出,不要加入任何主观评价。", model_config=model_config, ) risk_agent = ReActAgent( name="风险识别员", sys_prompt="""你根据事实清单识别潜在风险。 只输出 JSON 数组,每项包含 risk 和 reason 两个字段。""", model_config=model_config, ) advice_agent = ReActAgent( name="行动建议员", sys_prompt="你根据风险清单给出具体可执行的行动建议,每条建议对应一个风险项。", model_config=model_config, ) pipeline = SequentialPipeline( agents=[fact_agent, risk_agent, advice_agent] ) user_msg = Msg( name="user", role="user", content=""" 模拟项目X的背景材料: - 渠道A的获客成本环比上涨 15% - 渠道B的转化率稳定,但用户投诉物流时效 - 竞品近期降价 8%,主推套餐包 """, ) result = pipeline(user_msg) print(result.content)这段代码跑起来后,风险识别员拿到的是事实整理员的输出,行动建议员拿到的则是风险识别员的 JSON。整个流程完全不用手写“把上一个 Agent 的输出存到变量里再拼进下一个 Prompt”的逻辑。
3.3 如果想要每个环节的中间结果怎么办
串行 Pipeline 默认只把最终结果返回给主流程,但调试阶段你通常想看每一环的输出。我的做法是在 Pipeline 执行过程中,给每个 Agent 加一个轻量的消息钩子,把每次经过的消息打上当前 Agent 名和时间戳,再统一写到结构化日志里。
如果你不想自己写钩子,也可以直接读每个 Agent 的 Memory。ReActAgent在跑完一轮后,Memory 里会保存它上一次产出的Msg,你按顺序取出来打印即可。这个技巧在排查“某一环输出异常、但不知道是哪一环”时特别有用。
4. 我的踩坑笔记:标准化交互在真实业务里最容易碎掉的三个地方
标准化交互听起来很理想,真落地的时候,我在这条 24.4 分支上踩过不少坑,下面三个最有代表性。
4.1 同一个 Msg 对象被多个 Agent 复用,改着改着就串了
有段时间我图省事,在并行分支里直接让两个下游 Agent 共用同一个上游Msg对象。结果 A 分支的 Agent 在metadata里加了个"checked": true,B 分支后置的逻辑再读同一个对象时,也能看到这个字段,于是 B 误以为自己也完成了校验。
这种问题非常隐蔽,因为它不报错,只是行为悄悄偏了。后来我定了一条铁律:任何Msg在传给下一个 Agent 之前,要么调用复制方法生成独立对象,要么只读不可改。统一约定“新增字段必须放到新的 Msg 里”,从根上避免了共享可变对象带来的线程安全和逻辑串扰。
4.2 群聊场景下,模型不知道该回应谁,回复经常串台
当成批 Agent 在一个循环里轮流发言时,很容易出现“串台”现象:某个 Agent 明明应该回复刚才那条风险提示,结果因为 Memory 里同时塞了另外两条无关消息,模型就答非所问。
我的修复方案是给Msg增加一条conversation_id元信息。每条消息在进入 Agent 的 Memory 之前,先按会话线重新分组,拼装 Prompt 时只取当前会话线的历史。这样一来,即使物理上所有 Agent 共享同一个 Memory 容器,逻辑上每一轮对话还是封闭的。框架本身提供了一些群聊编排能力,但我在这个分支上实践下来,把“会话线识别”做进消息头部是最稳的。
4.3 输出契约定得再标准,也挡不住模型把 JSON 包在 Markdown 代码块里
这是我接触大模型以来遇到最频繁的小问题。你用系统提示词要求 Agent“只输出 JSON”,它也基本照做了,但偶尔会在外面包一层 ```json 代码块。如果下游是另一个模型,这层外壳还能勉强读;如果下游是规则引擎,解析直接炸。
我后来写了一个很简单的净化函数,每次在把 Agent 输出转成 JSON 之前先走一遍:
import json def extract_json(text: str): text = text.strip() if text.startswith("```"): text = text.split("```", 2)[1] if text.startswith("json"): text = text[4:].strip() return json.loads(text)这个函数不复杂,但它把一种非常常见的模型输出风格差异给抹平了。多智能体系统里每一层都净化一次,比在最终入口做一个大而全的解析器要稳得多。
5. 从Demo到生产环境,我在AgentScope之外补做的七处加固
跑通 Demo 只是第一步。真正把它放到生产链路里,我发现 AgentScope 只解决了“智能体怎么交互”的上层问题,下面这些工程项还得自己补。
5.1 消息持久化,回放永远排第一
线上多智能体系统出问题,最痛苦的是没法复现。我给每个经过主流程的Msg都加了落库逻辑,消息 ID、来源 Agent、目标 Agent、时间、完整内容全部存下来。出了问题就直接按run_id把整条链路重放一遍,定位速度比看日志快一个量级。
5.2 幂等重跑,从断点接续而不是整链重来
长链路里一个 Agent 超时,如果整条 Pipeline 重新跑,前面几个 Agent 的模型费用和时间都白白浪费。我按run_id + step_index做幂等判断,已经跑过的环节直接读缓存结果,只在失败节点重试。这个机制上线后,重试成本大约降了七成。
5.3 输出审查,Agent 的出参不能直接当入参
模型输出不可控,即使提示词写得再死,也会偶尔冒出不安全内容。我在每一个 Agent 的输出进入下一个 Agent 之前,都加了一道轻量过滤:敏感词拦截、JSON 结构校验、长度上限检查。三道都过了才继续流转。
5.4 超时与降级,模型不是永远可靠的模块
我统一给模型调用设置了 30 秒超时。超时之后不是直接报错,而是降级走一条规则模板,用提前写好的兜底结果继续跑。虽然降级结果质量一般,但至少整条链路不会因为单个环节卡住而死掉。
5.5 记忆压缩,防止上下文无限膨胀
多轮协作跑久了,Memory 里的历史消息会越堆越多,最终把上下文窗口塞满。我的方案是设置一个触发阈值,当历史消息超过一定条数时,用模型把旧对话压缩成带关键结论的摘要,再替换掉原始记录。代价是会丢失部分细节,但换来的是长跑稳定性。
5.6 审计日志,把每一步的“元数据”记下来
生产环境里光有结果日志不够,我还会记录每条Msg的 token 消耗、模型名称、耗时、置信度分位。这些东西在复盘某条链路为什么突然变慢、某些 Agent 为什么开始答非所问时非常有用。
5.7 回归用例,把“偶发性”变成“可追踪”
模型输出有随机性,多智能体更容易放大这种随机性。我把业务里最有代表性的几十个输入沉淀成回归用例,每次改完 Prompt 或框架版本都全量跑一遍。只要有一条链路输出格式不对,立刻就能发现是哪次改动引入的。
最后说句实在话:AgentScope 不是万能的,它最值钱的地方是把“交互标准化”这件事从口头约定变成了框架约束。如果你的场景只是两个 Agent 顺序调用模型,手写几行代码可能也够用;但一旦进入多人多轮、并行分支、结果回流这些复杂编排,一套统一的 Msg 消息和明确的 Agent 生命周期,会帮你省下大量 Debug 时间。我接下来还会继续在这条 24.4 分支上折腾消息版本兼容和回放工具,有结论了再回来更新。