第 13 章第二部分：架构原理

Runner 与批量执行

Single Runner 7 步编排、run_agent 工厂、Batch Runner 的 SQLite 后端、并发控制、Resume/Retry

最后更新：2026年10月6日 Nexus Runner Batch SQLite 编排

Single Runner：7 步编排

nexus/runner/single.py（546 行）是框架的入口——把配置变成执行。

async def run(run_input: RunInput) -> RunOutput:
    # 0. 加载 imports 中声明的模块
    for module_path in run_input.imports:
        Registry.import_module(module_path)

    # 1. 创建 RuntimeProvider
    runtime_provider = _create_runtime_provider(run_input)

    # 2. 创建 OpenAI Clients
    openai_clients = &#123;key: create_client(spec) for key, spec in run_input.openai_clients.items()&#125;

    # 3. 构建 run_agent 工厂函数
    run_agent, agent_capture = _build_run_agent_factory(run_input, task_needs_agent)

    # 4. 构建 Task Input
    task_input = _build_input(task_input_type, &#123;**task_config, run_agent=run_agent&#125;)

    # 5. 创建 Span Processors
    processors = _create_processors(run_input.processors)

    # 6. 加载 Cache
    cache_dict = load_cache(cache_config.file_path) if cache_enabled else None

    # 7. 执行
    async with ExecutionContext(...) as ctx:
        with processor_context(processors):
            with cache_context(cache_dict):
                task_output = await task_fn(task_input)

run_agent 工厂

Runner 不是直接把 Agent 函数传给 Task，而是构建一个工厂函数。工厂函数封装了 Agent 的基础配置（从 YAML），Task 只需要传业务参数：

async def run_agent(**task_kwargs) -> AgentRunOutput:
    merged_kwargs = &#123;**run_input.agent_config, **task_kwargs&#125;
    agent_input = _build_input(agent_input_type, merged_kwargs)
    result = await agent_fn(agent_input)
    capture.output = result  # 捕获 Agent 输出供 RunOutput 使用
    return result

RunInput / RunOutput

RunInput 是完全可序列化的（Pydantic BaseModel）——这使得分布式调度成为可能。

RunOutput 包含：

task_output：Task 函数的返回值
agent_output：Agent 函数的返回值（如果调用了 run_agent）
processor_results：各 Processor 的结果
cache_stats：缓存统计
success + error：执行状态

OmegaConf 配置解析

RunInput.from_file() 使用 OmegaConf 解析 YAML，支持：

环境变量插值：${oc.env:API_KEY}
自由文本保护：task_config.instance、agent_config.user_prompt 等字段的 ${...} 不会被解析

Batch Runner

nexus/runner/batch/（7 个文件，~95KB）实现了批量执行。

两阶段架构

Phase 1: Generation
  generator.py → generate_configs() → [BatchItem, ...]
  → 写入 SQLite database

Phase 2: Execution
  从 database 读取 pending/failed 的 runs
  → asyncio.Semaphore(max_concurrent)
  → 并行执行 single.run()
  → 写回 database

SQLite 后端

所有数据存在一个 SQLite 文件中（BatchDatabase，19.8KB）：

表	内容
`runs`	配置 + 状态 + 输出 + 错误 + 时间戳
`trace_events`	Span 事件（如果 enable_trace_db=True）
`log_chunks`	日志块（如果 enable_log_capture=True）
`metadata`	批次元数据

SQLite 的好处：单文件、无服务器、支持并发读写（WAL 模式）、天然支持 resume。

一处缺口值得点出：这四张表都没有 schema 版本字段。跨版本读写同一个 results.db 时无从判断兼容性，§scp 一文件迁移调试的便利性会被版本错配抵消——OpenAI Agents SDK 的 RunState 就显式维护了 CURRENT_SCHEMA_VERSION 来兜这件事（见下文佐证段）。建议在 metadata 表补一个版本号。📌 Nexus 内部是否已存在该字段，本章未体现，需对照 BatchDatabase 实际 schema 复核。

更准确地说，该补的是三件套而不是一个字段。 OpenAI Agents SDK 的 run_state.py 顶部写明了四条 schema 策略（2026-10-06 回源核对，逐字）：① 已发布版本保持可读；② 未发布版本可在发布前重编号或合并；③ to_json() 始终写 CURRENT_SCHEMA_VERSION；④ “Forward compatibility is intentionally fail-fast (older SDKs reject newer or unsupported versions)“。落到实现上是三样东西：

metadata 表加 schema 版本号；
维护一张 version → 一行摘要的登记表（源码注释原文要求 “Keep this mapping in chronological order. Every schema bump must add a one-line summary here.”），并用启动期断言强制——CURRENT_SCHEMA_VERSION 必须在摘要表内有对应项，且每个受支持版本必须有非空摘要；
读取时遇到未知版本直接 fail-fast 拒绝，而不是”尽力解析”。

只做第 (1) 条的后果是：跨版本读到旧库时不报错，但读到的数据结构含义已经变了——这比直接拒绝更难排查。

SQLite-for-Agent 设计决策论证

Nexus 选择 SQLite 而非分布式中间件（如 Kafka + Redis + PostgreSQL）作为 Agent 批量执行的后端，这一决策背后有系统性的工程论证：

Agent 的 Single-Tenant Focus 特性：Agent 批量执行通常是单用户/单项目场景，分布式中间件的多租户能力是伪需求。SQLite 的单文件模型天然匹配单租户场景
延迟比吞吐量更关键：LLM API 调用已是毫秒-秒级延迟，额外的网络跳转（如 Redis 查询任务状态）是纯粹的惩罚。SQLite 的进程内访问将任务状态查询延迟降至微秒级
WAL 读写分离可支撑数百并行子 Agent：WAL 模式允许多个读连接与一个写连接并行工作，足以覆盖 Nexus Batch Runner 的并发规模
RETURNING 子句实现原子任务申领：UPDATE runs SET status='running' WHERE run_id=? RETURNING * 一条 SQL 完成状态变更和数据读取，无需分布式锁
Savepoint 支持精确断点恢复：SAVEPOINT batch_checkpoint + ROLLBACK TO batch_checkpoint 实现子任务粒度的断点恢复，比分布式事务更轻量
scp 一文件迁移调试：scp results.db remote:debug/ 即可迁移全部执行数据，而分布式方案需要同步多个数据存储

这一设计哲学在多个生产级 Agent 框架中得到验证——OpenClaw 将 ACP/subagent/cron/后台 CLI 四种执行体统一到 SQLite-backed 任务账本，Hermes Kanban 用 ~/.hermes/kanban.db 做多 Agent 看板编排。SQLite-for-Agent 已从”简单场景的简单选择”演变为”Agent 基础设施的有意识设计决策”。

第三条佐证来自沙箱编排层：K8s agent-sandbox 在 v1.0.5（2026-10-01）集成了 OpenAI Agents（#1388，含 workspace hydration 与 session resumption，并给出 Ray RLlib PPO 端到端示例），v1.0.4（2026-09-24）增强了 RL SDK 的并发隔离。沙箱编排与 Agent SDK 两个方向同时向 RL 批量执行场景收敛，与本章”SQLite 足以覆盖 Agent 批量执行规模”的判断相互印证（此为本书分析推论 📌）。

这一判断如今有了官方 SDK 级的佐证：OpenAI Agents SDK 把 SQLite 列为 Session 层的一等实现——agents.memory.SQLiteSession，extensions/memory 下另有 AdvancedSQLiteSession、AsyncSQLiteSession、RedisSession、SQLAlchemySession、MongoDBSession、DaprSession、EncryptedSession；持久化边界则由可序列化的 RunState 承担，to_json() 快照带 CURRENT_SCHEMA_VERSION（2026-10-06 查 src/agents/run_state.py 得 1.18，对应 openai-agents v0.23.1 / 2026-10-02；2026-09-16 时该值为 1.17），用于对接 Temporal / Restate / DBOS。本章的选型论证由此从社区实践升到官方 SDK 背书。

持久化的边界：resume 不等于不重复副作用

上面六条讲的是 SQLite 为什么够用，还差一条反向的边界。本章的状态机（pending → running → success/failed/timeout/cancelled）恢复的是 run 级状态，不是工具副作用：--retry-only 重试一个”工具已执行成功、但结果写回 DB 前崩溃”的 run 时，整个 run 从头重跑；若该 run 的 Task 含写操作（写文件、发请求、拉镜像），副作用就重复一次。

对照 Zylos Research 2026-04 调研给出的七条持久化原语，可以看清 Nexus 覆盖到哪里：

原语	Batch Runner	说明
事件历史 / 步骤日志	❌	`trace_events` 默认关闭（`enable_trace_db=True` 才落盘），`log_chunks` 同理
持久化步骤结果	✅ 部分	由 run 级 success/failed 体现，非步骤级
持久化定时器 / sleep	❌	无
外部事件等待	❌	无
重试策略	✅	`--retry-only --retry-failed 3`
幂等键	❌	本章 `runs` 表为「配置 + 状态 + 输出 + 错误 + 时间戳」，未见 attempt 计数与幂等键字段 📌
子工作流 / 子任务	❌	无

同一份调研对”会话记忆 ≠ 持久化执行”的区分正好点破这里的位置：保存聊天历史能让 Agent “记得”它在做什么，证明不了哪条命令真的跑过、哪次重试会不会重复某个副作用。results.db 落在两者之间——比纯会话记忆强（有 run 级状态与时间戳），比持久化执行弱（无步骤日志与幂等键）。

还有一类对象值得单列：执行环境的身份与状态。 上面七原语讲的都是”步骤与副作用”，而 OpenAI Agents SDK 的 RunState 在 1.16 / 1.17 / 1.18 三个版本里，依次把 Docker 网络隔离状态、Docker 容器标签与当前响应生成项归属、compaction 元数据（含确认与模型交换证据，供压缩重试用）以及具名 Vercel 沙箱与精确执行 ID 纳入持久化快照。也就是说，在这条产品线上”持久化执行”不仅包含步骤与幂等，还包含这次跑在哪个环境上。

对 Nexus 的含义：除了”重试会不会重复副作用”，还缺一条”这次跑在哪个沙箱上”——本章前述 runs 表的字段描述（配置 + 状态 + 输出 + 错误 + 时间戳）中未见 provider / 镜像 / 容器标识一类记录 📌（Nexus 无公开源可核，需对照 BatchDatabase 实际 schema）。对 RL 训练数据采集这类需要复现的场景，后者与幂等键同等重要。同一对象的另一面见 Ch14 §Compaction 作为一等可观测事件——那里主张把 compaction 状态做成可观测事件，这里主张把它放进持久化快照，合起来才是完整主张。

实践含义只有一句：重试前确认该 Task 的工具是幂等的。要往前走，成本最低的一步是在 runs 表加 attempt 计数字段；加幂等键要求工具本身支持，改动大得多。

📌 待核实：上表”缺哪几项”依据本章对 runs 表字段的描述推断，Nexus 无公开源可核，未对照 BatchDatabase 实际 schema。

结果契约：跑完之后交出去的是什么

上面讲的都是”结果存在哪”（SQLite 四表）。还有一问没答：跑完之后交出去的那个东西是什么格式？ 对”结果只给自己看”的场景这不是问题；对要跟别人对比、要进论文、要给下游训练管线消费的场景，它就是问题。

Harbor 把每个 trial 的产物固定成一组路径（models/trial/paths.py）：

jobs/<job>/<trial>/
├── trial.log                 # 执行日志
└── verifier/
    ├── reward.txt            # 0 / 1（或连续值）
    ├── reward.json
    └── ctrf.json             # CTRF 标准测试报告

trial.log 与 reward.txt 不稀奇，稀奇的是 ctrf.json——CTRF（Common Test Report Format）是跨语言、跨测试框架的通用测试报告 JSON schema。把它列进默认产物意味着一次 run 的结果可以直接喂给任何吃 CTRF 的工具（CI 报告、聚合看板、可视化），中间不需要再写一层转换。

对照 Nexus：导出是 nexus export --format jsonl | csv | full，full 目录为 config.json / output.json / metadata.json / processors/*.json / summary.json——全是自定义格式。差异的后果有三条：

跨框架对比要靠人肉对齐。 同一 Agent 在 Nexus 与 Harbor 各跑一遍，两边结果文件没有公共 schema，比较只能靠手动搬数字。
结果进不了现成的 CI / 报告工具链。 summary.json 不是任何标准报告格式，想挂上去就得自己写转换器。
外部难以复核。 第 10 章提出过”若要向外部证明 RL 训练数据的采集过程，签名化的执行历史是比单个 SQLite 文件更强的证据链”——但那讲的是证明，本节讲的是出示：CTRF 解决的是”结果能被别人用标准工具直接读走”。

需要说明的是：这不等于 Nexus 该把存储换成 CTRF。 SQLite 作为执行后端（WAL / resume / 并发申领）与导出格式是两个层次的决定，前者没有理由动。要补的只是导出层多一个标准格式的 target——把 summary.json 同时写成 CTRF 是一层薄适配器，不动四表结构，也不影响现有三种导出。

⚠️ 本节 Harbor 产物路径核于 v0.24.0 / 2026-10-10。另需分清：CTRF 解决的是结果可交换，不是结果可信——它同样不带签名、不带环境指纹。要证明”这个分数是在这个环境里跑出来的”仍要靠第 10 章那条签名化执行历史的路线。两者互补，不可互相替代。

并发控制

semaphore = asyncio.Semaphore(batch_config.max_concurrent)

async def _run_one(item):
    async with semaphore:
        result = await run_single(item.config)
        db.save_result(item.run_id, result)

Conflict Groups

batch_config.enable_conflict_groups=True 时，metadata.conflict_groups 中相同组的任务串行执行。用于共享资源的场景（如同一个 Docker 镜像不能并行拉取）。

DAG 依赖编排方向

当前 Batch Runner 仅支持两种编排模式——Semaphore 并行 + Conflict Groups 串行。然而，Agent 工作流中常见更复杂的依赖关系：任务 B 依赖任务 A 的输出（而非仅资源冲突），需要 fan-out/fan-in 编排。

SilkLoom Core 的 add_node(..., depends_on=[...]) 模式提供了 DAG 依赖编排的参考：每个节点声明其依赖的前置节点，调度器自动计算拓扑序并行执行无依赖节点。与 Conflict Groups 的关键区别在于：

维度	Conflict Groups	DAG 依赖
解决的问题	资源冲突（同镜像不并行）	逻辑依赖（B 需要 A 的输出）
调度方式	同组串行，异组并行	拓扑排序，无依赖节点并行
数据传递	无（各 run 独立）	前置节点输出 → 后继节点输入

在 Nexus 中扩展 DAG 支持，可在 generator.py 的 config schema 中增加 depends_on 字段，Batch Runner 在 Phase 2 执行时按拓扑序调度。但需注意：DAG 编排引入的复杂性（循环检测、部分失败回滚、中间结果传递）可能偏离 Nexus “简单可靠” 的设计哲学，需权衡必要性。

这道权衡题在过去一个季度已经有了现成答案——durable execution 赛道把上面三件”复杂事”做成了标准件：

本章担心的复杂性	业界现成解法	代表实现
循环检测	编译期 / 注册期图校验	Temporal：工作流定义即代码，重放时命令序列与已记录历史不符即报 non-determinism error（规则 TMPRL1100）
部分失败回滚	日志化步骤 + 已完成步骤重放	Restate journal、Inngest / Hatchet step memoization
中间结果传递	持久化步骤结果 + KV 状态	Restate Virtual Object、DBOS `@step` 返回值落 Postgres
长时间等待人工 / 外部事件	持久化定时器 + awakeable	Restate awakeable、Inngest `step.waitForEvent`（零计算挂起）

Restate 官方把这类引擎收敛为四条一致原则：日志化步骤（外部交互的结果在返回给应用代码之前先落盘）、带幂等的自动重试（已完成步骤不重跑，重放已记录结果）、持久化定时器与信号（sleep、定时任务、人工审批等待可在崩溃与重启后存活，且不占用活进程）、可恢复性（任何在途执行可被任意健康 worker 透明接管）。

由此多出一条比自建 DAG 更轻的路径：不把 DAG 做进 Batch Runner，而是在需要 DAG 的场景把单个 run 交给外部 durable engine 托管，Nexus 只保留 RunInput 的可序列化契约——本章开头那句”RunInput 是完全可序列化的……这使得分布式调度成为可能”恰好就是现成的接口边界。是否接受这层外部依赖属编辑判断，需与 Nexus 团队确认 📌。

再往前一步是可验证执行：Dapr 1.18（2026-06）以工作流历史签名的形式把它引入运行时，Diagrid Catalyst 2.0（2026-07）推广到十个主流 Agent 框架。对 Nexus 的意义在于，批量执行的结果目前只能靠 results.db 自证；若将来需要向外部证明 RL 训练数据的采集过程，签名化的执行历史是比单个 SQLite 文件更强的证据链。📌 该条的时间与能力描述来自单一中文综述转述（2026-10-06 复测 releases.atom：1.18 主线最新稳定为 v1.18.4 / 2026-09-09；此后 v1.18.5-rc.1 于 2026-09-29、v1.19.0-rc.1 于 2026-10-01 已发布，1.19 尚未 GA；Dapr 官方 release notes 仍未逐条核对），属”值得知道”而非”应该做”。

Resume / Retry

# 续跑：跳过已成功的
nexus run-batch -d results.db

# 只重试失败的
nexus run-batch -d results.db --retry-only --retry-failed 3

数据库中每个 run 有状态：pending → running → success/failed/timeout/cancelled。

Inspector

batch/inspector.py（32.7KB）提供了丰富的批次检视能力——nexus inspect-batch 可以查看每个 run 的状态、输出、日志、Trace。