你一定會遇到這個時刻你的 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 當前最缺的是哪一塊工具治理、中斷恢復還是經驗記憶可以從最小問題開始補課。