当业务场景从“单次问答”走向“多步骤自动执行”时大模型本身其实解决不了全部问题。真正困难的地方在于一个任务可能包含几十次工具调用、中间结果保存、异常恢复、超时控制、权限校验以及执行过程回放。这个领域最近经常被提及的两个关键词一个是 Agentic Runtime一个是 Long-Horizon Reasoning。本文就以 Argus 这一类“通用代理运行时”为切入点拆解它的核心设计思路并给出一个可直接运行的最小工程实现帮助你理解长任务代理到底应该怎么做。1. 背景为什么需要 Agentic Runtime1.1 单次模型调用与长任务执行之间的鸿沟很多同学第一次接触大模型应用时都会写这样的代码response model.chat(帮我分析这份销售数据)这种调用方式适合“单轮问答”但真实业务很少只有一句话。比如读取三个不同位置的销售数据文件对数据做清洗、格式化按地区汇总对比上月变化生成一份带图表的报告把报告发送给指定人员。这五个步骤如果只靠一次模型调用完成效果通常很不稳定。模型要么漏掉中间步骤要么在生成长文本时丢失目标。更现实的方案是把任务拆成多段执行并把每一段的结果保存下来供下一步使用。这中间缺一个专门负责“编排、执行、恢复”的层这个层就是 Agentic Runtime。1.2 什么是 Agentic RuntimeAgentic Runtime可以理解为“AI Agent 的操作系统”。它不负责具体业务逻辑而是负责维护任务会话状态调度模型与工具控制执行循环记录执行轨迹处理超时、失败和恢复提供安全与权限拦截。如果你对比 JVM 和 Java 程序的关系就更容易理解了。JVM 负责内存、线程、异常等底层机制Java 程序只需要写业务逻辑。Agentic Runtime 对 Agent 来说也是同样的角色。1.3 Long-Horizon Reasoning 的四个特征Long-Horizon Reasoning 中文常翻译为“长视野推理”。它不是指“问题很长”而是指步骤多任务需要数十步甚至上百步才能完成依赖外部工具模型不能只靠内部知识需要查询数据库、调用 API、读写文件状态持续变化每一步的结果会影响后续决策执行过程不可完全预知中间可能遇到数据缺失、接口超时、权限不足等情况。一旦任务具备以上特征简单的“先后调用几个工具”就不够用了必须引入带状态、带恢复机制的运行时。1.4 Argus 的定位通用、可扩展、面向长时间任务Argus 这个名字代表着一种设计理念做一个通用的代理运行时而不是某个具体业务的工作流引擎。通用意味着不绑定某一个模型不内置某个行业工具执行循环与业务逻辑解耦支持不同任务类型复用同一套运行时。本文后面的实现就是按照这个方向从零构建一个最小可运行的 Argus 风格代理运行时。你可以把它作为理解 Agentic Runtime 的骨架也可以在此基础上扩展成生产级系统。2. Argus 的核心设计运行时代替硬编码工作流2.1 关键问题拆解在设计一个面向长视野推理的运行时之前要先回答下面几个问题问题传统做法运行时做法任务状态放哪里散落在局部变量里集中到 SessionContext下一步动作由谁决定写死在代码顺序里由 Planner 动态决策工具如何扩展直接调用函数注册到 ToolRegistry失败怎么办整体重跑记录轨迹、支持恢复怎么观察进度手动打印日志每步记录 execution steps这些问题就是 Argus 架构的最小需求集合。2.2 模块与职责一个标准代理运行时通常包含以下模块Session Manager创建、恢复、保存会话Context Manager维护任务上下文和中间结果Planner决定下一步执行什么Executor执行模型调用或工具调用Tool Hub统一管理工具注册与调用Checkpoint持久化任务状态Guard权限、超时、配额控制。模块之间职责必须清晰。特别是 Planner 和 Executor 不能混在一起否则任务一旦复杂代码会迅速失控。2.3 统一的执行循环无论任务具体是什么执行过程都可以抽象成下面的循环1. 读取当前上下文 2. Planner 根据上下文生成下一步动作 3. Executor 执行动作 4. 把执行结果写回上下文 5. 判断任务是否完成 6. 如果未完成写 Checkpoint 后回到第 1 步这样做最大的好处是业务任务只需要替换 Planner 和工具集运行时本身完全复用。2.4 对比传统实现维度硬编码工作流Agentic Runtime流程可读性流程写在代码里流程由数据驱动动态决策很难实现Planner 支持异常恢复需要手动处理Checkpoint 重放扩展新任务改主流程新增工具和 Planner可观测性依赖日志散点有完整执行轨迹从工程角度看Agentic Runtime 的核心不是“调用大模型”而是“管理长时间执行过程”。3. 环境准备与最小工程结构3.1 运行环境说明本文示例使用 Python 3核心代码只依赖标准库不依赖任何外部模型 API。这样做的原因是让重点落在运行时机制上而不是 API 调用细节。版本要求Python 3.9 或更高版本建议使用 venv 创建虚拟环境操作系统不限Windows、Linux、macOS 均可。如果你的项目中已经使用了其他 Python 版本代码本身兼容性较高只需注意dataclasses和类型注解语法即可。3.2 目录结构argus-demo/ ├── main.py ├── runtime.py ├── planner.py ├── tools.py ├── data/ │ ├── source_a.json │ └── source_b.json └── output/目录说明runtime.py上下文、工具注册表、代理运行时、Checkpoint 逻辑planner.py决策器决定每一步执行哪个工具tools.py具体业务工具main.py程序入口组装运行时并执行任务data源数据目录output结果与检查点目录由程序自动创建。3.3 依赖说明本文示例只使用标准库os json time traceback dataclasses typing不需要pip install任何第三方包。在真实项目中你可以把这里的 Runtime 与 FastAPI、Redis、消息队列等整合但核心机制是相通的。4. 关键实现从上下文、工具到执行循环4.1 SessionContext让任务状态成为一等公民在编写实际代码前先要理解一个核心概念长任务必须有“会话上下文”。如果没有上下文执行到第 8 步时第 3 步的结果就已经丢失了。SessionContext用 dataclass 来定义# runtime.py import time import json import os from dataclasses import dataclass, field, asdict from typing import Any, Dict, List, Optional dataclass class SessionContext: session_id: str task: str step: int 0 status: str running # running / finished / failed memory: Dict[str, Any] field(default_factorydict) steps: List[dict] field(default_factorylist) def append_step(self, action: str, output: Any None, error: str None): self.steps.append({ step: self.step, action: action, output: output, error: error, ts: time.time(), }) self.step 1字段含义session_id一次任务会话的唯一标识task任务描述便于人工查看step当前执行步数status任务状态用于恢复判断memory存放中间结果steps每一步的执行轨迹是排错与审计的关键。4.2 ToolRegistry工具不是散落的函数如果业务工具散落在各个模块中运行时很难统一管理权限和调用。合理的做法是建立工具注册表。# runtime.py class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict[str, Any]] {} def register(self, name: str, description: str): def decorator(func): self._tools[name] { func: func, description: description, } return func return decorator def call(self, name: str, args: Dict[str, Any]) - Any: if name not in self._tools: raise KeyError(ftool not found: {name}) return self._tools[name][func](**args) def list_tools(self): return {name: info[description] for name, info in self._tools.items()}注册表的好处工具名称全局唯一调用统一走call()方法可以在前后插入日志、鉴权、限流逻辑后续接入权限系统时只需在call()中增加检查。4.3 Planner把“下一步动作”与执行解耦Planner 负责根据当前上下文生成“下一个动作”。它决定了代理的执行策略。# planner.py class DemoPlanner: def plan_next(self, ctx): if a_rows not in ctx.memory: return { tool: read_source_file, args: {file_path: data/source_a.json}, store_as: a_rows, } if b_rows not in ctx.memory: return { tool: read_source_file, args: {file_path: data/source_b.json}, store_as: b_rows, } if merged not in ctx.memory: return { tool: merge_sales, args: { first_rows: ctx.memory[a_rows][rows], second_rows: ctx.memory[b_rows][rows], }, store_as: merged, } if summary not in ctx.memory: return { tool: summarize_sales, args: {rows: ctx.memory[merged][rows]}, store_as: summary, } if saved not in ctx.memory: return { tool: save_result, args: { result: ctx.memory[summary], output_path: output/result.json, }, store_as: saved, } return {done: True}这里我使用一个规则型 Planner 做演示。生产环境通常把plan_next替换为大模型调用让模型根据上下文动态决策。无论哪种方式对外接口是一致的。动作字典中的store_as表示把工具输出写入上下文的哪个 key。这个设计让中间结果能够被后续步骤引用。4.4 AgenticRuntime循环、超时与最大步数运行时是本系统的核心它负责驱动整个循环。# runtime.py class AgenticRuntime: def __init__(self, tools: ToolRegistry, max_steps: int 30, timeout_seconds: int 600): self.tools tools self.max_steps max_steps self.timeout_seconds timeout_seconds def run(self, ctx: SessionContext, planner) - SessionContext: start time.time() while ctx.step self.max_steps: if time.time() - start self.timeout_seconds: ctx.status failed ctx.append_step(action__system__, errortimeout) return ctx try: action planner.plan_next(ctx) except Exception as exc: ctx.status failed ctx.append_step(action__planner__, errorstr(exc)) return ctx if action.get(done): ctx.status finished ctx.append_step(action__done__) return ctx tool_name action.get(tool) args action.get(args, {}) try: output self.tools.call(tool_name, args) if action.get(store_as): ctx.memory[action[store_as]] output ctx.append_step(actiontool_name, outputoutput) except Exception as exc: ctx.status failed ctx.append_step(actiontool_name, errorstr(exc)) return ctx ctx.status failed if ctx.status running else ctx.status ctx.append_step(action__max_steps__, errormax steps reached) return ctx这段代码实现了四个关键机制步数保护防止任务进入死循环超时保护防止任务无限运行异常捕获把错误写进轨迹而不是直接崩溃结果回写每一步的产物自动写入 memory。4.5 Checkpoint失败后能恢复长任务最怕“跑到一半断电”。Checkpoint 机制把上下文持久化到磁盘重启后可恢复。# runtime.py def save_checkpoint(ctx: SessionContext, checkpoint_dir: str output) - str: os.makedirs(checkpoint_dir, exist_okTrue) path os.path.join(checkpoint_dir, f{ctx.session_id}.json) with open(path, w, encodingutf-8) as f: json.dump(asdict(ctx), f, ensure_asciiFalse, indent2) return path def load_checkpoint(session_id: str, checkpoint_dir: str output) - Optional[SessionContext]: path os.path.join(checkpoint_dir, f{session_id}.json) if not os.path.exists(path): return None with open(path, r, encodingutf-8) as f: raw json.load(f) return SessionContext( session_idraw[session_id], taskraw.get(task, ), stepraw.get(step, 0), statusraw.get(status, running), memoryraw.get(memory, {}), stepsraw.get(steps, []), )恢复逻辑很简单下次启动时加载该 session 的 Checkpoint跳过 memory 中已有的步骤继续执行未完成的工作。这就是“断点续跑”的基础。5. 完整实战多步骤销售数据汇总任务5.1 需求描述假设我们有两个数据文件分别来自两个区域的销售记录。任务是把它们合并计算销售总额并把结果写入最终文件。整个任务有 4 个工具步骤读文件 A读文件 B合并数据汇总计算并保存。这个例子虽然简单但完整展示了 Agentic Runtime 的执行循环。5.2 准备数据文件创建data/source_a.json[ {region: east, amount: 1200.5, order_id: A001}, {region: east, amount: 430.2, order_id: A002} ]创建data/source_b.json[ {region: west, amount: 800.0, order_id: B001}, {region: west, amount: 260.75, order_id: B002} ]5.3 编写工具函数创建tools.py# tools.py import json import os from pathlib import Path def read_source_file(file_path: str) - dict: 读取 JSON 源文件返回统一结构。 if not os.path.exists(file_path): raise FileNotFoundError(fsource file not found: {file_path}) with open(file_path, r, encodingutf-8) as f: rows json.load(f) if not isinstance(rows, list): raise ValueError(ffile {file_path} must contain an array) return {rows: rows} def merge_sales(first_rows: list, second_rows: list) - dict: 合并两个销售数据数组。 merged list(first_rows) list(second_rows) return {rows: merged} def summarize_sales(rows: list) - dict: 计算销售总金额和记录数。 total sum(float(row[amount]) for row in rows) return {total_amount: total, record_count: len(rows)} def save_result(result: dict, output_path: str) - dict: 把最终结果写入 JSON 文件。 Path(output_path).parent.mkdir(parentsTrue, exist_okTrue) with open(output_path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) return {saved_to: output_path}5.4 组装运行时创建main.py# main.py from runtime import ToolRegistry, AgenticRuntime, SessionContext, save_checkpoint from tools import read_source_file, merge_sales, summarize_sales, save_result from planner import DemoPlanner def main(): # 1. 创建工具注册表并注册工具 registry ToolRegistry() registry.register(read_source_file, 读取 JSON 销售数据源)(read_source_file) registry.register(merge_sales, 合并两个销售数据数组)(merge_sales) registry.register(summarize_sales, 计算销售总金额)(summarize_sales) registry.register(save_result, 保存最终结果)(save_result) # 2. 创建会话上下文 ctx SessionContext( session_iddemo-001, task汇总两个销售数据源并生成报告, ) # 3. 创建运行时并执行 runtime AgenticRuntime(toolsregistry, max_steps10, timeout_seconds30) ctx runtime.run(ctx, DemoPlanner()) # 4. 输出结果 if ctx.status finished: checkpoint_path save_checkpoint(ctx) print(task finished.) print(summary:, ctx.memory[summary]) print(checkpoint:, checkpoint_path) print(steps:, ctx.step) else: print(task failed.) for step in ctx.steps[-3:]: print(step) if __name__ __main__: main()5.5 运行与验证在项目根目录执行python main.py预期输出类似task finished. summary: {total_amount: 2691.45, record_count: 4} checkpoint: output/demo-001.json steps: 5同时output/demo-001.json会保存完整的执行轨迹。用编辑器打开后可以看到 memory 中有a_rows、b_rows、merged、summary、saved等字段steps 中记录了每一步使用了哪个工具。这里“steps: 5”的意思是读 A、读 B、合并、汇总、保存四个业务步骤加上最后的__done__标记共 5 条轨迹记录。6. 长时间任务的可观测性、恢复与控制6.1 用 steps 记录完整轨迹长任务最容易出的问题是“不知道卡在哪一步”。在 Argus 风格运行时中每一步都会被写入steps列表记录动作、输出、错误和时间戳。这套轨迹可以做三件事排错定位到具体是哪一步失败审计回溯一个任务执行了哪些工具回放按步骤重新执行或分析。如果接入日志系统建议在append_step时同步输出结构化日志比如 JSON 格式方便导入 Elasticsearch 等分析平台。6.2 断点恢复流程当任务执行到一半进程崩溃时恢复流程如下检查output/{session_id}.json是否存在如果存在调用load_checkpoint恢复上下文检查 status 是否为running调用runtime.run(ctx, planner)继续执行因为 memory 中已有中间结果未完成的步骤会被跳过。真实生产环境中Checkpoint 不一定要保存到本地文件可以是 Redis、数据库或对象存储。核心原则是所有中间状态都要可重建。6.3 超时、取消与并发控制运行时需要提供的控制维度包括整个任务超时已在timeout_seconds中实现单步工具超时可以在ToolRegistry.call()中给每个工具包一层超时装饰器手动取消外部系统通过检查 status 字段或向运行时发送取消信号并发限制同一个 session 不应被多个进程同时执行需要引入分布式锁。这里特别提醒在真实项目中千万不要让同一个 session 被重复调度否则会出现“重复写文件”“重复调接口”等问题。6.4 安全边界Agent 执行工具时权限控制至关重要。建议在工具调用前增加校验1. 这个 session 是否有权调用该工具 2. 参数是否在白名单范围内 3. 是否涉及敏感操作需要二次确认运行时只是执行框架它本身不替业务做安全决策但必须提供安全拦截的扩展点。7. 常见问题与排查思路在实际使用代理运行时时下面这些问题是高频出现的。问题现象常见原因解决思路任务永远跑不完没有最大步数保护设置 max_steps检查 Planner 是否产生终止信号中间结果丢失上下文只存在局部变量把状态集中到 SessionContext并开启 Checkpoint同一个任务重复执行多个调度器并发处理增加 session 状态判断和分布式锁工具报错但不知道在哪里没有执行轨迹检查 steps 列表定位最后一个非成功步骤恢复后从零开始没有加载 Checkpoint启动时先调用 load_checkpoint 判断是否已有上下文Planner 死循环条件判断遗漏在 Planner 中增加“已完成字段”集合检查排查时可以做一个最小复现脚本只跑一个工具确认工具本身正常再去怀疑运行时逻辑。这里再强调一个容易踩坑的点Planner 的plan_next必须保证“一定能返回 done 条件”。如果只写“如果 A 不存在就执行 A”但 A 始终不存在任务就会一直循环。建议在每次plan_next前设置一个已完成任务列表逐项检查。8. 最佳实践与工程建议8.1 规划器设计规则型 Planner 适合步骤固定的任务性能高、可解释性强LLM Planner 适合开放任务但必须设置最大重试次数无论哪种 Planner都建议返回tool、args、store_as三个字段保持接口稳定对 LLM 的输出要做校验防止模型生成不存在的工具名。8.2 工具设计每个工具只做一件事入参必须显式声明工具返回值尽量使用 dict统一结构方便写入 memory工具的副作用要写清楚是否会改数据库是否会发消息是否删除文件建议为每个工具增加 manifest描述入参、出参、权限级别。8.3 状态管理session_id 需要全局唯一可用 UUIDCheckpoint 写入要具备原子性避免写到一半崩溃如果 Checkpoint 数据量大可以只保存“必要中间结果”不保存完整原始数据考虑给 Checkpoint 增加版本号防止结构变更后无法加载。8.4 可观测性每步都要有 trace_id方便把日志串起来工具调用前后要记录耗时失败时不仅记录错误信息还要记录当时的 action 和 memory 快照如果跑了多轮重试保留重试次数避免无限重试。8.5 生产落地建议不要把所有 Agent 逻辑写到一个超大类里模块拆分是底线上线前做故障演练杀掉进程验证 Checkpoint 恢复对涉及外部系统变更的工具先在小流量下灰度涉及删除、写入、支付等高风险操作时必须有审批或人工确认环节所有配置优先走环境变量或配置中心避免硬编码。9. 从 Demo 到可用的代理运行时本文写了一个很小的 Argus 风格运行时它不依赖任何第三方库却完整覆盖了代理运行时的核心循环上下文、Planner、ToolRegistry、执行循环和 Checkpoint。你完全可以把它当成一个骨架在此基础上继续扩展把DemoPlanner替换成大模型 Planner把ToolRegistry.call()中增加权限校验把save_checkpoint从本地文件改为 Redis在AgenticRuntime.run()中增加单步超时和重试策略。如果你正在设计自己的 Agent 系统我建议先从这个最小闭环开始不要一上来就引入复杂的编排框架。把“状态管理”和“执行轨迹”这两个基础能力做扎实再逐步叠加模型能力系统会稳定很多。Long-Horizon Reasoning 的难点从来不只是模型聪明不聪明而是执行过程是否可控、可恢复、可审计。把这三点想清楚你的代理运行时就已经及格了。