交互状态一般不在单一位置保存:会话对象持有当前交互上下文,世界或场景实例持有可变的运行时状态,持久化层只保留被显式导出或快照过的部分。长跑之后想回看中间状态,先按关键词在源码里定位“谁持有状态、什么时候写”,再把状态导出成独立文件,最后用导出文件回放一段交互,确认记录下来的内容能还原。下面给的检索词、字段表、代码骨架都按通用结构写,具体类名和函数名需要结合手上的 LingBot-World 2.0 代码确认。
回看交互状态的关键不是保存整份,而是先定位状态持有者和写入时机,再按间隔导出摘要、按需导出整份快照。适用场景是长跑会话的排障与复现;动作是检索写入点、导出状态、写采样记录;验证方式是用导出文件重新加载并跑一步,比对关键字段与状态哈希;边界是状态对象可能含句柄或连接等不可序列化字段,导出前需过滤,导出过频会拖慢主循环。
在源码中检索状态相关变量与写入调用
先在仓库根目录用一组关键词扫一遍,比通读全部源码快。建议把“持有”和“写入”两类词分开搜,命中结果按文件归类:
grep -rn `--include`=*.py -E "world_state|session_state|interaction_state|runtime_state" .
grep -rn `--include`=*.py -E "def (step|tick|advance|apply_action|handle_event|update)" .
grep -rn `--include`=*.py -E "snapshot|checkpoint|serialize|to_dict|save_state|restore" .
把命中结果整理成一张定位表:文件路径、类名或函数名、角色(持有状态 / 写入状态 / 只读状态 / 序列化)。判断依据主要看三点:该变量是否在每步循环或事件回调里被赋值或原地修改;写入点是否与玩家或智能体的动作一一对应;同一变量是否既出现在初始化函数,也出现在更新函数中。初始化函数里出现的通常是默认值,更新路径里出现的才是随交互变化的部分。
写入时机一般有两层:一是每帧或每步的 tick,频率高但单次变化小;二是交互事件触发,比如收到动作、切换场景、回合结束,频率低但变化大。回看时更有价值的是后者,导出点优先挂在事件回调收尾处,而不是 tick 内部,这样导出的状态才是一个语义完整的中间点。若要确认落在哪一层,可以在两处各打一行日志,跑一次典型交互,看哪个写入点与页面或日志上观察到的变化同步。
确认状态结构里哪些字段会随交互变化
不要凭字段名猜。较省事的办法是用相同起点跑两次交互,每次结束时把状态转成字典并排序输出,再做对比:
import json
def dump_sorted(state, path):
with open(path, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, sort_keys=True, default=str)
两次输出用 diff 或 JSON 比较工具对照,只在一侧出现、或值不同的字段就是动态部分;两边完全一致的字段基本可归为静态配置,采样时不必重复记录。字段分类表可以按下面这个模板列:
| 字段名 | 类型 | 来源(初始化 / 事件 / 每步) | 两次运行是否变化 | 是否纳入导出 |
|---|---|---|---|---|
| turn | int | 事件 | 是 | 是 |
| entities | list | 每步 | 是 | 摘要记数量,快照记全量 |
| scene_id | str | 事件 | 否(同一起点) | 否 |
| config.* | dict | 初始化 | 否 | 否 |
需要提醒的是,结构里可能出现嵌套对象、时间戳、随机种子或引用句柄。时间戳和随机数会让每次对比都显示“有变化”,但它们未必代表交互状态,建议先排除再判断。带句柄、连接、文件描述符的字段通常不可序列化,导出前要过滤或替换成占位值,否则装载阶段会直接报错。
加一段状态导出代码并落盘到指定目录
导出函数独立成模块,不要塞进主循环内部逻辑,方便按需调用。骨架如下,目录通过环境变量或配置项覆盖:
import os, json, time
EXPORT_DIR = os.environ.get("LBW_STATE_DIR", "./state_dump")
def dump_state(state, run_id, seq, tag="turn", keep=200):
os.makedirs(EXPORT_DIR, exist_ok=True)
payload = {"run_id": run_id, "seq": seq, "tag": tag, "ts": time.time(), "state": state}
path = os.path.join(EXPORT_DIR, f"state_{run_id}_{seq:06d}_{tag}.json")
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, default=str)
os.replace(tmp, path) # 先写临时文件再改名,避免外部读到半截文件
return path
文件命名建议带三段:run_id、递增序号、触发标签,例如 state_runA_000012_turn.json。同一次运行的序号单调递增,回放时按序号排序即可还原时间线。落盘时机选在交互事件收尾、场景切换前后、以及异常抛出之前各挂一处;覆盖策略上建议不覆盖历史文件,改用滚动保留,比如只留最近若干个序号,具体数量需要结合磁盘空间和单份状态大小确认。目录建议单独挂盘或加配额,避免长跑时把业务盘写满。
按固定间隔记录状态摘要而非整份状态
整份状态只在关键节点导出,日常采样记摘要。采样循环可以起在单独线程或任务里,按固定间隔读取当前状态并追加一行,尽量不要在主循环里做重计算:
import time, json, hashlib
SAMPLE_INTERVAL = 2.0 # 秒,按状态大小与磁盘能力调整
def summarize(state, seq, action=None):
return {
"seq": seq,
"ts": time.time(),
"action": action,
"turn": state.get("turn"),
"entities": len(state.get("entities") or []),
"flags": sorted(k for k, v in state.items() if isinstance(v, bool) and v),
"state_hash": hashlib.sha1(
json.dumps(state, sort_keys=True, default=str).encode()
).hexdigest()[:16],
}
def sampler(get_state, get_seq, out_path, stop_evt):
while not stop_evt.is_set():
rec = summarize(get_state(), get_seq())
with open(out_path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
time.sleep(SAMPLE_INTERVAL)
摘要字段建议固定成下面这几项,字段名保持稳定,后续才能按列比对:
| 字段 | 含义 | 典型用途 |
|---|---|---|
| seq | 采样序号 | 排序、定位时间线 |
| ts | 采样时间 | 与运行日志对齐 |
| turn | 交互回合 | 判断变化发生在哪一轮 |
| action | 触发动作 | 解释状态变化原因 |
| entities | 实体或对象数量 | 观察规模是否异常膨胀 |
| flags | 置位开关 | 观察状态机走到哪一步 |
| state_hash | 状态摘要哈希 | 快速判断是否发生变化 |
| dump_ref | 对应整份快照文件名 | 从摘要跳到快照 |
| err | 异常或告警标记 | 标记需要重点看的位置 |
哈希只用于变化判定,不要用它替代内容比较。如果某段时间 hash 一直不变但交互明显在推进,多半是采样点取在状态更新之前,或者状态被原地修改后没有回写。此时把采样点移到事件回调之后,或改成先取状态再判断,通常能定位到原因。
用导出的文件回放一段典型交互
回放分两步:先把导出文件读回成状态结构,再塞回引擎执行同一条动作,比较输出是否与当时记录一致。
import json
def load_state(path):
with open(path, encoding="utf-8") as f:
return json.load(f)["state"]
snap = load_state("./state_dump/state_runA_000012_turn.json")
engine.restore(snap) # 按你的引擎接口替换,可能是 load / apply / reset
out = engine.step(same_action) # 动作与导出时记录的那一条保持一致
for k in ("turn", "flags", "entities"):
print(k, snap.get(k), "->", out.get(k))
校验是否还原成功,通常看三层:结构层比较字段名和类型是否齐全;数值层比较关键标量与状态哈希;行为层用同一动作跑一步,看输出或日志是否与当时的摘要对得上。三层里行为层最有说服力,但也最容易受随机种子、时间和外部依赖影响,回放时建议固定随机种子,并把外部输入替换成录制值。如果装载时报序列化相关错误,多半是状态里混进了句柄或连接对象,回到导出那一步补过滤规则即可,不必改动引擎本身。