LangGraph持久化记忆：让Agent跨周期保持上下文

这是 5 篇系列文章的第三部分。第一部分构建了一个基于规则的单区域 Agent。第二部分引入了一个 LLM，用于两个狭窄的场景：读取运维笔记，以及解释决策。这两部分都没有改变一个基本事实：Agent 只查看一个区域一次，然后就停止了。

这是一个真正的问题。严重的供需失衡通常需要不止一个 15 分钟窗口来修复。但再次调用 Agent 时，一切从零开始：

没有已尝试过什么的记忆

无法判断这是第一轮还是第六轮

区域的各项数字会重置，而不是累计继承

第三部分用 MemorySaver 和 thread_id 解决这一问题。一个线程成为一个区域的持续故事。一次 .invoke() 调用成为一个 15 分钟的周期。

决策逻辑本身完全没有改变。detect_imbalance、classify_severity、resolved_imbalance 和 choose_best_policy 与第一部分中完全相同的确定性函数，直接复用。

内存在这里实际带来了什么：

区域的真实状态正确地向前延续，而不是重置。这包括司机、乘客，以及每个周期的自然补充。

一条笔记可以为未来的周期安排事项——"接下来一小时交通很差"、"活动下午 6 点开始"——而无需每次重复。LLM 提取一次这个事实。之后代码会自动应用它。

两个新节点被加在最前面。两者都是纯 Python——没有 LLM：

start_cycle ──▶ apply_scheduled_conditions ──▶ detect_imbalance ──▶ classify_severity ──▶ set_candidates
                                                                                                │
                                                                                   ┌────────────┴────────────┐
                                                                                   ▼ (balanced)               ▼ (deficit/surplus)
                                                                           trivial_do_nothing            reconcile_inputs  ← LLM #1
                                                                                   │                          ▼
                                                                                   │                   resolved_imbalance
                                                                                   │                          ▼
                                                                                   │                   choose_best_policy
                                                                                   │                          ▼
                                                                                   │                   generate_explanation ← LLM #2
                                                                                   │                          │
                                                                                   └──────────────┬───────────┘
                                                                                                    ▼
                                                                                           simulate_and_report

start_cycle 清除上一轮的原始对话。这可以防止消息日志在多次独立的 .invoke() 调用中无限增长。（历史记录仍然存在——它保存在自己的 history 字段中，而不是消息日志里。）它还会推进时钟，使供需使用正确小时数的基准。

apply_scheduled_conditions 检查过去笔记安排的事项是否应该在本周期过期或触发。没有 LLM 调用——只是将当前周期数与存储的到期时间或触发时间进行比较。

reconcile_inputs 也是扩展而非替换。它与第二部分中的单次调用相同。现在它还会获取声明的持续时间（"接下来一小时"）或未来的触发时间（"下午 6 点开始"），当笔记中提到时。

LLM 提取一次原始事实。apply_scheduled_conditions 在此后的每个周期自动应用它。

AgentState 在第二部分的基础上增加了四个字段：

initial_hour 和 cycle_number 将时钟锚定

history 累计每个周期的记录

scheduled_conditions 保存过去笔记为后续安排的任何事项

这些都不是 MemorySaver 本身——只是普通的状态。MemorySaver 才是让这些状态在同 thread_id 的多次独立 .invoke() 调用之间存活而不每次重置的关键：

class AgentState(TypedDict):
    zone:                 dict
    ops_note:             str
    initial_hour:         int   # set once on cycle 1, anchors the clock
    cycle_number:         int
    history:              Annotated[list[dict], operator.add]
    scheduled_conditions: dict  # e.g. {"traffic": {"expires_at_cycle": 5}}
    imbalance_ratio:      float
    imbalance_type:       str
    severity:             str
    candidate_policies:   list
    policy_evaluations:   dict
    policy_resolutions:   dict
    recommended_policy:   str
    explanation:          str
    messages:             Annotated[list[BaseMessage], add_messages]
    outcome:              dict
    outcome_delta:        dict
    report:               str

report_context_and_schedule 是第二部分的 report_context 工具，增加了两个字段。这让 LLM 可以报告时间安排，而不仅仅是当前状况——声明的持续时间，或未来的触发小时。

代码仍然拥有后续的所有算术：分钟转周期、小时转剩余周期数。LLM 只提取原始事实：

@tool
def report_context_and_schedule(
    rain_flag: bool,
    event_flag: bool,
    traffic_level: Literal["none", "light", "moderate", "heavy"],
    traffic_duration_minutes: int = 0,
    event_starts_at_hour: int = -1,
) -> str:
    """
    Call this exactly once with your interpretation of the ops note for THIS
    cycle. If the note doesn't affect a value, report the zone's current value
    unchanged.

    traffic_duration_minutes: ONLY if the note states how long this traffic
    condition will last (e.g. "for the next hour" -> 60). 0 if no duration is
    stated -- traffic_level then applies to this cycle only.

    event_starts_at_hour: ONLY if the note describes an event scheduled for a
    specific clock time that hasn't started yet (e.g. "starts at 6pm" -> 18).
    -1 if no future start time is stated.
    """
    return (
        f"rain_flag={rain_flag}, event_flag={event_flag}, traffic_level={traffic_level}, "
        f"traffic_duration_minutes={traffic_duration_minutes}, event_starts_at_hour={event_starts_at_hour}"
    )

接入 Checkpointer

这里的每个节点都是一个普通函数，与之前一样。唯一真正新的部分是 MemorySaver，在 compile() 时传入。

这正是将 zone/history/scheduled_conditions 从普通状态转变为可在同 thread_id 的多次独立 .invoke() 调用之间存活的东西：

llm = ChatOllama(model="qwen2.5:14b", temperature=0)
llm_with_reconcile_tool = llm.bind_tools([report_context_and_schedule])

def _reconcile(state):
    return reconcile_inputs(state, llm_with_reconcile_tool)

def _explain(state):
    return generate_explanation(state, llm)

g = StateGraph(AgentState)

g.add_node("start_cycle",                start_cycle)
g.add_node("apply_scheduled_conditions", apply_scheduled_conditions)
g.add_node("detect_imbalance",           detect_imbalance)
g.add_node("classify_severity",          classify_severity)
g.add_node("set_candidates",             set_candidates)
g.add_node("trivial_do_nothing",         trivial_do_nothing)
g.add_node("reconcile_inputs",           _reconcile)
g.add_node("resolved_imbalance",         resolved_imbalance)
g.add_node("choose_best_policy",         choose_best_policy)
g.add_node("generate_explanation",       _explain)
g.add_node("simulate_and_report",        simulate_and_report)

g.add_edge(START, "start_cycle")
g.add_edge("start_cycle", "apply_scheduled_conditions")
g.add_edge("apply_scheduled_conditions", "detect_imbalance")
g.add_edge("detect_imbalance",  "classify_severity")
g.add_edge("classify_severity", "set_candidates")
g.add_conditional_edges("set_candidates", route_llm_or_skip, {
    "trivial_do_nothing": "trivial_do_nothing",
    "reconcile_inputs":   "reconcile_inputs",
})
g.add_edge("reconcile_inputs",     "resolved_imbalance")
g.add_edge("resolved_imbalance",   "choose_best_policy")
g.add_edge("choose_best_policy",   "generate_explanation")
g.add_edge("generate_explanation", "simulate_and_report")
g.add_edge("trivial_do_nothing",   "simulate_and_report")
g.add_edge("simulate_and_report",  END)

app = g.compile(checkpointer=MemorySaver())

这里的 app 在本文其余部分中被重复使用。一个图，无数独立的区域故事——通过 thread_id 区分，而不是重建任何东西。

运行多个周期

MemorySaver 加 thread_id 就是全部机制。在同一个 thread_id 上再次调用 .invoke()，图会从上次调用停止的地方精确恢复。

只有第一次调用需要完整的初始状态。之后每次调用只需要新的内容——一条运维笔记，或者什么都没有。

Downtown Core，从供需失衡开始，没有运维笔记：

[Cycle 1] ratio=1.93 | policy=surge_pricing | wait 6.3min → 4.6min | resolved=NO
    Highest profit at $230.78, even though it did not resolve the imbalance.
[Cycle 2] ratio=1.41 | policy=surge_pricing | wait 4.6min → 4.3min | resolved=YES
    RESOLVED

两次调用，同一个 thread_id。第二次已经知道这是持续供需失衡的第 2 周期——无需手动传回任何东西。

跨周期安排条件

第一周期的运维笔记说接下来一小时交通很差——那是 4 个周期。reconcile_inputs 提取 traffic_duration_minutes=60。代码将其转换为 4 个周期，并安排到期时间。

无需进一步笔记。apply_scheduled_conditions 让 traffic_level="heavy" 保持活跃到第 4 周期，然后在第 5 周期自动恢复——完全按计划：

notes = {1: "There's a bad accident on the main road backing up traffic for the next hour."}
results = run_cycles(app, "downtown-traffic-duration", downtown, ops_notes=notes)

cycle=1 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}
cycle=2 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}
cycle=3 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}
cycle=4 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}
cycle=5 traffic=0.00 scheduled={}

第 1 周期的一条笔记影响了 5 个周期的行为。没有人需要在第 2 到 4 周期提醒 Agent 关于交通的情况。

未来触发器，以及覆盖已有安排

还有两种情况值得展示：

未来的触发器，而非即时纠正。"音乐会下午 6 点开始"——现在是下午 5 点。这意味着 event_flag 目前应保持 False，只有当那个小时真正到来时才切换为 True。

reconcile_inputs 提取 event_starts_at_hour=18。代码计算出距离那个时间还有多少个周期，并安排切换：

cycle=1 hour=17 event_flag=False
cycle=2 hour=17 event_flag=False
cycle=3 hour=17 event_flag=False
cycle=4 hour=17 event_flag=False
cycle=5 hour=18 event_flag=True
cycle=6 hour=18 event_flag=True

后续笔记覆盖更早的安排。假设在新笔记到来时，之前的安排尚未到期——"实际上事故已经清理了，交通恢复正常了。"它会完全替换掉之前的安排，而不是等待原始计时器。新的信息优先：

notes = {
    1: "Traffic is bad for the next hour due to an accident.",
    3: "Actually the accident has been cleared, traffic is back to normal.",
}

cycle=1 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}
cycle=2 traffic=1.00 scheduled={'traffic': {'expires_at_cycle': 5}}

到第 2 周期时，该区域已经自行解决了。所以安排实际上从未被覆盖过。但是这个机制——新笔记擦除并替换已有安排——与中途到达纠正时触发的机制是相同的。

策略实际上会改变吗？

到目前为止，每个例子都在第一次解决时就停止了。这掩盖了一个重要问题：Agent 会在区域平衡后继续涨价，还是真的会退让？

连续运行 20 个周期，不提前停止，直接回答这个问题：

cycle=1  type=deficit   ratio=1.93  policy=surge_pricing   drivers=16  riders=18  wait= 3.7min
cycle=2  type=balanced  ratio=1.12  policy=do_nothing      drivers=15  riders=19  wait= 4.4min
cycle=4  type=deficit   ratio=1.36  policy=surge_pricing   drivers=14  riders=15  wait= 3.8min
cycle=8  type=deficit   ratio=1.31  policy=surge_pricing   drivers=17  riders=18  wait= 3.5min
cycle=12 type=deficit   ratio=1.46  policy=surge_pricing   drivers=14  riders=14  wait= 3.3min
cycle=20 type=balanced  ratio=1.08  policy=do_nothing      drivers=11  riders=12  wait= 3.6min

在整个 20 个周期中，Downtown Core 在供需失衡（surge_pricing）和平衡（do_nothing）之间切换了四次。不是一次解决就完事——而是真实的、持续的反复拉锯。普通的漂移会周期性地将比率推回供需失衡线以上，然后 do_nothing 再次使其平衡。

整个过程中司机和乘客数量保持在稳定且有界的范围内——司机 11–17 人，乘客 12–21 人。没有漂移，没有堆积。

第二部分相比有什么变化

决策本身从未变得更聪明。choose_best_policy 与第一部分完全相同。

变化的是：Agent 现在可以区分"还在处理中，已经三个周期了"和"第一次看这个区域"。而且像"接下来一小时交通很差"这样的笔记不需要重复四次。

仍然缺失的部分：这里的每个决策仍然无人监督地执行，没有人在事前检查。第四部分从这里继续。

本系列代码：github.com/ebiarian/zone-balancing-ridesharing-langgraph-agent

下一篇——第四部分：使用 interrupt() 实现人机交互暂停，这样有风险或不寻常的决策不会在没有复核的情况下直接运行。