DevNull 知识库

Appearance

27｜会话持久化：状态怎么存 ​

上一篇把工具接进来了。这一篇解决多轮会话的状态存储。

第 8 篇建立了 SessionState，但那是在内存里。进程重启后状态就丢了，多 Worker 时每个进程有自己的内存，同一个会话的两次请求可能落到不同进程，第二轮看不到第一轮的上下文。本篇讲会话状态怎么持久化、怎么在多 Worker 间共享。

从内存到持久存储 ​

单进程本地调试时，内存里的 SessionState 能工作。问题是两个：进程重启后 SESSIONS 字典清空；多 Worker 时每个进程有自己的字典。

AIOps 概念图：从内存到持久存储

第一步是把状态存到磁盘或数据库。SQLite 对早期项目够用，单文件、不用起服务：

python
import sqlite3, json

conn = sqlite3.connect("sessions.db")
conn.execute("""
    CREATE TABLE IF NOT EXISTS sessions (
        session_id TEXT PRIMARY KEY,
        state_json TEXT,
        updated_at TEXT
    )
""")

def save_session(state: SessionState) -> None:
    record = {"session_id": state.session_id, "messages": state.messages, ...}
    conn.execute(
        "REPLACE INTO sessions (session_id, state_json, updated_at) VALUES (?, ?, ?)",
        (state.session_id, json.dumps(record, ensure_ascii=False), now()),
    )
    conn.commit()

def load_session(session_id: str) -> SessionState:
    row = conn.execute("SELECT state_json FROM sessions WHERE session_id = ?", (session_id,)).fetchone()
    if not row:
        return SessionState(session_id, topic="unknown")
    return SessionState.from_dict(json.loads(row[0]))

多 Worker 共享时换 Redis，读写快、天然支持并发：

python
import redis
r = redis.Redis(host="localhost", port=6379, db=0)

def save_session(state):
    r.set(f"session:{state.session_id}", json.dumps(state.to_dict()))

def load_session(session_id):
    data = r.get(f"session:{session_id}")
    return SessionState.from_dict(json.loads(data)) if data else SessionState(session_id, "unknown")

有状态和无状态的取舍 ​

会话存储有个根本取舍：有状态还是无状态。

有状态流程保留进度，能快速续跑——排查到一半被打断，恢复后接着查。但带来状态不一致风险：多个 Worker 同时改一份状态，后写覆盖先写。无状态流程便于复盘审计，每次请求自带完整上下文，但长任务效率低。

ops-assistant 用混合模式：会话状态（messages、turn_count、summary）持久化便于续跑，但每次请求的完整 trace 单独存到对象存储或日志系统，不进会话状态。trace 用于审计，会话状态用于续跑，两者分离。

竞态处理 ​

多 Worker 共享会话时必须处理竞态。两个 Worker 同时加载同一份状态，各自修改后写回，后写的覆盖先写的。

AIOps 概念图：竞态处理

text
Worker A: 加载 session-001 → 修改 → 保存
Worker B: 加载 session-001（此时还没被 A 改）→ 修改 → 保存（覆盖 A 的修改）

两种解法。

悲观锁。修改前先抢锁，抢到才能改：

python
def update_session_with_lock(session_id, modifier):
    lock_key = f"lock:session:{session_id}"
    token = r.set(lock_key, "locked", nx=True, ex=30)
    if not token:
        raise ConcurrentUpdateError(f"会话 {session_id} 正在被修改")
    try:
        state = load_session(session_id)
        new_state = modifier(state)
        save_session(new_state)
    finally:
        r.delete(lock_key)

乐观锁。不抢锁，保存时检查版本号，版本不对就重试：

python
def update_session_optimistic(session_id, expected_version, new_state):
    pipe = r.pipeline()
    pipe.watch(f"session:{session_id}")
    current = pipe.get(f"session:{session_id}")
    if json.loads(current)["version"] != expected_version:
        pipe.unwatch()
        return False   # 版本不对，重试
    pipe.multi()
    new_state.version = expected_version + 1
    pipe.set(f"session:{session_id}", json.dumps(new_state.to_dict()))
    pipe.execute()
    return True

悲观锁适合写冲突多的场景，乐观锁适合读多写少。运维排查一个会话同时被多个 Worker 改的情况不多，乐观锁通常够用。

存什么、不存什么 ​

Session 持久化的最大敌人不是存不进去，是存了太多。

很多系统把每一轮的完整消息列表都存进数据库，很快达到存储上限。内存中保留完整状态，磁盘/Redis 上只存摘要 + 最近几轮 + 关键 checkpoint。完整 trace 单独存到对象存储或日志系统，不占用高频访问的 Session 存储。

Session 里不能存不可序列化对象。数据库连接、文件句柄、线程锁无法 JSON 序列化，存进 Redis/SQLite 会失败。更隐蔽的是：某些对象看起来可序列化，但反序列化后状态不对（Pydantic 模型含 datetime 对象，不同版本解析行为不同）。Session 里只存纯数据（dict、list、str、int、float、bool），连接和句柄用时重建。

落盘频率 ​

会话落盘不是每次请求都写一次磁盘。频繁写盘引入 I/O 延迟和并发问题。

内存中维护状态，每几轮或会话结束时批量落盘；关键操作（写操作工具执行前）强制同步落盘，保证崩溃后可恢复。续跑时从最近 checkpoint 恢复，丢失最近几轮可以接受的话，落盘频率可以低一些。

Session 过期策略要考虑正在进行中的排查。固定 TTL（1 小时）可能导致用户正在排查时 Session 被清除。活跃 Session（最近 5 分钟有更新）自动续期，不活跃的按 TTL 过期，续期上限 4 小时防无限续期。

持久化之后 ​

会话状态能持久化了，多 Worker 能共享，进程重启能恢复。agent 现在能跨进程稳定工作。

下一篇讲最终产出：把排查过程组织成带证据的报告，以及支撑报告的 trace 和审计。