托管 Agent 執(zhí)行循環(huán)只是起點AgentRun 托管的更是企業(yè) AI 生產全鏈路在 LLM 應用開發(fā)領域我們常常陷入一個誤區(qū)認為只要封裝好while True: 觀察-思考-行動的執(zhí)行循環(huán)就能交付一個生產級 Agent。誠然循環(huán)是骨架但企業(yè)級 AI 系統(tǒng)的真正挑戰(zhàn)在于——如何將脆弱的、單次的、依賴上下文的推理過程嵌入到復雜、多租戶、可觀測、可回滾的企業(yè) IT 生態(tài)中。AgentRun正是從這個視角切入它托管的不是循環(huán)本身而是循環(huán)背后的全生命周期治理。### 從「循環(huán)」到「運行態(tài)」問題域的根本轉變傳統(tǒng) Agent 框架如 LangChain 的AgentExecutor解決的是「如何讓模型多次調用工具并收斂」。但企業(yè)場景下一次 Agent 調用往往涉及-多步驟狀態(tài)持久化用戶中斷后恢復或異步任務回調。-工具調用的審計與追溯每個步驟的輸入輸出、token 消耗、延遲。-故障恢復與人工介入當 Agent 反復失敗或置信度低時需要人工審批或回退。-版本化與灰度模型或 prompt 更新后如何保證舊任務不受影響。這些需求超出了「循環(huán)」的抽象范圍需要一個運行態(tài)Runtime層。AgentRun 將 Agent 執(zhí)行視為一個可暫停、可恢復、可觀測的分布式事務而非一個簡單函數(shù)調用。### 核心設計事件溯源驅動的狀態(tài)機AgentRun 的核心是一個持久化的執(zhí)行狀態(tài)機。它不直接存儲內存對象而是將所有 Agent 的決策點決策、工具調用、異常、人類反饋作為不可變事件追加到日志中。任何時刻Agent 的狀態(tài)都可以通過重放事件流重建。python# 示例 1定義 AgentRun 中的事件類型偽代碼使用 Pydantic 風格from enum import Enumfrom datetime import datetimefrom typing import Any, Optionalclass EventType(str, Enum): AGENT_START agent_start TOOL_CALL tool_call TOOL_RESULT tool_result LLM_STEP llm_step HUMAN_APPROVAL_REQUIRED human_approval_required HUMAN_APPROVAL_GIVEN human_approval_given AGENT_END agent_end AGENT_ERROR agent_errorclass AgentEvent: def __init__(self, run_id: str, type: EventType, payload: dict[str, Any], timestamp: datetime None): self.run_id run_id self.type type self.payload payload self.timestamp timestamp or datetime.utcnow() def to_dict(self): return { run_id: self.run_id, type: self.type.value, payload: self.payload, timestamp: self.timestamp.isoformat() }# 事件流存儲層簡化版class EventStore: def __init__(self): self._events: list[AgentEvent] [] def append(self, event: AgentEvent): self._events.append(event) # 實際生產環(huán)境會寫入 Kafka/PostgreSQL/EventStore def replay(self, run_id: str) - list[AgentEvent]: return [e for e in self._events if e.run_id run_id]通過事件溯源AgentRun 實現(xiàn)了三個關鍵能力1.確定性重放當線上 Agent 出現(xiàn)問題運維可以基于相同事件流在沙箱中重現(xiàn)定位是提示詞問題還是工具接口問題。2.時點恢復如果某個工具調用導致數(shù)據(jù)不一致可以回滾到該事件之前的狀態(tài)而不是整個任務失敗。3.并發(fā)控制多個消費者可以獨立讀取事件流實現(xiàn)監(jiān)控、審計、訓練數(shù)據(jù)采集的解耦。### 生產全鏈路從編排到治理AgentRun 的托管范圍遠超執(zhí)行循環(huán)它覆蓋了以下環(huán)節(jié)#### 1. 動態(tài)工具注冊與權限校驗企業(yè)環(huán)境下的工具調用必須經過權限校驗。AgentRun 在每次工具調用前會檢查調用者身份通過 JWT 傳遞的 tenant_id、工具的白名單、以及該調用是否在預算策略內。python# 示例 2帶權限與預算校驗的工具調用包裝器class SecureToolWrapper: def __init__(self, tool_func, allowed_roles: set[str], max_cost_per_call: float, usage_client): self.tool tool_func self.allowed_roles allowed_roles self.max_cost max_cost_per_call self.usage usage_client async def call(self, *args, user_ctx: dict, **kwargs): # 1. 角色校驗 if user_ctx.get(role) not in self.allowed_roles: raise PermissionError(fRole {user_ctx.get(role)} not allowed) # 2. 預算預檢查基于外部計費服務 pre_cost self.usage.estimate_cost(self.tool.__name__, args) if pre_cost self.max_cost: # 觸發(fā)審批流程而不是直接拒絕 approval_id await self.usage.request_approval( run_iduser_ctx[run_id], tool_nameself.tool.__name__, estimated_costpre_cost ) # 阻塞等待人工審批通過 WebSocket/回調 approved await self._wait_for_approval(approval_id, timeout300) if not approved: raise BudgetExceededError(Approval rejected) # 3. 調用真實工具并記錄用量 result await self.tool(*args, **kwargs) self.usage.record(self.tool.__name__, input_tokenskwargs.get(_tokens, 0), costpre_cost) return result async def _wait_for_approval(self, approval_id: str, timeout: int): # 實際實現(xiàn)會訂閱 Redis 發(fā)布/訂閱 或 Kafka 主題 # 此處簡化 return True#### 2. 可觀測性與追蹤AgentRun 自動為每個步驟生成 OpenTelemetry Span將 LLM 調用、工具調用、狀態(tài)轉換全部串聯(lián)。這不僅是監(jiān)控更是訓練數(shù)據(jù)回流的基礎——每個成功的 Agent 軌跡都可以被標記為「正樣本」用于后續(xù)模型微調。#### 3. 人類介入的編排當 Agent 連續(xù) 3 次工具調用返回錯誤或者置信度低于閾值AgentRun 會暫停執(zhí)行并生成一個「人工審批任務」。人工反饋會作為新的事件注入事件流Agent 從暫停點恢復而非重新開始。#### 4. 版本策略與灰度發(fā)布AgentRun 將「模型版本」和「提示詞版本」作為一等公民。每次執(zhí)行都會綁定具體的版本號發(fā)布新版本時通過事件流中的版本標簽可以精確統(tǒng)計新舊版本的勝出率如成功率、耗時、用戶滿意度。### 為什么這是「全鏈路」而非「循環(huán)」傳統(tǒng)循環(huán)只關心llm - tool - llm的周轉而 AgentRun 關心的是-執(zhí)行前配額、鑒權、模型選擇、緩存命中。-執(zhí)行中動態(tài)工具注冊、超時熔斷、降級策略、并發(fā)控制。-執(zhí)行后成本歸因、異常分析、自動回放、數(shù)據(jù)標注。以金融場景為例一個 Agent 處理貸款審批時每一步工具調用如查征信、驗資產都必須有審計日志。AgentRun 的事件溯源天然滿足合規(guī)要求而普通循環(huán)則無法追溯「為什么模型在這個節(jié)點選擇了這個工具」。### 代碼示例一個完整的 AgentRun 托管執(zhí)行過程下面展示如何用 AgentRun 的 API 包裝一個簡單的帶人工審批的 Agentpythonimport asynciofrom agentrun import AgentRun, ToolSpecasync def main(): # 初始化 AgentRun 運行時連接事件存儲、權限服務、監(jiān)控 runtime AgentRun( event_storeEventStore(), # 實際用 PostgreSQL usage_clientUsageClient(), model_providerOpenAIProvider(versiongpt-4o-2024-05), ) # 注冊兩個工具 async def search_db(query: str) - str: return fResults for {query} async def request_approval(amount: float) - str: # 此工具會觸發(fā)人工審批返回審批ID return fapproval_{amount} runtime.register_tool(ToolSpec( funcsearch_db, namesearch_db, allowed_roles{analyst, admin}, cost_per_call0.01 )) runtime.register_tool(ToolSpec( funcrequest_approval, namerequest_approval, allowed_roles{admin}, cost_per_call0.0 )) # 啟動一個執(zhí)行任務自動進入事件循環(huán) run_id await runtime.start_agent( task查一下客戶的信用評分如果低于600則需要人工審批, user_ctx{role: analyst, tenant_id: t1} ) # 模擬等待執(zhí)行完成實際會通過回調通知 await asyncio.sleep(5) # 查詢執(zhí)行狀態(tài)從事件流重建 events runtime.event_store.replay(run_id) for e in events: print(f{e.timestamp} | {e.type} | {e.payload}) # 輸出: # ... | agent_start | {task: ...} # ... | tool_call | {tool: search_db, args: {query: ...}} # ... | tool_result | {result: ...} # ... | llm_step | {tokens: 123, model: gpt-4o} # ... | human_approval_required | {reason: 信用分600} # 然后程序暫停等待人工審批...asyncio.run(main())上述代碼中start_agent內部會持續(xù)運行循環(huán)但所有狀態(tài)變更都被持久化。當request_approval被調用時AgentRun 自動進入「等待狀態(tài)」不會占用計算資源直到外部系統(tǒng)通過 API 注入審批結果。### 總結AgentRun 的哲學是Agent 的執(zhí)行循環(huán)只是最內層的核心真正的價值在于圍繞它的支撐體系。它把 Agent 從「一個聰明的函數(shù)」升級為「一個可治理的企業(yè)服務」。通過事件溯源、權限控制、人工介入、版本管理AgentRun 解決了 AI 生產中最棘手的「黑盒不可控」問題。當你的 Agent 需要面向真實業(yè)務時不要只關注循環(huán)的效率更要關注循環(huán)之外的全鏈路治理——這正是 AgentRun 的立足點也是企業(yè) AI 落地的關鍵。