托管 Agent 执行循环只是起点AgentRun 托管的更是企业 AI 生产全链路在 LLM 应用开发领域我们常常陷入一个误区认为只要封装好while True: 观察-思考-行动的执行循环就能交付一个生产级 Agent。诚然循环是骨架但企业级 AI 系统的真正挑战在于——如何将脆弱的、单次的、依赖上下文的推理过程嵌入到复杂、多租户、可观测、可回滚的企业 IT 生态中。AgentRun正是从这个视角切入它托管的不是循环本身而是循环背后的全生命周期治理。### 从「循环」到「运行态」问题域的根本转变传统 Agent 框架如 LangChain 的AgentExecutor解决的是「如何让模型多次调用工具并收敛」。但企业场景下一次 Agent 调用往往涉及-多步骤状态持久化用户中断后恢复或异步任务回调。-工具调用的审计与追溯每个步骤的输入输出、token 消耗、延迟。-故障恢复与人工介入当 Agent 反复失败或置信度低时需要人工审批或回退。-版本化与灰度模型或 prompt 更新后如何保证旧任务不受影响。这些需求超出了「循环」的抽象范围需要一个运行态Runtime层。AgentRun 将 Agent 执行视为一个可暂停、可恢复、可观测的分布式事务而非一个简单函数调用。### 核心设计事件溯源驱动的状态机AgentRun 的核心是一个持久化的执行状态机。它不直接存储内存对象而是将所有 Agent 的决策点决策、工具调用、异常、人类反馈作为不可变事件追加到日志中。任何时刻Agent 的状态都可以通过重放事件流重建。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) # 实际生产环境会写入 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 实现了三个关键能力1.确定性重放当线上 Agent 出现问题运维可以基于相同事件流在沙箱中重现定位是提示词问题还是工具接口问题。2.时点恢复如果某个工具调用导致数据不一致可以回滚到该事件之前的状态而不是整个任务失败。3.并发控制多个消费者可以独立读取事件流实现监控、审计、训练数据采集的解耦。### 生产全链路从编排到治理AgentRun 的托管范围远超执行循环它覆盖了以下环节#### 1. 动态工具注册与权限校验企业环境下的工具调用必须经过权限校验。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: # 触发审批流程而不是直接拒绝 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): # 实际实现会订阅 Redis 发布/订阅 或 Kafka 主题 # 此处简化 return True#### 2. 可观测性与追踪AgentRun 自动为每个步骤生成 OpenTelemetry Span将 LLM 调用、工具调用、状态转换全部串联。这不仅是监控更是训练数据回流的基础——每个成功的 Agent 轨迹都可以被标记为「正样本」用于后续模型微调。#### 3. 人类介入的编排当 Agent 连续 3 次工具调用返回错误或者置信度低于阈值AgentRun 会暂停执行并生成一个「人工审批任务」。人工反馈会作为新的事件注入事件流Agent 从暂停点恢复而非重新开始。#### 4. 版本策略与灰度发布AgentRun 将「模型版本」和「提示词版本」作为一等公民。每次执行都会绑定具体的版本号发布新版本时通过事件流中的版本标签可以精确统计新旧版本的胜出率如成功率、耗时、用户满意度。### 为什么这是「全链路」而非「循环」传统循环只关心llm - tool - llm的周转而 AgentRun 关心的是-执行前配额、鉴权、模型选择、缓存命中。-执行中动态工具注册、超时熔断、降级策略、并发控制。-执行后成本归因、异常分析、自动回放、数据标注。以金融场景为例一个 Agent 处理贷款审批时每一步工具调用如查征信、验资产都必须有审计日志。AgentRun 的事件溯源天然满足合规要求而普通循环则无法追溯「为什么模型在这个节点选择了这个工具」。### 代码示例一个完整的 AgentRun 托管执行过程下面展示如何用 AgentRun 的 API 包装一个简单的带人工审批的 Agentpythonimport asynciofrom agentrun import AgentRun, ToolSpecasync def main(): # 初始化 AgentRun 运行时连接事件存储、权限服务、监控 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: # 此工具会触发人工审批返回审批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 )) # 启动一个执行任务自动进入事件循环 run_id await runtime.start_agent( task查一下客户的信用评分如果低于600则需要人工审批, user_ctx{role: analyst, tenant_id: t1} ) # 模拟等待执行完成实际会通过回调通知 await asyncio.sleep(5) # 查询执行状态从事件流重建 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内部会持续运行循环但所有状态变更都被持久化。当request_approval被调用时AgentRun 自动进入「等待状态」不会占用计算资源直到外部系统通过 API 注入审批结果。### 总结AgentRun 的哲学是Agent 的执行循环只是最内层的核心真正的价值在于围绕它的支撑体系。它把 Agent 从「一个聪明的函数」升级为「一个可治理的企业服务」。通过事件溯源、权限控制、人工介入、版本管理AgentRun 解决了 AI 生产中最棘手的「黑盒不可控」问题。当你的 Agent 需要面向真实业务时不要只关注循环的效率更要关注循环之外的全链路治理——这正是 AgentRun 的立足点也是企业 AI 落地的关键。
