实用指南：LangGraph内存——管理方面的完整实操手册

以下笔记为《LangGraph Memory：管理智能体记忆的完整实用指南》提供了一条实际操作路径。重点在于契约、校验机制以及可直接插入的代码占位符，而非激励性表述。 在阅读概述部分时，首先写下契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤失败时，故障应能指向单一责任点，而非复杂的流程链。

教程中从未提及的记忆问题

在教程中无人提及的内存问题，若将其视为可测量的对象来处理效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

内存架构：首先需要理解的四点

《内存景观：首先需要理解的四点》作为可度量的结构来理解效果最佳。在扩大范围之前，先记录一份成功的案例、一个失败案例以及回滚说明。 在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致程序无法继续运行。

User sends message
        ↓
Graph loads State from Checkpointer (using thread_id)
        ↓
Memory Management runs (trim / filter / summarize)
        ↓
LLM receives cleaned messages
        ↓
LLM response added to State
        ↓
Checkpointer saves updated State (to memory / SQLite / Postgres)

模块2扩展：具备内存感知能力的状态

模块2扩展：将“内存感知状态”视为可测量的对象时，其效果最佳。在扩大范围之前，需记录一份理想状态样本、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 保持系统状态的扁平化与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。 模块2扩展：将“内存感知状态”视为可测量的对象时，其效果最佳。在扩大范围之前，需记录一份理想状态样本、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一的责任模块，而非复杂的流程链。

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

标准的内存感知状态

对于“标准内存感知状态”，在修改代码之前需明确输入参数、该步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功标准，杜绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。

# ── MODULE 2: STATE (Memory-Aware Version) ──────────────────
from langgraph.graph import MessagesState
# Option A: The shorthand (recommended for most agents)
# MessagesState is a pre-built TypedDict with just the messages field.
# Use it when you don't need custom fields.
class State(MessagesState):
    summary: str   # The only addition needed for memory management

管理对话长度的三种策略

在实施“管理对话长度的三种策略”时，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本信息，可避免在从演示环境切换到共享环境时出现意外费用。对于会产生费用或更改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

策略1：消息过滤

对于策略1：消息过滤，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的连接方式并不足以确保业务的完整性。 对于策略1：消息过滤，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。

模式A：仅过滤LLM能看到的内容（状态不变）

在采用模式A：仅过滤LLM能看到的内容（状态不变）时，首先需列出规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关产物命名，明确成功判定标准，并拒绝默许的部分完成。 缓存稳定的系统指令和工具架构。重复发送相同的前置信息是导致资源浪费的常见原因。

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Calls the LLM with only the last 2 messages, but state keeps the full history."""

    # Filter: only send last 2 messages to the LLM
    # State is NOT modified - full history stays in the checkpoint
    recent_messages = state["messages"][-2:]

    response = llm.invoke(recent_messages)
    return {"messages": [response]}

模式B：RemoveMessage——从状态中永久删除

在实现模式B“RemoveMessage——从状态中永久删除”时，首先需明确相关约定：所需的输入参数、成功信号，以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在代码从演示环境转向共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次LLM调用收费。

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import RemoveMessage

# ── MODULE 4: NODES ──────────────────────────────────────────
def filter_messages_node(state: State) -> dict:
    """A dedicated node that removes all but the last 2 messages from state.

    RemoveMessage works by ID: you pass it the ID of a message you want deleted,
    and the add_messages reducer handles the actual deletion. This is NOT the same
    as just slicing the list - it permanently removes from the checkpoint.
    """

    # Build a list of RemoveMessage objects for every message we want to delete
    # state["messages"][:-2] = all messages EXCEPT the last 2
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {"messages": delete_messages}  # Reducer sees these and deletes them

def chat_model_node(state: State) -> dict:
    """After filtering, the state already has a short message list."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("filter", filter_messages_node)
graph_builder.add_node("chat_model", chat_model_node)
# Filter runs FIRST, then chat_model sees the cleaned state
graph_builder.add_edge(START, "filter")
graph_builder.add_edge("filter", "chat_model")
graph_builder.add_edge("chat_model", END)
graph = graph_builder.compile(checkpointer=MemorySaver())

策略2：使用trim_messages进行基于令牌的裁剪

在使用 trim_messages 实施策略2：基于令牌的截断功能时，首先需明确接口规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。 在使用 trim_messages 实施策略2：基于令牌的截断功能时，首先需明确接口规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任模块，而非复杂的整体结构。

管道。

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import trim_messages

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Uses token-based trimming to control exactly how much history the LLM sees."""

    trimmed = trim_messages(
        state["messages"],

        # Maximum number of tokens to allow (tune this to your model and budget)
        max_tokens=1000,

        # "last" = keep the MOST RECENT messages (the ones that fit within max_tokens)
        # This is almost always what you want
        strategy="last",

        # The LLM used to count tokens. Use the same model you're chatting with
        # for accurate counting. This does NOT make an API call - just counts.
        token_counter=ChatOpenAI(model="gpt-4o"),

        # False = never cut a message in half. Either include it fully or drop it.
        # Always set this to False in production to avoid feeding partial tool results.
        allow_partial=False,

        # True = always include the most recent HumanMessage even if it alone
        # exceeds max_tokens. Prevents the agent from receiving an empty context.
        include_system=True,
    )

    response = llm.invoke(trimmed)
    return {"messages": [response]}

关键参数快速参考

将“关键参数快速参考”视为可度量的结构来使用效果最佳。在扩大范围之前，先收集一份优秀的转录文本、一个故障案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，且在中断后会导致无法继续处理。

策略3：基于LLM的摘要生成（生产环境标准）

策略3：基于大语言模型的摘要生成（生产环境标准）在被视为可度量的指标时效果最佳。在扩大应用范围之前，先记录一份理想的输出文本、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外账单。为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限能防止演示环境变成收费陷阱。

步骤1：明确需求（模块2）

第一步：将（模块2）视为可度量的结构时，其效果最佳。在扩大范围之前，需记录一份理想的运行日志、一个故障案例以及回滚说明。 将配置信息与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放于一个位置，以便操作人员无需查看整个结构即可进行审计。 保持结构状态的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致恢复失败。

class State(MessagesState):
    summary: str   # Stores the running summary. Empty string by default.

第二步：两个节点（模块4）

第二步：将“两个节点”（模块4）视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一个成功的用例、一个失败案例以及回滚说明。同时将正常流程和恢复流程记录下来。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致流程无法继续。

def call_model(state: State) -> dict:
    """The main agent node. Uses summary as context if it exists."""

    summary = state.get("summary", "")

    if summary:
        # If we have a summary, prepend it as a system message
        # This gives the LLM the compressed history without the raw messages
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        # No summary yet — just use the raw messages
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}
def summarize_conversation(state: State) -> dict:
    """Triggered when conversation gets long. Compresses history into a summary,
    then deletes the old raw messages to keep state lean."""

    existing_summary = state.get("summary", "")

    # Build the summarization prompt.
    # If a summary already exists, extend it with new messages.
    # This is the "rolling summary" pattern - it never reads all messages from scratch.
    if existing_summary:
        summary_message = (
            f"This is the summary of our conversation so far:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages above. "
            f"Keep it concise - focus on facts and decisions that matter for future turns."
        )
    else:
        summary_message = (
            "Create a concise summary of the conversation above. "
            "Focus on key facts, user intent, and any decisions made. "
            "This summary will replace the full message history."
        )

    # Call the LLM with all current messages + the summarization instruction
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = llm.invoke(messages)

    # After summarizing, delete all messages except the last 2.
    # Why keep 2? So the LLM has immediate conversational context
    # (it can see what was just said) even after summarization runs.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,    # Updated summary stored in state
        "messages": delete_messages      # Old messages removed from state
    }

第三步：路由逻辑（模块5）

第三步：路由逻辑（模块5）若被视为可度量的对象，其效果会更好。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出现故障时，故障应能指向单一责任模块，而非复杂的流程链。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。

from typing import Literal
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """After the main agent responds, decide: keep going or trigger summarization?

    The threshold (6 messages here) is your tuning knob.
    Too low = you summarize frequently, losing some nuance.
    Too high = you let the context grow before compressing.
    For most production agents, 6–10 messages is a good starting range.
    """

    if len(state["messages"]) > 6:
        return "summarize_conversation"

    return "__end__"

第四步：图结构构建（模块6）

第4步：图结构组装（模块6）若被视为可测量的表面，效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续处理。

graph_builder = StateGraph(State)

graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
# After summarization, the conversation ends for this turn
graph_builder.add_edge("summarize_conversation", END)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

实际应用中的表现

“如何在实践中应用”这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态的结构化与类型化，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后还会导致数据恢复失败。

Turn 1–3:   Messages accumulate. should_continue → "__end__" each time.
Turn 4:     call_model runs. len(messages) > 6 → should_continue → "summarize_conversation"
            summarize_conversation:
              - LLM reads all 7 messages, generates a 2-sentence summary
              - state["summary"] = "User is troubleshooting a Python import error..."
              - Deletes messages[:-2] (keeps only last 2 raw messages)
Turn 5:     call_model prepends summary as SystemMessage + 2 recent messages
            LLM sees: [SystemMessage(summary), HumanMessage, AIMessage, HumanMessage(latest)]
            Token cost: 4 messages instead of potentially 20+

实现真正的持久内存：外部检查点

实现真正的持久内存：将外部检查点视为可度量的界面使用效果最佳。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一处，以便操作人员无需查看整个系统结构即可进行审计。 保持系统状态的扁平化与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致恢复失败。 实现真正的持久内存：将外部检查点视为可度量的界面使用效果最佳。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一的责任模块，而非复杂的流程链。

检查点接口

对于 Checkpointer 接口，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接配置并不等同于业务上的完整性。

# Development: everything in RAM (lost on restart)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# Local / Single-machine production: persists to a .db file on disk
from langgraph_checkpoint_sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("chat_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
# Cloud / Distributed production: persists to a Postgres database
from langgraph_checkpoint_postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@host:5432/dbname"
)
# This one line is the ONLY thing that changes between environments:
graph = graph_builder.compile(checkpointer=checkpointer)

SqliteSaver 深入解析

若要深入使用 SqliteSaver，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询的成本。提前了解成本情况，可避免在从演示环境切换到共享环境时出现意外费用。对于会耗费资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

# ── Full SqliteSaver Setup ───────────────────────────────────
import sqlite3
from langgraph_checkpoint_sqlite import SqliteSaver

# File-based: state persists across restarts
conn = sqlite3.connect(
    "agent_memory.db",
    check_same_thread=False  # Required for LangGraph's multi-threaded execution
)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)
# Usage is IDENTICAL to MemorySaver:
config = {"configurable": {"thread_id": "user-123"}}
response = graph.invoke({"messages": [HumanMessage(content="Hello")]}, config)
# Stop and restart your app. Run this again with the SAME thread_id:
# The agent picks up exactly where it left off.
response = graph.invoke({"messages": [HumanMessage(content="What did I just say?")]}, config)
# Agent: "You said 'Hello'."

何时使用 SqliteSaver 而非 PostgresSaver

在决定何时使用 SqliteSaver 而非 PostgresSaver 时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审核。 对于涉及资金支出或修改生产数据的操作，需经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。 在决定何时使用 SqliteSaver 而非 PostgresSaver 时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一的责任主体而非多个部分。

一个错综复杂的流程管道。

检查内存：get_state与get_state_history

在处理“检查内存：get_state与get_state_history”时，首先需明确契约内容：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关元素命名，定义成功判定标准，并杜绝无声的半完成状态。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。

# ── Inspecting Current State ─────────────────────────────────
config = {"configurable": {"thread_id": "user-123"}}
# Get the current state snapshot for this thread
state_snapshot = graph.get_state(config)
# What's in the snapshot?
print(state_snapshot.values["messages"])  # Current message list
print(state_snapshot.values.get("summary", ""))  # Summary, if any
print(state_snapshot.next)  # Which nodes would run next (empty list = graph is at END)
# ── Inspecting State History ─────────────────────────────────
# get_state_history gives you every checkpoint ever saved for this thread.
# Each checkpoint is a moment in time after a node ran.
for state in graph.get_state_history(config):
    print(f"Step: {state.metadata['step']}")
    print(f"Messages: {len(state.values['messages'])}")
    print(f"Next: {state.next}")
    print("---")

整合所有要素：完整的记忆管理代理

在编写《Putting It All Together: The Complete Memory-Managed Agent》时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。

# ============================================================
# LANGGRAPH MEMORY-MANAGED AGENT — COMPLETE TEMPLATE
# Combines: Summarization + External Persistence (SqliteSaver)
# ============================================================

# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
import sqlite3
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage, AIMessage, SystemMessage,
    BaseMessage, RemoveMessage
)
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState
from langgraph.checkpoint.memory import MemorySaver
# For production, swap MemorySaver for one of these:
# from langgraph_checkpoint_sqlite import SqliteSaver
# from langgraph_checkpoint_postgres import PostgresSaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── MODULE 2: STATE ─────────────────────────────────────────
class State(MessagesState):
    # summary stores the rolling compressed history.
    # It starts empty and gets written by summarize_conversation.
    summary: str

# ── MODULE 4: NODES ─────────────────────────────────────────
def call_model(state: State) -> dict:
    """Main reasoning node. Uses rolling summary for long-context efficiency."""

    summary = state.get("summary", "")

    if summary:
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}

def summarize_conversation(state: State) -> dict:
    """Compression node. Summarizes history and prunes old messages from state."""

    existing_summary = state.get("summary", "")

    if existing_summary:
        summary_prompt = (
            f"Current conversation summary:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages. "
            f"Be concise. Preserve facts, user intent, and key decisions."
        )
    else:
        summary_prompt = (
            "Summarize this conversation concisely. "
            "Focus on facts, user intent, and decisions made."
        )

    messages = state["messages"] + [HumanMessage(content=summary_prompt)]
    response = llm.invoke(messages)

    # Keep only the 2 most recent raw messages in state.
    # Everything older is now captured in the summary.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,
        "messages": delete_messages
    }

# ── MODULE 5: ROUTING ────────────────────────────────────────
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """Trigger summarization when conversation exceeds 6 messages.

    Tune this threshold based on your average message length and token budget.
    For GPT-4o at ~$0.005/1K tokens, 6 messages is a practical starting point.
    """
    if len(state["messages"]) > 6:
        return "summarize_conversation"
    return "__end__"

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
graph_builder.add_edge("summarize_conversation", END)
# ── CHECKPOINTER SELECTION ───────────────────────────────────
# Development (in-memory, lost on restart):
checkpointer = MemorySaver()
# Uncomment for local persistence (survives restarts):
# conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
# checkpointer = SqliteSaver(conn)
# Uncomment for cloud/production (requires PostgreSQL):
# checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
graph = graph_builder.compile(checkpointer=checkpointer)

# ── MODULE 7: ENTRYPOINT ─────────────────────────────────────
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "user-001"}}

    print("Memory-managed agent ready. Type 'exit' to quit, 'state' to inspect memory.\n")

    while True:
        user_text = input("You: ").strip()

        if not user_text or user_text.lower() == "exit":
            break

        # Debug command: inspect what the agent currently remembers
        if user_text.lower() == "state":
            snapshot = graph.get_state(config)
            print(f"\n[DEBUG] Messages in state: {len(snapshot.values['messages'])}")
            print(f"[DEBUG] Summary: {snapshot.values.get('summary', '(none)')}\n")
            continue

        response = graph.invoke(
            {"messages": [HumanMessage(content=user_text)]},
            config=config
        )

        print(f"Agent: {response['messages'][-1].content}\n")

内存策略决策指南

在研读《内存策略决策指南》时，首先需明确相关规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型接口。 在研读《内存策略决策指南》时，首先需明确相关规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体的功能模块，而非整个复杂的流程。

更新后的关键词参考卡

将更新后的关键词参考卡视为可度量的基准，效果最佳。在扩大范围之前，需记录一份优秀案例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默许部分完成的情况。 保持图表状态简洁且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，且在中断后会导致无法继续处理。

结论：内存是架构的核心，而非事后补充

结论：《内存即架构，而非事后补充》这一观点在被视为可测量的对象时最为有效。在扩大范围之前，先记录一个理想的运行案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。保持图结构简洁且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。

操作检查清单

在制定操作检查清单时，应在修改代码之前明确输入参数、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。

需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品不可或缺的部分，而非后续需要补充的内容。

对于会花费资金或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。

编写简短的操作手册：说明如何轮换密钥、如何清空队列、以及如何回滚上一次的处理操作。

应优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出错时，故障应能指向具体的责任模块，而非复杂的流程链。

对于会花费资金或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。

在升级技术栈之前，应先冻结版本，为关键流程记录完整的操作日志，并确认好回滚步骤。共享环境需要设置速率限制、进行租户身份验证，同时明确密钥轮换的负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。

d865d505a59e的批处理说明：不要将提供者密钥放入仓库中，设定每会话的令牌上限，并将转录内容存储在评估测试用例旁边，以便后续模型更换时仍能保持可比性。