我用一个点餐 Demo，串了一遍 Agent 应用开发

发表于 更新于 分类于 阅读次数： 本文字数： 14k 阅读时长 ≈ 12 分钟
通过一个可运行的点餐 Demo，串联 Tool Calling、MCP、A2A、AG-UI、HITL、Skill 路由与多 Agent 协作的完整开发链路。

这段时间看 Agent 相关的东西，一个很直接的感受是：概念越来越多，但单独看每个概念，又好像都不复杂。

比如 Tool Calling 是让模型选择函数，MCP 是连接工具，A2A 是 Agent 之间通信，AG-UI 负责前后端交互。介绍文章看了不少，真正把它们放到同一个项目里时，还是会遇到很多具体问题：请求从哪里进来，工具结果怎么回给模型，多个 Agent 怎么传递上下文，有副作用的动作怎样等待用户确认，用户中途取消以后后台任务要不要继续跑，以及工具越来越多以后，是否应该全部暴露给模型。

为了把这些问题串起来，我写了一个 AgentMesh Demo。业务入口仍然很简单：查食堂菜单、价格和库存，获取推荐并完成一次订餐。后来又加入了天气 Agent、跨会话用户偏好、HITL 确认、订单与审计、Skill 动态路由和一个最小 Skill 市场。

场景虽然小，但现在已经可以把 AG-UI、AgentScope、MCP、A2A 和一套完整的运行状态机接在一起。这篇文章不准备逐个解释协议，而是沿着一次请求看一遍：拿到一个大模型接口之后，怎样把它做成一个能交互、会调用工具、能委派任务、可以暂停确认，也能够持久化和审计的应用。

为什么还是一个点餐 Demo

一开始最容易做的是聊天页面：输入一句话，请求模型，然后把结果显示出来。

但这种形式很难把 Agent 和普通聊天的区别表现出来。为了让系统里真的出现工具调用和任务委派，场景至少需要三类能力：

  • 确定性查询，比如查价格、库存和某天菜单；
  • 需要理解意图的领域任务，比如推荐菜品、组织订餐请求；
  • 带副作用的动作，比如确认订单和扣减库存。

点餐正好同时覆盖这三类问题。

菜品价格和库存存在 SQLite 里，查出来是多少就是多少，不应该让模型猜。菜单推荐和订餐可以交给独立的 Food Agent；天气问题则交给 Weather Agent。订餐动作在真正执行前还要暂停 Run，让用户看到参数并确认。

项目目前会启动六个进程：

1
2
3
4
5
6
Web Frontend        输入、SSE 流式输出、工具状态、确认卡片、订单和 Skill 市场
Orchestrator        会话、Run 状态机、模型调用、Skill 路由、HITL 和工具编排
MCP Server          菜品目录查询、按星期列菜单、库存扣减
A2A Gateway         Agent 注册、鉴权、路由和请求转发
Food Agent          菜单、推荐和订餐领域处理
Weather Agent       当前天气、未来预报和穿衣建议

Skill Marketplace、Capability Registry 和 SQLite 持久化都在 Orchestrator 内部，不是额外进程。大致链路如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
用户
  │ RunAgentInput / SSE
  ▼
AG-UI Frontend (:3000)
  │
  ▼
AgentScope Orchestrator (:8000)
  ├─ Orchestrator SQLite：Run / Message / Order / Audit / Skill
  └─ SessionManager：会话锁 + 每消息路由
       └─ Skill Marketplace → SkillRouter → CapabilityRegistry
            └─ AgentScope Agent / ReAct
                 ├─ Local：时间、用户记忆
                 ├─ MCP：Catalog MCP Server (:7100) → Food Catalog SQLite
                 └─ A2A：Gateway (:6001)
                      ├─ Food Agent (:7004) ──MCP 查询菜单──→ Catalog MCP Server
                      └─ Weather Agent (:7005)

如果按职责而不是按进程看，这套 Demo 可以拆成五层：

层次 主要职责 项目中的实现
交互层 把一次长运行呈现给用户 AG-UI、SSE、前端确认卡片
推理层 决定直接回答还是调用工具 AgentScope ReAct Agent、LLM
能力控制层 决定本轮允许看到哪些能力 Skill Marketplace、SkillRouter、Capability Registry
执行层 真正读取数据或委派任务 Local Function、MCP Server、A2A Agent
状态与治理层 保存事实并约束副作用 Run 状态机、SQLite、HITL、审计、幂等

这个分层对我理解项目很有帮助。AG-UI、MCP、A2A 解决的是不同边界上的通信问题；AgentScope 负责模型推理循环；Skill 和 Capability 决定能力怎样进入模型工作台；Run、确认记录、订单和审计则保证模型之外的应用状态不会失控。

这里仍然没有消息队列、注册中心或者复杂的分布式基础设施。项目的目标不是展示一套生产架构，而是尽量少引入别的东西，把 Agent 调用链以及它周围必须处理的软件工程问题暴露出来。

先把项目跑起来

项目需要 Python 3.11、Node.js 20.19+ 和 npm。本文对应仓库 b1d3743（2026-08-29）这一版实现；后续如果代码继续演进，可以先用 git log -1 确认版本，再对照文中的目录和状态机。

1
2
3
git clone https://github.com/jiankunking/agentmesh-demo.git
cd agentmesh-demo
cp .env.example .env

在 .env 中配置模型：

1
2
3
LLM_API_KEY=your-api-key
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini

然后执行：

1
./scripts/start.sh

脚本会创建或同步 Python 虚拟环境、安装前端依赖、构建前端，并按依赖顺序启动六个组件。浏览器访问 http://127.0.0.1:3000 就可以开始测试，Orchestrator 的 Swagger 地址是 http://127.0.0.1:8000/docs。

可以用下面两个脚本查看状态和停止服务：

1
2
./scripts/status.sh
./scripts/stop.sh

第一次打开页面后，我建议不要马上翻所有代码，可以先试几类问题：

1
2
3
4
5
6
7
8
9
你是谁？
现在几点？
今天食堂有什么菜？
水煮牛肉多少钱，还有库存吗？
上海未来三天天气怎么样？
记住我不吃辣。
结合我的偏好推荐一道菜。
今天上海天气怎么样，食堂有什么适合我的菜？
帮我订一份水煮牛肉。

这些问题看起来都从同一个输入框进入，后台经过的路径却不同：

  • “你是谁”通常由模型直接回答；
  • “现在几点”调用 Local 工具；
  • 菜品价格、库存和按星期菜单走 MCP；
  • 推荐和订餐通过 A2A 调用 Food Agent；
  • 天气通过 A2A 调用 Weather Agent；
  • “记住我不吃辣”写入用户长期记忆；
  • 跨领域问题可能触发多个工具并行执行；
  • 订餐会让 Run 暂停，等待前端确认或拒绝。

前端侧边栏还可以看到最近订单和 Skill 市场。可以尝试停用天气 Skill，再发送天气问题，观察每条消息可见工具集的变化。

一次请求是怎么跑完的

以这个问题为例：

今天上海天气怎么样，食堂有什么适合我的菜？如果有水煮牛肉，再帮我订一份。

这句话同时涉及天气、菜单、用户偏好和订餐。理想情况下，系统需要调用 Weather Agent、MCP 菜品目录、长期记忆和 Food Agent；真正执行订餐前还要等待用户确认。

前端发起一次 Run

普通聊天接口经常是一问一答：前端提交文本，后端返回文本。

Agent 应用不太一样。模型可能先输出几句话，然后调用一个或多个工具；工具执行完成以后，模型继续输出；中间可能等待用户确认，也可能失败或被取消。因此，前端面对的是一次持续变化的运行，而不只是一个字符串响应。

项目中的前端使用 AG-UI 客户端发送 RunAgentInput，Orchestrator 通过 SSE 返回事件。页面同时消费两类消息：

类型 代表事件 用途
AG-UI 标准事件 RUN_STARTED、TEXT_MESSAGE_*、TOOL_CALL_*、RUN_FINISHED、RUN_ERROR 驱动文本、工具过程和 Run 生命周期
项目自定义事件 tool_confirmation_required、tool_confirmation_resolved 驱动 HITL 确认卡片

订单列表和 Skill 市场不属于 Agent 事件流，仍然通过普通 REST API 读取和修改。把“流式运行状态”和“可查询业务资源”分开以后，前端状态会清楚很多。

这也是我接入 AG-UI 后比较明显的一个认识：流式交互不只是把模型 token 一个个推到页面上。工具调用有开始和结束，整个任务有生命周期，有副作用的工具中间还会出现一次“暂停”。如果前端只处理文本流，模型一旦开始调用工具，用户看到的往往就是页面突然停住。

前端入口比较集中，主要代码在：

1
2
frontend/src/main.ts
frontend/src/styles.css

现在这两个文件除了聊天流，还包含确认卡片、取消操作、订单列表和 Skill 市场界面。想知道浏览器如何把协议事件转换成用户可见状态，从这里开始比较直观。

为什么确认事件需要一条“旁路”

HITL 有一个容易忽略的并发问题：确认发生在工具执行中，而工具中间件会 await 用户决定。此时正常的 AgentScope 事件流正停在工具调用位置；如果“需要确认”这个通知也依赖同一条被阻塞的流，前端永远看不到确认卡片，系统就会互相等待。

项目在 RunService.stream_run() 中为每次 Run 创建一个 asyncio.Queue，把确认事件作为 out-of-band 事件写入队列，再由 _interleaved_stream() 同时等待 AgentScope 事件和旁路事件：

1
2
3
AgentScope event stream ─┐
                         ├─ _interleaved_stream() ─→ SSE ─→ Frontend
HITL out-of-band queue ──┘

这样工具协程可以继续等待确认，而 SSE 协程仍然能先把 tool_confirmation_required 推到浏览器。这个细节看起来只是一个队列，实际上解决的是“控制事件不能被数据执行路径堵住”的问题。

Orchestrator 不只是转发模型请求

请求进入 Orchestrator 后，需要先建立这次运行的持久化上下文。

项目把 Conversation、Message 和 Run 分开保存：

  • Conversation 对应一个持续存在的会话；
  • Message 是用户或 assistant 的一条消息；
  • Run 是处理某次用户输入的执行过程。

一个会话中会有多次 Run。当前状态机是：

1
2
3
4
5
PENDING → RUNNING → SUCCEEDED
                  → FAILED
                  → CANCELLED
                  → AWAITING_CONFIRMATION → RUNNING（确认后执行或拒绝后跳过工具）
                                          → CANCELLED（用户取消）

AWAITING_CONFIRMATION 是后来加入的关键中间状态。它表示 Run 并没有结束，只是某个工具在等待用户决定。

这种拆分在刚做聊天 Demo 时可能显得有些多余，但加入取消、确认和失败处理以后就很有用。例如模型已经输出一部分内容，随后 MCP 调用超时；这时对话仍然存在，但本次 Run 失败了。那段未完成内容可以保留在执行记录里，却不应该作为成功消息进入下一轮模型上下文。

Orchestrator 收到请求后，大致会做这些事：

  1. 校验用户、会话、消息和 Run 标识；
  2. 创建或读取 Conversation；
  3. 保存用户消息并创建 PENDING Run；
  4. SSE 真正开始执行时把 Run 更新为 RUNNING；
  5. 按 (userId, threadId) 获取会话 Agent 和会话锁；
  6. 读取该用户当前安装且启用的 Skill；
  7. 根据本轮消息构建 Toolkit 并注入 Agent；
  8. 恢复此前成功完成的历史消息；
  9. 调用 AgentScope 开始流式推理；
  10. 根据工具、确认和最终结果更新 Run、assistant 消息、订单及审计记录。

这一段代码主要分布在：

1
2
3
4
services/orchestrator/routes.py
services/orchestrator/run_service.py
services/orchestrator/persistence.py
services/orchestrator/session.py

routes.py 是 HTTP 入口，真正的运行逻辑更多在 run_service.py。如果只看路由，很容易误以为 Orchestrator 只是做了一层协议转换。

在模型选工具之前，先由 Skill 缩小工具集

旧版本会把所有启用工具都放进 Toolkit。工具少时问题不大，数量增加后会出现两个问题：提示词变长，职责相近的工具也更容易被模型选错。

当前实现增加了一层 Skill：

1
2
3
4
5
6
7
8
9
10
11
用户本轮消息
    ↓
读取用户已安装且启用的 Skill
    ↓
SkillRouter 按关键词选择 Skill
    ↓
收集这些 Skill 引用的 Capability
    ↓
构建本轮 Toolkit
    ↓
交给 AgentScope / LLM

内置 Manifest 位于：

1
2
3
4
5
services/orchestrator/skill_manifests/
├── 00-basics.json
├── 10-food-service.json
├── 20-weather.json
└── 30-memory.json

当前有四组 Skill：

Skill 主要能力 路由示例
basics 当前时间 始终启用
food_service 菜品搜索、菜单、Food Agent 菜单、库存、推荐、订餐
weather Weather Agent 天气、气温、预报、穿衣
memory 保存和召回用户偏好 记住、喜欢、不吃、推荐

Skill 不是新的远程调用协议。它是 Orchestrator 内部对 Capability 的分组、授权和路由机制。真正执行时，底层仍然是 Local、MCP 或 A2A。

目前路由是关键词匹配。basics 是系统级、始终可用的 Skill；如果没有命中其他业务 Skill，当前实现会回退到用户全部已启用 Skill，避免路由规则遗漏后让 Agent 完全失去能力。这个回退提高了可用性，但也意味着它不是严格的最小权限策略：例如一条没有命中关键词的消息，可能重新看到全部已启用能力。关键词路由只是第一阶段，后续可以继续尝试 Embedding、小模型分类，或者直接把“未命中”改成显式拒绝与澄清。

代码里真正完成动态替换的动作很直接：

1
2
3
skill_router = self._get_skill_router(user_id)
routed_toolkit = skill_router.build_toolkit_for_message(user_message)
session.agent.toolkit = routed_toolkit

关键不在这三行本身，而在于它发生在每条消息进入 ReAct 循环之前，所以用户刚刚停用一个 Skill，下一条消息就会使用新的 Toolkit，不需要重建整个服务。

Capability 是协议之上的统一契约

Local、MCP 和 A2A 的调用方式不同，但在进入 AgentScope 之前都会先变成 CapabilityDefinition。这个对象把执行函数和治理元数据放在一起：

1
2
3
4
5
6
7
8
9
10
@dataclass(frozen=True, slots=True)
class CapabilityDefinition:
    name: str
    capability_type: CapabilityType  # LOCAL / MCP / A2A
    description: str
    handler: Callable
    enabled: bool = True
    read_only: bool = True
    requires_confirmation: bool = False
    timeout_seconds: float = 30.0

这样 Skill 只需要引用 Capability 名称，不需要关心底层是进程内函数、MCP 工具还是远程 Agent；HITL、审计和能力清单也有了统一的元数据入口。需要注意的是，read_only=false 不等于一定要确认：remember_preference 会写数据库，但当前不要求 HITL；是否确认还要结合动作风险，而不是只看有没有写操作。

模型当前可以调用哪些工具

默认配置会注册并启用八个面向模型的真实工具：

1
2
3
4
5
6
7
8
get_current_time        Local，查询服务器当前时间
search_food_catalog     MCP，按名称、分类或描述搜索菜品
get_food_item           MCP，按稳定 ID 查询价格和库存
list_menu_by_day        MCP，按星期获取当天菜单
call_food               A2A，调用 Food Agent，需要用户确认
call_weather            A2A，调用 Weather Agent，只读
remember_preference     Local，保存用户偏好
recall_preferences      Local，搜索用户历史偏好

此外还保留了模拟 web_search 和 execute_python，但默认不暴露，需要显式打开 DEMO_TOOLS_ENABLED。

MCP Server 还有一个 deduct_stock 写工具。它不会直接暴露给 LLM，而是在用户确认订单、订单数据成功持久化后由 Orchestrator 调用。这种设计把“模型决定业务意图”和“应用执行受控写操作”分开了。

模型拿到本轮 Toolkit 后，会选择直接回答，或者返回 tool_calls。工具结果再进入下一轮模型请求，直到生成最终答案。AgentScope 在这里负责 ReAct 循环和流式事件；当模型一次返回多个工具调用时，也可以并发执行彼此独立的工具。

代码中设置了最大迭代次数，避免模型不断调用工具却无法结束。不过限制迭代次数只是最后一道保护，更关键的仍然是工具边界、描述和 Skill 路由。如果两个工具职责重叠，模型很容易在它们之间选错。

MCP 在这里负责什么

“水煮牛肉还有几份”是一个确定性问题。

库存存放在 SQLite 里，模型不应该凭训练数据或者对话上下文回答。Orchestrator 会调用 MCP 工具，MCP Server 再查询菜品目录数据库。

1
2
3
4
5
Orchestrator
    ↓ search_food_catalog / get_food_item / list_menu_by_day
MCP Server
    ↓ SQL
Food Catalog SQLite

这个调用完全可以直接写成一个 HTTP API。之所以使用 MCP，是为了看清楚工具能力如何被声明、发现和调用，以及同一套工具接口如何被不同的 AI Host 使用。

在这个 Demo 里，MCP 并不承担推理。它接收明确参数，执行查询或库存事务，然后返回结构化结果。模型负责判断什么时候查，MCP Server 负责返回数据库中的真实数据。

相关代码在：

1
2
3
services/orchestrator/mcp_client.py
services/mcp_server/main.py
services/mcp_server/catalog.py

可以重点观察四件事：

  • Orchestrator 如何通过 Streamable HTTP 连接 MCP Server；
  • MCP 工具如何转换成模型可以调用的 Capability；
  • traceId、runId 和 toolInvocationId 如何传到 MCP 请求；
  • 订单确认后，应用如何调用未暴露给模型的 deduct_stock 完成受控写操作。

第一次启动 MCP Server 时，会根据配置创建菜品目录 SQLite。可以直接调用工具查询 food-001，验证返回的价格和库存，而不必每次都经过模型。

A2A 在这里负责什么

“帮我推荐一道菜”不只是拿到一行数据库记录，还需要组合菜单、日期和用户请求。项目把这部分能力拆成独立的 Food Agent，再由 Orchestrator 通过 A2A 委派。

这里需要特别澄清：当前 Food Agent 和 Weather Agent 都不是各自再调用一次大模型的 LLM Agent。 它们是实现了 Agent Card、message/send / message/stream 和任务响应格式的独立 A2A 服务，内部主要使用规则和确定性代码。整个项目里真正执行 ReAct 推理的是 Orchestrator 中的 AgentScope Agent。

这并不影响 A2A 的演示价值。A2A 解决的是“怎样发现、调用和隔离一个远程任务执行者”，被调用方可以是 LLM Agent、工作流、规则引擎，甚至是旧业务系统的适配层。先用确定性子 Agent，反而更容易单独观察协议、路由和状态传播；以后要把 Food Agent 替换成拥有独立模型和工具的 Agent，Orchestrator 侧的委派边界可以保持不变。

现在 A2A Gateway 后面有两个 Agent：

1
2
food       菜单、推荐和订餐
weather    当前天气、未来预报和穿衣建议

Orchestrator 调用 call_food 或 call_weather 时，会构造 A2A JSON-RPC 请求，经 Gateway 转发给对应 Agent。请求里除了用户文本，还会携带：

1
2
3
4
5
6
messageId
contextId
traceId
runId
threadId
toolInvocationId

contextId 用来保持 A2A 侧上下文，其余标识主要用于关联一次请求在多个服务中的日志。

A2A Gateway 当前负责校验请求、检查 API Key、从静态注册表找到 Agent 地址，然后根据 message/send 或 message/stream 转发请求。Food Agent 处理菜单、推荐或订餐时，还会继续调用 MCP Server 获取当天真实菜单，因此调用链可能出现一层嵌套：

1
2
3
4
5
6
7
Orchestrator / LLM
    ↓ call_food（A2A）
A2A Gateway
    ↓ message/send
Food Agent
    ↓ list_menu_by_day（MCP）
MCP Server → Food Catalog SQLite

这说明 MCP 和 A2A 并不是二选一：Orchestrator 可以把任务委派给另一个 Agent，而那个 Agent 仍然可以使用 MCP 工具完成自己的工作。相关代码在：

1
2
3
services/a2a_gateway/main.py
services/food_agent/main.py
services/weather_agent/main.py

这个 Gateway 仍然是轻量实现，只覆盖 Demo 所需的 A2A 子集，并不是完整的 Agent 平台。Agent Registry 是代码与环境变量配置，不支持后台动态上架、版本管理、能力协商或租户隔离。不过即使只是这样，也能把 Orchestrator 和领域 Agent 的地址、鉴权及转发逻辑分开。

天气数据目前是十个城市的模拟数据，这一点很重要：Weather Agent 的调用链是真实的 A2A 调用，但它背后的数据源不是线上天气服务。

Skill、MCP 和 A2A 不是同一层东西

项目加入 Skill 以后，更容易把几个概念混在一起。

概念 解决的问题 项目中的例子
Skill 本轮应该给模型哪些能力，以及用户是否安装/启用了这些能力 food_service、weather、memory
MCP 如何标准化连接确定性工具或数据源 菜品搜索、价格、库存、按星期菜单
Agent 能独立处理领域任务的执行者 Food Agent、Weather Agent
A2A Agent 之间如何传递任务和结果 Orchestrator 经 Gateway 委派给子 Agent

Food Agent 技术上也可以包成一个 MCP 工具，菜品查询也可以再套一层 Agent，但这样会让边界变得模糊。

我现在主要看两点：

第一，被调用方是否需要独立推理或领域上下文。查询数据库有明确输入输出，不需要再调用一次模型，适合做工具；推荐菜品、处理复杂订餐意图，可能有自己的提示词、上下文和工具，更适合作为 Agent。

第二，这项能力是否需要独立演进。如果它有单独的模型、权限、发布节奏和领域逻辑，拆成 Agent 会更自然。如果只是一个函数，就没必要为了使用 A2A 再增加一个服务。

可以把 MCP 理解成“使用工具”，A2A 理解成“找另一个人协作”，Skill 则是在出发前决定“这次把哪些工具和协作者放进工作台”。这个类比不够严谨，但做架构选择时比较直观。

真正花时间的不是把协议接通

AG-UI、MCP 和 A2A 的基本链路跑通以后，项目其实还不能算完整。后面花时间更多的是一些看起来不太“Agent”的问题。

同一个会话不能随便并发

如果用户在上一条消息还没有处理完时又发了一条，两次推理可能同时读写上下文。

项目按 (userId, threadId) 创建 Agent 和 asyncio.Lock。同一个会话串行执行，不同会话可以并发。AG-UI 中间件、工具调用上下文和 HITL 确认处理器则使用 contextvars 隔离，避免多个并发 SSE 请求共享中间变量。

这里还要区分两个层次：

  • 同一个会话的不同 Run 串行，保证历史顺序；
  • 同一个 Run 内，如果模型返回多个独立 tool_calls，AgentScope 可以并发执行。

因此，“上海天气怎么样，今天食堂吃什么”可以并行调用 Weather Agent 和菜单 MCP；如果其中某个工具需要确认，也只应该阻塞对应工具，而不是污染其他并行任务的上下文。

失败内容不能直接进入下一轮对话

假设模型已经输出一半，工具调用突然失败。如果把这段不完整文本加入历史，下一轮模型可能会把它当成已经完成的回答。

项目只把成功完成的消息恢复到 AgentScope 上下文。失败和取消的 Run 仍然有记录，但不会污染后续推理。这里需要区分“为了排查而保存”和“适合作为模型上下文”是两件事。

用户离开以后，任务也应该停下来

流式请求中，用户可能点击取消，也可能直接关闭页面。

项目通过 ActiveRunRegistry 保存 runId 到 asyncio.Task 的关系。收到取消请求后，会记录取消时间并调用 Task.cancel()，让取消沿当前 await 链传播。

这个实现目前只适用于单进程。多实例部署以后，任务可能运行在另一个实例，需要共享任务系统或单独的取消信号通道，不能继续依赖内存映射。

服务重启后，状态不能撒谎

Orchestrator 重启以后，内存中的模型、工具任务和确认信号已经消失，但 SQLite 里可能还留着 PENDING、RUNNING 或 AWAITING_CONFIRMATION。

项目启动时会把这些遗留 Run 标记为 FAILED，同时让过期确认记录进入终态。它不会假装任务仍在运行，也不会自动恢复到中断位置。

自动续跑当然更理想，但需要可持久化工作流、幂等工具和更完整的恢复机制。对这个 Demo 来说，先保证状态真实更重要。

日志需要能串起来

一次用户请求可能调用多轮模型、多个工具，还可能经过 A2A Gateway、Food Agent、Weather Agent 和 MCP Server。只有一个 traceId 还不够，因为它无法区分同一次 Run 中的多轮操作。

项目使用了几类标识：

1
2
3
4
5
6
7
traceId             整条调用链
runId               一次 Agent 运行
threadId            一个会话
llmCallId           某一轮模型请求
toolInvocationId    某一次工具调用
rpcId               一次 A2A JSON-RPC 请求
contextId           A2A 会话上下文

所有组件统一输出 [FLOW] 摘要，相关 Header 会从 Orchestrator 传到 A2A Gateway、领域 Agent 和 MCP Server。scripts/trace.sh 可以按 traceId 聚合日志。

有了这些标识，才比较容易回答：到底是模型慢、MCP 慢、确认还没完成，还是某个 Agent 根本没有收到请求。

有副作用的工具怎样等待用户确认

原文最早写到这里时，readOnly 和 requiresConfirmation 还只是注册表中的预留字段。当前版本已经把 HITL 闭环做完了。

call_food 被标记为：

1
2
readOnly = false
requiresConfirmation = true

当 AgentScope 准备执行这个工具时，Orchestrator 不会直接放行，而是执行下面的流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
AgentScope  → Orchestrator：准备执行 call_food(args)
Orchestrator→ Orchestrator：保存 confirmation_request
Orchestrator→ Orchestrator：RUNNING → AWAITING_CONFIRMATION
Orchestrator→ Frontend：通过 SSE 发送 tool_confirmation_required
Frontend    → 用户：展示工具名和参数

用户确认：
用户        → Frontend：确认执行
Frontend    → Orchestrator：POST /runs/{runId}/confirm
Orchestrator→ Orchestrator：AWAITING_CONFIRMATION → RUNNING
Orchestrator→ Food Agent：经 A2A 执行 call_food
Food Agent  → Orchestrator：返回文本结果 + ORDER_DATA

用户拒绝：
用户        → Frontend：拒绝执行
Frontend    → Orchestrator：POST /runs/{runId}/reject
Orchestrator→ AgentScope：返回 ToolResultBlock(DENIED)
AgentScope  → Frontend：告知用户操作未执行

确认请求默认 300 秒后过期。用户确认后，原来的 Run 从暂停位置继续；拒绝后，工具不会执行，但 Agent 会收到一个明确的 DENIED 结果，而不是把拒绝误判成系统异常。

Food Agent 完成订餐后会返回一段结构化 ORDER_DATA。Orchestrator 提取数据并写入 orders 表，以 toolInvocationId 作为幂等键，避免同一次工具调用被重复持久化；之后再通过 MCP 的 deduct_stock 扣减库存。deduct_stock 不出现在模型 Toolkit 中，模型只能表达“我要下单”，真正的库存写操作仍由应用代码控制。

当前实现是“先持久化订单，再逐项 best-effort 扣库存”；某一项扣减失败只记录警告，不会回滚订单。这对 Demo 足够直观，但不是生产级事务。真实系统需要明确订单状态、库存预占、失败补偿和幂等重试，或者通过 Saga / Outbox 把跨服务一致性显式建模。

所有 Local、MCP 和 A2A 工具调用还会写入 audit_log，记录输入/输出摘要、耗时和 SUCCESS、FAILED、CANCELLED、DENIED 等状态。可以通过下面的接口查看：

1
2
3
4
GET /api/v1/orders
GET /api/v1/orders/{orderId}
GET /api/v1/audit
GET /api/v1/audit?runId={runId}

这里说的“订单”是 Demo 自己 SQLite 中的持久化业务记录，不是接入支付或外部订单中心的真实交易。

当前确认粒度仍然是整个 call_food Capability。因为它同时承担推荐和订餐，某些只读的 Food Agent 请求也可能被要求确认。更合理的生产设计是进一步拆分只读的 recommend_food 和有副作用的 place_order，只让后者进入 HITL。

长期记忆不是把所有历史都塞进 Prompt

项目新增了两个 Local 工具：

1
2
remember_preference
recall_preferences

用户说“记住我不吃辣”时，Agent 可以把偏好写入 user_preferences 表；之后在另一个会话中询问推荐，Agent 可以通过 SQLite FTS5 搜索相关偏好。

这和把全部聊天历史一直塞进模型上下文不同。会话历史解决当前对话连贯性，长期记忆保存跨会话仍然有价值的结构化信息。两者都要有边界，否则不仅 Prompt 越来越长，也很难让用户知道系统究竟记住了什么。

当前 Demo 的记忆能力仍然很简单：没有记忆合并、冲突解决、遗忘策略、敏感信息分类和用户可视化管理。它更像是在展示“长期记忆应该作为显式工具和数据模型存在”，而不是把它藏在一个无限增长的 Prompt 里。

Skill 市场解决的是安装、授权和启停

只有 SkillRouter 还不够。不同用户可能希望启用不同能力，Skill 本身也需要版本、发布者、权限和说明。

当前版本增加了一个 Skill Marketplace MVP。每个 Skill 使用 JSON Manifest 描述：

1
2
3
4
5
6
7
id / name / version / publisher / category
routing.keywords / alwaysOn
capabilities
providerTypes
permissions
instructions
defaultInstalled / system

SQLite 的 skill_installations 表保存用户级安装、启用状态和已授予权限。Orchestrator 在每条消息路由前重新读取，因此安装、停用、启用或卸载后无需重启。

不过这里的 permissions 目前主要用于 Manifest 声明、安装时完整性校验和界面展示，还不是一个独立的运行时策略引擎。真正暴露能力时，核心判断仍是 Skill 是否安装/启用、Capability 是否启用。生产环境如果需要细粒度授权，还要增加基于用户、租户、资源和参数的策略检查，不能把“Manifest 里写了权限”直接等同于“权限已经被强制执行”。

相关 API 包括：

1
2
3
4
5
6
GET    /api/v1/skill-marketplace
GET    /api/v1/skill-marketplace/{skillId}
POST   /api/v1/skill-marketplace/{skillId}/install
POST   /api/v1/skill-marketplace/{skillId}/enable
POST   /api/v1/skill-marketplace/{skillId}/disable
DELETE /api/v1/skill-marketplace/{skillId}

builtin.basics 是系统 Skill，不能停用或卸载，保证 Agent 至少保留基础能力。

这里有一个我认为比较重要的安全边界：当前市场 Skill 只能引用 Orchestrator 已经注册的 Local、MCP 或 A2A Capability，不会下载并在主进程中执行第三方任意代码。因此它更接近“能力目录 + 用户授权 + 动态路由”，还不是一个可以上传代码包并自动部署 Provider 的开放市场。

如果以后真的支持第三方 Skill，需要继续补签名、审核、版本约束、依赖校验、Provider 健康检查、撤回机制，以及容器或 WASM 级别的隔离，不能把远程代码直接 import 到 Orchestrator。

如果从头读代码，我会按这个顺序

第一步先看前端和 HTTP 路由：

1
2
frontend/src/main.ts
services/orchestrator/routes.py

先确认请求格式、SSE 响应、取消、确认、订单和 Skill 市场 API，不要急着钻进模型细节。

第二步看 Run 怎么执行：

1
2
3
4
5
services/orchestrator/run_service.py
services/orchestrator/run_control.py
services/orchestrator/session.py
services/orchestrator/persistence.py
services/orchestrator/models.py

这一部分可以看到会话锁、历史恢复、状态迁移、取消、确认等待、订单和审计持久化。

第三步看 Skill 如何决定本轮工具集：

1
2
3
services/orchestrator/skills.py
services/orchestrator/skill_marketplace.py
services/orchestrator/skill_manifests/*.json

先理解用户安装状态、Manifest 和关键词路由，再看 Toolkit 是怎样按消息动态替换的。

第四步看能力注册和工具执行：

1
2
3
4
5
6
services/orchestrator/capabilities.py
services/orchestrator/tool_registry.py
services/orchestrator/tool_impls.py
services/orchestrator/tool_trace_middleware.py
services/orchestrator/confirmation.py
services/orchestrator/mcp_client.py

这里重点对比 search_food_catalog、call_weather 和 call_food：一个走 MCP，一个走只读 A2A，一个走需要 HITL 的 A2A，但最终都会以工具形式进入 AgentScope。

最后再看下游服务：

1
2
3
4
5
services/mcp_server/main.py
services/mcp_server/catalog.py
services/a2a_gateway/main.py
services/food_agent/main.py
services/weather_agent/main.py

如果想确认自己对代码的理解是否正确，可以直接看测试。当前代码覆盖 AG-UI 并发隔离、Capability Registry、MCP、Food Agent、A2A Gateway、Run 状态机、持久化、HITL、订单、并行编排、Skill 路由和 Skill 市场。可以在项目创建的 Python 3.11 虚拟环境中运行后端测试，再执行前端 TypeScript/Vite 生产构建：

1
2
.venv/bin/python -m unittest discover -s tests -v
npm run build --prefix frontend

可以继续做的几个实验

把项目跑起来只是第一步。如果想用它继续熟悉 Agent 开发，我觉得下面这些改动比继续看概念文章更有用。

拆分推荐和下单能力

把当前 call_food 拆成只读的 recommend_food 和有副作用的 place_order。观察 Capability 元数据、Skill Manifest、系统提示、HITL 和审计如何一起变化。

把关键词 SkillRouter 换成语义路由

可以先用 Embedding 做 Top-K，再尝试一个小模型做意图分类。重点比较误路由、延迟、成本和回退策略，而不只是看“能不能匹配”。

实现 Skill 组合链

例如把“按偏好推荐午餐”显式拆成：

1
recall_preferences → list_menu_by_day → filter → call_food

这样 Skill 就不只是工具分组，而会开始接近可持久化 Workflow。

接入真实天气或订单系统

把模拟天气替换成真实 API，或者把 Demo 订单发送到外部测试系统。外部依赖加入后，重试、限流、幂等、密钥和错误分类都会变得更具体。

人为制造失败

把 MCP 地址改错、让 Food Agent 超时、在确认期间重启 Orchestrator，或者在请求过程中停止 Gateway。观察 Run 状态、前端提示、确认记录和审计日志是否一致。

正常路径只能证明功能能跑，失败路径更容易暴露状态设计的问题。

尝试多实例

当前取消、会话锁和 HITL 信号依赖单个 Orchestrator 进程。启动多个实例后，可以尝试用 Redis 或任务队列实现分布式锁、会话共享和取消/确认信号广播。

目前没有做的事情

这个项目主要用来观察协议和状态如何配合，因此仍然有不少地方有意保持简单：

  • 菜品目录、Run、订单、审计、用户记忆和 Skill 安装状态都使用 SQLite；
  • 天气数据是本地模拟数据，不是真实天气 API；
  • Capability Registry 仍是代码定义和环境变量开关，修改后需要重启；
  • Skill Manifest 来自项目内 JSON，没有第三方上传、审核、签名和远程包部署；
  • Skill 权限目前是声明与安装校验元数据，还没有独立的运行时策略执行点；
  • SkillRouter 使用关键词匹配，未命中时会回退到全部已启用 Skill，还不是严格的最小权限路由；
  • call_food 的确认粒度较粗，推荐和订餐尚未拆成不同能力；
  • Food Agent 和 Weather Agent 目前是规则驱动的 A2A 服务，不是独立 LLM 推理 Agent；
  • 订单持久化与库存扣减不是原子事务，库存失败只记录日志，没有补偿流程；
  • 取消、会话锁和确认信号只支持单个 Orchestrator 进程；
  • 服务重启后会把中断任务标记为失败，不会自动续跑；
  • 没有正式身份系统、租户隔离、RBAC 和细粒度数据权限；
  • 模拟搜索和 Python 工具默认关闭，不能当作生产能力。

项目也没有覆盖 RAG、模型训练、微调、系统化评测和推理优化。把这些内容全部塞进一个 Demo，项目反而会失去重点。

如果准备对外部署，API Key、CORS、HTTPS、限流、密钥管理、日志脱敏、工具权限和数据保留策略都需要重新检查。README 和 .env.example 中的默认配置主要面向本机实验。

和 Hermes 对照：核心运行骨架其实一样

把前面的调用链、状态和能力边界都展开以后，回过头再对照 Hermes 的核心对话流程，会发现两套系统虽然组件名称和工程边界不同，核心运行骨架其实是一样的：接收用户输入，组装 Prompt 和记忆，进入 LLM 与工具交替执行的迭代循环，最后返回答案并持久化状态。

Hermes 核心对话流程（重绘）

这里要区分两种保护：迭代上限用于防止 Agent 无限调用工具，达到上限后应安全终止；上下文上限约束下一轮送入模型的 token 数，接近阈值时才进行上下文压缩。上下文压缩不能解决迭代次数耗尽的问题。

把图中的节点映射到 AgentMesh Demo，关系会更直观：

Hermes 中的环节 AgentMesh Demo 中的对应实现
入口层 AG-UI Frontend 将 RunAgentInput 交给 Orchestrator，并通过 SSE 消费运行事件
Agent 引擎启动 Orchestrator 创建 Run，SessionManager 获取会话 Agent 与会话锁
组装系统提示词与预取记忆 AgentScope Agent 加载系统指令、成功会话历史、用户偏好，以及 SkillRouter 为本轮筛选出的 Toolkit
流式调用 LLM AgentScope 驱动模型流式推理，并持续产生文本、工具调用和 Run 生命周期事件
判断纯文本还是工具调用 模型直接生成最终文本，或者返回一个或多个 tool_calls
工具调度与执行 Capability Registry 将调用分发到 Local Function、MCP Server 或 A2A Agent；高风险动作先经过 HITL
工具结果写回历史 tool_result 回到 Agent 上下文，触发下一轮模型推理
迭代与上下文治理 两者都有迭代上限和上下文边界；Hermes 图中显式画出了上下文压缩，本 Demo 当前主要依靠最大迭代次数、只恢复成功历史和显式长期记忆控制上下文
返回响应与持久化 Orchestrator 通过 SSE 返回结果，并把 Conversation、Message、Run、Order、Audit 和用户偏好写入 SQLite

所以更准确地说，AgentMesh Demo 与 Hermes 相同的是 Agent Runtime 的主循环，而不是外围组件逐项相同。Hermes 把这条循环画成一个通用 Agent 内核；这个项目则把同一条循环落实到 AG-UI、AgentScope、MCP、A2A、Skill、HITL 和 SQLite 上，并额外强调协议边界、能力治理、副作用确认与可审计状态。

最后

做到现在，我对 Agent 应用的理解反而没有以前那么“模型中心”了。

模型仍然是最重要的决策节点，但一个请求能否可靠完成，还取决于前端事件、Skill 路由、工具边界、会话状态、HITL、幂等、取消传播、下游超时、长期记忆和日志关联。模型只负责其中一段，剩下的大部分仍然是熟悉的软件工程问题，只是调用链里多了不确定的模型输出。

如果刚开始接触 AgentScope、MCP 或 A2A，我不建议先把每份协议文档从头背一遍。可以先找一条具体请求，把它从页面一路跟到 SkillRouter、模型、工具、子 Agent 和数据库，再跟着 AG-UI 事件返回。之后再故意加一次取消、拒绝或超时。链路跑明白以后，再去看协议细节，会更容易知道每个字段和状态为什么存在。

AgentMesh Demo 仍然只是一个用于学习和实验的小项目，离生产系统还有不少距离。但它已经不只是“模型调用几个工具”的截图 Demo，而是一套可以观察 Run 状态、用户确认、订单幂等、审计、长期记忆和能力路由如何协作的参考实现。

项目地址：https://github.com/jiankunking/agentmesh-demo