失业一年了，天天想着怎么翻身。前四课我们已经能搭建单Agent和多Agent团队了，这节课我们要解决一个非常关键的问题：如何让Agent真正记住之前的对话和中间状态，从而处理多轮、复杂的长期任务。

本课目标

• 彻底搞清楚LangGraph中状态持久化的正确方式
• 掌握checkpointer和thread_id的使用方法
• 构建一个带记忆的多轮编程助手Agent，能记住项目需求、已写代码和用户偏好
• 理解为什么不加checkpointer状态就会丢失，以及如何正确实现长期记忆

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core

核心代码实战

代码段1：定义支持记忆的State

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek
from langchain_core.tools import tool

# ==================== 带记忆的State ====================
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    project_requirements: Annotated[str, "用户最初的项目需求，会一直记住"]
    current_code: Annotated[str, "当前已经写好的代码，会持续迭代"]
    user_preferences: Annotated[str, "用户编码偏好，例如：使用TypeScript、注重性能、代码风格简洁"]

代码段2：定义编程助手Agent节点

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.6)

def coding_agent_node(state: State):
    system_prompt = f"""
    你是一个经验丰富的全栈编程助手。
    
    项目需求（请始终记住）：
    {state.get('project_requirements', '暂无')}
    
    用户编码偏好：
    {state.get('user_preferences', '暂无')}
    
    当前已有代码（请基于此继续修改或扩展）：
    {state.get('current_code', '尚未开始编写')}
    
    请根据用户最新指令，继续推进编程任务。
    """
    
    # 将系统提示与历史消息合并
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    
    return {"messages": [response]}

代码段3：构建带Checkpointer的多轮编程工作流（核心）

# 创建Checkpointer（内存版，生产环境可换成SqliteSaver）
checkpointer = MemorySaver()

workflow = StateGraph(State)

workflow.add_node("agent", coding_agent_node)

# 这里我们不加工具，专注于演示记忆机制（后面可以轻松扩展工具）
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)   # 单节点演示，重点在记忆

# 编译时必须传入checkpointer，才能实现状态持久化
graph = workflow.compile(checkpointer=checkpointer)

# ==================== 多轮对话测试（关键演示） ====================

config = {"configurable": {"thread_id": "project_thread_001"}}

# 第一轮：用户提出项目需求
result1 = graph.invoke({
    "messages": [HumanMessage(content="帮我用Python写一个FastAPI用户管理系统，要求支持注册、登录和JWT认证")],
    "project_requirements": "使用Python + FastAPI + JWT认证的用户管理系统",
    "current_code": "",
    "user_preferences": "代码结构清晰、添加详细注释、使用异步风格"
}, config=config)

print("第一轮完成，当前代码长度：", len(result1.get("current_code", "")))

# 第二轮：用户继续提修改需求（同一个thread_id）
result2 = graph.invoke({
    "messages": [HumanMessage(content="在登录接口中增加速率限制，每分钟最多10次请求")],
}, config=config)   # 使用同一个thread_id，状态会自动恢复！

print("\n=== 第二轮最终输出（Agent应该记住第一轮的需求和偏好） ===")
print(result2["messages"][-1].content)

运行结果说明
正确使用checkpointer + thread_id后：

• 第二轮中，Agent 能记住 第一轮的项目需求（FastAPI + JWT）
• 能记住用户偏好（异步风格 + 详细注释）
• 会在已有代码基础上继续修改，而不是从零开始

小练习（正好2道）

• 练习1（基础）
  把MemorySaver()改成SqliteSaver.from_conn_string("./checkpoints.db")，实现本地文件持久化。重启Python后，用同一个thread_id测试是否还能记住第一轮的需求。

• 练习2（进阶）
  在State中增加一个字段iteration_count: int，每次agent执行后自动+1。当iteration_count >= 5时，在路由中强制结束流程。修改State和代码实现。

本课小结
本课核心是使用checkpointer实现状态持久化。不加checkpointer时，每次调用都是全新会话，状态走完即丢失；加上checkpointer + thread_id后，同一个线程可以跨多轮持续保留State（需求、代码、偏好等），从而实现真正的多轮复杂工作流。这也是从玩具级Agent走向生产级系统的必备技术。

下节预告
第6课：Human-in-the-loop（人机交互）与中断机制
我们会让人类可以随时介入Agent的工作流（审批代码、提供信息、手动修改等），实现真正的人机协作。下节见！

如果觉得这篇有用，欢迎点赞和关注，一起玩转 LangGraph！