你一定会遇到这个时刻你的 Agent 在演示环境里跑得丝滑顺畅一旦接进真实业务要么在工具调用之间反复横跳要么一次请求把上下文塞爆要么干脆在一个失败动作上无限重试。问题往往不在大模型本身而在于你的 Agent 只有“智能”没有“结构”。“Let Them”这个说法最近在 Agent 开发者圈子里被反复提及。它不是在鼓励你放手让 Agent 乱跑而是指一种新的分工方式让 Agent 做它擅长的自主推理与行动同时由开发者搭好边界、协议、工具注册表和中断机制。UMA 正是这套思路下的一种典型架构实践。本文不会绑定某个特定产品或版本而是从工程实现的角度拆解 UMA 的核心设计并用可运行的 Python 代码带你从零搭建一个具备统一消息协议、工具治理、记忆管理和中断恢复能力的 Agent 运行时。1. UMA 是什么先给一个明确判断UMA 在不同社区和技术文档里会展开为不同全称常见的解释包括 Unified Model Adapter、Unified Message Architecture、Universal Multi-Agent Architecture。不管名称怎么变落到工程层面其实是同一件事UMA 是一层连接大模型、工具、记忆和应用业务的运行时抽象。它不是一个新的模型不是一句更复杂的 Prompt也不是某个云厂商的托管产品。UMA 的核心贡献是让 Agent 在运行时拥有统一的消息格式、统一的工具调用入口、统一的状态保存机制并且在需要时允许开发者安全地中断和恢复执行。很多人在搭建 Agent 时第一反应是先把大模型 API 调通然后在 Prompt 里塞一堆工具说明。这种做法在单轮任务里完全没有问题但一旦任务需要多步推理、多次工具调用甚至要跨天恢复状态代码就会迅速腐化。典型症状包括工具调用的结果散落在各个局部变量里没有统一的数据结构。Agent 返回的 JSON 中字段名不一致解析逻辑越来越脆。没有工具超时、重试和权限校验Agent 一调用外部服务就出事。没有中断能力一旦发现 Agent 走偏只能让它继续跑完或由外部进程强制 kill。UMA 解决的就是这些问题。它把 Agent 的开发从“写 Prompt 调 API”提升到了“设计协议 构建运行时”的工程化层次。维度普通 LLM 调用手写 Agent 循环UMA 分层输入一段 Prompt自由拼接消息统一消息类型工具调用无直接在循环里写死工具注册中心状态管理无状态局部变量持久化上下文中断恢复不支持不支持支持暂停、恢复可观测性依赖日志零散打印统一消息流追踪生产治理弱弱超时、重试、权限、额度这里的核心判断是Agent 能不能在真实业务里落地取决于运行时设计而不是模型参数。UMA 的价值正在于把大模型的“不可控智能”装进一套“可控工程外壳”里。2. 构建 Agent 时最容易踩的三个坑在展开 UMA 的架构之前我们先对齐三个常见的失败模式。理解了这些坑你就能明白为什么 UMA 要设计成后面那个样子。2.1 把智能全部押在 Prompt 上很多人认为 Agent 能力弱是因为 Prompt 写得不够好。于是不断往系统提示词里追加规则“你必须仔细思考”“你要一步一步分析”“工具调用失败后要重试不超过三次”。这确实能改善一部分行为但 Prompt 无法保证结构。模型可能在某一次响应里忘了遵循规则可能把工具参数拼错格式也可能在多次调用后上下文过长把早期的关键信息冲掉。结构性的问题必须用结构性的方案解决。超时、重试、参数校验、状态持久化这些不应该写在 Prompt 里而应该在运行时强制完成。2.2 让 Agent 裸奔没有工具治理Agent 要完成任务几乎必然要调用外部工具搜索引擎、数据库、订单系统、内部 API。如果你把这些工具直接暴露给大模型任何一次参数错误都可能造成真实影响。比如 Agent 调用一个删除接口因为参数解析失误把id123传成了idall。这样的问题靠模型自身很难完全规避。更好的做法是让工具调用经过一个注册中心由注册中心统一处理参数白名单、超时、重试、权限校验和结果规范化。2.3 只有“开始”没有“暂停”部分 Agent 框架设计成一次运行从入口一路执行到结束中间没有人工介入点。这在简单问答场景还可以接受但在真实业务流程里非常危险。Agent 可能会不小心确认一筆不该确认的订单或者调用了一个需要人工复核的接口。所以现代 Agent 架构特别强调 interrupt 能力执行过程中Agent 可以主动暂停等待用户确认、补充信息或修正方向开发者也可以基于规则主动打断执行把控制权交还给业务系统。这就是最近讨论里高频出现的 “deep agents interrupt” 概念。中断不是失败而是 Agent 与外部世界协作的正式接口。3. UMA 的核心模块设计UMA 作为一个运行时抽象通常包含以下六个核心模块。3.1 统一消息协议Agent 运行过程中会涉及多类消息系统指令、用户输入、模型回复、工具调用请求、工具返回结果。如果每一类消息都用不同的数据结构代码会越来越难维护。UMA 的做法是定义一套统一消息类型所有环节都通过这一种数据结构传递信息。消息中至少包含角色、内容、时间戳如果要支持工具调用还需要携带工具调用 ID 和工具名称。3.2 工具注册中心工具注册中心负责维护 Agent 可以调用的全部工具列表。每个工具注册时声明名称、描述、参数结构和处理函数。大模型看到的工具清单从这里生成运行时也通过这里完成工具调用。这个模块的价值在于把工具调用变成可插拔机制新增一个工具不需要改 Agent 主循环只要向注册中心注册即可。3.3 任务循环任务循环是 Agent 的主进程。它不断执行“读取消息 - 调用模型 - 解析输出 - 执行动作或结束”的循环直到任务完成、达到最大步数或触发中断条件。循环需要控制两个关键指标最大步数和单次工具调用超时时间。没有步数上限Agent 可能陷入死循环没有超时控制一个卡住的工具会拖垮整个任务。3.4 记忆与上下文管理大模型上下文窗口有限无法承载无限长的历史消息。UMA 需要在运行时管理上下文哪些消息需要保留哪些可以压缩哪些可以归档到外部记忆存储中。对于复杂任务推荐把长期记忆和短期上下文分开。短期上下文只保留最近几轮必要消息长期记忆则存到向量数据库或键值存储中。3.5 中断恢复机制中断恢复是 UMA 与其他简单 Agent 框架最大的区别。运行时需要支持两类中断主动中断Agent 认为需要用户确认时暂停执行并等待外部输入。被动中断开发者根据业务规则强制暂停例如检测到敏感操作、超时或成本超限。恢复执行时运行时应该从最近一个检查点继续而不是从头开始。这就要求消息列表和上下文状态可以序列化、可以持久化。3.6 可观测与追踪Agent 的调试比普通后端服务更困难因为模型输出有随机性。UMA 需要记录完整的消息流转过程每一轮模型返回了什么、选择了哪个工具、参数是什么、工具返回了什么。这些记录可以输出到日志系统也可以对接 OpenTelemetry 等追踪工具。4. 环境准备与项目结构为了让后面的示例可以顺利运行先准备环境。本文以 Python 3.10 为例不绑定特定大模型厂商。你可以使用 OpenAI SDK也可以使用兼容 OpenAI 接口的本地模型服务例如 Ollama 或 vLLM。建议创建如下项目结构uma-agent-demo/ ├── core/ │ ├── __init__.py │ ├── message.py # 统一消息定义 │ ├── registry.py # 工具注册中心 │ └── runtime.py # Agent 运行时与任务循环 ├── tools/ │ └── search_tool.py # 示例工具 ├── clients/ │ └── llm_client.py # 大模型客户端封装 ├── requirements.txt └── main.py # 程序入口最小依赖如下openai1.0.0 pydantic2.0.0如果你希望不依赖外部 API 也能跑通示例可以把llm_client.py改成本地 Mock 实现本文第 5 节会提供可替换方案。4.1 统一消息定义创建core/message.py定义统一消息类型。# core/message.py from dataclasses import dataclass, field from typing import Optional import time dataclass class UMessage: role: str content: str tool_call_id: Optional[str] None tool_name: Optional[str] None timestamp: float field(default_factorytime.time)这里定义了四种角色的消息system表示系统指令user表示用户输入assistant表示模型回复tool表示工具返回结果。工具调用相关的tool_call_id和tool_name用于把模型发起的工具调用和工具结果关联起来。4.2 工具注册中心创建core/registry.py实现工具注册和调用。# core/registry.py from dataclasses import dataclass from typing import Any, Callable, Dict, Optional import json dataclass class ToolSpec: name: str description: str handler: Callable[..., Any] parameters: dict timeout: float 10.0 class ToolRegistry: 工具注册中心 def __init__(self) - None: self._tools: Dict[str, ToolSpec] {} def register(self, spec: ToolSpec) - None: if spec.name in self._tools: raise ValueError(ftool already registered: {spec.name}) self._tools[spec.name] spec def get_schema(self) - list: tools [] for spec in self._tools.values(): tools.append({ type: function, function: { name: spec.name, description: spec.description, parameters: spec.parameters, }, }) return tools def invoke(self, name: str, arguments: dict) - str: spec self._tools.get(name) if spec is None: return json.dumps({status: error, message: funknown tool: {name}}) try: result spec.handler(**arguments) return json.dumps({status: ok, result: result}, ensure_asciiFalse) except Exception as exc: return json.dumps({status: error, message: str(exc)})工具注册中心的核心价值在于两件事一是把工具清单统一输出给大模型二是所有工具调用都经过同一个入口便于后续添加超时、重试和权限控制。5. 从零实现一个 UMA 风格 Agent5.1 大模型客户端封装创建clients/llm_client.py。这里提供一个 OpenAI 兼容的客户端以及一个用于本地演示的 MockClient。# clients/llm_client.py from typing import Any, Dict, List class UniClient: 统一的模型调用客户端适配 OpenAI 兼容接口 def __init__(self, model: str gpt-4o-mini, base_url: str | None None): try: from openai import OpenAI except ImportError: raise RuntimeError(请安装 openai 客户端pip install openai) self.client OpenAI(base_urlbase_url) self.model model def chat( self, messages: List[Dict[str, Any]], tools: List[Dict[str, Any]], ) - Dict[str, Any]: response self.client.chat.completions.create( modelself.model, messagesmessages, toolstools or None, ) message response.choices[0].message tool_calls message.tool_calls or [] if tool_calls: call tool_calls[0] if call.function: return { type: tool_call, name: call.function.name, arguments: json_loads_safe(call.function.arguments), } return { type: finish, content: message.content or , } def json_loads_safe(text: str) - dict: import json try: return json.loads(text) except Exception: return {}MockClient 不需要网络也不消耗任何额度适合在 CI 或本地环境快速验证 UMA 运行时逻辑。# clients/mock_client.py class MockClient: 本地 Mock 模型只在指定轮次调用工具其余轮次直接结束。 def __init__(self, tool_plan: list): self._plan tool_plan self._step 0 def chat(self, messages, tools): step self._step self._step 1 if step len(self._plan): plan self._plan[step] return { type: tool_call, name: plan[name], arguments: plan[arguments], } return { type: finish, content: 任务已完成。 }5.2 Agent 运行时创建core/runtime.py实现任务循环、步数控制和中断机制。# core/runtime.py from typing import Any, Dict, List, Optional from .message import UMessage from .registry import ToolRegistry class UmaRuntime: UMA Agent 运行时 def __init__( self, llm_client: Any, registry: ToolRegistry, system_prompt: str, max_steps: int 10, ): self.llm_client llm_client self.registry registry self.messages: List[UMessage] [UMessage(rolesystem, contentsystem_prompt)] self.max_steps max_steps self.current_step 0 def _to_model_messages(self) - List[Dict[str, str]]: return [{role: msg.role, content: msg.content} for msg in self.messages] def run(self, user_input: str) - str: self.messages.append(UMessage(roleuser, contentuser_input)) while self.current_step self.max_steps: self.current_step 1 response self.llm_client.chat( self._to_model_messages(), self.registry.get_schema(), ) if response[type] finish: return response.get(content, ) if response[type] tool_call: tool_name response[name] arguments response.get(arguments) or {} # 中断点敏感工具调用前挂起 if not self._before_tool_call(tool_name, arguments): return f工具调用被拦截{tool_name} tool_result self.registry.invoke(tool_name, arguments) self.messages.append(UMessage( roleassistant, content, tool_call_idtool_name, tool_nametool_name, )) self.messages.append(UMessage( roletool, contenttool_result, tool_call_idtool_name, )) return 达到最大步骤数任务未完成。 def _before_tool_call(self, tool_name: str, arguments: Dict[str, Any]) - bool: 安全钩子业务方可以在这里加入人工审批或规则拦截 return True class InterruptibleRuntime(UmaRuntime): 带中断恢复能力的运行时 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.checkpoint: Optional[Dict[str, Any]] None def save_checkpoint(self) - None: self.checkpoint { messages: [ { role: m.role, content: m.content, tool_call_id: m.tool_call_id, tool_name: m.tool_name, } for m in self.messages ], current_step: self.current_step, } def restore_checkpoint(self) - None: if self.checkpoint is None: return self.messages [ UMessage( roleitem[role], contentitem[content], tool_call_iditem[tool_call_id], tool_nameitem[tool_name], ) for item in self.checkpoint[messages] ] self.current_step self.checkpoint[current_step]这段代码有三个关键点在run的循环体里每一轮都会根据当前消息列表调用大模型并传入工具清单。如果模型返回tool_call运行时从注册中心调用对应工具并把结果追加为tool角色消息供模型在下一轮参考。_before_tool_call是一个安全钩子业务方可以实现人工审批、敏感操作拦截等逻辑。这也是 interrupt 机制最简单的落地形态在工具调用之前决定是否放行。5.3 注册一个示例工具创建tools/search_tool.py注册一个模拟搜索工具。# tools/search_tool.py import time from core.registry import ToolRegistry, ToolSpec def search_news(keyword: str, limit: int 3) - list: 模拟搜索新闻返回假数据演示工具调用过程 time.sleep(0.2) return [ {title: f{keyword} 最新进展Agent 架构持续演进, source: demo-news}, {title: fUMA 风格运行时在多 Agent 场景的实践, source: demo-tech}, ][:limit] def register_search_tool(registry: ToolRegistry) - None: registry.register(ToolSpec( namesearch_news, description搜索最近关于某个关键词的新闻标题, handlersearch_news, parameters{ type: object, properties: { keyword: { type: string, description: 搜索关键词, }, limit: { type: integer, description: 返回条数, default: 3, }, }, required: [keyword], }, ))6. 运行示例与效果验证6.1 使用 MockClient 验证运行时创建main.py使用 MockClient 模拟两轮工具调用验证运行时流程。# main.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.mock_client import MockClient from tools.search_tool import register_search_tool registry ToolRegistry() register_search_tool(registry) # 构造一个工具调用计划第一轮搜索第二轮结束 client MockClient([ { name: search_news, arguments: {keyword: UMA Agent, limit: 2}, }, ]) runtime UmaRuntime( llm_clientclient, registryregistry, system_prompt你是一个智能助手可以通过工具获取信息。, max_steps5, ) result runtime.run(帮我搜索一下 UMA Agent 的最新新闻) print(最终结果:, result) print(消息数量:, len(runtime.messages))运行命令cd uma-agent-demo python main.py预期输出最终结果: 任务已完成。 消息数量: 5这个消息数量对应1 条系统消息 1 条用户消息 1 条助手工具调用 1 条工具返回 1 条助手最终回复符合预期。6.2 接入真实大模型把 MockClient 换成 UniClient并把工具计划替换成真实的大模型输出。# main_real.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.llm_client import UniClient from tools.search_tool import register_search_tool registry ToolRegistry() register_search_tool(registry) client UniClient(modelgpt-4o-mini) runtime UmaRuntime( llm_clientclient, registryregistry, system_prompt你是一个智能助手当用户询问新闻时请先调用 search_news 工具再根据结果总结。, max_steps5, ) result runtime.run(帮我搜索 UMA Agent 的最新新闻) print(最终结果:, result) for msg in runtime.messages: print(f[{msg.role}] {msg.content[:80]})如果模型决定调用search_news运行时会自动执行工具并把结果回传给模型最终模型会基于工具结果生成回答。6.3 验证中断钩子把UmaRuntime换成带中断拦截的子类模拟敏感工具调用被拦截。# main_interrupt.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.mock_client import MockClient from tools.search_tool import register_search_tool class ApprovalRuntime(UmaRuntime): def _before_tool_call(self, tool_name, arguments): # 模拟人工审批工具名称包含 delete 就拒绝 if delete in tool_name: print(f人工审批未通过拦截工具调用: {tool_name}) return False print(f审批通过放行工具: {tool_name}) return True registry ToolRegistry() register_search_tool(registry) client MockClient([ {name: search_news, arguments: {keyword: UMA, limit: 1}}, ]) runtime ApprovalRuntime( llm_clientclient, registryregistry, system_prompt你是智能助手。, max_steps3, ) result runtime.run(执行一次搜索) print(结果:, result)运行后会看到工具调用先经过审批钩子再进入注册中心执行。这就是 UMA 中断机制的最小实现。7. 常见问题与排查思路问题现象可能原因排查方式解决方案Agent 一直重复调用同一个工具工具结果没有正确回传给模型打印消息列表确认是否追加 tool 角色消息检查运行时中工具结果追加逻辑模型返回 JSON 解析失败大模型输出不符合函数调用格式查看模型原始响应内容使用官方 function calling 接口不要自己解析文本 JSON工具调用超时外部服务响应慢在注册中心内打印耗时给 ToolSpec 增加异步超时控制Agent 达到最大步数仍未结束任务过于复杂或模型没有收敛增加日志输出每步动作提高 max_steps或拆分任务为多个 Agent上下文越来越长导致费用飙升每轮消息都堆积到上下文里统计 messages 数量实现上下文裁剪或消息摘要中断后状态丢失没有持久化消息和当前步骤检查 checkpoint 实现把 checkpoint 序列化到 Redis 或数据库真正的坑往往出现在工具层。一个工具返回结构不稳定会直接让模型在下游推理时产生幻觉。排查顺序建议是先看模型原始输出再看工具返回结果最后看消息历史。8. 从“能跑”到“会学”自改进 Agent 的演进方向UMA 解决的是 Agent 的骨架问题。骨架搭好之后下一步要面对的是经验复用和持续改进。近期关于 self-improving agents 的讨论很多代表性思路是从“单个任务执行”走向“经验积累与自我演化”。具体做法是Agent 完成一次任务后把任务背景、工具调用序列、成功经验和失败教训写入记忆库。下一次遇到类似任务时运行时先检索记忆库把历史经验注入到消息流中帮助模型避开上次的坑。# 伪代码经验写入与检索 class ExperienceMemory: def save(self, task_id, steps, success): ... def retrieve(self, task_desc, top_k3): # 可以用向量数据库做相似度检索 return [成功后记得对金额字段做二次校验。]这种“自我到元演化”的思路本质上是在 Agent 之上再加一层元认知回路Agent 不仅执行任务还产生关于自身行为的知识。不过在工程落地时要注意经验库的质量比数量重要。写入错误经验会导致模型在下一次任务中重复犯同样的错误。给开发者的建议是分阶段推进。第一阶段先完成工具调用、中断和日志第二阶段加入上下文摘要与缓存第三阶段再考虑经验记忆库。不要一开始就上复杂体系否则排查问题的成本会超过框架带来的收益。9. 工程化最佳实践9.1 工具层工具函数必须幂等。至少做到重复调用不会产生副作用。工具返回结构要固定。建议统一返回 JSON 对象并保证顶层字段稳定。在注册中心统一处理超时。不要让单个工具决定整个 Agent 的可用性。9.2 运行时层永远设置最大步数。没有步数上限的 Agent 不适合生产环境。给每一轮模型调用生成唯一 trace_id方便问题追踪。使用持久化消息队列存储工具结果避免进程重启后状态丢失。9.3 安全与权限层敏感工具调用前必须经过人工审批或规则引擎校验。遵循最小权限原则Agent 只能访问完成任务所必需的接口。在生产环境中用显式白名单控制 Agent 可以调用的工具集合。9.4 成本控制层对长上下文做压缩尤其是工具返回结果可以截断或摘要后再回传模型。为单任务设置 token 预算超过预算立即中断。对工具调用次数做配额统计异常增长时触发告警。9.5 部署与回滚Agent 的 Prompt 和工具列表应该纳入版本管理任何修改都要走发布流程。模型升级前用历史测试集做回归验证重点看工具调用格式是否变化。线上 Agent 必须保留历史消息快照以便出现事故时可回溯、可回滚。10. 总结与后续学习方向UMA 的价值不在某个炫酷的 API而在于把 Agent 从“一次模型调用”提升为“一个可治理的运行时系统”。本文讲清楚了它的核心模块统一消息、工具注册中心、任务循环、中断恢复和可观测性并且用一个最小 Python 实现验证了完整流程。如果你正在开发 Agent 应用下一步建议不要急着上多 Agent 编排先把单 Agent 的运行时打磨扎实为所有工具加上超时、重试和参数校验。实现至少一个中断点确认人工审批流程能正常放行和拦截。把完整消息流转记录接入日志平台。建立一套基于真实业务的回归测试集防止模型升级导致工具调用回归。Agent 的能力会随模型迭代不断进步但工程化的基础不会过时。你觉得自己的 Agent 当前最缺的是哪一块工具治理、中断恢复还是经验记忆可以从最小问题开始补课。 UMA架构解析：从零构建可中断、可治理的Agent运行时
你一定会遇到这个时刻你的 Agent 在演示环境里跑得丝滑顺畅一旦接进真实业务要么在工具调用之间反复横跳要么一次请求把上下文塞爆要么干脆在一个失败动作上无限重试。问题往往不在大模型本身而在于你的 Agent 只有“智能”没有“结构”。“Let Them”这个说法最近在 Agent 开发者圈子里被反复提及。它不是在鼓励你放手让 Agent 乱跑而是指一种新的分工方式让 Agent 做它擅长的自主推理与行动同时由开发者搭好边界、协议、工具注册表和中断机制。UMA 正是这套思路下的一种典型架构实践。本文不会绑定某个特定产品或版本而是从工程实现的角度拆解 UMA 的核心设计并用可运行的 Python 代码带你从零搭建一个具备统一消息协议、工具治理、记忆管理和中断恢复能力的 Agent 运行时。1. UMA 是什么先给一个明确判断UMA 在不同社区和技术文档里会展开为不同全称常见的解释包括 Unified Model Adapter、Unified Message Architecture、Universal Multi-Agent Architecture。不管名称怎么变落到工程层面其实是同一件事UMA 是一层连接大模型、工具、记忆和应用业务的运行时抽象。它不是一个新的模型不是一句更复杂的 Prompt也不是某个云厂商的托管产品。UMA 的核心贡献是让 Agent 在运行时拥有统一的消息格式、统一的工具调用入口、统一的状态保存机制并且在需要时允许开发者安全地中断和恢复执行。很多人在搭建 Agent 时第一反应是先把大模型 API 调通然后在 Prompt 里塞一堆工具说明。这种做法在单轮任务里完全没有问题但一旦任务需要多步推理、多次工具调用甚至要跨天恢复状态代码就会迅速腐化。典型症状包括工具调用的结果散落在各个局部变量里没有统一的数据结构。Agent 返回的 JSON 中字段名不一致解析逻辑越来越脆。没有工具超时、重试和权限校验Agent 一调用外部服务就出事。没有中断能力一旦发现 Agent 走偏只能让它继续跑完或由外部进程强制 kill。UMA 解决的就是这些问题。它把 Agent 的开发从“写 Prompt 调 API”提升到了“设计协议 构建运行时”的工程化层次。维度普通 LLM 调用手写 Agent 循环UMA 分层输入一段 Prompt自由拼接消息统一消息类型工具调用无直接在循环里写死工具注册中心状态管理无状态局部变量持久化上下文中断恢复不支持不支持支持暂停、恢复可观测性依赖日志零散打印统一消息流追踪生产治理弱弱超时、重试、权限、额度这里的核心判断是Agent 能不能在真实业务里落地取决于运行时设计而不是模型参数。UMA 的价值正在于把大模型的“不可控智能”装进一套“可控工程外壳”里。2. 构建 Agent 时最容易踩的三个坑在展开 UMA 的架构之前我们先对齐三个常见的失败模式。理解了这些坑你就能明白为什么 UMA 要设计成后面那个样子。2.1 把智能全部押在 Prompt 上很多人认为 Agent 能力弱是因为 Prompt 写得不够好。于是不断往系统提示词里追加规则“你必须仔细思考”“你要一步一步分析”“工具调用失败后要重试不超过三次”。这确实能改善一部分行为但 Prompt 无法保证结构。模型可能在某一次响应里忘了遵循规则可能把工具参数拼错格式也可能在多次调用后上下文过长把早期的关键信息冲掉。结构性的问题必须用结构性的方案解决。超时、重试、参数校验、状态持久化这些不应该写在 Prompt 里而应该在运行时强制完成。2.2 让 Agent 裸奔没有工具治理Agent 要完成任务几乎必然要调用外部工具搜索引擎、数据库、订单系统、内部 API。如果你把这些工具直接暴露给大模型任何一次参数错误都可能造成真实影响。比如 Agent 调用一个删除接口因为参数解析失误把id123传成了idall。这样的问题靠模型自身很难完全规避。更好的做法是让工具调用经过一个注册中心由注册中心统一处理参数白名单、超时、重试、权限校验和结果规范化。2.3 只有“开始”没有“暂停”部分 Agent 框架设计成一次运行从入口一路执行到结束中间没有人工介入点。这在简单问答场景还可以接受但在真实业务流程里非常危险。Agent 可能会不小心确认一筆不该确认的订单或者调用了一个需要人工复核的接口。所以现代 Agent 架构特别强调 interrupt 能力执行过程中Agent 可以主动暂停等待用户确认、补充信息或修正方向开发者也可以基于规则主动打断执行把控制权交还给业务系统。这就是最近讨论里高频出现的 “deep agents interrupt” 概念。中断不是失败而是 Agent 与外部世界协作的正式接口。3. UMA 的核心模块设计UMA 作为一个运行时抽象通常包含以下六个核心模块。3.1 统一消息协议Agent 运行过程中会涉及多类消息系统指令、用户输入、模型回复、工具调用请求、工具返回结果。如果每一类消息都用不同的数据结构代码会越来越难维护。UMA 的做法是定义一套统一消息类型所有环节都通过这一种数据结构传递信息。消息中至少包含角色、内容、时间戳如果要支持工具调用还需要携带工具调用 ID 和工具名称。3.2 工具注册中心工具注册中心负责维护 Agent 可以调用的全部工具列表。每个工具注册时声明名称、描述、参数结构和处理函数。大模型看到的工具清单从这里生成运行时也通过这里完成工具调用。这个模块的价值在于把工具调用变成可插拔机制新增一个工具不需要改 Agent 主循环只要向注册中心注册即可。3.3 任务循环任务循环是 Agent 的主进程。它不断执行“读取消息 - 调用模型 - 解析输出 - 执行动作或结束”的循环直到任务完成、达到最大步数或触发中断条件。循环需要控制两个关键指标最大步数和单次工具调用超时时间。没有步数上限Agent 可能陷入死循环没有超时控制一个卡住的工具会拖垮整个任务。3.4 记忆与上下文管理大模型上下文窗口有限无法承载无限长的历史消息。UMA 需要在运行时管理上下文哪些消息需要保留哪些可以压缩哪些可以归档到外部记忆存储中。对于复杂任务推荐把长期记忆和短期上下文分开。短期上下文只保留最近几轮必要消息长期记忆则存到向量数据库或键值存储中。3.5 中断恢复机制中断恢复是 UMA 与其他简单 Agent 框架最大的区别。运行时需要支持两类中断主动中断Agent 认为需要用户确认时暂停执行并等待外部输入。被动中断开发者根据业务规则强制暂停例如检测到敏感操作、超时或成本超限。恢复执行时运行时应该从最近一个检查点继续而不是从头开始。这就要求消息列表和上下文状态可以序列化、可以持久化。3.6 可观测与追踪Agent 的调试比普通后端服务更困难因为模型输出有随机性。UMA 需要记录完整的消息流转过程每一轮模型返回了什么、选择了哪个工具、参数是什么、工具返回了什么。这些记录可以输出到日志系统也可以对接 OpenTelemetry 等追踪工具。4. 环境准备与项目结构为了让后面的示例可以顺利运行先准备环境。本文以 Python 3.10 为例不绑定特定大模型厂商。你可以使用 OpenAI SDK也可以使用兼容 OpenAI 接口的本地模型服务例如 Ollama 或 vLLM。建议创建如下项目结构uma-agent-demo/ ├── core/ │ ├── __init__.py │ ├── message.py # 统一消息定义 │ ├── registry.py # 工具注册中心 │ └── runtime.py # Agent 运行时与任务循环 ├── tools/ │ └── search_tool.py # 示例工具 ├── clients/ │ └── llm_client.py # 大模型客户端封装 ├── requirements.txt └── main.py # 程序入口最小依赖如下openai1.0.0 pydantic2.0.0如果你希望不依赖外部 API 也能跑通示例可以把llm_client.py改成本地 Mock 实现本文第 5 节会提供可替换方案。4.1 统一消息定义创建core/message.py定义统一消息类型。# core/message.py from dataclasses import dataclass, field from typing import Optional import time dataclass class UMessage: role: str content: str tool_call_id: Optional[str] None tool_name: Optional[str] None timestamp: float field(default_factorytime.time)这里定义了四种角色的消息system表示系统指令user表示用户输入assistant表示模型回复tool表示工具返回结果。工具调用相关的tool_call_id和tool_name用于把模型发起的工具调用和工具结果关联起来。4.2 工具注册中心创建core/registry.py实现工具注册和调用。# core/registry.py from dataclasses import dataclass from typing import Any, Callable, Dict, Optional import json dataclass class ToolSpec: name: str description: str handler: Callable[..., Any] parameters: dict timeout: float 10.0 class ToolRegistry: 工具注册中心 def __init__(self) - None: self._tools: Dict[str, ToolSpec] {} def register(self, spec: ToolSpec) - None: if spec.name in self._tools: raise ValueError(ftool already registered: {spec.name}) self._tools[spec.name] spec def get_schema(self) - list: tools [] for spec in self._tools.values(): tools.append({ type: function, function: { name: spec.name, description: spec.description, parameters: spec.parameters, }, }) return tools def invoke(self, name: str, arguments: dict) - str: spec self._tools.get(name) if spec is None: return json.dumps({status: error, message: funknown tool: {name}}) try: result spec.handler(**arguments) return json.dumps({status: ok, result: result}, ensure_asciiFalse) except Exception as exc: return json.dumps({status: error, message: str(exc)})工具注册中心的核心价值在于两件事一是把工具清单统一输出给大模型二是所有工具调用都经过同一个入口便于后续添加超时、重试和权限控制。5. 从零实现一个 UMA 风格 Agent5.1 大模型客户端封装创建clients/llm_client.py。这里提供一个 OpenAI 兼容的客户端以及一个用于本地演示的 MockClient。# clients/llm_client.py from typing import Any, Dict, List class UniClient: 统一的模型调用客户端适配 OpenAI 兼容接口 def __init__(self, model: str gpt-4o-mini, base_url: str | None None): try: from openai import OpenAI except ImportError: raise RuntimeError(请安装 openai 客户端pip install openai) self.client OpenAI(base_urlbase_url) self.model model def chat( self, messages: List[Dict[str, Any]], tools: List[Dict[str, Any]], ) - Dict[str, Any]: response self.client.chat.completions.create( modelself.model, messagesmessages, toolstools or None, ) message response.choices[0].message tool_calls message.tool_calls or [] if tool_calls: call tool_calls[0] if call.function: return { type: tool_call, name: call.function.name, arguments: json_loads_safe(call.function.arguments), } return { type: finish, content: message.content or , } def json_loads_safe(text: str) - dict: import json try: return json.loads(text) except Exception: return {}MockClient 不需要网络也不消耗任何额度适合在 CI 或本地环境快速验证 UMA 运行时逻辑。# clients/mock_client.py class MockClient: 本地 Mock 模型只在指定轮次调用工具其余轮次直接结束。 def __init__(self, tool_plan: list): self._plan tool_plan self._step 0 def chat(self, messages, tools): step self._step self._step 1 if step len(self._plan): plan self._plan[step] return { type: tool_call, name: plan[name], arguments: plan[arguments], } return { type: finish, content: 任务已完成。 }5.2 Agent 运行时创建core/runtime.py实现任务循环、步数控制和中断机制。# core/runtime.py from typing import Any, Dict, List, Optional from .message import UMessage from .registry import ToolRegistry class UmaRuntime: UMA Agent 运行时 def __init__( self, llm_client: Any, registry: ToolRegistry, system_prompt: str, max_steps: int 10, ): self.llm_client llm_client self.registry registry self.messages: List[UMessage] [UMessage(rolesystem, contentsystem_prompt)] self.max_steps max_steps self.current_step 0 def _to_model_messages(self) - List[Dict[str, str]]: return [{role: msg.role, content: msg.content} for msg in self.messages] def run(self, user_input: str) - str: self.messages.append(UMessage(roleuser, contentuser_input)) while self.current_step self.max_steps: self.current_step 1 response self.llm_client.chat( self._to_model_messages(), self.registry.get_schema(), ) if response[type] finish: return response.get(content, ) if response[type] tool_call: tool_name response[name] arguments response.get(arguments) or {} # 中断点敏感工具调用前挂起 if not self._before_tool_call(tool_name, arguments): return f工具调用被拦截{tool_name} tool_result self.registry.invoke(tool_name, arguments) self.messages.append(UMessage( roleassistant, content, tool_call_idtool_name, tool_nametool_name, )) self.messages.append(UMessage( roletool, contenttool_result, tool_call_idtool_name, )) return 达到最大步骤数任务未完成。 def _before_tool_call(self, tool_name: str, arguments: Dict[str, Any]) - bool: 安全钩子业务方可以在这里加入人工审批或规则拦截 return True class InterruptibleRuntime(UmaRuntime): 带中断恢复能力的运行时 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.checkpoint: Optional[Dict[str, Any]] None def save_checkpoint(self) - None: self.checkpoint { messages: [ { role: m.role, content: m.content, tool_call_id: m.tool_call_id, tool_name: m.tool_name, } for m in self.messages ], current_step: self.current_step, } def restore_checkpoint(self) - None: if self.checkpoint is None: return self.messages [ UMessage( roleitem[role], contentitem[content], tool_call_iditem[tool_call_id], tool_nameitem[tool_name], ) for item in self.checkpoint[messages] ] self.current_step self.checkpoint[current_step]这段代码有三个关键点在run的循环体里每一轮都会根据当前消息列表调用大模型并传入工具清单。如果模型返回tool_call运行时从注册中心调用对应工具并把结果追加为tool角色消息供模型在下一轮参考。_before_tool_call是一个安全钩子业务方可以实现人工审批、敏感操作拦截等逻辑。这也是 interrupt 机制最简单的落地形态在工具调用之前决定是否放行。5.3 注册一个示例工具创建tools/search_tool.py注册一个模拟搜索工具。# tools/search_tool.py import time from core.registry import ToolRegistry, ToolSpec def search_news(keyword: str, limit: int 3) - list: 模拟搜索新闻返回假数据演示工具调用过程 time.sleep(0.2) return [ {title: f{keyword} 最新进展Agent 架构持续演进, source: demo-news}, {title: fUMA 风格运行时在多 Agent 场景的实践, source: demo-tech}, ][:limit] def register_search_tool(registry: ToolRegistry) - None: registry.register(ToolSpec( namesearch_news, description搜索最近关于某个关键词的新闻标题, handlersearch_news, parameters{ type: object, properties: { keyword: { type: string, description: 搜索关键词, }, limit: { type: integer, description: 返回条数, default: 3, }, }, required: [keyword], }, ))6. 运行示例与效果验证6.1 使用 MockClient 验证运行时创建main.py使用 MockClient 模拟两轮工具调用验证运行时流程。# main.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.mock_client import MockClient from tools.search_tool import register_search_tool registry ToolRegistry() register_search_tool(registry) # 构造一个工具调用计划第一轮搜索第二轮结束 client MockClient([ { name: search_news, arguments: {keyword: UMA Agent, limit: 2}, }, ]) runtime UmaRuntime( llm_clientclient, registryregistry, system_prompt你是一个智能助手可以通过工具获取信息。, max_steps5, ) result runtime.run(帮我搜索一下 UMA Agent 的最新新闻) print(最终结果:, result) print(消息数量:, len(runtime.messages))运行命令cd uma-agent-demo python main.py预期输出最终结果: 任务已完成。 消息数量: 5这个消息数量对应1 条系统消息 1 条用户消息 1 条助手工具调用 1 条工具返回 1 条助手最终回复符合预期。6.2 接入真实大模型把 MockClient 换成 UniClient并把工具计划替换成真实的大模型输出。# main_real.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.llm_client import UniClient from tools.search_tool import register_search_tool registry ToolRegistry() register_search_tool(registry) client UniClient(modelgpt-4o-mini) runtime UmaRuntime( llm_clientclient, registryregistry, system_prompt你是一个智能助手当用户询问新闻时请先调用 search_news 工具再根据结果总结。, max_steps5, ) result runtime.run(帮我搜索 UMA Agent 的最新新闻) print(最终结果:, result) for msg in runtime.messages: print(f[{msg.role}] {msg.content[:80]})如果模型决定调用search_news运行时会自动执行工具并把结果回传给模型最终模型会基于工具结果生成回答。6.3 验证中断钩子把UmaRuntime换成带中断拦截的子类模拟敏感工具调用被拦截。# main_interrupt.py from core.registry import ToolRegistry from core.runtime import UmaRuntime from clients.mock_client import MockClient from tools.search_tool import register_search_tool class ApprovalRuntime(UmaRuntime): def _before_tool_call(self, tool_name, arguments): # 模拟人工审批工具名称包含 delete 就拒绝 if delete in tool_name: print(f人工审批未通过拦截工具调用: {tool_name}) return False print(f审批通过放行工具: {tool_name}) return True registry ToolRegistry() register_search_tool(registry) client MockClient([ {name: search_news, arguments: {keyword: UMA, limit: 1}}, ]) runtime ApprovalRuntime( llm_clientclient, registryregistry, system_prompt你是智能助手。, max_steps3, ) result runtime.run(执行一次搜索) print(结果:, result)运行后会看到工具调用先经过审批钩子再进入注册中心执行。这就是 UMA 中断机制的最小实现。7. 常见问题与排查思路问题现象可能原因排查方式解决方案Agent 一直重复调用同一个工具工具结果没有正确回传给模型打印消息列表确认是否追加 tool 角色消息检查运行时中工具结果追加逻辑模型返回 JSON 解析失败大模型输出不符合函数调用格式查看模型原始响应内容使用官方 function calling 接口不要自己解析文本 JSON工具调用超时外部服务响应慢在注册中心内打印耗时给 ToolSpec 增加异步超时控制Agent 达到最大步数仍未结束任务过于复杂或模型没有收敛增加日志输出每步动作提高 max_steps或拆分任务为多个 Agent上下文越来越长导致费用飙升每轮消息都堆积到上下文里统计 messages 数量实现上下文裁剪或消息摘要中断后状态丢失没有持久化消息和当前步骤检查 checkpoint 实现把 checkpoint 序列化到 Redis 或数据库真正的坑往往出现在工具层。一个工具返回结构不稳定会直接让模型在下游推理时产生幻觉。排查顺序建议是先看模型原始输出再看工具返回结果最后看消息历史。8. 从“能跑”到“会学”自改进 Agent 的演进方向UMA 解决的是 Agent 的骨架问题。骨架搭好之后下一步要面对的是经验复用和持续改进。近期关于 self-improving agents 的讨论很多代表性思路是从“单个任务执行”走向“经验积累与自我演化”。具体做法是Agent 完成一次任务后把任务背景、工具调用序列、成功经验和失败教训写入记忆库。下一次遇到类似任务时运行时先检索记忆库把历史经验注入到消息流中帮助模型避开上次的坑。# 伪代码经验写入与检索 class ExperienceMemory: def save(self, task_id, steps, success): ... def retrieve(self, task_desc, top_k3): # 可以用向量数据库做相似度检索 return [成功后记得对金额字段做二次校验。]这种“自我到元演化”的思路本质上是在 Agent 之上再加一层元认知回路Agent 不仅执行任务还产生关于自身行为的知识。不过在工程落地时要注意经验库的质量比数量重要。写入错误经验会导致模型在下一次任务中重复犯同样的错误。给开发者的建议是分阶段推进。第一阶段先完成工具调用、中断和日志第二阶段加入上下文摘要与缓存第三阶段再考虑经验记忆库。不要一开始就上复杂体系否则排查问题的成本会超过框架带来的收益。9. 工程化最佳实践9.1 工具层工具函数必须幂等。至少做到重复调用不会产生副作用。工具返回结构要固定。建议统一返回 JSON 对象并保证顶层字段稳定。在注册中心统一处理超时。不要让单个工具决定整个 Agent 的可用性。9.2 运行时层永远设置最大步数。没有步数上限的 Agent 不适合生产环境。给每一轮模型调用生成唯一 trace_id方便问题追踪。使用持久化消息队列存储工具结果避免进程重启后状态丢失。9.3 安全与权限层敏感工具调用前必须经过人工审批或规则引擎校验。遵循最小权限原则Agent 只能访问完成任务所必需的接口。在生产环境中用显式白名单控制 Agent 可以调用的工具集合。9.4 成本控制层对长上下文做压缩尤其是工具返回结果可以截断或摘要后再回传模型。为单任务设置 token 预算超过预算立即中断。对工具调用次数做配额统计异常增长时触发告警。9.5 部署与回滚Agent 的 Prompt 和工具列表应该纳入版本管理任何修改都要走发布流程。模型升级前用历史测试集做回归验证重点看工具调用格式是否变化。线上 Agent 必须保留历史消息快照以便出现事故时可回溯、可回滚。10. 总结与后续学习方向UMA 的价值不在某个炫酷的 API而在于把 Agent 从“一次模型调用”提升为“一个可治理的运行时系统”。本文讲清楚了它的核心模块统一消息、工具注册中心、任务循环、中断恢复和可观测性并且用一个最小 Python 实现验证了完整流程。如果你正在开发 Agent 应用下一步建议不要急着上多 Agent 编排先把单 Agent 的运行时打磨扎实为所有工具加上超时、重试和参数校验。实现至少一个中断点确认人工审批流程能正常放行和拦截。把完整消息流转记录接入日志平台。建立一套基于真实业务的回归测试集防止模型升级导致工具调用回归。Agent 的能力会随模型迭代不断进步但工程化的基础不会过时。你觉得自己的 Agent 当前最缺的是哪一块工具治理、中断恢复还是经验记忆可以从最小问题开始补课。