外观
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 列表。
每一轮对话都会发生三步:
- 把用户输入追加为
{"role": "user", "content": ...}。 - 把完整
messages传给 Agent。 - 把模型回答追加为
{"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()