企业官网建设流程全解析

1. 从一个真实请求链路说起：A2A 到底在解决什么问题

假设你对着一个智能助手说了一句：“帮我查一下上个月华东区的销售数据，做个同比分析，然后生成一份 PPT 发给我。”这句话对人来说稀松平常，但对系统来说，它同时包含了数据检索、数值计算、文档生成、消息投递四类完全不同的能力需求。如果把这些能力全部塞进一个模型里，结果通常是：模型在某一环上掉链子，整条链路就崩了。

A2A（Agent-to-Agent）要解决的核心问题，就是把一件复杂的事拆给多个各有所长的 Agent，让它们像一个小团队一样协作完成。这里的关键词有三个：选谁来做（Agent 选择）、怎么配合（协作机制）、做完的事记在哪（记忆管理）。这三个问题不解决，多 Agent 系统就只是一堆各自为政的接口调用，谈不上“协作”。

我在实际项目里踩过最典型的坑，就是一开始把 A2A 理解成“多个 Agent 互相调 API”。结果做出来的东西，A 调 B、B 调 C，调用链一长，谁该负责什么完全说不清，出错之后连日志都对不上。后来才明白，A2A 的本质不是“调用”，而是基于能力描述的任务分派与状态同步。每个 Agent 要先说清楚“我能干什么、我干不了什么”，调度方才有可能做出正确选择。

这篇文章我会按真实请求的推进顺序来讲：先讲 Agent 怎么被选中，再讲它们之间怎么协作，然后重点讲记忆到底该放在哪一层，最后给出一套可以照着搭的最小实现思路。适合已经了解大模型基础调用、想往多 Agent 方向落地的开发者，也适合正在评估“要不要上多 Agent”的技术负责人。

提示：A2A 不是银弹。如果你的任务本身是单轮的、能力单一的，硬拆成多 Agent 只会增加延迟和故障点。判断标准很简单——任务是否需要两种以上差异明显的能力，且这些能力由不同模型或工具更擅长。

2. Agent 选择：不是“谁最强”，而是“谁最合适”

2.1 能力描述卡：让调度方有据可依

Agent 选择的第一步，是每个 Agent 都得有一张能力描述卡。这张卡不是随便写句“我擅长数据分析”就完事，它需要结构化，至少包含四个字段：能力域、输入格式、输出格式、限制条件。

我一般会把它设计成类似下面这样的结构：

{ "agent_id": "data_analyst_01", "capability": ["sales_data_query", "trend_analysis", "ratio_calculation"], "input_schema": {"region": "string", "period": "string"}, "output_schema": {"metrics": "object", "chart_data": "array"}, "constraints": {"max_rows": 100000, "timeout_sec": 30} }

为什么要这么细？因为调度方在做选择时，靠的不是“感觉这个 Agent 行”，而是把当前任务的输入输出和候选 Agent 的 schema 做匹配。输入对不上、输出接不住、超出限制条件，直接排除。这一步能过滤掉大部分错误选择。

2.2 选择策略：规则匹配、语义匹配与混合打分

实际落地时，Agent 选择通常有三种策略，各有适用场景。

规则匹配最简单，就是按能力标签硬匹配。任务需要trend_analysis，就找 capability 里带这个标签的 Agent。优点是快、可预测；缺点是标签体系一旦不统一，匹配就会失效。我建议在项目早期就用规则匹配，先把链路跑通。

语义匹配是用向量相似度，把任务描述和 Agent 能力描述都转成向量，算余弦相似度。它适合任务描述比较模糊的场景，比如用户说“帮我看看数据有没有异常”，语义匹配能把它关联到anomaly_detection能力上。但它的缺点是不稳定，同一个任务换个说法，选出来的 Agent 可能就变了。

混合打分是我目前最推荐的方案：先用规则匹配做硬过滤，把不满足 schema 和约束的 Agent 剔除；再在剩下的候选里用语义相似度排序；最后叠加一个历史成功率权重。公式大致是：

final_score = 0.5 * semantic_similarity + 0.3 * success_rate + 0.2 * (1 / avg_latency)

这个权重不是拍脑袋定的。语义相似度占大头，是因为它决定了“能力对不对口”；成功率占三成，是因为一个能力对口但经常失败的 Agent 不该被优先选；延迟权重最小，是因为在多数业务里，正确性比速度更重要。你可以根据自己业务调整，但一定要把成功率纳入考量，否则系统会反复选中那个“看起来最匹配、实际最不靠谱”的 Agent。

2.3 一个容易忽略的点：Agent 的“拒绝能力”

我在项目里吃过一个亏：某个 Agent 被选中后，发现任务超出了它的能力边界，但它没有拒绝，而是硬着头皮返回了一个看似合理、实则错误的结果。这种错误最难排查，因为链路上没有任何报错。

后来我给所有 Agent 加了一个显式拒绝机制：当输入不满足 schema、或任务超出 capability 范围时，Agent 必须返回一个标准化的reject响应，而不是硬答。调度方收到 reject 后，会重新走一遍选择流程，选下一个候选。

这个机制的价值在于：让失败变得可见。宁可让系统多绕一圈，也不要让错误结果悄悄流到下游。这是多 Agent 系统和单模型系统在工程上最大的区别之一。

3. 协作机制：从“互相调用”到“任务编排”

3.1 三种协作拓扑：链式、星型与网状

Agent 之间的协作，本质上是在决定控制流怎么走。常见的拓扑有三种。

链式是 A 做完交给 B，B 做完交给 C。它适合步骤严格串行的任务，比如“取数 → 计算 → 生成报告”。优点是逻辑清晰，缺点是任何一环失败，整条链就断，且无法并行。

星型是有一个中心调度 Agent，所有子 Agent 都只和它通信。子 Agent 之间不直接联系。这是我最推荐的拓扑，因为控制权集中，状态好管理，出错好定位。中心调度方掌握全局任务状态，知道每一步进行到哪、谁在做、结果是什么。

网状是 Agent 之间可以任意通信。它理论上最灵活，但工程上最难维护——调用关系一旦复杂，就会出现循环调用、状态不一致、死锁等问题。除非你有非常明确的理由，否则不要一上来就做网状。

3.2 任务状态机：让协作过程可追踪

星型拓扑能跑起来的关键，是中心调度方维护一个任务状态机。每个任务从创建到完成，会经历若干状态：pending（待分派）、dispatched（已分派）、running（执行中）、succeeded（成功）、failed（失败）、rejected（被拒绝）。

状态机的价值不只是“记录”，而是驱动下一步动作。比如一个任务处于rejected状态，调度方就知道要重新选择 Agent；处于failed状态，就要判断是重试还是降级；处于succeeded，就把结果写入记忆，并触发下游任务。

我一般会把状态机持久化到一张表里，字段包括task_id、parent_task_id、agent_id、status、input、output、created_at、updated_at。这样任何一个请求进来，我都能顺着parent_task_id把整条协作链路还原出来。排查问题时，这张表比任何日志都好用。

3.3 结果传递：结构化数据优先，自然语言兜底

Agent 之间传什么，是个很容易被忽视但影响巨大的问题。我的经验是：能用结构化数据就用结构化数据，自然语言只作为兜底。

原因很直接。自然语言有歧义，A Agent 说“数据略有增长”，B Agent 可能理解成 5%，也可能理解成 50%。而结构化数据{"growth_rate": 0.052}没有歧义。在多 Agent 链路里，歧义会被逐级放大，最后结果完全不可控。

具体做法是：每个 Agent 的输出 schema 里，核心字段必须是结构化的；如果确实需要传递解释性内容，就单独放一个notes字段，且下游 Agent 默认不依赖它做判断。这样既保留了可读性，又保证了链路的确定性。

注意：不要为了“看起来智能”而让 Agent 之间用自然语言自由对话。自由对话在演示里很酷，在生产里是灾难。控制流必须由状态机驱动，而不是由模型“聊”出来。

4. 记忆放在哪里：三层记忆的职责划分

4.1 为什么记忆不能只放一个地方

多 Agent 系统里，“记忆”这个词被用得很泛。有人把对话历史叫记忆，有人把向量库叫记忆，有人把任务状态也叫记忆。如果不把记忆分层，就会出现两个极端：要么所有 Agent 都去读同一份巨大的上下文，又慢又贵；要么每个 Agent 各记各的，协作时对不上。

我的做法是把记忆分成三层：会话记忆、任务记忆、长期记忆。它们服务不同的对象，存储不同的内容，生命周期也完全不同。

4.2 会话记忆：面向用户的短期上下文

会话记忆是面向用户的，记录的是这次交互里用户说过什么、系统回过什么。它的生命周期通常就是一次会话，会话结束就可以归档或丢弃。

它放在哪里？我一般放在调度方，而不是各个子 Agent。因为用户只和调度方对话，子 Agent 不需要知道用户上一句说了什么。调度方在分派任务时，只把和当前任务相关的那部分上下文提取出来，作为任务输入传给子 Agent。这样既保证了子 Agent 拿到的信息是干净的，也避免了把整段对话历史塞进每个 Agent 的上下文里。

4.3 任务记忆：协作链路的中间状态

任务记忆是面向协作的，记录的是每个任务的输入、输出、状态、依赖关系。它就是我前面说的那张状态机表。

这层记忆必须持久化，而且必须能被所有参与协作的 Agent 按权限读取。为什么？因为下游 Agent 经常需要上游的结果。比如生成报告的 Agent，需要拿到分析 Agent 算出的指标。如果任务记忆不共享，下游就只能重新算一遍，既浪费又可能算出不一样的结果。

任务记忆的读取要有边界：每个 Agent 只能读和它当前任务直接相关的上游结果，不能随便翻整条链路。这既是性能考虑，也是安全考虑。

4.4 长期记忆：跨会话的知识沉淀

长期记忆是面向系统的，记录的是跨会话、跨任务沉淀下来的知识。比如某个用户偏好什么样的报告格式、某类数据查询常用的过滤条件、某个 Agent 在什么情况下容易失败。

长期记忆通常用向量库加结构化存储的组合来实现。向量库负责语义检索，结构化存储负责精确查询。它的写入要谨慎——不是什么都要记，只记那些未来会被复用的信息。我见过一些项目把所有中间结果都往长期记忆里塞，结果检索时噪声极大，反而降低了效果。

4.5 记忆写入时机：成功才写，失败要记原因

记忆写入的时机也有讲究。我的原则是：任务成功时写结果，任务失败时写原因。

成功的结果写入任务记忆，供下游使用；同时如果这个结果有复用价值，再写入长期记忆。失败的任务，结果不写，但失败原因一定要写——是输入不合法、超时、还是能力不匹配。这些失败原因积累起来，就是优化 Agent 选择策略的依据。前面说的“历史成功率权重”，数据就是从这来的。

5. 搭一套最小可用的 A2A 系统：从零到跑通

5.1 组件清单与职责边界

要跑通一套最小 A2A 系统，你需要的组件其实不多，但每个的职责必须清晰。

• 调度器（Orchestrator）：接收用户请求，拆解任务，选择 Agent，维护状态机，管理会话记忆。
• Agent 注册中心：存放所有 Agent 的能力描述卡，供调度器查询。
• 任务存储：持久化任务状态与结果，即任务记忆。
• 长期记忆库：向量库加结构化库，存跨会话知识。
• 若干 Agent 实例：各自实现一个能力域，暴露统一的调用接口。

这里最容易出问题的是调度器和 Agent 的职责边界。我的建议是：调度器只负责“分派和协调”，不负责“具体计算”；Agent 只负责“执行和返回”，不负责“决定下一步做什么”。一旦 Agent 开始自己决定调用谁，控制流就散了。

5.2 一次完整请求的推进过程

还是用开头那个例子：“查上个月华东区销售数据，做同比分析，生成 PPT 发给我。”

调度器收到请求后，先做任务拆解，得到三个子任务：数据查询、同比分析、报告生成与投递。然后它查注册中心，为每个子任务匹配 Agent。数据查询匹配到data_analyst，同比分析也匹配到它（因为它有ratio_calculation能力），报告生成匹配到report_writer。

接着调度器按依赖关系排序：数据查询先跑，拿到结果后触发同比分析，分析完成后再触发报告生成。每一步的状态都写入任务存储。报告生成 Agent 需要分析结果时，从任务存储里读，而不是重新算。

整个过程中，会话记忆留在调度器，任务记忆在任务存储，长期记忆只在需要时读写。三层记忆各司其职，链路清晰。

5.3 关键接口设计示例

Agent 的统一调用接口，我一般设计成这个样子：

def invoke(agent_id, task_input, context): agent = registry.get(agent_id) if not schema_match(task_input, agent.input_schema): return {"status": "rejected", "reason": "input_schema_mismatch"} try: result = agent.execute(task_input, context) if not schema_match(result, agent.output_schema): return {"status": "failed", "reason": "output_schema_mismatch"} return {"status": "succeeded", "output": result} except TimeoutError: return {"status": "failed", "reason": "timeout"}

这个接口里有两个校验点：入口校验输入 schema，出口校验输出 schema。入口校验防止 Agent 接到不该接的任务，出口校验防止 Agent 返回不符合约定的结果。这两个校验是多 Agent 系统稳定性的底线，千万别省。

5.4 跑通之后必须补的三件事

链路跑通只是开始。我踩过的坑告诉我，下面三件事不补，系统上线后一定出问题。

第一件是超时与重试策略。每个 Agent 调用都要设超时，超时后不能无限重试，要有次数上限和退避。我一般设 2 次重试，间隔指数增长。

第二件是幂等性。同一个任务可能因为重试被多次执行，Agent 的执行逻辑必须保证幂等，否则会重复写入数据。做法是给每个任务一个唯一task_id，Agent 执行前先查这个 id 是否已处理过。

第三件是可观测性。每个任务的开始、结束、状态变更都要打点，最好能按parent_task_id串成完整链路。没有这个，线上出问题你只能靠猜。

6. 那些文档里不会写的踩坑经验

6.1 Agent 数量不是越多越好

我一开始觉得 Agent 越多越灵活，结果注册中心里堆了十几个能力高度重叠的 Agent，调度器选择时经常在几个“差不多”的 Agent 之间反复横跳，同一个任务两次执行选中的 Agent 都不一样，结果自然不稳定。

后来我做了合并：能力重叠度超过 70% 的 Agent 直接合并，只保留能力边界最清晰的那几个。合并之后，选择稳定性明显提升。经验是：Agent 的粒度应该由能力边界决定，而不是由实现方便决定。

6.2 上下文膨胀是隐形杀手

多 Agent 协作时，很容易把越来越多的上下文往 Agent 里塞，觉得“信息越多判断越准”。实际上，上下文一长，模型注意力会被稀释，关键信息反而被淹没，而且成本直线上升。

我的做法是给每个 Agent 的输入上下文设一个硬上限，只传和当前任务直接相关的字段。需要更多信息时，让 Agent 主动去任务记忆里查，而不是一股脑塞进去。这个“按需拉取”的模式，比“全量推送”稳定得多。

6.3 失败要快速暴露，不要静默降级

有些系统为了“用户体验”，在某个 Agent 失败时悄悄降级，返回一个凑合的结果。这在多 Agent 链路里非常危险，因为降级结果会作为下游的输入，错误被逐级放大，最后你根本不知道问题出在哪一环。

我的原则是：失败就失败，明确返回失败状态，让调度器决定是重试、换 Agent 还是终止。宁可让用户看到一次明确的失败，也不要让一个错误结果悄悄流完全程。

6.4 记忆的读写权限要收紧

任务记忆是共享的，但共享不等于谁都能读写。我见过因为权限没收紧，某个 Agent 误写了另一个任务的结果，导致下游拿到脏数据的情况。

做法很简单：每个 Agent 只能写自己任务的输出，只能读自己任务依赖的上游输出。这个规则在调度器层面强制执行，Agent 自己无权越界。多花十分钟做权限校验，能省掉后面几天的排查时间。

7. 关于 A2A 选型与落地的一点个人判断

回到最开始的问题：什么时候该上 A2A，什么时候不该上。我的判断标准是看任务的能力异构程度。如果一件事需要两种以上差异明显的能力，且这些能力用不同模型或工具实现效果更好，那 A2A 值得上。如果只是单轮问答或者单一能力，硬拆只会增加复杂度。

选型上，我倾向于先规则后语义、先星型后网状、先单机后分布。规则匹配能跑通就别急着上向量检索，星型拓扑能解决就别碰网状，单机能扛住就别急着拆服务。多 Agent 系统的复杂度是乘法增长的，每加一个维度，排查难度就翻一倍。

记忆这块，我的核心体会是：会话记忆归调度器，任务记忆归存储层，长期记忆归系统。三层不要混，混了就会又慢又乱。任务记忆是重中之重，它既是协作的纽带，也是排查的依据，值得你花最多精力去设计。

最后分享一个我一直在用的小技巧：给每个 Agent 加一个“自检”步骤，在执行前先判断自己能不能做这个任务，不能做就明确拒绝。这个自检看起来多了一步，但它把大量潜在的错误挡在了执行之前，整体稳定性反而更高。多 Agent 系统的可靠性，往往不是靠“让每个 Agent 更聪明”，而是靠“让每个 Agent 更清楚自己的边界”。