Agent 中断后怎么继续：检查点与持久状态

{"next_step": 1, "data": {"normalized": "hello agent"}}

mkdir "F:\学习\l4-checkpoints" -Force
cd "F:\学习\l4-checkpoints"

mkdir "F:\学习\l4-checkpoints" -Force
cd "F:\学习\l4-checkpoints"

"""将文本处理进度保存到SQLite，在新进程中继续执行。"""
import argparse
import hashlib
import json
import sqlite3


class Conflict(ValueError):
    pass


class Store:
    def __init__(self, path, workflow_version='v1'):
        self.workflow_version = workflow_version
        self.db = sqlite3.connect(path, isolation_level=None, timeout=5)
        self.db.row_factory = sqlite3.Row
        self.db.execute('''CREATE TABLE IF NOT EXISTS runs(
            run_id TEXT PRIMARY KEY, owner TEXT NOT NULL, input TEXT NOT NULL,
            fingerprint TEXT NOT NULL, workflow TEXT NOT NULL,
            next_step INTEGER NOT NULL, revision INTEGER NOT NULL, data TEXT NOT NULL)''')

    def close(self):
        self.db.close()

    def create(self, run_id, owner, text):
        # 输入摘要用于检查同一个任务编号是否仍对应原来的文本。
        digest = hashlib.sha256(text.encode('utf-8')).hexdigest()
        self.db.execute('INSERT OR IGNORE INTO runs VALUES(?,?,?,?,?,0,0,?)',
                        (run_id, owner, text, digest, self.workflow_version, '{}'))
        row = self.get(run_id, owner)
        if row['fingerprint'] != digest or row['workflow'] != self.workflow_version:
            raise Conflict('same run requires same input and workflow')
        return row

    def get(self, run_id, owner):
        # 按任务和归属一起读取，随后把JSON字符串还原成字典。
        row = self.db.execute('SELECT * FROM runs WHERE run_id=? AND owner=?', (run_id, owner)).fetchone()
        if row is None:
            raise PermissionError('run unavailable')
        value = dict(row)
        value['data'] = json.loads(value['data'])
        return value

    def advance(self, run_id, owner, expected_revision, crash_before_commit=False):
        # 结果、下一步和版本在同一个事务里更新。
        self.db.execute('BEGIN IMMEDIATE')
        try:
            state = self.get(run_id, owner)
            if state['workflow'] != self.workflow_version or state['revision'] != expected_revision:
                raise Conflict('stale checkpoint or workflow mismatch')
            step, data = state['next_step'], state['data']
            # 第0步规范化空白，第1步统计单词，第2步表示已经完成。
            if step == 0:
                data['normalized'] = ' '.join(state['input'].split())
            elif step == 1:
                data['report'] = f"words={len(data['normalized'].split())}"
            elif step == 2:
                self.db.execute('COMMIT')
                return state
            else:
                raise Conflict('unknown step')
            changed = self.db.execute('''UPDATE runs SET next_step=?, revision=revision+1, data=?
                WHERE run_id=? AND owner=? AND revision=?''',
                (step+1, json.dumps(data), run_id, owner, expected_revision)).rowcount
            if changed != 1:
                raise Conflict('checkpoint write conflict')
            if crash_before_commit:
                # 主动触发异常，用来观察尚未提交的更新如何回滚。
                raise RuntimeError('simulated interruption before commit')
            self.db.execute('COMMIT')
        except BaseException:
            if self.db.in_transaction:
                self.db.execute('ROLLBACK')
            raise
        return self.get(run_id, owner)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--db', required=True)
    parser.add_argument('--phase', choices=['step', 'resume'], required=True)
    args = parser.parse_args()
    store = Store(args.db)
    try:
        state = store.create('r1', 'alice', '  hello   agent  ')
        if args.phase == 'step' and state['next_step'] == 0:
            state = store.advance('r1', 'alice', state['revision'])
        elif args.phase == 'resume':
            while state['next_step'] < 2:
                state = store.advance('r1', 'alice', state['revision'])
        print(json.dumps({key: state[key] for key in ['run_id', 'workflow', 'next_step', 'revision', 'data']}, ensure_ascii=False))
    finally:
        store.close()


if __name__ == '__main__':
    main()

cd "F:\学习\l4-checkpoints"
ls
python -X utf8 checkpoint_lab.py --db lesson.sqlite --phase step

python -X utf8 checkpoint_lab.py --db lesson.sqlite --phase resume

{"run_id": "r1", "workflow": "v1", "next_step": 1, "revision": 1, "data": {"normalized": "hello agent"}}
{"run_id": "r1", "workflow": "v1", "next_step": 2, "revision": 2, "data": {"normalized": "hello agent", "report": "words=2"}}
{"run_id": "r1", "workflow": "v1", "next_step": 2, "revision": 2, "data": {"normalized": "hello agent", "report": "words=2"}}

BEGIN IMMEDIATE
state = get(run_id, owner)
check(state.workflow, expected_revision)
data = run_one_pure_step(state)
UPDATE runs SET data=?, next_step=?, revision=revision+1
    WHERE run_id=? AND owner=? AND revision=?
COMMIT

python -X utf8 -m unittest -v test_checkpoint_lab.py