> ## Documentation Index
> Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 检查点与分支

> 使用 QitOS 检查点机制保存、恢复和分支 Agent 运行。

# 检查点与分支

QitOS 检查点让你可以在任意步骤保存 Agent 状态，在中断后恢复运行，以及从某个步骤分叉运行来探索替代路径——所有这些都不需要从头重新运行。

这对以下场景至关重要：

* **长时间运行的任务**，可能被中断（网络错误、预算限制）
* **分支实验**，你想从特定步骤探索"如果…会怎样？"
* **时间旅行调试**，你从更早的检查点恢复状态

## Step 1: 基本检查点

创建 `CheckpointStore` 并配置 Engine 使用它：

```python theme={null}
from qitos.checkpoint import InMemoryCheckpointStore, CheckpointConfig

# 创建内存存储（使用 SqliteCheckpointStore 进行持久化）
store = InMemoryCheckpointStore()

# 传递给 Engine
from qitos.engine import Engine
from my_agent import MyAgent

agent = MyAgent()
engine = Engine(agent, checkpoint_store=store, auto_approve=True)

# 运行 Agent — 检查点自动保存
result = engine.run("分析数据集")

# 访问最新检查点
config = CheckpointConfig(thread_id=result.run_id)
latest = store.get_tuple(config)
print(f"Checkpoint at step {latest.checkpoint.step}")
```

跨会话持久化存储，使用 `SqliteCheckpointStore`：

```python theme={null}
from qitos.checkpoint import SqliteCheckpointStore

store = SqliteCheckpointStore("./checkpoints.db")
```

## Step 2: 恢复运行

如果运行被中断，从上次检查点恢复：

```python theme={null}
from qitos.checkpoint import CheckpointConfig

# thread_id 标识运行（通常是 run_id）
config = CheckpointConfig(thread_id="my-run-001")

# 恢复运行
result = engine.run("继续分析", checkpoint_config=config)
```

Engine 从最新检查点恢复状态，从中断处继续执行。

## Step 3: 分叉检查点

分叉从现有检查点创建新分支，允许你探索替代路径而不影响原始运行：

```python theme={null}
from qitos.checkpoint import fork_checkpoint, CheckpointConfig

# 指向特定检查点（例如第 3 步）
source_config = CheckpointConfig(
    thread_id="my-run-001",
    checkpoint_id="abc123",  # 要分叉的检查点
)

# 分叉到新线程（真正分支）
fork_config = fork_checkpoint(store, source_config, new_thread_id="my-run-001-branch-a")

# 从分叉状态以不同提示运行 Agent
result = engine.run(
    "尝试不同策略",
    checkpoint_config=fork_config,
)
```

原始运行不受影响。分叉运行从相同状态开始，但可以走不同路径。

## Step 4: 时间旅行（同线程分叉）

如果省略 `new_thread_id`，分叉替换同线程内的当前状态——适用于"撤销并重试"场景：

```python theme={null}
from qitos.checkpoint import fork_checkpoint, CheckpointConfig

# 在同一线程内分叉（时间旅行）
config = CheckpointConfig(
    thread_id="my-run-001",
    checkpoint_id="step2-checkpoint-id",
)
time_travel_config = fork_checkpoint(store, config)

# 从更早状态重新运行
result = engine.run("从第 2 步重新开始，修改指令", checkpoint_config=time_travel_config)
```

## Step 5: 持久化模式

控制检查点持久化的激进程度：

```python theme={null}
from qitos.checkpoint import DurabilityMode

# SYNC: 继续前写入磁盘（最安全，较慢）
engine = Engine(agent, checkpoint_store=store, checkpoint_durability=DurabilityMode.SYNC)

# 默认: 检查点已保存，但持久化取决于存储实现
```

## Step 6: 状态版本追踪

追踪检查点之间哪些状态字段发生了变化：

```python theme={null}
from qitos.checkpoint import StateVersionTracker

tracker = StateVersionTracker()

# 每步之后，追踪变化
config = CheckpointConfig(thread_id="my-run-001")
tuple_ = store.get_tuple(config)
versions = tuple_.checkpoint.state_versions
print(f"State field versions: {versions}")
# 例如 {'scratchpad': 3, 'current_step': 5, 'final_result': 1}
```

***

<CardGroup cols={2}>
  <Card href="/zh/tutorials/critic-system" title="Critic 系统" description="运行时控制 Agent 行为质量" />

  <Card href="/zh/tutorials/hook-lifecycle" title="Hook 生命周期" description="观察和扩展 Engine 循环" />
</CardGroup>