如果你现在还在用 for 循环加一堆 if else 去串多个 Agent,建议先停下来看看 AgentScope 2.0 的变化。前阵子我把内部一个多 Agent 任务系统重构了一遍,正好赶上 HiClaw 加入 AgentScope 生态,和 CoPaw 一起把多 Agent 这块的基础设施补齐了。以前每个 Agent 要自己管工具调用、会话上下文、子任务回传,代码里全是胶水和回调;现在用 HiClaw 做工具接入,CoPaw 做协作编排,AgentScope 2.0 做运行时,三件事分得很干净。这篇文章不打算讲概念,就聊聊这套架构的完整推演、落地过程和最容易踩的坑,尤其适合正在搞多 Agent 研发、被工具调用和共享记忆折磨过的同学。先声明,代码示例基于我当时验证的版本,AgentScope 2.0 后续 API 可能有所调整,但核心设计是稳定的。
1. 为什么需要一套“多 Agent 的基础设施”
1.1 多 Agent 应用不是多个 Agent 排队
很多项目所谓多 Agent,其实只是用一个 for 循环依次调用多个 Agent。比如先让 A 生成提纲,再把提纲丢给 B 写正文,C 最后改一版。这种写法本质上还是单链路,只是把原本的一个大 Prompt 拆成了好几段。真正需要多 Agent 协作的场景,不是排队,而是并发、分工、回退、同步。我举个例子:做一个市场分析报告,数据分析 Agent 要查数据库,行业研究 Agent 要搜索网页,文案 Agent 要根据前两个 Agent 的输出写稿子。这三个 Agent 不是简单排队,数据 Agent 可能要等研究 Agent 拉回一批行业关键词再查得更准,文案 Agent 中途发现数据缺了一个维度,还得回调数据 Agent 补充。这个过程中谁先谁后、谁的数据会影响到谁、失败之后如何降级,都需要基础设施来兜底。
如果这些逻辑全部写在业务代码里,你会看到大量状态变量和回调。每个 Agent 自己维护工具连接、会话历史、失败重试,代码很快变得没法维护。我最早的自研方案就是这种状态机加消息队列的混合体,上线第一个月还能跑,等 Agent 从 3 个涨到 8 个,逻辑就开始拧成一团。所以把“多 Agent 怎么协作”这件事从业务里抽出来,下沉到框架层,是所有多 Agent 应用走到一定规模之后的必经之路。这就是我们要说的“多 Agent 基础设施”。
1.2 从自研调度框架迁移到 AgentScope 2.0
我早期自研的调度框架核心是一个状态机,定义了 TODO、RUNNING、WAITING、DONE 几个状态,Agent 之间通过内部消息体通信。这套东西在小规模下确实能跑,但很快暴露了四个问题。第一,消息格式是私有的,新 Agent 接入要读懂我自创的字段。第二,工具调用和 Agent 调用没有统一抽象,有的 Agent 自己直接发 HTTP,有的走内部函数,日志格式对不上。第三,可观测性基本为零,出问题只能开 DEBUG 日志看时序。第四,没有现成的共享记忆,Agent 之间上下文全靠消息传递,一长任务就丢。
后来我对比了几个开源框架,最后选了 AgentScope 2.0。原因很直接:它把 Agent、消息、工具这三层抽象做得干净,消息模型天然是 JSON-like 的,运行时自带分布式调度能力,而且 2.0 版本的扩展点很清晰。更重要的是,社区刚好在推 HiClaw 和 CoPaw 这两个配套组件,一个解决工具接入,一个解决协作与记忆,正好补上我当时最头疼的两块。迁移之后,状态机那套代码删掉了将近一半,取而代之的是统一的消息流转和可插拔的工具网关。整个架构变成了三个层次,反而是最省心的。
2. HiClaw:把“工具调用”变成一等公民
2.1 HiClaw 是什么:统一工具接入层
HiClaw 在 AgentScope 生态里的定位,是一个工具接入网关。可以理解成把散落在各个 Agent 里的工具调用逻辑,统一收口到一个地方管理。它支持注册外部 HTTP API、内部函数、数据库查询,甚至 RPA 操作。名字我看过社区解释,大概是“Hand Claw”的变体,意思是给 Agent 装一只真正能干活的“手”。日常开发中,我见过太多项目死在工具调用这一环:工具参数没定义清楚、鉴权 key 散落在配置文件里、外部服务超时把整个 Agent 拖死。HiClaw 要做的事情,就是把这些横切问题全部标准化。
从实现角度,HiClaw 的核心概念是“工具注册表”。每个工具注册时至少提供四样东西:名称、输入 schema、可调用对象、策略配置。你会问,这不就是普通的工具类吗?区别在于 HiClaw 把调用的全过程管理起来了。比如外部 API 超时后自动重试,重试的间隔用指数退避;比如不同工具的鉴权方式统一走一个密钥中心,而不是每个 Agent 自己去读环境变量;再比如工具执行过程自动记录 trace,方便事后排查。这些能力单独实现都不难,但要每个 Agent 项目都自己实现一遍,就非常浪费。HiClaw 的价值在于,它把这些能力做成基础设施,让业务 Agent 专注于“用什么工具”,而不是“怎么调工具”。
2.2 HiClaw 加入 AgentScope 后带来的核心能力
HiClaw 加入 AgentScope 后,最直观的变化是所有 Agent 都能通过标准消息调用工具,不再关心工具内部是 HTTP 还是本地函数。我列一下我实际用下来最受益的几点:
- 统一工具注册中心:Agent 只需要声明需要哪些工具,运行时通过 HiClaw 拿到工具对象,不需要自己维护 API 地址和连接池。
- 鉴权与密钥管理:工具需要的 token、AK/SK 统一托管,Agent 侧只传业务参数,避免密钥随代码分发。
- 超时、重试、熔断:每个工具可以独立配置 timeout 和重试次数,失败超过阈值自动熔断,不会拖垮整个任务链路。
- 流式输出和可观测性:工具返回结果可以按流式消息回传,同时每次调用都会记录耗时、入参、出参,方便做性能分析。
- Agent as Tool:这是 AgentScope 2.0 里最妙的设计,把 subagent 当作一个另类的 tool 进行调用。HiClaw 的工具模型天然支持这种包装,worker 的输入输出定义成 schema,supervisor 调用 worker 和调用某个接口完全一致。
这里面我要重点展开一下“Agent as Tool”。很多人第一次接触主从模式,都会想着用对话接口去让 supervisor“聊”一个子任务,比如“你帮我分析一下这个数据,然后把结果发给我”。这个模式的问题在于,对话本身是开放的,没有强约束,worker 可能理解偏,也可能回传一个非结构化的文本,supervisor 还得再做一轮解析。而把 subagent 视为另类的 tool,本质上是把协作变成“函数调用”。调用方给出结构化输入,被调用方返回结构化输出,中间的状态、记忆、重试全部由框架接管。这样主从模式就不再是聊天机器人之间的对话,而是一个工作流引擎在调度子任务,可靠性和可控性完全不一样。
2.3 集成实操:最小可运行示例
下面是一个我这边验证过的最小集成例子。假设我们有一个 search_web 工具,通过 HiClaw 注册,然后在一个 Agent 里调用。
# demo_hiclaw.py from agentscope.agent import AgentBase from agentscope.message import Msg import hiclaw # 注册一个外部HTTP工具 hiclaw.register_tool( name="search_web", input_schema={ "query": {"type": "string", "description": "搜索关键词"}, "top_k": {"type": "integer", "description": "返回条数", "default": 5}, }, endpoint="https://internal-search.example.com/api/v1/query", auth_token="token-from-centers", timeout=8.0, max_retry=2, ) class ResearchAgent(AgentBase): def __init__(self, name="research_agent", **kwargs): super().__init__(name=name, **kwargs) # Agent声明需要哪些工具,运行时从HiClaw拿 self._search = hiclaw.get_tool("search_web") def reply(self, x: Msg = None) -> Msg: result = self._search.call(query=x.content, top_k=5) return Msg(name=self.name, content=result, role="assistant")代码本身不复杂,但有几个细节值得注意。register_tool 里的 input_schema 不是给机器看的装饰,而是给 LLM 做参数抽取用的。如果你不写清楚字段类型和描述,模型经常会把参数传错。timeout 和 max_retry 这两个参数一定要根据工具的真实耗时来定,不是越大越好。我曾经给一个搜索接口设了 60 秒超时,结果某个慢查询把整个 Agent 拖住了,上游任务全部排队。后来改成 8 秒超时加 2 次重试,慢查询直接失败,由业务层决定下一步,整体反而更快。这就是基础设施带来的“失败要让上游尽快知道,而不是无限等待”。
3. CoPaw:协作与共享记忆的编排层
3.1 CoPaw 的定位:不是调度器,是“协作协议”
如果说 HiClaw 解决的是 Agent 怎么“干活”,那 CoPaw 解决的就是 Agent 之间怎么“协作”。我之前踩过一个坑,就是自己去写一个中央调度器,把所有 Agent 的调用顺序、依赖关系写死在一个编排引擎里。结果 Agent 一多,编排引擎本身变成了一个新的复杂系统。CoPaw 的理念不太一样,它不是一个集中式调度器,而是一套协作协议和共享基础设施。它告诉你多个 Agent 之间可以用哪些标准模式配合,比如主从模式、竞争模式、流水线模式,并且把协作需要的上下文传递、共享记忆、结果回传这些能力统一实现好了。
主从模式是我用得最多的模式。一个 supervisor 负责拆解任务,多个 worker 各自完成子任务,worker 的结果汇总回 supervisor。CoPaw 对这个模式的支持很直接:每个 worker 可以像工具一样注册成可调用对象,supervisor 通过函数签名去调用,调用结果自动记录到任务上下文里。这样带来的好处是,supervisor 不用关心 worker 内部到底用了哪个模型、调了哪些工具,只关心“这个子任务按时返回了结构化结果”。用公司里的话说,就是“只管派活,不管具体怎么干”。
3.2 多 Agent 共享记忆的实现思路
多 Agent 共享记忆是个听着很高端、做起来很容易翻车的功能。CoPaw 的共享记忆方案,简单说就是“分区加显式声明”。每个 Agent 默认有私有记忆,只有显式声明为 shared 的 key 才会进入共享区。共享区可以接入不同的存储后端,我用过的三种方案对比如下。
| 存储方案 | 适合场景 | 典型延迟 | 一致性 |
|---|---|---|---|
| 内存字典 | 单进程调试、小规模任务 | 微秒级 | 单进程内一致,多进程不行 |
| Redis | 跨 Agent 实时共享、缓存、短任务上下文 | 毫秒级 | 需要自己处理版本覆盖 |
| PGVector 或向量库 | 长期业务记忆、语义召回 | 十毫秒到百毫秒 | 强一致取决于数据库事务 |
我个人推荐的搭配是:短期任务上下文用 Redis,长期业务记忆和语义检索用 PGVector。为什么不用内存?因为 AgentScope 2.0 支持分布式 Runtime,跨进程的 Agent 不能共享内存,一旦上分布式,内存共享就失效了。为什么用 Redis 时要小心?因为多个 Agent 同时读改写同一个 key 很容易互相覆盖。CoPaw 的做法是给共享记忆加版本号,写入时带上 expected_version,版本不匹配就重试或报冲突。这样相当于给 Redis 包了一层乐观锁,虽然增加了一点实现成本,但能避免很多脏数据问题。
3.3 主从模式下 subagent 作为一个另类 tool 的调用模式
关于“subagent 作为另类的 tool”这种设计,我在第 2 节提过概念,这里给出一个 CoPaw 常见的写法。假设我们有个 trend_worker,专门做行业趋势分析,要把它注册成可调用工具,只需要定义输入输出,然后包裹一层调用函数。
import copaw @copaw.register_agent_tool def analyze_trends(question: str, window_days: int = 30) -> dict: """分析趋势,返回结构化报告""" return copaw.call_subagent( agent_id="trend_worker", payload={ "question": question, "window_days": window_days, }, timeout=60, result_schema={ "trend": {"type": "string", "description": "趋势结论"}, "confidence": {"type": "float", "description": "置信度"}, }, )这里的关键是 result_schema。worker 返回的内容会按照这个 schema 做一次校验和清洗,不合规的返回会触发重试。这种做法的好处非常明显:supervisor 拿到的一定是结构化的、可以继续参与后续计算的结果,而不是一段需要再解析的自然语言。在主从模式里,如果 worker 返回的是那种“好的,我已经分析了,结果是……”的大段文本,下游 agent 还得再调一轮模型来抽取字段,整个链路既慢又容易出错。把 subagent 当成 tool 调用之后,supervisor 和 worker 之间的契约就变成了类似 RPC 的接口契约,这比任何口头约定都可靠。
3.4 HiClaw 与 CoPaw 的分工与协作
HiClaw 和 CoPaw 经常被放在一起说,但它们管的事情完全不同。HiClaw 管的是“对物的调用”,包括外部 API、数据库、内部函数;CoPaw 管的是“对 Agent 的调用”,包括主从协作、上下文传播、共享记忆。我用一个实际调用链来说明:
用户提问进入 supervisor,supervisor 先通过 CoPaw 把任务拆成子任务;第一个子任务交给 research worker,research worker 在内部通过 HiClaw 调用搜索和数据库工具;工具结果写入 CoPaw 的共享记忆;第二个子任务交给 write worker,write worker 从共享记忆读取 research worker 的结果,再调用 HiClaw 的写作模板工具生成内容;最后 supervisor 汇总两个 worker 的输出返回给用户。
这条链路里,HiClaw 负责确保每一次工具调用有超时、有重试、有日志;CoPaw 负责确保 worker 之间的结果能正确传递、不会相互污染。两者一个管“工具基础设施”,一个管“协作基础设施”,合在一起,才是完整的“多 Agent 基础设施”。在实际部署时,我会给 HiClaw 和 CoPaw 各自独立的日志主题,这样排查问题时能快速区分是工具问题还是协作问题。这一点看起来很小,真到线上排查事件时是真的救命。
4. 实操:基于 AgentScope 2.0 + HiClaw + CoPaw 搭建多 Agent 应用
4.1 整体架构与组件清单
这一节把前面的理论串起来,做一个可以实际交付的最小系统。先说架构,我采用一个 supervisor 加两个 worker 的设计,worker 分别负责“资料收集”和“报告撰写”。资料收集 worker 会调用 HiClaw 注册的两个工具:search_web 和 query_db;报告撰写 worker 从共享记忆里读取资料,最终的产物写回共享记忆。CoPaw 负责 supervisor 与 worker 之间的调度协议,HiClaw 负责所有工具调用。
组件清单如下:AgentScope 2.0 运行时、HiClaw 工具网关、CoPaw 协作核心、Redis(短期共享记忆)、PGVector(长期记忆向量库)、两个大模型 API 配置(可以都是同一个模型,但角色不同)。模型配置我建议至少区分 temperature:supervisor 用低 temperature 保证规划稳定,worker 用稍高的 temperature 让产出更丰富。这个细节很多同学会忽略,其实影响很大。
4.2 配置与代码实现步骤
先装依赖。我当时的环境是 Python 3.10:
pip install agentscope[distributed,realtime] hiclaw copaw然后建一个 agentscope.json,配置模型和运行时。
{ "runtime": { "name": "demo_runtime", "mode": "local", "max_workers": 8 }, "model": { "api_key": "your-api-key", "model_name": "gpt-4o-mini", "temperature": 0.3 }, "memory": { "backend": "redis", "url": "redis://localhost:6379/0" } }接着初始化运行时、注册工具和 worker。核心步骤如下:
from agentscope.manager import Manager from agentscope.message import Msg import hiclaw import copaw mgr = Manager() mgr.init(config_file="agentscope.json") # 注册工具 hiclaw.register_tool( name="search_web", input_schema={"query": {"type": "string"}}, endpoint="http://search-service/api", timeout=8.0, max_retry=2, ) hiclaw.register_tool( name="query_db", input_schema={"sql": {"type": "string"}}, endpoint="http://db-service/query", timeout=5.0, max_retry=1, ) # 注册worker为agent tool @copaw.register_agent_tool def collect_materials(topic: str) -> dict: return copaw.call_subagent( agent_id="collector", payload={"topic": topic}, timeout=60, result_schema={"summary": {"type": "string"}}, )这里有个小坑。register_agent_tool 的定义里,输入参数名要和 LLM 抽取时的描述保持一致。如果你把一个参数命名为 question,但在 prompt 里给模型看的工具描述写的是 topic,模型就会经常传错。我在项目中就把所有工具的参数命名统一成和自然语言描述一致,明显提高了调用准确率。
最后定义 supervisor,让它调用 collect_materials 和 write_report 两个“工具”,把整个流程串起来。关键点在于 supervisor 不需要关心 worker 内部细节,它看到的路由就是两个函数。
4.3 超时与重试参数的估算方法
工具和 subagent 的超时不能拍脑袋。我一般会分三步来定。第一步,先压测单次调用的 P95 耗时。比如 search_web 压测下来 P95 是 2 秒,但偶尔会到 5 秒,那就把 timeout 定在 8 秒,覆盖到 P995。第二步,考虑重试次数。如果调用失败后,重试的成功率在前两次最高,第三次开始边际效益很低,那 max_retry 设 2 就够了。第三步,考虑端到端任务的可容忍延迟。如果一个任务允许最长 30 秒,里面要串行调 3 个工具,每个工具 timeout 8 秒,理论最坏是 24 秒,重试一旦触发就会超过 30 秒,这时应该把工具改成并行调用,或者降低单工具 timeout,而不是硬加 max_retry。
subagent 的超时要更大一些,因为它的成本里包含了一次或多次模型生成,还要加上它内部调工具的时间。我建议把 subagent 的 timeout 设置为它内部所有关键路径的 P95 之和乘以 1.5,再向上取整到 10 秒的倍数。算下来通常 60 到 120 秒比较合适。这套估算方法不精确,但能避免两个极端:超时太短导致正常慢请求被切断,超时太长导致故障被隐藏。
4.4 评估这套基础设施是否到位
系统搭完后,要有一组数据来判断是否真的比原来好。我主要