1. 从单兵作战到团队协作:为什么Agent架构正在经历一次根本性转变
过去一年多,我一直在折腾各种智能体项目,从最早的单一对话助手,到后来能调用工具、能读写文件、能自主规划任务的Personal Agent,踩过的坑可以说遍布整个开发链路。最开始那种“一个Agent包打天下”的思路确实很爽——你给它一个目标,它自己拆解、自己执行、自己反思,看起来就像给电脑装了个大脑。但用得越多,我越发现一个尴尬的事实:单个Agent的能力天花板,远比我们想象中来得更早。
这个天花板不是模型不够聪明造成的,而是架构本身决定的。一个Personal Agent无论怎么优化,它本质上还是一个“串行执行器”——它一次只能专注做一件事,上下文窗口有限,工具集越堆越臃肿,一旦任务链条拉长,它就开始丢三落四、前后矛盾。我印象特别深的一次,是让一个Agent帮我完成“调研某个技术方案、写一份对比报告、再生成一份演示大纲”这样的复合任务,结果它在调研阶段就把上下文塞满了,到了写报告环节已经完全忘了前面查到的关键数据,最后输出的东西前后对不上,只能推倒重来。
这种体验逼着我去思考一个问题:我们是不是把太多职责压在了一个Agent身上?现实世界里,一个靠谱的团队从来不是靠一个全能超人,而是靠分工明确的多个角色协作完成的。调研有调研的人,写作有写作的人,审核有审核的人,每个人专注自己那一块,通过清晰的接口传递信息。Agent系统其实也该如此——这就是Agent-to-Agent(后面我简称A2A)架构的核心思路:把一个大而全的Personal Agent,拆解成多个职责单一、通过标准协议互相通信的专职Agent。
这个转变的意义,我觉得不亚于从“单机程序”到“分布式系统”的跨越。Personal Agent解决的是“一个人怎么把事做完”,而A2A解决的是“一群Agent怎么把事做好”。前者拼的是单点能力,后者拼的是协作效率和信息流转的可靠性。对于已经在用Personal Agent做自动化、做工作流、做内容生产的人来说,理解这个转变不是赶时髦,而是实实在在能解决你当前遇到的瓶颈——上下文爆炸、任务串行太慢、单点故障、能力无法复用,这些问题在A2A架构下都有对应的解法。
接下来我会把这套东西拆开讲透:从Personal Agent的典型架构和它的固有局限讲起,再讲A2A的通信机制怎么设计、角色怎么划分、协议怎么定,然后给出一套可以照着搭的最小可行方案,最后把我踩过的那些坑和排查经验整理出来。不管你是刚接触Agent的新手,还是已经写过几个Personal Agent想往多Agent方向走的老手,应该都能从里面找到能直接抄作业的部分。
2. Personal Agent的架构真相与它的三个硬伤
2.1 一个典型Personal Agent到底由什么组成
在聊A2A之前,得先把Personal Agent这个东西讲清楚,不然很多人对它的理解还停留在“套个壳调用大模型”的层面。一个真正能干活、能自主完成任务的Personal Agent,通常包含这么几个核心模块,我用自己搭过的一个内容处理Agent来举例说明。
最底层是模型层,也就是大语言模型本身,负责理解意图、生成内容、做决策。这一层现在基本是标准化的,选一个能力够用的模型就行,不是架构设计的重点。往上一层是记忆层,分短期记忆和长期记忆——短期记忆就是当前对话的上下文窗口,长期记忆一般是向量数据库或者结构化的知识库,用来存那些跨会话需要保留的信息。再往上是工具层,这是Personal Agent真正“能动手”的关键,包括文件读写、网络请求、代码执行、数据库查询等等,每个工具都是一个可以被模型调用的函数。最上面是规划与执行层,也就是所谓的Agent Loop:接收目标、拆解任务、选择工具、执行动作、观察结果、反思调整,循环往复直到任务完成。
这个结构看起来挺完整,我一开始也觉得只要把每一层都做扎实,Agent就能无限扩展能力。但实际跑起来才发现,问题恰恰出在这个“全塞在一起”的设计上。因为所有模块共享同一个上下文窗口,所有决策都由同一个模型实例做出,所有工具都挂在同一个执行器上,这就导致了一个根本性的矛盾:能力越强,上下文越挤;任务越长,一致性越差。
2.2 硬伤一:上下文窗口是稀缺资源,不是免费午餐
第一个硬伤,也是最要命的,就是上下文窗口的消耗问题。很多人做Personal Agent的时候有个错觉,觉得上下文窗口越大越好,模型厂商也在不断把窗口往大了做,好像窗口大了问题就解决了。但实际用下来你会发现,窗口再大也架不住任务链条的消耗,而且窗口越大,模型对中间信息的注意力反而越容易涣散。
我拿一个具体的数字来说明。假设你让一个Personal Agent完成“读取一份50页的文档、提取关键信息、生成摘要、再基于摘要写一篇分析文章”这样的任务。光是读取文档这一步,50页大概就是两三万字,直接吃掉一大块上下文。提取关键信息的过程又会产生一堆中间推理文本,生成摘要再占一部分,等到写分析文章的时候,前面读过的原文细节其实已经很难被模型准确回忆了。我实测过,一个128K窗口的模型,在处理这种长链条任务时,到了后半段对前半段信息的引用准确率会明显下降,经常出现“张冠李戴”的情况——把文档A的数据安到文档B的结论上。
这个问题的本质在于,Personal Agent把“信息处理”和“信息消费”混在了同一个上下文里。它既要记住原始素材,又要记住中间推理,还要记住最终产出,所有东西挤在一个篮子里,篮子再大也会满。而A2A架构的解法很直接:让不同的Agent各自持有自己需要的上下文,通过消息传递只交换“结论”而不是“全过程”。调研Agent读完文档后,只把结构化的关键信息传给写作Agent,写作Agent根本不需要看到那50页原文,它的上下文里只有干净的素材摘要。这一下就把上下文压力分散开了。
2.3 硬伤二:串行执行让任务速度卡在单点
第二个硬伤是执行效率。Personal Agent的Agent Loop本质上是串行的:想一步、做一步、看一步、再想下一步。这种模式在任务简单的时候没问题,但一旦任务里包含多个可以并行的子任务,串行执行就成了瓶颈。
举个例子,你要做一个“竞品分析”的任务,需要同时调研五家竞品的信息。Personal Agent的做法通常是一家一家来:先查A,整理A的信息,再查B,整理B的信息……五家查完,时间全花在等待上了。如果这五家的调研之间没有依赖关系,完全可以并行去做,但单Agent架构做不到,因为它只有一个执行线程,一次只能推进一个子任务。
有人可能会说,那我用异步调用、用多线程不就行了?问题是Personal Agent的决策是依赖上下文的,多个子任务并行执行会产生多个并行的上下文分支,这些分支最后要合并回同一个上下文,合并的时候又是一堆冲突和冗余。所以单Agent架构下,并行不是不能做,而是做了之后上下文管理会变得极其复杂,得不偿失。A2A架构天然就是并行的:五个调研Agent可以同时开工,各自维护自己的上下文,最后把结果汇总给一个协调Agent就行,干净利落。
2.4 硬伤三:能力复用几乎为零,每个Agent都在重复造轮子
第三个硬伤是能力复用的问题。在Personal Agent的世界里,每个Agent都是一个独立的整体,你为A项目写的调研Agent,很难直接拿到B项目里用,因为它的工具集、提示词、记忆结构都是跟具体任务绑死的。结果就是每做一个新任务,都要重新配一套工具、重新调一轮提示词,重复劳动特别多。
这个问题的根源在于,Personal Agent没有把“能力”抽象成独立的、可组合的单元。它更像是一个手工作坊,每个产品都是从头到尾手工打造。而A2A架构的思路是把它变成流水线:每个Agent是一个标准化的“工位”,有明确的输入输出接口,只要接口对得上,这个工位就能被任意组装到不同的流水线里。一个擅长做信息抽取的Agent,既可以用在竞品分析流程里,也可以用在合同审核流程里,不需要重写。
这三个硬伤叠加在一起,就解释了为什么很多Personal Agent项目做到一定程度就推不动了——不是模型不行,是架构到了极限。而A2A要解决的,正是这三个结构性问题。
3. Agent-to-Agent的核心机制:通信、角色与协议设计
3.1 A2A到底“通”的是什么:消息传递的基本模型
理解A2A,最关键的是理解“Agent之间到底在传递什么”。很多人第一反应是“传递对话”,就像人和人聊天一样你一句我一句。但实际做下来你会发现,Agent之间传递对话是效率极低的做法,因为对话里充满了冗余信息、客套话和上下文依赖,两个Agent如果靠自然语言对话来协作,光是理解对方在说什么就要消耗大量token,还容易产生歧义。
我实践下来比较靠谱的做法是结构化消息传递。也就是说,Agent之间传递的不是自由文本,而是带有明确schema的结构化数据。比如调研Agent给写作Agent的消息,不是“我调研了A公司,它的情况是这样的……”,而是一个JSON对象,里面明确包含company_name、key_findings、data_points、confidence这些字段。写作Agent拿到这个结构,直接就能用,不需要再去解析自然语言。
这种设计的好处是显而易见的。第一,信息密度高,没有废话,每个字段都是有用的。第二,解析成本低,接收方不需要用模型去理解,直接按字段读取就行。第三,可校验,字段缺失或者格式不对,一眼就能看出来,不用等模型“理解错了”才发现问题。第四,可追溯,每条消息都有明确的结构,出问题的时候容易定位是哪个环节传错了。
当然,结构化消息也不是万能的。有些任务确实需要传递比较灵活的信息,比如创意类的、开放性的内容,这时候可以退化成半结构化的形式,主体是结构化字段,但保留一个free_text字段放补充说明。我的经验是,能结构化就尽量结构化,实在不行再退而求其次。
3.2 角色怎么划分:不是越多越好,而是边界越清晰越好
A2A架构里,角色划分是最考验设计功力的地方。新手最容易犯的错是“按功能堆角色”——看到任务里有调研、有写作、有审核,就设三个Agent,看起来挺合理。但实际跑起来会发现,角色之间的边界模糊,经常出现职责重叠或者信息断档。
我总结下来,角色划分的核心原则是按“信息处理阶段”划分,而不是按“任务名称”划分。什么意思呢?一个Agent应该对应一个明确的信息处理阶段,这个阶段的输入是什么、输出是什么、需要什么能力,都应该是清晰的。如果两个Agent的输入输出有大量重叠,那它们大概率应该合并;如果一个Agent内部又包含了多个明显不同的处理阶段,那它大概率应该拆分。
拿内容生产流程举例,我现在的划分是这样的:采集Agent负责从各种来源获取原始素材,它的输出是清洗过的原始数据;提炼Agent负责从原始数据里抽取关键信息和观点,输出是结构化的要点;写作Agent负责把要点组织成连贯的文章,输出是初稿;审核Agent负责检查初稿的事实准确性和逻辑一致性,输出是修改意见或通过标记。这四个角色的边界非常清晰,每个角色的输入输出都是明确定义的结构化数据,互相之间不会打架。
这里有个反直觉的点:角色数量不是越多越好。我见过有人把流程拆成十几个Agent,结果协调成本高得吓人,消息在Agent之间传来传去,光协调就消耗了大量资源,整体效率反而不如三个Agent。我的经验是,一个A2A系统里,核心角色控制在3到7个之间比较合适,超过这个数量就要考虑是不是划分过细了。
3.3 通信协议怎么定:让Agent之间“说同一种语言”
角色划分好之后,下一个关键问题是通信协议。如果每个Agent都用自己的一套消息格式,那协作就会变成一场灾难——A发给B的消息B看不懂,B发给C的消息C又理解错了。所以A2A系统必须有一套统一的通信协议,规定好消息的格式、字段、类型和语义。
我目前用的协议设计大概是这样的:每条消息包含header和payload两部分。header里有message_id、sender、receiver、timestamp、message_type、correlation_id这些元信息,用来做消息路由和追踪。payload里是具体的业务数据,根据message_type不同而不同。比如message_type是research_result的时候,payload里就是调研结果的结构化数据;message_type是review_feedback的时候,payload里就是审核意见。
这套协议看起来简单,但有几个细节特别重要。第一个是correlation_id,它用来把一次任务的所有相关消息串起来,方便追踪和调试。没有这个字段,消息一多你就分不清哪条消息属于哪个任务了。第二个是message_type的枚举管理,所有可能的消息类型必须提前定义好,不能临时发明,否则接收方不知道怎么处理。第三个是版本管理,协议一旦定下来,后续修改要兼容旧版本,不然升级的时候整个系统都会崩。
提示:协议设计阶段多花一天时间,能省后面一周的调试时间。我一开始图省事,消息格式定得很随意,结果跑到后面各种字段对不上、类型不匹配,排查起来极其痛苦,最后不得不推倒重来。
3.4 协调机制:谁来决定下一步该谁干
有了角色和协议,还需要一个“指挥”。A2A系统里,通常需要一个协调Agent(或者叫Orchestrator)来负责调度。它的职责是:接收总目标,拆解成子任务,分派给对应的Agent,收集结果,决定下一步。
协调Agent的设计有两种常见模式。一种是中心化协调,所有决策都经过协调Agent,它像一个项目经理,谁干完了、下一步该谁干,都由它说了算。这种模式的好处是逻辑清晰、容易调试,坏处是协调Agent容易成为瓶颈,而且一旦它挂了整个系统就停了。另一种是去中心化协调,Agent之间直接通信,通过某种共识机制来决定下一步。这种模式更灵活、更健壮,但实现复杂度高很多,调试也困难。
我的建议是,除非你有非常明确的理由,否则优先用中心化协调。去中心化听起来很酷,但实际项目里,中心化协调的简单可靠远比去中心化的灵活性重要。我早期尝试过让Agent之间自由协商,结果就是消息满天飞,出了问题根本不知道是谁的锅,最后老老实实改回中心化。
4. 从零搭一套最小可行的A2A系统
4.1 环境准备与技术选型
讲完原理,该动手了。这一节我给出一个最小可行的A2A系统搭建方案,你可以照着复现。先说技术选型,我的原则是“用最少的依赖跑通核心流程”,不要一上来就上重型框架。
模型层:选一个支持结构化输出(function calling或者JSON mode)的模型就行。结构化输出能力是A2A的基础,因为Agent之间传的是结构化消息,模型必须能稳定地生成符合schema的输出。如果模型不支持结构化输出,你就得自己写解析和校验逻辑,麻烦很多。
通信层:最小方案用HTTP + JSON就够了,不需要上消息队列。每个Agent暴露一个HTTP接口,接收消息、处理、返回结果。等系统规模大了、并发高了,再考虑引入消息队列做异步解耦。我一开始就上了消息队列,结果发现调试极其麻烦,消息发出去不知道被谁消费了,后来退回HTTP同步调用,简单直接。
存储层:短期用内存字典就行,长期记忆用SQLite或者轻量级的向量库。不要一上来就上分布式数据库,那是给自己找麻烦。
编排层:协调Agent用一个简单的状态机来实现,定义好每个状态、每个状态下可以执行的动作、状态之间的转移条件。状态机的好处是逻辑清晰,不会出现“Agent不知道该干嘛”的情况。
4.2 定义消息协议与Agent接口
第一步是定义消息协议。我用Python的dataclass来定义,清晰又不容易出错:
from dataclasses import dataclass, field from typing import Any, Dict, Optional from datetime import datetime import uuid @dataclass class Message: sender: str receiver: str message_type: str payload: Dict[str, Any] message_id: str = field(default_factory=lambda: str(uuid.uuid4())) correlation_id: Optional[str] = None timestamp: str = field(default_factory=lambda: datetime.now().isoformat())然后是Agent的基类接口,所有Agent都实现这个接口:
class BaseAgent: def __init__(self, name: str): self.name = name def handle(self, message: Message) -> Message: raise NotImplementedError def can_handle(self, message_type: str) -> bool: raise NotImplementedError这个接口设计的关键是can_handle方法,它让协调Agent能快速判断某个Agent能不能处理某类消息,不用去猜。我见过一些设计里没有这个方法,协调Agent只能靠硬编码的映射表来决定谁处理什么,一旦新增Agent就要改协调逻辑,很不好维护。
4.3 实现三个核心Agent
接下来实现三个最核心的Agent:采集、提炼、写作。为了演示,我用简化的逻辑,实际项目里每个Agent内部可以很复杂。
采集Agent负责接收一个采集任务,返回原始素材:
class CollectorAgent(BaseAgent): def can_handle(self, message_type: str) -> bool: return message_type == "collect_request" def handle(self, message: Message) -> Message: topic = message.payload["topic"] # 实际项目里这里会调用搜索、爬取等工具 raw_material = self._collect(topic) return Message( sender=self.name, receiver=message.sender, message_type="collect_result", payload={"topic": topic, "raw_material": raw_material}, correlation_id=message.correlation_id )提炼Agent接收原始素材,返回结构化要点:
class ExtractorAgent(BaseAgent): def can_handle(self, message_type: str) -> bool: return message_type == "extract_request" def handle(self, message: Message) -> Message: raw_material = message.payload["raw_material"] # 调用模型做信息抽取,输出结构化要点 key_points = self._extract(raw_material) return Message( sender=self.name, receiver=message.sender, message_type="extract_result", payload={"key_points": key_points}, correlation_id=message.correlation_id )写作Agent接收要点,返回初稿:
class WriterAgent(BaseAgent): def can_handle(self, message_type: str) -> bool: return message_type == "write_request" def handle(self, message: Message) -> Message: key_points = message.payload["key_points"] draft = self._write(key_points) return Message( sender=self.name, receiver=message.sender, message_type="write_result", payload={"draft": draft}, correlation_id=message.correlation_id )这三个Agent的实现都很简单,但关键在于它们的输入输出都是结构化的,互相之间通过消息类型来匹配,不依赖自然语言理解。这就是A2A和Personal Agent最本质的区别。
4.4 协调Agent的状态机实现
协调Agent是整个系统的大脑,我用状态机来实现。状态机的核心是定义清楚每个状态和转移条件:
class Orchestrator: def __init__(self, agents): self.agents = {agent.name: agent for agent in agents} self.state = "idle" self.context = {} def run(self, goal: str): self.context["goal"] = goal self.state = "collecting" self._dispatch("collect_request", {"topic": goal}) while self.state != "done": self._step() def _dispatch(self, message_type: str, payload: dict): for agent in self.agents.values(): if agent.can_handle(message_type): msg = Message( sender="orchestrator", receiver=agent.name, message_type=message_type, payload=payload, correlation_id=self.context.get("correlation_id") ) result = agent.handle(msg) self._on_result(result) return raise ValueError(f"没有Agent能处理消息类型: {message_type}") def _on_result(self, result: Message): if result.message_type == "collect_result": self.context["raw_material"] = result.payload["raw_material"] self.state = "extracting" self._dispatch("extract_request", {"raw_material": self.context["raw_material"]}) elif result.message_type == "extract_result": self.context["key_points"] = result.payload["key_points"] self.state = "writing" self._dispatch("write_request", {"key_points": self.context["key_points"]}) elif result.message_type == "write_result": self.context["draft"] = result.payload["draft"] self.state = "done"这个状态机虽然简单,但已经能跑通完整的流程了。实际项目里,状态机会更复杂,会有错误处理、重试、并行分支等逻辑,但核心思路是一样的:用明确的状态和转移条件来驱动整个流程,而不是让Agent自己决定下一步干什么。
4.5 跑通第一个A2A任务
把上面的代码组装起来,就可以跑第一个任务了:
if __name__ == "__main__": agents = [ CollectorAgent("collector"), ExtractorAgent("extractor"), WriterAgent("writer") ] orchestrator = Orchestrator(agents) orchestrator.run("智能体协作架构的技术趋势") print(orchestrator.context["draft"])跑通之后你会发现,整个流程非常清晰:协调Agent按顺序调用三个Agent,每个Agent只做自己那一小块,上下文压力被分散了,每个Agent的提示词也可以针对自己的任务专门优化。这就是A2A架构的威力——它把一个复杂任务拆成了三个简单任务,每个简单任务都更容易做好。
5. 实操中踩过的坑与排查经验
5.1 消息丢失与重复处理
A2A系统跑起来之后,最先遇到的问题往往是消息层面的。我遇到最多的是两种情况:消息丢失和消息重复处理。
消息丢失通常发生在Agent处理失败但没有正确返回错误的时候。比如某个Agent内部抛了异常,但没有把异常包装成错误消息返回给协调Agent,协调Agent就一直在等结果,整个流程卡死。解决方法是所有Agent的handle方法必须用try-except包起来,任何异常都要转成结构化的错误消息返回,不能让异常穿透出去。
消息重复处理则发生在重试逻辑设计不当的时候。协调Agent发现某个Agent超时了,就重新发一次消息,但第一次的消息其实已经被处理了,只是返回慢了,结果同一个任务被处理了两次。解决方法是在消息里加message_id,Agent处理前先检查这个id有没有处理过,处理过就直接返回缓存结果。这个去重逻辑看起来简单,但能省掉很多莫名其妙的bug。
5.2 结构化输出不稳定
第二个大坑是模型的结构化输出不稳定。你明明在提示词里写了“请输出JSON格式”,模型有时候还是会输出一段自然语言加JSON的混合体,或者字段名拼错,或者该是数组的地方给了个字符串。这种问题在单Agent场景下还能靠后处理兜底,但在A2A场景下会直接导致下游Agent解析失败。
我的应对策略有三层。第一层是用模型原生的结构化输出能力,比如function calling或者JSON mode,这比靠提示词约束靠谱得多。第二层是加校验和重试,收到输出后先按schema校验,不通过就带着错误信息让模型重新生成,一般重试一两次就能过。第三层是设计容错的下游,下游Agent在解析消息时,对缺失字段要有默认值,对类型不对的字段要尝试转换,不要一遇到格式问题就崩。
提示:结构化输出的稳定性跟提示词的清晰度强相关。我实测下来,把schema直接写在系统提示词里,并且给出一个完整的示例输出,比只描述字段含义的稳定性高很多。
5.3 协调Agent成为瓶颈
第三个坑是协调Agent的性能瓶颈。当任务变多、Agent变多之后,所有消息都要经过协调Agent,它很快就扛不住了。表现是任务排队、响应变慢,严重的时候协调Agent自己就超时了。
解决思路有两个方向。一个是把协调逻辑做轻,协调Agent只负责路由和状态管理,不做任何重计算,所有重活都下放给专职Agent。另一个是引入并行,对于没有依赖关系的子任务,协调Agent可以同时派发给多个Agent,等所有结果回来再继续。并行化之后,整体吞吐量能提升好几倍。
不过并行也带来了新的问题:结果合并的顺序和一致性。我的做法是给每个并行分支分配一个branch_id,结果回来的时候按branch_id归位,确保合并逻辑是确定的,不会因为返回顺序不同导致结果不一致。
5.4 常见问题速查表
下面这张表是我整理的高频问题和对应解法,遇到问题可以先查这张表:
| 问题现象 | 可能原因 | 排查方向 | 解决方法 |
|---|---|---|---|
| 流程卡死不动 | Agent异常未返回 | 检查各Agent日志 | 异常包装成错误消息返回 |
| 同一任务重复执行 | 重试无去重 | 检查message_id | 加处理去重缓存 |
| 下游解析失败 | 结构化输出不稳定 | 检查上游输出格式 | 加schema校验和重试 |
| 协调Agent变慢 | 消息全走中心 | 看协调Agent负载 | 下放重计算、引入并行 |
| 结果前后矛盾 | 上下文传递有误 | 追踪correlation_id | 检查消息字段映射 |
| Agent互相等待 | 依赖关系成环 | 画依赖图 | 打破环或引入超时 |
5.5 几个让我印象深刻的排查案例
说两个具体的排查经历,都是花了不少时间才定位到的。
第一个是“结果莫名其妙丢失”。现象是协调Agent明明收到了写作Agent的返回,但最终输出里就是没有内容。查了半天发现,写作Agent返回的消息里correlation_id是空的,协调Agent在归位结果的时候按correlation_id匹配,匹配不上就把结果丢了。问题出在写作Agent内部重新构造消息的时候忘了带上correlation_id。这个bug的教训是:消息的元信息在传递过程中必须原样保留,任何重新构造消息的地方都要检查元信息有没有丢。
第二个是“Agent之间死锁”。现象是两个Agent互相等对方的消息,谁也不动。查下来发现是依赖关系设计成了环:A等B的结果才能继续,B又等A的结果才能继续。这种问题在单Agent架构下不会出现,因为单Agent是串行的,但在A2A架构下,如果依赖关系没设计好,很容易成环。解决办法是在设计阶段就把依赖关系画成有向无环图,如果发现有环,就要重新设计流程,或者引入超时机制打破死锁。
6. A2A架构的扩展方向与我的实践体会
6.1 从固定流程到动态编排
上面搭的最小系统是固定流程的——采集、提炼、写作,顺序写死。但实际任务往往不是固定流程,有的任务需要先审核再写作,有的任务需要跳过提炼直接写作。这时候就需要动态编排能力。
动态编排的核心是让协调Agent能根据任务特点自己决定调用哪些Agent、以什么顺序调用。实现方式通常是把每个Agent的能力描述成结构化的元数据,协调Agent拿到任务后,先做一个规划,生成一个执行计划,然后按计划执行。这个规划过程本身可以用模型来做,但要注意加约束,不能让模型规划出系统里不存在的Agent或者不合法的流程。
我实践下来的经验是,动态编排适合任务类型多变的场景,固定流程适合任务类型稳定的场景。如果你的业务就是固定的那几种任务,别折腾动态编排,固定流程更可靠。只有当任务类型经常变、而且变化无法提前枚举的时候,动态编排才值得投入。
6.2 Agent能力的复用与组合
A2A架构最大的价值之一,就是Agent能力可以复用。一个写好的提炼Agent,不应该只能用在内容生产流程里,它应该能被任何需要信息提炼的流程调用。要做到这一点,关键是把Agent的能力接口设计得足够通用。
具体来说,Agent的输入输出不应该跟具体业务绑死。提炼Agent的输入应该是“一段文本”,输出应该是“结构化要点”,而不是“一篇关于某主题的文章”和“某主题的要点”。业务相关的部分应该由协调Agent在调用时通过参数传入,而不是硬编码在Agent内部。这样设计之后,同一个提炼Agent就能被复用到各种场景里。
我现在维护着一个Agent库,里面有十几个通用Agent,做新项目的时候大部分时候是组合现有Agent,只有少数特殊需求才写新Agent。这种复用带来的效率提升是非常明显的,一个新项目从想法到跑通,有时候半天就够了。
6.3 监控与可观测性
A2A系统比单Agent系统复杂得多,没有好的监控基本没法维护。我现在的做法是给每条消息都打上完整的追踪信息,包括correlation_id、message_id、sender、receiver、timestamp、duration,全部落到日志里。然后做一个简单的可视化面板,能按correlation_id把一次任务的所有消息串起来看,一眼就能看出流程走到哪一步、哪一步慢了、哪一步出错了。
这个监控体系看起来是额外投入,但实际省下的排查时间远超投入。没有它的时候,一个跨Agent的问题可能要查半天;有了它,大部分问题几分钟就能定位。
6.4 我个人的几点体会
折腾A2A这段时间,有几个体会特别深,分享给准备往这个方向走的人。
第一,不要为了A2A而A2A。如果你的任务用单Agent就能很好地完成,那就用单Agent,别硬拆。A2A解决的是单Agent解决不了的问题——上下文爆炸、串行太慢、能力无法复用。如果这些问题你还没遇到,说明你的任务复杂度还没到需要A2A的程度。
第二,协议和接口设计值得多花时间。我前面反复强调这一点,是因为我自己在这上面吃过亏。协议设计得好,后面加Agent、改流程都很轻松;协议设计得差,每加一个Agent都是一场灾难。宁可前期多花两天把协议定清楚,也不要后期花两周去填坑。
第三,从最小系统开始,逐步演进。不要一上来就设计一个包含十几个Agent、支持动态编排、带完整监控的复杂系统。先搭一个三个Agent的固定流程,跑通,再根据实际需求一点点加东西。我见过太多人一上来就搞大架构,结果卡在某个细节上,整个项目就搁置了。
第四,调试工具比功能更重要。A2A系统的调试难度比单Agent高一个数量级,因为问题可能出在任何一个Agent、任何一条消息上。所以从一开始就要把日志、追踪、可视化这些基础设施做好,它们不是锦上添花,是雪中送炭。
最后分享一个小技巧:在开发阶段,我会给每个Agent加一个“dry run”模式,这个模式下Agent不真正执行任务,只是把收到的消息和将要返回的消息打印出来。这样在调试流程的时候,可以快速验证消息流转是否正确,不用等每个Agent真正跑完。这个模式帮我省了大量时间,尤其是在流程刚搭好、还没调通的时候。
这个方向后续还能继续扩展,比如引入Agent的版本管理、支持Agent的热插拔、做Agent能力的自动发现等等。但这些都是后话,把基础的多Agent协作跑通、跑稳,才是当前最值得投入的事情。