跳到正文

07. Memory 与 State

学习目标

  • 理解短期消息历史为什么能让 Agent “记住”当前会话。
  • 区分 Memory、RAG 和 workflow state,避免把它们混成一个概念。
  • 知道哪些信息适合保存,哪些信息不应该保存。
  • 识别上下文增长、隐私和错误记忆带来的风险。

Memory 不是魔法

大模型本身不会自动记住你上一次运行脚本时说过什么。它能使用的内容,来自本次调用传入的上下文。最常见的短期 Memory 就是一组消息历史:用户说过什么、助手回答过什么,然后在下一次调用时一起交给模型。

运行示例:

bash
uv run python examples/07_memory_agent_cli.py

这个脚本维护一个 messages: list[dict[str, str]]。每次用户输入后,脚本把用户消息追加进去,再把完整消息列表传给 agent.invoke(...)。模型回答后,脚本把助手消息也追加进去。这样下一轮提问时,模型能看到前面的对话。

这种 Memory 是“可见历史”,不是永久记忆。关闭程序后,列表就消失了。即使把历史写入数据库,模型也不会自动读取,仍然需要应用在合适的时候把相关内容取出来并放入上下文。

Memory、RAG、State 的区别

Memory 关注“当前会话里已经发生了什么”。例如用户刚说了自己的偏好,下一轮追问“那按这个偏好推荐呢?”时,消息历史能提供上下文。

RAG 关注“外部知识里有什么”。例如知识库、产品文档、工单记录或私有规范。RAG 通常需要检索相关片段,再把检索结果交给模型。它不是简单保存聊天记录,而是为了让模型基于可追溯资料回答。

State 关注“工作流当前走到哪一步”。在 LangGraph 这类图工作流里,state 可以保存问题、草稿、工具结果、审批状态、重试次数等结构化字段。它不一定是给模型看的自然语言历史,也可能只给程序逻辑使用。

简单判断:

  • Memory:这轮对话需要参考前面聊过的内容。
  • RAG:回答需要查外部资料或长期知识库。
  • State:程序需要记录流程中间结果和下一步路由依据。

风险

  • 上下文增长:消息越多,调用越慢、成本越高,也更容易把无关内容塞给模型。
  • 隐私泄露:不要无差别保存密码、令牌、身份证号、个人隐私或客户敏感数据。
  • 错误记忆:用户或模型说过的内容不一定是真的,不能把聊天记录当作事实来源。
  • 指令污染:历史消息里可能包含过期指令、恶意提示或误导性要求。
  • 状态混乱:把会话 Memory、RAG 结果和 workflow state 都塞进同一个字符串,会让调试变困难。

适合保存的是用户明确希望延续的偏好、任务上下文、当前会话的必要摘要和工作流中间结果。不适合保存的是敏感凭据、一次性验证码、未经确认的事实、无关闲聊和已经过期的操作指令。

代码阅读重点

阅读 examples/07_memory_agent_cli.py 时,重点看 messages 列表。

每一轮对话都会发生三步:

  1. 把用户输入追加为 {"role": "user", "content": ...}
  2. 把完整 messages 传给 Agent。
  3. 把模型回答追加为 {"role": "assistant", "content": ...}

这就是最小 memory。它没有数据库、没有向量检索、没有自动总结,只是把当前会话历史重新交给模型。简单但直观,适合理解 memory 的本质。

什么时候需要更复杂的 memory

当出现下面情况时,简单消息列表就不够了:

  • 会话很长,超过模型上下文窗口。
  • 需要跨会话记住用户偏好。
  • 需要删除或更正历史信息。
  • 需要把短期对话和长期事实区分开。
  • 需要审计谁在什么时候写入了哪些记忆。

这时可以考虑摘要 memory、数据库、向量检索或 LangGraph 的持久化能力。但不要过早引入。先明确你到底要保存什么,以及保存多久。

实验练习

  • 先告诉 Agent “我的名字是小林”,再问“我叫什么?”。
  • 注释掉追加 assistant 消息的代码,观察多轮对话有什么变化。
  • 连续聊十轮,观察回答是否开始变慢或偏离主题。

自测问题

  • memory 和 RAG 都能提供上下文,它们的来源有什么不同?
  • 为什么不能无限保留所有历史消息?
  • workflow state 保存的是聊天内容,还是流程运行状态?

深入理解:Memory 的三种层级

真实 Agent 里的 memory 通常分成三层。

第一层是短期消息历史。它保存当前会话刚发生的对话,适合处理“刚才那个”“按我上面说的”这类引用。

第二层是摘要记忆。当对话很长时,可以把早期消息压缩成摘要,只保留任务目标、用户偏好和关键决策。摘要会损失细节,但能控制上下文长度。

第三层是长期记忆。它跨会话保存用户偏好、项目背景或历史事实。长期记忆必须谨慎,因为错误信息一旦保存,会在未来持续影响模型。

本教程只实现第一层,因为它最直观,也最适合初学者理解上下文如何进入模型。

Memory 设计的关键问题

设计 memory 前先回答四个问题:

  • 保存什么:原始消息、摘要、结构化偏好,还是工具结果?
  • 保存多久:当前会话、一天、一个项目周期,还是永久?
  • 谁能修改:用户、模型、工具,还是只能由程序规则写入?
  • 如何删除:用户要求删除时,系统是否能真正清理?

如果这些问题没有答案,就不要急着做长期 memory。

反模式

  • 把所有历史原样保存并每次发送给模型。
  • 把模型猜测当成长期事实保存。
  • 把敏感信息写入普通日志或向量库。
  • 不区分用户明确偏好和模型临时推断。
  • 没有删除机制。

Memory 会让 Agent 看起来更“懂用户”,但也会放大错误和隐私风险。越长期的 memory,越需要明确权限和审计。

完整示例

python
from __future__ import annotations

import sys
from pathlib import Path

sys.path.append(str(Path(__file__).resolve().parents[1]))

from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_ollama import ChatOllama

from examples.common.config import load_config, ollama_client_kwargs
from examples.common.ollama_check import assert_model_available, assert_ollama_running


def _friendly_error(exc: RuntimeError, model: str) -> str:
    message = str(exc)
    if message.startswith("Cannot reach Ollama."):
        return "无法连接 Ollama。请先打开 Ollama 应用,或运行 `ollama serve`,并检查 OLLAMA_BASE_URL。"
    if message.startswith("Cannot read Ollama model list"):
        return "无法读取 Ollama 模型列表。请检查 Ollama 是否正常运行,以及 OLLAMA_BASE_URL 是否正确。"
    if message.startswith("Ollama model list"):
        return "Ollama 返回的模型列表格式异常。请检查 OLLAMA_BASE_URL 是否指向 Ollama。"
    if message.startswith("Ollama model `"):
        return f"当前配置的 Ollama 模型未安装。请先运行 `ollama pull {model}`。"
    return message


def _final_message_content(agent_response: dict) -> str:
    messages = agent_response.get("messages", [])
    if not messages:
        return ""

    final_message = messages[-1]
    if isinstance(final_message, AIMessage):
        return str(final_message.content)
    return str(getattr(final_message, "content", final_message))


def main() -> None:
    config = load_config()
    try:
        assert_ollama_running(config.ollama_base_url)
        assert_model_available(config.ollama_base_url, config.ollama_model)
    except RuntimeError as exc:
        print(f"错误:{_friendly_error(exc, config.ollama_model)}")
        return

    model = ChatOllama(
        model=config.ollama_model,
        base_url=config.ollama_base_url,
        client_kwargs=ollama_client_kwargs(),
        temperature=0,
    )
    agent = create_agent(
        model=model,
        tools=[],
        system_prompt=(
            "你是一个简洁的中文助手。回答时要使用当前请求中可见的对话历史;"
            "如果用户追问上一轮内容,要结合消息历史回答,不要假装拥有不可见的长期记忆。"
        ),
    )

    messages: list[dict[str, str]] = []

    print("LangChain Memory 与消息历史示例。输入 `exit` 或 `quit` 退出。")
    while True:
        user_input = input("\n你:").strip()
        if user_input.lower() in {"exit", "quit"}:
            break
        if not user_input:
            continue

        messages.append({"role": "user", "content": user_input})
        try:
            response = agent.invoke({"messages": messages})
        except Exception as exc:
            print(f"错误:Agent 调用失败:{exc}")
            messages.pop()
            continue

        answer = _final_message_content(response)
        messages.append({"role": "assistant", "content": answer})
        print(f"AI:{answer}")


if __name__ == "__main__":
    main()