消息驱动多智能体系统核心设计：从消息协议到工程实战

拿到“hermes-agent”这个项目名，圈内人第一反应多半会心一笑——Hermes（赫尔墨斯）本就是希腊神话里的信使神，负责传递消息、引导旅人、接洽边界。把它和“agent”放一起，意图几乎是写在脸上的：这是一个以消息驱动为核心的多智能体项目。我最初接触这个代号时，以为是某个团队的内部工具，后来自己上手搓了一遍类似方案，才发现这里面的设计取舍比想象中多得多。

这篇文章我想用自己实际搭建和踩坑的经历，把“hermes-agent”这类消息型Agent系统的核心设计、关键实现和实战细节拆开讲透。不管你是想自己搭一套轻量级Agent编排框架，还是想在现有项目里引入Agent协作机制，这篇都能给你一套可直接参考的思路。我会把重点放在消息协议、任务编排、工具调用这三件最关键的事上，再附上我部署上线之后的排障实录。内容偏工程向，但每个概念我都会用最直白的方式解释清楚。

1. 先想清楚：hermes-agent到底解决什么问题

名字不过是代号，真正值得琢磨的，是为什么要做这样一个agent系统，以及它和市面上的主流Agent框架差在哪里。动手之前如果没把这层想透，后面大概率会被各种边角问题拖垮。

1.1 名字背后的定位：信使型Agent不生产消息，只负责精确传递

如果你看过Agent类项目的命名规律，会发现大家特别喜欢用神名、星名、炼金术名词。比如阿波罗、普罗米修斯、雅典娜之类的。但Hermes这个命名有一个很强的暗示：这个Agent的核心职责是“通信与协调”，而不是“生成与创造”。

什么意思？传统的Agent项目，核心是“一个大模型+一堆工具”，用户问一句，Agent内部规划、调用工具、返回结果。这是单智能体模式。而hermes-agent这类名字指向的是另一种形态：多个Agent并行存在，各有分工，通过消息通信协作完成一个复杂任务。也就是说，它的核心不是“一个更聪明的AI助手”，而是“一组能互相传递消息的AI员工”。

我在实际设计里，把Agent分成了三类角色：

• 协调型Agent：类似项目经理，负责拆解任务、分配任务、汇总结果。
• 执行型Agent：类似一线员工，专注调用某个工具或处理某类数据。
• 网关型Agent：类似前台，负责对接外部请求、翻译协议、转发消息。

这三类角色之间的消息传递，就是hermes-agent的核心骨架。如果你也准备做一个多Agent项目，强烈建议先按这个思路把角色划分清楚，而不是一上来就写大模型Prompt。角色不清，后续的权限、队列、重试机制全部会乱。

1.2 现有的Agent方案，卡在了哪几个地方

聊这个话题之前，我先声明：OpenAI的Assistants API、LangChain的AgentExecutor、AutoGen、CrewAI这些开源方案我都实际用过，都是好东西，也能跑通Demo。但真要放到生产环境跑一周以上，会遇到一些很现实的问题。

第一个是编排偏向单机串行。LangChain的Agent基本上是一个Agent内部走完“思考→行动→观察”的循环，虽然也能多工具切换，但很难做到多Agent并行、分头干活再归并结果。你可以用LangGraph硬塞状态机，但写起来并不轻松。

第二个是消息语义太薄弱。很多框架里的Agent之间没有真正的消息流，只是函数调用嵌套。你很难回答一个问题：如果Agent A给Agent B发了一条指令，B执行成功/失败之后，A怎么感知？重试策略挂在哪一层？这些都依赖隐式的层层返回，排查问题的时候特别痛苦。

第三个是上下文越滚越胖。每个Agent都倾向于把对话历史塞进Prompt，跑几天之后token开销暴涨，响应变慢，甚至超出上下文窗口。这个问题我在后面排障章节会专门讲，它是我见过的Agent项目里最普遍的隐性杀手。

hermes-agent要做的，就是在消息层面把这些痛点用工程手段解决掉。说白了，它不是在“提高模型智能”，而是在“规范Agent之间的协作秩序”。

1.3 为什么我决定走轻量级自研路线

如果你也在纠结“用现成框架还是自己搭”，我的结论是：重度依赖大模型框架，不如把核心机制握在自己手里。这不是叛逆，是被坑出来的判断。

拿我当时的项目来说，需求是企业内部工单自动分发与处理：工单进来之后，系统要识别类型、匹配负责部门、提取关键字段、生成处理建议。看起来简单，但涉及六个不同业务系统的对接，每个系统有不同的认证方式、字段标准、错误返回。用现成框架，我得给每个工具写适配器，还得处理框架自身的升级兼容问题。

后来我决定只保留两个底层依赖：一个是模型SDK，一个是消息队列。其余全是自己的代码。核心思路就是：Agent本质上是一个“消息处理单元”，它接收消息、处理消息、发送消息。整个系统就是一个消息处理网络。这个思路后来证明非常省心，因为业务逻辑全部收敛在消息处理器里，而非散落在框架回调里。

所以这个项目的核心结论是：Agent框架的竞争力不在“它支持多少种模型”，而在“它把消息机制做得有多干净”。

2. 核心设计拆解：消息、任务与工具三件事

hermes-agent这类系统，不管名字怎么变，内在绕不开三个基础问题：Agent之间怎么通信？一个任务怎么被拆解和跟踪？Agent怎么调用外部工具？这三件事我在第一版里全做砸过，第二版才逐渐沉淀出稳定模式。下面一个个讲清楚。

2.1 消息总线：Agent之间到底怎么“说话”

Agent之间通信看起来简单，不就是互相发请求吗？但我实际做下来发现，简单RPC方式有三个坑：

• 强耦合：A想调用B，必须知道B的地址、接口、参数格式。Agent一多，互相依赖变成蜘蛛网。
• 难追溯：一条链路跨了四五个Agent，中途谁处理了什么，没有日志根本查不出来。
• 难重试：B临时挂了，A的重试逻辑如果写在RPC调用里，代码会被重试逻辑淹没。

我的解法是引入消息队列作为通信底座。每个Agent的消息发送方和接收方是不见面的，发送方把消息投递到对应的消息队列，接收方从自己的队列里拉取处理。这个模式本身不新奇，但放在Agent系统里有几个好处特别明显：

一是削峰填谷。业务高峰期工单蜂拥而至，协调Agent来不及处理时，消息会先在队列里排队，不会把系统拖垮。二是天然重试。消息消费失败可以重返队列，间隔递增重试，不需要Agent之间各自实现重试。三是审计复盘。每条消息自带消息ID、来源Agent、目标Agent、时间戳、状态，出了问题可以从头到尾回放整个消息链路。

我用的通信协议很简单，类似下面这个JSON结构：

{ "message_id": "msg_20240520001", "version": "1.0", "sender": "coordinator.main", "receiver": "executor.extractor", "message_type": "task.dispatch", "task_id": "task_20240520003", "payload": { "ticket_id": "TKT-7789", "content": "客户反馈登录超时，希望尽快处理", "priority": "high" }, "created_at": "2024-05-20T10:00:01Z", "trace_id": "trace_88f2a1" }

每个字段都有讲究。sender和receiver用“点分命名”，方便按模块路由；message_type是消息类型，Agent根据类型决定走哪套处理逻辑；task_id用于关联同一个业务任务下的多条消息；trace_id则贯穿所有相关消息，排障时的关键索引。这套协议我建议原样复制到你的项目里，这是实践出来的通用结构。

2.2 任务分解与状态机：一次复杂任务是怎么被跟踪的

消息只负责把话传到，但一次完整任务往往涉及多个Agent的接力。比如工单处理任务：协调Agent拆解出“分类”“字段提取”“部门匹配”“建议生成”四个子任务，分别发给四个执行Agent，最后还要汇总结果。这个过程中的状态管理，是第一版最头疼的地方。

一版我用了数据库表“task”字段存状态，但多个Agent并发更新同一条记录时，经常出现覆盖写，状态直接从“处理中”变成“已完成”，其实还有个子任务没跑完。二版我改成给每个Task维护一张独立的状态流转表，每条状态变更都是一条记录，而不是更新同一行。这其实就是事件溯源思路的简化版。

具体状态机我设计成：

PENDING -> DISPATCHED -> PROCESSING -> SUCCEEDED | | v v FAILED PARTIAL

这里有个容易被忽略的设计点：一个任务被拆成多个子任务时，父任务不能简单用“成功/失败”这种二元状态。我引入了一个PARTIAL状态，表示“部分子任务成功，部分失败”。如果成功子任务数量超过阈值，父任务进入人工复核队列；如果失败比例过高，父任务整体回滚到待重拆状态。这套机制上线后，工单处理的“半途而废”问题基本绝迹了。

另外强烈建议在每个Agent里维护一个“任务看板”，即使不上前端，也要在日志里按task_id聚合输出进度。我遇到过调试时最崩溃的情况就是：一个任务跑了十个步骤，中间某步失败，日志里却只看到十行孤立的INFO。后来我把关键节点的日志统一打上task_id前缀，再用日志工具的“按字段聚合”功能拉出来看，整个流程一目了然。

2.3 工具注册与调用协议：让Agent正确使用外部能力

Agent再聪明，不接工具就是纸上谈兵。但是“接工具”这件事，比大多数人想的要麻烦。大模型输出的是自然语言，工具要执行的是结构化参数，中间的翻译任务，比提示词工程更考验系统设计。

我定的工具调用协议分为三层。第一层是工具元信息注册，每个工具提供一个JSON Schema，声明名称、描述、入参结构、出参结构。这里要特别强调：描述信息一定要写清楚“什么时候用、什么时候别用”，否则模型经常在无关场景下瞎调工具。比如我注册了一个“查询天气”的工具，描述里只写“查询天气”，结果模型在算优惠券金额时也去调用它。

第二层是参数清洗与校验。模型返回的参数经常是“尽力而为”的，类型可能不对、字段可能缺失、枚举可能超出范围。我在工具执行前加了一道硬校验，类似下面的伪代码逻辑：

def validate_tool_args(tool_schema, raw_args): """ 对模型返回的原始参数做类型强转和必填项校验。 注意：这个环节不能省，模型给出的参数比想象中更不可靠。 """ import jsonschema from jsonschema import ValidationError try: jsonschema.validate(instance=raw_args, schema=tool_schema) return raw_args, None except ValidationError as e: # 尝试一次智能修复：缺失字段补默认值，错误类型强制转换 fixed_args = smart_repair(tool_schema, raw_args) if fixed_args is not None: return fixed_args, None return None, str(e)

第三层是统一异常返回。工具执行结果不管是成功还是失败，都以标准格式返回给Agent，包括状态码、结果数据、错误信息、耗时。我遇到过执行成功但结果格式不规范的情况，比如数据库查询返回了Decimal类型，JSON序列化直接报错，Agent收到的就是一段没头没尾的异常。后来所有工具返回前都要经过normalize_result()统一转成字符串、数字、布尔、列表、字典这五种基础类型。

3. 实操：从零搭一套可用的消息型Agent系统

理论讲完，直接上实操。我尽量把每个步骤写成可以照抄的命令和代码，但你最好先理解每一步在干什么，不要只做复制粘贴。我以Python为例，因为生态最成熟，排查问题也最方便。

3.1 基础选型与目录结构

技术选型上我给一套经过实际验证的组合，不会过度依赖重型组件：

目录结构我按“消息处理单元”的模型来组织：

hermes-agent/ ├── hermes/ │ ├── core/ │ │ ├── message.py # 消息协议定义 │ │ ├── bus.py # 消息总线封装 │ │ ├── state.py # 任务状态机 │ │ └── router.py # 消息路由 │ ├── agents/ │ │ ├── base.py # Agent基类 │ │ ├── coordinator.py # 协调型Agent │ │ ├── executor.py # 执行型Agent │ │ └── gateway.py # 网关型Agent │ ├── tools/ │ │ ├── registry.py # 工具注册表 │ │ └── weather.py # 示例工具 │ └── config/ │ └── settings.py # 全局配置 ├── tests/ ├── requirements.txt └── README.md

这个目录的划分遵循一条原则：协议层、控制层、执行层严格分离。core里不写任何业务逻辑，agents里不直接操作数据库，tools里不感知消息协议。这样任何一个Agent替换或工具调整，都不会牵动全局。

3.2 实现一个最小可运行的Agent基类

Agent基类是整套系统的心脏，我把它简化到只剩“接收消息→处理→发送结果”三条路径，但把扩展点全部留好：

import abc import json import logging from typing import Any, Callable logger = logging.getLogger(__name__) class BaseAgent(abc.ABC): """ Agent基类。 子类只需实现 handle_message() 方法，即可拥有完整的消息收发能力。 生命周期由外部AgentRunner管理，Agent自身不启动线程。 """ def __init__(self, name: str, model_client: Any, tool_registry: Any): self.name = name self.model_client = model_client self.tool_registry = tool_registry self.current_trace_id = None @abc.abstractmethod def handle_message(self, message: dict) -> dict: """ 处理单条消息。必须返回标准响应字典： { "status": "success" | "failure", "data": "...", "error": "..." # 仅失败时填充 } """ raise NotImplementedError def receive_and_execute(self, message: dict) -> dict: """ 统一入口：记录日志、调用子类业务、捕获异常、输出标准响应。 所有Agent的消息入口都必须走这个方法，便于统一埋点。 """ self.current_trace_id = message.get("trace_id", "") logger.info( "[%s] 收到消息 message_id=%s type=%s", self.name, message.get("message_id"), message.get("message_type"), ) try: result = self.handle_message(message) if result.get("status") == "failure": logger.warning("[%s] 处理失败 task_id=%s reason=%s", self.name, message.get("task_id"), result.get("error")) return result except Exception as exc: logger.exception("[%s] 处理异常 task_id=%s", self.name, message.get("task_id")) return { "status": "failure", "error": f"{type(exc).__name__}: {str(exc)}", }

基类里我特意加了current_trace_id，每个Agent在处理任何消息前都会记录这个ID。这样排查问题的时候，可以用一个trace_id把参与整个链路的所有Agent日志全部拉出来，按时间排序，就能复现一次任务的完整生命周期。这个习惯学会了，Agent排障效率能提升一大截。

接下来是协调Agent的简化实现，它的任务是：收到一个工单→调大模型分类→根据分类决定触发哪个执行Agent→把结果汇总返回：

class CoordinatorAgent(BaseAgent): """ 协调型Agent：拆解任务，分发给执行Agent，再聚合结果。 这里仅演示单个子任务的流程，多个子任务用并发池处理。 """ def handle_message(self, message: dict) -> dict: content = message.get("payload", {}).get("content", "") task_id = message.get("task_id", "") # 步骤1：调用大模型进行任务分类 category = self.classify_ticket(content) # 步骤2：根据分类选择下游执行Agent并发送消息 if category == "login_issue": downstream = "executor.login" elif category == "refund_request": downstream = "executor.refund" else: return { "status": "failure", "error": f"无法识别的工单类别: {category}", } downstream_result = self.send_to_agent( receiver=downstream, task_id=task_id, payload={"ticket_content": content}, ) # 步骤3：返回聚合结果 return { "status": downstream_result.get("status", "failure"), "data": downstream_result, } def classify_ticket(self, content: str) -> str: # 实际项目里这里调用大模型，这里用规则代替说明逻辑 if "登录" in content or "超时" in content: return "login_issue" if "退款" in content or "退钱" in content: return "refund_request" return "unknown"

注意这里我没有把大模型调用逻辑写在handle_message里，而是抽成了独立方法。原因是：分类逻辑以后很容易演化成“先检索知识库再调用模型”，如果耦合在主流程里，改动风险很大。

3.3 走通一次完整的跨Agent协作流水线

启动整套系统，需要在入口脚本里完成三件事：初始化消息总线、注册所有Agent、把Agent挂到对应的消息消费者上。以下是一个可直接运行的最小示例：

import redis from hermes.core.bus import MessageBus from hermes.agents.coordinator import CoordinatorAgent from hermes.agents.executor import ExecutorAgent def main(): # 1. 初始化消息总线，基于Redis Streams实现 redis_conn = redis.Redis(host="localhost", port=6379, decode_responses=True) bus = MessageBus(redis_conn) # 2. 初始化Agent并注册工具 tool_registry = { "fetch_user_info": fetch_user_info, "create_ticket_comment": create_ticket_comment, } coordinator = CoordinatorAgent( name="coordinator.main", model_client=model_client, tool_registry=tool_registry, ) executor = ExecutorAgent( name="executor.login", model_client=model_client, tool_registry=tool_registry, ) # 3. 订阅对应消息队列 bus.subscribe("queue.coordinator", coordinator.receive_and_execute) bus.subscribe("queue.executor.login", executor.receive_and_execute) # 4. 投递一条测试工单 bus.publish("queue.coordinator", { "message_id": "msg_test_001", "message_type": "ticket.new", "task_id": "task_test_001", "payload": {"content": "用户反馈登录一直超时，请求协助"}, "trace_id": "trace_test_001", }) # 注意：这里阻塞一段时间等待消费者线程处理，实际运行时会有事件循环 import time time.sleep(3) if __name__ == "__main__": main()

整个流程跑起来之后，你可以在Redis里用命令查看消息流：

XLEN queue.coordinator XREAD COUNT 5 STREAMS queue.executor.login 0

这会返回执行Agent处理后的消息内容。如果一切正常，你应该能看到status: success的标准响应。如果看到不正常的返回，优先检查消息体里的字段名——我在实际调试中，有一半以上问题都是字段名大小写或拼写不一致导致的，比如taskId传进了Java风格字段，而Python这边只认识task_id。

4. 上线后踩过的坑和排查技巧

这章我按真实的发生频率排序，把我在生产环境里遇到的四个最大问题，以及对应的排查思路完整写出来。这些东西在官方文档和教程里基本看不到，属于只能靠实战喂出来的经验。

4.1 并发冲突：多个Agent同时消费，任务状态被互相覆盖

问题现象：系统跑了两天，突然出现“任务已完成，但子任务全部未执行”的诡异情况。查数据库发现，任务记录里的状态字段是“完成”，但子任务关联的消息根本没有被下游Agent消费。

排查过程：先是怀疑消息发丢了，检查消息队列一看，消息还在队列里躺着。后来才意识到不是消息丢，而是任务状态被提前“完成”了。原因是我当时的父任务判断逻辑是这样的：

if all_child_tasks_done(): mark_parent_done()

多个执行Agent并发完成任务时，A完成子任务1后，判断发现所有子任务都“已完成”（其实B和C还没开始，只是初始状态是“完成”），于是直接标记父任务完成。这是一个典型的并发初值语义错误：初始态和终态使用同一个值，导致并发判断误判。

修复方案分两步。第一步，把子任务初始状态改成PENDING，明确区分“还没开始”和“已完成”。第二步，在判断父任务完成时，加一个短暂的“冷却时间”，等待所有子任务的消息都进入终态后再做汇总判断。这里用到了Redis的原子递增计数，等所有子任务回调到位后，再触发一次父任务状态机更新。

4.2 上下文超长：Prompt越滚越胖，慢到你怀疑人生

问题现象：Agent上线一周后，响应时间从2秒涨到30秒，token费用翻了好几倍。排查发现，每个Agent在处理新消息时，都会把“历史对话记录”完整带进Prompt。一个Agent处理几千条消息后，历史少说也有十几万token，每次调用模型都是在做超长文本处理。

这个问题的根源在于：我错误地把“对话历史的完整记忆”当成了Agent的默认需求，而实际业务只需要“最近几条消息+当前任务上下文”。修复方式是把Prompt上下文回收机制做成分层策略：

• 短期上下文：保留当前任务相关的所有消息，最多不超过20条。
• 中期上下文：保留最近30分钟内其他任务的摘要，用3到5句话压成摘要存入。
• 长期上下文：不进入Prompt，只在需要时通过检索调用。

我还做了一道“上下文净化”逻辑：在大模型调用前，自动移除Prompt里的URL、常见无意义语气词、以及大段重复的系统提示词。这一招让token开销直接降低了40%。如果你的项目也遇到类似问题，优先检查：是不是有Agent在每次消息里都带上了全局历史？如果是，赶紧切成“任务级上下文”，而不是“会话级上下文”。

4.3 工具调用失败：模型产生了结构化结果，但你的代码看不懂

问题现象：Agent明明调用了工具，但工具侧抛出一个奇怪的异常：list index out of range。查日志发现，模型返回的参数是{"city": "Shanghai, China"}，而工具里写的是用city_list.index(city)去匹配城市列表，结果当然匹配不到。

排查思路：这类问题表面上是代码鲁棒性不足，实际上是参数清洗层形同虚设。我后来在工具调用前加了三道保险：

1. 对模型返回的所有字符串参数做strip()，去除首尾空白和特殊字符。
2. 对模型返回的枚举值做模糊匹配，比如“上海”和“Shanghai”能映射到同一个城市ID。
3. 对模型返回的数组参数做长度校验，一旦发现长度和业务预期不一致，立即返回“参数格式错误”的标准异常，而不是让下游代码裸奔。

更重要的一点是，工具本身的异常信息必须能反哺给模型。当工具执行失败时，我会把标准化的错误信息返回给Agent，并让它重新构造一次工具调用参数。这个“重试-反思-再调用”的循环，是Agent系统的自我修正能力核心。如果没有这层，工具一报错整个任务就断了。

4.4 依赖冲突：Agent没挂，环境先崩了

问题现象：新加的Agent依赖requests-html，结果和原来的requests版本冲突，核心Agent直接被新依赖顶得无法启动。线上系统从“能跑”变成“全挂”，只隔了一次pip install。

这个坑说出来有点丢人，但它确实在真实开发中反复出现。我的修复方案是三步强制规范：

• 所有依赖必须固定精确版本，禁止使用requests>=2.0这类宽松写法。每次锁版本后，统一在干净环境里验证一次。
• 核心Agent与实验型Agent拆到不同的虚拟环境里，用独立的消息队列连接，互不共享依赖。
• 引入pip-tools或uv这类依赖锁定工具，自动生成requirements.lock文件，部署时严格按照锁文件安装。

后来我把Agent系统的每个模块都做成了独立的微服务容器，彻底解决了依赖互踩的问题。代价是部署复杂度上升，但换来的是“爆炸半径”可控：就算某个工具型Agent容器崩了，核心协调Agent还能继续跑，其他任务不受影响。

4.5 日志追踪：没有统一trace_id，排障等于大海捞针

最后这节我不说代码，说一个更重要的工作习惯。我见过太多Agent项目，日志倒是打得满天飞，但没有一条能被串联起来。A Agent打印“收到消息”，B Agent打印“处理完成”，C Agent打印“返回结果”，你根本不知道这些日志属于哪一个任务。

我定了一条铁律：所有Agent、所有环节、所有日志，必须携带同一个trace_id。具体做法是，在消息进入系统的第一个入口处生成trace_id，然后所有下游Agent在记录日志时，都从消息体里取trace_id并打印到日志上下文中。这样用日志系统的按字段过滤功能，输入一个trace_id，就能把一条业务链路的全部日志一次性拉出来。

我甚至给关键消息链路加了“看板日志”：定时输出当前未完成任务数、各Agent队列积压数、失败消息TOP10。每次有异常，先看看板，再按trace_id钻取详细日志，基本能在五分钟内定位问题源头。这套方法我想推荐给所有做Agent工程化的开发者：先让日志可追踪，再谈优化和扩展。

5. 从Demo到生产，我的最后几点体会

项目做完了，感触最深的反而不是技术本身，而是几个在过程中反复出现的认识。

第一，Agent系统的复杂度不在模型，而在不确定性管理。传统程序只要输入确定，输出必然确定。但Agent系统里，大模型输出天然带有随机性，甚至同一个Prompt两次调用结果都不同。你要是没有解决好“输出不确定性”带来的连锁反应，系统永远处于随时失控的边缘。我的做法是：所有模型输出都要经过校验、清洗、修复这三道关卡，宁可多写点防御代码，也不要让脏数据流入业务层。

第二，消息协议要往“未来三年”设计。第一版我的消息字段只够当时用，上线第二周提需求加“来源渠道”，我就得改协议、改Agent、改数据库，折腾了整整一天。后来反思，凡是涉及跨Agent传递的字段，设计时至少要留出扩展后缀，比如metadata、source_channel这类通用字段，宁可暂时用不上，也不要等需要时再去全局改动。

第三，不要迷信“全自动”。Agent再强，也不是每个环节都能完全托付。我在系统里专门设计了一个人工复核队列，凡是置信度低于阈值、或者多次重试失败的任务，都会自动转给人工处理。上线一个月的数据是：约12%的任务走人工复核，但恰恰是这12%的兜底，让整个系统在用户那里攒下了“靠谱”的口碑。任何一个Agent项目，如果没有人工兜底，那它只是技术演示，不是生产系统。

如果你是刚开始接触这类消息型Agent，我最后给一个最实际的建议：第一个版本不要追求多Agent百花齐放。先做一个协调Agent加两个执行Agent的最小闭环，把消息协议、状态机、工具调用链路这三件事跑得滚瓜烂熟，再往上加角色。地基打牢了，房子怎么盖都稳。