Day 37 / 共 68 天 · 阶段 6 Agent 核心

状态 & 断点续跑

昨天(Day 36)学了记忆——Agent 怎么记住信息。今天讲把这些信息装进一块"共享黑板"(state),再学会给运行过程随时存档(checkpointer)、崩了能断点续跑、甚至中途暂停等人拍板(interrupt)。这几个词就是明天(Day 38)手写 Agent、以及后天(Day 40)LangGraph 的地基。

📍 你在阶段 6(Agent 核心 D33-38)的位置
D33 Agent 是什么 D34 ReAct D35 规划 D36 记忆 D37 状态&断点续跑 D38 手写最小 Agent
💡 用一个类比兜住今天(今天全程沿用「打游戏闯关」的世界观) 跑一个 Agent = 打一局带存档的游戏state 状态 = 你的存档栏(记着等级、金币、走到第几关、背包里有啥);每走一步(节点) = 打一个小关卡,打完把成果写回存档栏;checkpointer 存档器 = 游戏的自动存档功能,每过一关帮你存一次;断点续跑 = 电脑崩了、明天再开,读档从上次那关继续,不用从头打;interrupt 中断 = 剧情里的"重大选择",游戏暂停弹窗等你点"是/否",你选完才继续。今天你从"只会一口气跑完"升级成"能存档、能续跑、能等人"。
L01

为什么 Agent 需要"状态"这个东西

🤔 痛点前几天的 Agent 主循环里,变量东一个西一个:历史消息一个变量、工具结果又一个变量、当前计划再一个……步骤一多,谁改了谁、传给下一步要带哪些,乱成一团。换成多个环节(先搜资料→再写稿→再校对),更是没法交接。
💡 本质状态(state)就是把"这一局运行里所有要记的东西"收进同一个存档栏。每个环节都从这块存档读、往这块存档写,不再各存各的。像打游戏:等级、金币、进度全在一个存档里,任何关卡都看得到、改得动。
👶 一句话"状态"= 一个大字典,记着这局 Agent 到现在为止的全部家当。以前是散装变量,现在打包成一个存档。
📝 举个例子:一个"查天气再穿衣建议"的 Agent 状态长这样
{
  "messages": ["用户: 明天穿什么?", "助手: 我查一下天气"],
  "city": "北京",
  "weather": "明天 5℃ 有风",
  "step": 2,
  "done": false
}
随便哪个环节打开这个存档,都能立刻知道:聊到哪了、查的哪个城市、天气结果、走到第几步、结束没。
L02

state = 一块"共享黑板"

🤔 痛点Agent 里常有好几个环节接力干活。搜资料的、写稿的、校对的,怎么把成果传给下一位?一层层用函数参数传,参数越滚越多,谁都嫌麻烦。
💡 本质把 state 想成教室墙上一块大黑板:每个环节上台,先看黑板(读到目前的进度),干完活把结果写到黑板上,下一位上台照样先看黑板。谁都不用手把手交接,黑板就是唯一的"交接台"。这就是共享状态
一块共享黑板,各环节都读它、写它 📋 state 共享黑板 messages / city / weather / step 环节A 查天气 环节B 出建议 环节C 回复
图注:环节之间不直接对话,全靠黑板交接。加一个新环节,也只要它会读写黑板即可,改动最小。
# 用一个字典当"黑板",最朴素的共享状态
state = {"city": "北京", "weather": None, "advice": None}

def 查天气(s):                       # 环节A:读 city,把结果写回黑板
    s["weather"] = f"{s['city']} 明天 5℃ 有风"
    return s

def 出建议(s):                       # 环节B:读 weather,写 advice
    s["advice"] = "冷且有风,穿厚外套 🧥"
    return s

state = 查天气(state)                # 每个环节:进来读黑板、出去写黑板
state = 出建议(state)
print(state["advice"])              # → 冷且有风,穿厚外套 🧥
L03

状态怎么更新:覆盖 vs 追加

🤔 痛点每个环节写黑板,是把旧值擦掉重写,还是接在后面?比如天气结果覆盖没问题,但对话历史要是每次覆盖,前面聊的全丢了!
💡 本质不同字段有不同"写法":像 city/weather 这种是覆盖(存档栏里"当前金币"直接改成新数);像 messages 对话历史是追加(像"任务日志",只往后加一行,不擦旧的)。框架里给某字段标一句"这个要追加",框架就会自动帮你 += 而不是替换。
👶 类比存档栏金币数 = 覆盖(打完关直接变成新数值);闯关日志 = 追加(每过一关多一行记录,从不删旧的)。搞清哪个字段该覆盖、哪个该追加,是用状态最容易踩的坑。
# 覆盖:直接改成新值
state["weather"] = "北京 明天 5℃"   # 旧天气被新的替换掉,合理

# 追加:历史要保留,用 += 接在后面(别用 = 覆盖!)
state.setdefault("messages", [])
state["messages"] += ["用户: 明天穿什么?"]   # 加一行
state["messages"] += ["助手: 建议穿厚外套"]  # 再加一行,前面的还在
print(state["messages"])   # → ['用户: 明天穿什么?', '助手: 建议穿厚外套']
后面学 LangGraph 时(Day 40),你会看到给字段加一个 Annotated[list, add] 的标记,意思正是"这个字段请用追加,别覆盖"——现在理解"覆盖 vs 追加"这层意思就够了。
L04

checkpointer:给运行过程"自动存档"

🤔 痛点一个 Agent 跑了 8 步,第 6 步调 API 时网断了,程序崩了。难道从第 1 步重跑?前面调 API 花的钱和时间全打水漂,用户也早跑了。
💡 本质checkpointer(存档器)= 游戏的自动存档功能。每走完一步,就把当时的整块 state 存一份下来(存内存、存文件、或存数据库)。这样任何一步之后都有一个"存档点",随时能读回来。它是"断点续跑"和"中断等人"能实现的前提。
每过一关,存一次档 步骤1 步骤2 步骤3 💾存档 💾存档 💾存档
图注:有了每一步的存档点,崩在第3步之后,就能从第3步的存档接着跑,前两步不白干。

👶 小白:存档存在哪?会不会占很多地方?

👨‍🏫 老师:看你选。练习阶段存在内存里最简单(程序一关就没了,够用);要真能"关机明天再续",就存到文件或数据库(比如 SQLite)。每次存的就是那个 state 字典,通常很小。框架会用一个 thread_id(想成"存档栏位编号")区分不同用户/不同会话的存档,互不串。

L05

断点续跑:崩了从上次那步接着来

🤔 痛点长任务最怕跑一半挂掉:调外部服务超时、机器重启、临时报错。没有续跑,每次都得从头,长任务基本没法上线。
💡 本质断点续跑= 拿最近的存档读档继续。因为每步都存了 state,重启后只要"用同一个存档栏位编号再跑一次",框架发现已经有存档,就跳过已完成的步骤,从断点往下接。像游戏读档:不是从头,是从你上次站的那关。
📝 举个例子:手写版模拟"读档续跑"
import json, os
ARCHIVE = "save.json"                       # 存档文件

def 存档(state):                             # 每步存一次
    with open(ARCHIVE, "w") as f:
        json.dump(state, f, ensure_ascii=False)

def 读档():                                  # 有存档就读回来,没有就新开一局
    if os.path.exists(ARCHIVE):
        return json.load(open(ARCHIVE))
    return {"step": 0, "result": []}

state = 读档()                               # ← 关键:先尝试读档
steps = ["查天气", "选衣服", "生成回复"]
for i in range(state["step"], len(steps)):   # 从 step 处接着跑,不是从 0
    print("执行:", steps[i])
    state["result"].append(steps[i])
    state["step"] = i + 1
    存档(state)                              # 每步存档;这里崩了,下次接着跑
print("完成:", state["result"])
第一次跑到一半 Ctrl+C 掐掉,再运行一次——它会跳过已完成的步骤,从断点继续。这就是"断点续跑"的内核,LangGraph 把这套自动化了。
真实项目里你几乎不用手写这些,框架的 checkpointer 一开就自带续跑。但亲手写一遍,你才真懂框架在背后替你干了什么——这正是明天(Day 38)手写 Agent 的意义。
L06

interrupt:中途暂停,等人拍板

🤔 痛点有些动作很危险——Agent 要"删库""发邮件给全体客户""下单花钱"。全自动放它干,出事就晚了。可总不能人一直盯着吧?
💡 本质interrupt(中断/人在环路)= 游戏里的重大选择弹窗:走到危险步,Agent 主动暂停并存档,把"我打算这么干,批准吗?"抛给人。人点"同意"→读档继续;点"改一下"→改完再续。核心靠的还是存档:能暂停在半路、还能原样接回来,全因为 state 被存住了。这就是"human-in-the-loop(人在环路)"。
危险动作前,暂停等人点头 Agent 想发邮件 ⏸ 暂停+存档"批准吗?" ✅ 同意 → 读档续跑 ✏️ 修改 → 改后再续
图注:暂停不是"停掉重来",而是"冻结现场、等指令、再原样解冻"。危险动作、花钱动作最该这么设计。
👶 一句话interrupt = "先别动,等我批"。它让 Agent 在关键处交出方向盘给人——既保留自动化的省事,又守住"人说了算"的安全线。Day 41、Day 54 会专门讲怎么设计这道"审批闸"。
🔗 想深入 state / checkpointer / interrupt 的真实写法?去看《LangGraph 20 天精讲》→(学完回来继续 Day 38)
L07

今日小结 + 动手 10 分钟

🧠 今天你应该能回答

  • 为什么要把散装变量收成一块 state?"共享黑板"是什么意思?
  • 更新状态时,"覆盖"和"追加"分别用在什么字段?(金币 vs 闯关日志)
  • checkpointer(存档器)干嘛?为什么它是续跑和中断的前提?
  • "断点续跑"怎么做到不从头跑?(读最近存档,跳过已完成步)
  • interrupt(人在环路)解决什么问题?哪些动作最该用?

✋ 动手 10 分钟:做一个"能读档续跑"的迷你流程

把 L05 的代码敲进 day37.py,跑起来,然后在中途按 Ctrl+C 掐断,再运行一次,观察它是不是从断点继续:

import json, os, time
ARCHIVE = "save.json"
def 存档(s): json.dump(s, open(ARCHIVE, "w"), ensure_ascii=False)
def 读档(): return json.load(open(ARCHIVE)) if os.path.exists(ARCHIVE) else {"step": 0, "done": []}

state = 读档()
steps = ["搜集资料", "写初稿", "校对", "发布"]
for i in range(state["step"], len(steps)):
    print("正在:", steps[i]); time.sleep(2)   # 假装每步很慢,方便你 Ctrl+C 打断
    state["done"].append(steps[i]); state["step"] = i + 1
    存档(state)                                # 每步存档
print("全部完成:", state["done"])
os.remove(ARCHIVE)                             # 收尾删档,下次重新开始

体会一下:打断→再跑→自动跳过已完成的步。你就亲手复现了 checkpointer 的核心。

明日预告 · Day 38:概念齐了,明天不用任何框架,亲手写一个最小 Agent——一个 while 循环 + 一张工具表 + 让 LLM 决定"下一步调哪个工具",约 60 行代码跑通"想→动手→看→再想"的完整主循环。今天的 state 会成为那个循环里传来传去的"黑板"。写完这一个,你就真正理解了后面所有框架到底在替你做什么。
← Day 36 · 记忆 Day 38 · 手写最小 Agent →