Run

Run 是 Agent 一次完整的任务执行实例，从用户输入到最终输出的完整生命周期。它包含多个 Step，每个 Step 是一次推理-动作循环。

#type / concept #status / evergreen #tech / ai #tech / architecture

[!info] related notes

所属 MOC: Agent Runtime MOC

上游概念: Agent Runtime

并列概念: Step, Run State, Trace

下游概念: Checkpoint

Run

一句话定义

Run 是 Agent 一次完整的任务执行实例，从接收用户输入到生成最终输出（或失败/取消）的完整生命周期。一个 Run 包含多个 Step，每次 LLM 推理 + 工具执行构成一个 Step。

它解决什么问题

Agent 的执行不是一次 API 调用，而是一个可能跨越多轮推理、多次工具调用的长过程。需要一个抽象来管理这个过程的：

生命周期：什么时候开始、什么时候结束
状态追踪：当前处于什么阶段
资源管理：token 消耗、时间消耗
错误处理：某步失败怎么办
可观测性：执行过程的完整记录

核心原理

Run 的生命周期

pending → running → completed
                  → failed
                  → cancelled
                  → timeout

Run 的数据结构

@dataclass
class Run:
    run_id: str                    # 唯一标识
    session_id: str                # 所属会话
    status: RunStatus              # 当前状态
    created_at: datetime           # 创建时间
    started_at: datetime           # 开始执行时间
    completed_at: datetime         # 完成时间

    # 输入
    input_message: str             # 用户输入
    context: ContextBundle         # 上下文包

    # 执行过程
    steps: list[Step]              # 所有步骤
    current_step: int              # 当前步骤索引

    # 输出
    output: str                    # 最终输出
    tool_calls: list[ToolCall]     # 所有工具调用

    # 资源消耗
    total_tokens: int              # 总 token 消耗
    total_duration_ms: int         # 总耗时

    # 错误
    error: Optional[Error]         # 错误信息

Run 与 Step 的关系

Run (一次任务执行)
├── Step 1: LLM 推理 → 决定调用搜索工具
├── Step 2: 执行搜索工具 → 结果回传 LLM
├── Step 3: LLM 推理 → 决定调用分析工具
├── Step 4: 执行分析工具 → 结果回传 LLM
└── Step 5: LLM 推理 → 生成最终回复

典型工程实现

Run Manager

class RunManager:
    def __init__(self, agent_engine, checkpoint_store):
        self.engine = agent_engine
        self.checkpoint_store = checkpoint_store

    async def start_run(self, input_message: str, context: ContextBundle) -> Run:
        run = Run(
            run_id=generate_id(),
            session_id=context.session_id,
            status=RunStatus.PENDING,
            input_message=input_message,
            context=context,
        )
        await self.checkpoint_store.save(run)
        return run

    async def execute_run(self, run: Run) -> AsyncIterator[Event]:
        run.status = RunStatus.RUNNING
        run.started_at = datetime.now()

        try:
            async for event in self.engine.run(run.context):
                # 记录事件
                run.steps[-1].events.append(event)

                # 更新 token 消耗
                if hasattr(event, 'tokens'):
                    run.total_tokens += event.tokens

                # 定期保存 checkpoint
                if should_checkpoint(run):
                    await self.checkpoint_store.save(run)

                yield event

            run.status = RunStatus.COMPLETED

        except CancelledError:
            run.status = RunStatus.CANCELLED

        except TimeoutError:
            run.status = RunStatus.TIMEOUT

        except Exception as e:
            run.status = RunStatus.FAILED
            run.error = Error.from_exception(e)

        finally:
            run.completed_at = datetime.now()
            run.total_duration_ms = (run.completed_at - run.started_at).total_seconds() * 1000
            await self.checkpoint_store.save(run)

常见设计模式

1. 同步 Run

用户发送消息 → 等待 Run 完成 → 返回结果。适合简单场景。

2. 异步 Run

用户发送消息 → 立即返回 run_id → 通过 SSE 推送进度 → Run 完成后推送最终结果。

3. 可恢复 Run

Run 中断后，从最近的 Checkpoint 恢复执行。

常见坑

不做状态持久化: 服务重启后丢失所有进行中的 Run
不设超时: Run 可能无限执行
不追踪 token: 无法监控成本
Run 和 Session 混淆: 一个 Session 可以有多个 Run
不做并发控制: 同一个 Session 同时启动多个 Run

和其他概念的关系

vs Step: Run 包含多个 Step
vs Run State: Run State 是 Run 的状态机
vs Trace: Trace 记录 Run 的执行过程
vs Checkpoint: Checkpoint 是 Run 的状态快照
vs Agent Runtime: Runtime 管理 Run 的生命周期

总结

Run 是 Agent 执行的基本调度单元。理解 Run 的生命周期，才能做好 Agent 的状态管理和错误恢复。

Run