聊天记录、事实数据、工作流程状态以及检查点分别是四种不同的存储方式。

14部分中的第9部分：独立的聊天记录、保存的事实以及可恢复的工作流数据

这是系列文章中的第九篇，共14篇，旨在指导用户将LangChain从首次模型调用逐步发展为可投入生产的系统。后续文章会将完整系统转化为面试练习工具。

上一篇文章介绍了运行手册搜索功能：查询经过筛选的语料库，保留来源元数据，并阻止引用搜索未返回文档的建议。

有值班人员询问系统是否能够“记住”明天的事件。这个问题的描述不够具体——是要保存聊天记录？团队偏好设置？标签及检索到的内容？还是等待审批的写入操作？又或是图结构中的所有中间字段？人们往往将这些内容统归为一个模糊的类别，但实际上每个部分都需要独立的键、过期时间、访问控制规则以及故障处理策略。

本部分将这四个概念分开阐述：

chat history
  ordered messages for one conversation

saved facts
  selected application data about a user or accountworkflow state
  the current named values for one runcheckpoint
  a saved snapshot of workflow state that can be loaded later

将之前的消息放入静态的LangChain可执行环境中并不能自动实现暂停与继续的功能。持久线程和快照源自LangGraph的状态及检查点模型。

当前待解决的问题

以大家熟悉的故障案例作为示例：

在14:05版本发布后，来自欧盟地区的checkout请求失败。checkout-api的日志显示数据库拒绝了新的连接请求。

应为每种类型的记录分配独立的键空间：

chat session:    chat:INC-2048
user facts:      user-17
workflow thread: ticket:INC-2048

如果将故障标识符当作个人标识符来处理，就会导致无关的命名空间相互混淆。同样地，使用单一的共享记录列表也会使不同的案例混为一谈。

首先，别再说“内存”了

明确指出你所指的记录名称。

聊天历史

可以采用有序列表的形式呈现：

human: The failure began after 14:05.
assistant: I recorded the start time.
human: The failed requests are only in the EU region.
assistant: I added the affected region to the investigation context.

当您后续从这些轮次中组合提示语时，该序列便具有支撑作用。

已保存的事实

包括以下选定字段：

{
  "team": "commerce-platform",
  "timezone": "America/Los_Angeles"
}

这些事实可以保存超过单次对话的时间。只能通过明确的应用规则来保留它们，而不能通过抓取模型生成的每一条内容来实现。

工作流状态

某个任务流程的当前数据：

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

随着步骤的执行，状态会发生变化。

检查点

可以把检查点视为一个冻结的工作流状态快照，以及运行时继续执行所需的相关记录。您可以加载某个线程的最新快照，在暂停后恢复工作，查看某一步所获取的信息，以及在崩溃后进行恢复。进程本地的保存机制在退出时会消失；而生产环境中的恢复则需要基于数据库的保存机制。

检索并不属于上述任何一种情况

精心整理的运行手册索引其实是一个搜索语料库。在发生故障时打开它并不会将检索结果转换为聊天记录。片段内容也不应自动升级为永久性的个人资料信息。嵌入索引并非检查点数据库。即便单个HTTP请求涉及多个存储，也需将它们分开处理。

固定链默认会记住什么

在独立的调用之间，除非应用程序主动注入或保存上下文，否则该链不会记住任何内容。

此调用：

result = chain.invoke(current_input)

不会自动转发之前的输入或输出。你可以自行添加先前的对话内容，或者使用旧的历史记录封装方式。在本次评测的LangChain版本中，RunnableWithMessageHistory会发出警告，并引导新任务使用LangGraph进行持久化存储。

对于静态链而言，将历史记录管理在应用程序代码中通常更易于理解：

read permitted messages
  -> select the messages needed for this request
  -> call the chain
  -> store the new turn under the correct session ID

这就是配套代码的作用。

项目结构

第9部分的快照包含：

langchain-helpdesk/
├── app.py
├── checkpoint_graph.py
├── facts.py
├── history.py
└── tests/
    └── test_state.py

安装包：

python -m pip install -U langchain-core langgraph pydantic pytest

该示例没有调用任何提供者接口。

步骤1：按会话存储聊天消息

创建history.py文件：

from dataclasses import dataclass, field

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
)
@dataclass
class ChatHistoryStore:
    histories: dict[str, list[BaseMessage]] = field(
        default_factory=dict
    )    def read(self, session_id: str) -> list[BaseMessage]:
        return list(self.histories.get(session_id, []))    def add_turn(
        self,
        session_id: str,
        user_text: str,
        reply_text: str,
    ) -> None:
        history = self.histories.setdefault(session_id, [])
        history.extend(
            [
                HumanMessage(content=user_text),
                AIMessage(content=reply_text),
            ]
        )    def prior_turn_count(self, session_id: str) -> int:
        return len(self.histories.get(session_id, [])) // 2

histories将每个会话映射到一个有序列表中。read方法会创建副本，以防止调用者通过副作用向存储中添加内容。add_turn用于记录人类用户与助手的对话对。prior_turn_count会将列表长度减半，因为示例仅存储完整的对话对。实际应用中的实时转录内容还包含工具消息、未完成的对话以及错误信息——不要假设生产环境中的数据会保持整齐的配对结构。

历史记录需要保留规则

永久保留每一轮对话并非产品功能。策略必须明确规定可保留的内容、保留时长、允许查看的人员、需遮蔽的字段、删除机制，以及进入下一次模型调用时会传递多少轮对话记录。过大的历史数据会浪费令牌和成本。摘要虽能提供帮助，但也可能产生错误——应将其视为具有明确来源规则的衍生内容。

步骤2：单独存储选定的事实

创建 facts.py 文件：

from dataclasses import dataclass, field

@dataclass
class UserFactsStore:
    records: dict[str, dict[str, str]] = field(
        default_factory=dict
    )    def put(self, user_id: str, key: str, value: str) -> None:
        self.records.setdefault(user_id, {})[key] = value    def get(self, user_id: str) -> dict[str, str]:
        return dict(self.records.get(user_id, {}))

按 user_id 对事实进行索引，切勿使用会话ID或事件ID。仅接受带名称的字段；绝不能将整段对话记录存储在同一个键下。真正的 put 操作路径需要具备白名单机制、验证功能、授权控制以及审计记录。模型提示并不等同于持久化存储的授权。

步骤3：定义工作流状态

过渡到具有状态的工作流。在 checkpoint_graph.py 中使用 TypedDict：

from operator import add
from typing import Annotated, TypedDict

class TicketWorkflowState(TypedDict, total=False):
    ticket_id: str
    details: str
    classification: str
    recommendation: str
    audit: Annotated[list[str], add]

total=False 可让字段保持缺失状态，直到有节点为其赋值。审计事件会使用还原器：

Annotated[list[str], add]

当某个节点返回多个审计行时，还原器会进行拼接而非覆盖。需有意识地选择合适的还原器——事件流用追加方式，标量字段用替换方式。

第4步：编写小型确定性节点

教学用图基于普通Python实现，因此可以清晰地看到检查点行为：

def classify_node(state: TicketWorkflowState) -> TicketWorkflowState:
    details = state["details"].lower()

    if "database" in details or "connection refused" in details:
        category = "database"
    elif "access" in details or "role" in details:
        category = "access"
    else:
        category = "unknown"    return {
        "classification": category,
        "audit": [f"classified:{category}"],
    }

每个节点会读取状态并返回修改后的数据；它从不修改传入的字典。推荐节点则用于处理分类结果：

def recommend_node(state: TicketWorkflowState) -> TicketWorkflowState:
    category = state["classification"]

    if category == "database":
        recommendation = (
            "Compare database settings with the last good release."
        )
    elif category == "access":
        recommendation = (
            "Confirm the requested role and current access policy."
        )
    else:
        recommendation = "Ask a person to classify the ticket."    return {
        "recommendation": recommendation,
        "audit": ["recommendation_created"],
    }

这些都是普通的Python函数——本节重点在于状态管理与持久化，而非分类器的准确率。

第5步：构建图结构

from langgraph.graph import END, START, StateGraph

def build_checkpointed_graph(checkpointer=None):
    builder = StateGraph(TicketWorkflowState)
    builder.add_node("classify", classify_node)
    builder.add_node("recommend", recommend_node)
    builder.add_edge(START, "classify")
    builder.add_edge("classify", "recommend")
    builder.add_edge("recommend", END)    return builder.compile(
        checkpointer=checkpointer or InMemorySaver()
    )

StateGraph(TicketWorkflowState)将共享状态与类型化字典关联起来。节点和边决定了顺序；compile函数会验证结构并连接检查点。流程始终保持线性——该图之所以有效，是因为状态和检查点是首要考虑的因素，而非因为其结构复杂。

第6步：为每个工作流分配线程ID

def thread_config(thread_id: str) -> dict[str, dict[str, str]]:
    return {"configurable": {"thread_id": thread_id}}

运行该任务：

config = thread_config("ticket:INC-2048")

result = graph.invoke(
    {
        "ticket_id": "INC-2048",
        "details": (
            "checkout-api reports database connection refused"
        ),
        "audit": ["ticket_received"],
    },
    config,
)

thread_id用于划分检查点历史记录。在无关的事故之间重复使用同一线程会导致状态在它们之间泄露。

第7步：读取保存的状态

snapshot = graph.get_state(config)
print(snapshot.values)

其中包含的值为：

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

该快照仅包含工作流数据——它既不是配置存储，也不是操作手册集合。

InMemorySaver能做什么和不能做什么

它仅在 Python 进程运行期间保留检查点——这适用于单元测试和笔记本。但检查点不会在进程重启后仍然存在，也无法跨服务副本使用，更无法满足数据保留、加密或备份的需求。现有文档建议将生产环境中的内存管理及可恢复线程功能交给基于数据库的存储方案，如 Postgres。

生产环境配置：

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = build_checkpointed_graph(checkpointer)

连接密钥、数据迁移、连接池管理以及资源清理等工作仍属于应用程序的职责范围。请勿将数据库 URI 放入已提交的源代码中。

LangChain 的终点与 LangGraph 的起点

何时使用固定版本的 LangChain 即可

当处理流程是固定的；单个请求无需人工干预即可完成；重新启动整个请求没有问题；中间步骤的状态无需持久化保存；普通应用代码就能存储所需的小量历史数据时，使用固定版本的 LangChain 就足够了。

何时选择 LangGraph 更为合适

控制流通过命名状态进行分支或循环处理；必须在运行过程中获得人工批准；后续可在同一线程上继续工作；进程重启时必须保留待处理状态；操作人员需要可查看的快照；恢复时应从保存的点继续，而非重新开始。

目前的create_agent辅助函数已经会返回一个位于LangGraph上的智能体。只需提供一个检查点，后续的运行流程就会由此继续——这是预期的架构设计，而非偶然的漏洞。

检查点并非审计日志

检查点的存在是为了让运行过程能够继续。审计日志的存在则是为了让安全与业务审核人员能够还原各项操作。二者有时会包含相同的字段，但其用途截然不同。审计记录应注明请求人、所使用的工具、审批人、执行的参数、结果以及时间戳。切勿将内部序列化的快照视为符合合规要求的审计轨迹。

检查点与副作用

持久化状态并不能使外部写入操作具备幂等性。如果进程更新了某个工单却在下一个检查点之前崩溃，恢复过程可能会重复执行该写入操作。因此需要使用幂等性键或“已应用”检查机制。应将副作用处理放在审批之后；为稳定运行的操作添加标识；并明确记录重试规则。下一阶段应在工单写入之前暂停执行，然后判断是批准还是拒绝。

测试隔离性

在配套快照中包含三项离线测试。

消息历史记录相互独立

history.add_turn("chat:first", "First note", "First reply")
history.add_turn("chat:first", "Second note", "Second reply")
history.add_turn("chat:second", "Other ticket", "Other reply")

assert history.prior_turn_count("chat:first") == 2
assert history.prior_turn_count("chat:second") == 1

已保存的事实并非聊天消息

facts.put("user-17", "team", "commerce-platform")

assert facts.get("user-17") == {
    "team": "commerce-platform"
}
assert history.read("user-17") == []

不同工单线程的检查点相互独立

first, first_config = run_ticket(
    graph,
    "INC-2048",
    "checkout-api reports database connection refused",
)
second, second_config = run_ticket(
    graph,
    "INC-2050",
    "identity-api denied an access role request",
)

assert graph.get_state(first_config).values["ticket_id"] == "INC-2048"
assert graph.get_state(second_config).values["ticket_id"] == "INC-2050"

执行步骤：

pytest -q

预期结果：

3 passed

它们能够证明命名空间边界和线程隔离性——但在使用 InMemorySaver 时无法保证数据库的持久性。

常见错误

仅使用一个全局历史记录列表

这样会导致不同用户或事件之间的数据发生冲突。务必用经过身份验证且范围受限的标识符来标记历史记录。

将所有模型语句都视为事实保存

模型应谨慎生成数据。仅通过经过验证的写入路径来保存允许存储的字段。

将机密信息存储在状态中

快照会被复制、查看并保留。应将机密信息存放在安全保管处，仅传递其引用地址。

用 thread_id 作为授权依据

线程 ID 只能用于查找状态，无法证明调用者有权限读取该状态。需单独进行授权处理。

将向量存储称为“长期记忆”

这一口号掩盖了所有权问题与数据删除行为。应明确记录的名称、创建者、查询路径以及删除规则。

期望通过检查点修复错误步骤

快照会保留所有已写入的内容，包括错误信息。验证与测试依然是必不可少的。

第9部分的结果

四种有明确名称的存储边界：

session ID -> ordered chat messages
user ID    -> selected saved facts
thread ID  -> current workflow state
checkpoint -> persisted workflow snapshot

静态链仍适用于单次处理的任务。而当需要持久化状态、暂停、恢复功能时，LangGraph才是更合适的方案。接下来是第一个真正的智能体决策——只读工具可以自动运行，而任务变更则需人工审批。

文档核对：已于2026年8月26日根据LangChain的短期记忆相关文档及LangGraph的持久化功能进行核查。相关包的API已发生变更。

延伸阅读：LangChain短期记忆功能、LangChain智能体、LangGraph持久化功能。

生产帮助台系统通常需要同时使用这四种存储机制：用于当前工程师的会话级聊天缓冲区、用于保存持久化配置的用户级事实存储、用于管理工单流程的线程级工作流状态，以及可用于搜索的运行手册索引——该索引绝不能同时充当历史记录或检查点。在代码评审中明确界定这些存储边界，可以避免将所有数据都塞进一个标为“memory”的Redis列表这种常见错误。在引入新团队成员时，要求他们画出这四个存储区域并标注对应的键名；如果他们无法做到，说明该设计还不具备中断/恢复功能。

当后续引入人工审批功能（第10部分）时，检查点将成为工作流在等待期间的暂停位置。聊天记录会独立继续保存，这样工程师就可以提出澄清问题，而不会改变正在处理的写入内容。除非有明确规则要求复制某个字段，否则相关数据不会进入中断处理路径。正是这种分离机制，才避免了“午休后继续处理”变成“将整个对话重新录入工单更新中”的情况。

跨存储库混合使用保留策略

聊天记录、持久化事实、工作流检查点以及运行手册嵌入内容几乎从不使用相同的保留计时机制。为了“简化操作”而让它们保持一致，通常会违背隐私删除要求或事件回放需求。应当为这四类数据分别设定计时机制、指定负责人以及定义删除入口——即便其中两类目前指向同一个Redis实例也是如此。

将弃用警告视为可选事项

当库发出提示，表示历史数据封装器将迁移到LangGraph持久化系统时，应将其视为一种设计信号。在已弃用的路径上推出新的帮助台功能，只会让日后不得不在截止日期前重新编写代码。对于可能暂停执行的流程，建议采用检查点模型。

忽视 reducer 也是架构的一部分

团队们会花费数小时讨论字段名称，却随意将追加reducer附加到本应被替换的字段上。几周后就会出现分类重复或审计记录被删除的错误。应在与TypedDict相同的PR中一起审查reducer。