第7章 没记性的助理谁敢用
第6章我们让Agent学会调用工具了。但还有一个扎心的问题:你跟它聊了十分钟,转头它就把你叫什么、刚才聊了啥全忘了。没有记忆的Agent,就像一个每天失忆的助理——你敢用它吗?这一章,我们给Agent装上记忆系统。
7.1 人的记忆分短期和长期,Agent也一样
7.1.1 三种记忆,三种用途
人的记忆不是一种东西——你有"刚才说了什么"的短期记忆,有"我叫什么名字"的长期记忆,还有"去年生日那天发生了什么"的情景记忆。
Agent也一样。它的记忆系统可以分成三种:
| 记忆类型 | 存什么 | 能记多久 | 典型例子 |
|---|---|---|---|
| 短期记忆 | 当前对话的上下文 | 当前会话 | "你刚才说想看北京的天气" |
| 长期记忆 | 用户偏好、历史记录 | 跨会话 | "你偏好经济舱、靠窗座位" |
| 知识记忆 | 外部文档、数据库 | 持久化 | "公司的退款政策是7天内无条件退" |
7.1.2 短期记忆:LLM的"脑容量"
短期记忆最简单——就是把之前的对话塞进Prompt里。
python
messages = [
{"role": "user", "content": "北京天气怎么样?"},
{"role": "assistant", "content": "北京今天晴,25度。"},
{"role": "user", "content": "那明天呢?"} # 没有"北京",但LLM知道你在问北京
]LLM看到前面提到了"北京",就知道"明天"指的是"北京明天"。这就是短期记忆——让LLM记住前面聊了什么。
但问题很明显:每轮对话都塞进去,上下文窗口迟早被撑爆。GPT-4的上下文窗口虽然有128K token,但如果一个用户跟你聊了100轮,全存进去,token费用会高得离谱。
所以短期记忆需要"管理",后面会讲。
7.1.3 长期记忆:跨会话的"用户档案"
当你第二次打开Agent,它还记得你上次说过什么——这就是长期记忆。
最简单的实现:用数据库存用户的偏好。
python
# 用户第一次说
save_user_preference(user_id="u123", key="seat_preference", value="靠窗")
# 第二次打开Agent
pref = get_user_preference("u123", "seat_preference")
# "靠窗"
# Agent自动带上:"帮你搜了靠窗的座位"真正的长期记忆比这个复杂——需要自动提取关键信息、更新过时的偏好、处理冲突。但这些原理是一样的:把用户说过的重要信息存下来,下次会话时加载。
7.2 LangChain 的 Memory 系统
上面讲了记忆的概念分类——短期、长期、知识。那代码里怎么落地?LangChain 提供了几种开箱即用的 Memory 类,让你不用从头写记忆管理逻辑。
| 类型 | 描述 | 适用场景 |
|---|---|---|
| BufferMemory | 保存所有对话历史 | 简单场景 |
| ConversationSummaryMemory | 自动总结对话内容 | 长对话 |
| ConversationBufferWindowMemory | 只保留最近 N 轮 | 限制上下文长度 |
| RunnableWithMessageHistory | 多会话隔离管理 | 多用户服务 |
7.2.1 BufferMemory:记住每一句话
最直白的记忆——把每轮对话都存下来。
python
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain.agents import create_openai_functions_agent, AgentExecutor
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chat_history = InMemoryChatMessageHistory()
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的助手。"),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_functions_agent(llm, tools=[], prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=[], memory=chat_history)
# 多轮对话 — Agent 记得之前说过什么
response1 = agent_executor.invoke({"input": "我叫张三"})
print(response1["output"]) # 很高兴认识你,张三!
response2 = agent_executor.invoke({"input": "我叫什么名字?"})
print(response2["output"]) # 你叫张三!
# 查看记忆内容
print(chat_history.messages)
# [HumanMessage(content='我叫张三'), AIMessage(content='很高兴认识你,张三!'),
# HumanMessage(content='我叫什么名字?'), AIMessage(content='你叫张三!')]7.2.2 ConversationSummaryMemory:对话太长就自动总结
BufferMemory 的问题是对话长了内存爆炸。SummaryMemory 自动把历史对话压缩成摘要:
python
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm, return_messages=True)
memory.save_context(
{"input": "我叫李四,喜欢编程"},
{"output": "你好李四!编程是一项很棒的技能!"}
)
memory.save_context(
{"input": "我最近在学Python"},
{"output": "Python是个好选择,入门简单功能强大!"}
)
# 内存中存的是摘要,而不是原始对话全文
print(memory.load_memory_variables({})["history"])
# "用户叫李四,喜欢编程,最近在学Python。助手已友好地回应,推荐了Python。"7.2.3 ConversationBufferWindowMemory:只保留最近N轮
滑动窗口——自动丢弃太老的对话,保证内存不爆炸:
python
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=3, return_messages=True)
# 存入5轮对话,但只保留最近3轮
for i in range(5):
memory.save_context(
{"input": f"第{i+1}轮问题"},
{"output": f"第{i+1}轮回答"}
)
print(memory.load_memory_variables({})["history"])
# 只包含第3、4、5轮——第1、2轮已被自动丢弃7.2.4 RunnableWithMessageHistory:多用户会话隔离
真实业务里,你需要给不同用户分别维护记忆,互不串线:
python
from langchain_core.runnables.history import RunnableWithMessageHistory
chat_history_store = {}
def get_history(session_id: str):
"""获取或创建指定会话的历史"""
if session_id not in chat_history_store:
chat_history_store[session_id] = InMemoryChatMessageHistory()
return chat_history_store[session_id]
chain_with_history = RunnableWithMessageHistory(
chain,
get_history,
input_messages_key="message",
history_messages_key="history",
)
# 用户 1 的对话
print(chain_with_history.invoke(
{"message": "我叫王五"},
config={"configurable": {"session_id": "user_001"}}
).content)
# 很高兴认识你,王五!
# 用户 2 的对话(完全独立,互不影响)
print(chain_with_history.invoke(
{"message": "我叫赵六"},
config={"configurable": {"session_id": "user_002"}}
).content)
# 你好赵六!
# 用户 1 继续对话
print(chain_with_history.invoke(
{"message": "我叫什么名字?"},
config={"configurable": {"session_id": "user_001"}}
).content)
# 你叫王五!核心思路:用一个 chat_history_store 字典,按 session_id 管理各自独立的记忆。
7.3 Agent 的类型与模式
前面讲了Agent需要记忆,但Agent本身是怎么工作的?它凭什么能"自己决定做什么"?这一节拆开来看。
7.3.1 Agent 的本质公式
Agent = 模型(大脑)+ 工具(手脚)+ 循环(思考→行动→观察)
和Chain不一样——Chain是预设好的死流程,Agent是LLM自己决定每一步做什么:
| 特性 | Chain | Agent |
|---|---|---|
| 执行方式 | 固定顺序 | 动态决策 |
| 下一步 | 预先定义 | 模型决定 |
| 工具调用 | 不支持 | 支持 |
| 循环 | 单次执行 | 多轮循环 |
| 适用场景 | 固定流程 | 复杂任务 |
7.3.2 ReAct:边想边做
ReAct(Reasoning + Acting)是Agent最经典的思考模式:
- 思考(Think):分析当前情况,决定要不要调工具
- 行动(Action):调用工具获取信息
- 观察(Observe):拿到工具返回的结果
- 判断:任务完成了就返回,没完成继续循环
对比纯推理(Chain-of-Thought),ReAct 能"上网查"获取实时信息;对比纯行动,ReAct 有明确的思考过程,决策更透明、更可解释。
7.3.3 四种Agent类型
LangChain 提供了几种预置 Agent 类型,各有适用场景:
| Agent 类型 | 特点 | 适用场景 |
|---|---|---|
| OpenAI Functions | 使用 OpenAI 原生 function calling | OpenAI 模型首选 |
| ReAct | 标准 ReAct 循环 | 通用场景 |
| Self-Ask | 包含自我追问 | 复杂推理 |
| Conversational | 对话式,自带记忆 | 聊天机器人 |
OpenAI Functions Agent:
python
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
llm = ChatOpenAI(model="gpt-4o", temperature=0)
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun()
tools = [search, wikipedia]
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手,擅长搜索信息并回答问题。"),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"), # Agent 的"草稿纸"
])
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = agent_executor.invoke({"input": "特斯拉的CEO是谁?"})
print(result["output"])ReAct Agent:
python
from langchain.agents import create_react_agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印思考过程,方便调试
max_iterations=5 # 防止死循环
)
result = agent_executor.invoke({"input": "LangChain是什么时候发布的?"})agent_scratchpad 是Agent的"草稿纸"——它记录思考过程、工具调用记录,模型用它来知道自己"想到哪了"。
7.4 让Agent"临时抱佛脚"也能答对边
7.4.1 LLM的知识盲区
不管你的LLM多强,它都有一个致命缺陷:训练数据有截止日期。
- GPT-4 的训练数据截止到某个时间点,不知道之后发生的事情
- 任何LLM都不知道你公司的内部文档
- 你的产品手册、退款政策、FAQ——LLM从未见过
如果用户问你"你们的退款政策是什么",Agent不能瞎编,它必须去查你们公司的真实文档。
这就是 RAG(Retrieval-Augmented Generation,检索增强生成) 的用武之地。
7.4.2 RAG的工作流程
RAG的思路很简单:
回答问题前,先去知识库里找到相关资料,拿着资料再回答。
像一个开卷考试——你可以翻书,但不能瞎编。
第一步:把用户的"退款政策是什么"转成一个向量(一串数字,代表语义)。
第二步:跟知识库里所有文档的向量做相似度计算,找到最相关的。
第三步:取出最相关的3-5条,作为"参考资料"。
第四步:把参考资料和用户问题一起塞给LLM。
第五步:LLM基于这些真实资料生成回答。
7.4.3 代码演示
python
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# 1. 加载文档
loader = TextLoader("company_policy.txt")
documents = loader.load()
# 2. 切分文档(每段500字,重叠50字保证上下文不丢失)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)
# 3. 转成向量,存入ChromaDB
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
# 4. 检索相关文档
query = "退款政策是什么?"
relevant_docs = vectorstore.similarity_search(query, k=3)
# 5. 拼接Prompt,让LLM基于文档回答
context = "\n".join([doc.page_content for doc in relevant_docs])
prompt = f"""基于以下文档内容回答问题。如果文档中没有相关信息,请说"未找到相关信息"。
文档内容:
{context}
问题:{query}
"""7.5 RAG 进阶:LangChain 实战
上一节展示了RAG的骨架代码,但真正在生产中使用LangChain做RAG,你还需要知道更多细节。这一节补全 LangChain RAG 生态里的关键组件。
7.5.1 不止 TXT:各种文档加载器
LangChain 提供了丰富的文档加载器,支持各种格式:
| 类型 | 加载器 | 说明 |
|---|---|---|
PyPDFLoader | PDF 文件 | |
| Word | Docx2txtLoader | Word 文档 |
| Markdown | UnstructuredMarkdownLoader | Markdown 文件 |
| CSV | CSVLoader | CSV 文件 |
| 网页 | WebBaseLoader | 网页内容 |
| 目录 | DirectoryLoader | 批量加载文件夹 |
python
# pip install langchain-community pypdf python-docx
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader
from langchain_community.document_loaders import WebBaseLoader, DirectoryLoader
# PDF
loader = PyPDFLoader("document.pdf")
pages = loader.load() # 返回 Document 对象列表,每页一个
# Word
loader = Docx2txtLoader("document.docx")
docs = loader.load()
# 网页
loader = WebBaseLoader("https://python.langchain.com/docs/")
docs = loader.load()
# 批量加载目录
loader = DirectoryLoader("./documents/", glob="**/*.pdf", loader_cls=PyPDFLoader)
docs = loader.load()7.5.2 分割器的高级用法
之前用的 RecursiveCharacterTextSplitter 还有更多参数可以调:
python
from langchain.text_splitter import RecursiveCharacterTextSplitter, TokenTextSplitter
# 字符分割 —— 指定分隔符优先级
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
length_function=len,
separators=["\n\n", "\n", "。", "!", "?", " ", ""], # 优先按段落切
)
# Token 分割 —— 按 token 数量切分,更精准
token_splitter = TokenTextSplitter(
chunk_size=500, # 最大 token 数
chunk_overlap=50, # 重叠 token 数
)chunk_size 选择建议:100-300 tokens 适合简单问答;500-1000 tokens 适合大多数场景(推荐);1000+ tokens 适合需要更多上下文的复杂任务。
7.5.3 更多嵌入提供商
除了 OpenAI,LangChain 还支持多种嵌入模型:
python
from langchain_openai import OpenAIEmbeddings
# OpenAI(在线)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector = embeddings.embed_query("你好,世界") # 1536维向量python
# 本地运行(Ollama)
from langchain_community.embeddings import OllamaEmbeddings
embeddings = OllamaEmbeddings(model="nomic-embed-text")
# Cohere(多语言)
from langchain_cohere import CohereEmbeddings
embeddings = CohereEmbeddings(model="embed-multilingual-v3.0", cohere_api_key="your-key")
# HuggingFace(开源模型)
from langchain_huggingface import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh")| 提供商 | 模型 | 特点 |
|---|---|---|
| OpenAI | text-embedding-3-small/large | 通用,效果好 |
| BGE | bge-large-zh | 中文表现优秀,开源 |
| Cohere | embed-multilingual-v3.0 | 多语言 |
| Ollama | nomic-embed-text | 本地运行,离线可用 |
7.5.4 FAISS:大规模向量检索
之前用了 ChromaDB,LangChain 还原生支持 FAISS(Meta开源的向量检索库):
python
# pip install faiss-cpu
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
vectorstore = FAISS.from_documents(
documents=chunks,
embedding=OpenAIEmbeddings()
)
# 保存到本地
vectorstore.save_local("faiss_index")
# 下次加载(不用重新计算向量)
loaded_vectorstore = FAISS.load_local(
"faiss_index",
OpenAIEmbeddings(),
allow_dangerous_deserialization=True # 仅在信任的来源时使用
)
# 带分数的检索
results = vectorstore.similarity_search_with_score("LangChain的Agent是什么?", k=3)
for doc, score in results:
print(f"相似度: {score:.4f} | 内容: {doc.page_content[:100]}...")7.5.5 用 LCEL 方式构建 RAG 链
LangChain 的 LCEL(LangChain Expression Language)让你用 | 管道拼装 RAG 链:
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
RAG_PROMPT = ChatPromptTemplate.from_template("""根据以下上下文回答问题。如果上下文中没有相关信息,请说明不知道。
上下文:
{context}
问题:{question}
回答:""")
# LCEL 管道:检索 → 拼上下文 → 填提示词 → LLM → 输出
rag_chain = (
{"context": retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])),
"question": RunnablePassthrough()}
| RAG_PROMPT
| llm
| StrOutputParser()
)
result = rag_chain.invoke("LangChain的Agent是什么?")
print(result)7.5.6 多查询检索与上下文压缩
多查询检索器——从一个问题生成多个查询角度,综合检索结果:
python
from langchain.retrievers.multi_query import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm,
include_original=True
)
# 例:"Agent是什么?" → 自动生成多个查询:
# - "LangChain中Agent的作用"
# - "如何使用Agent"
# 综合多个查询的检索结果上下文压缩——检索后过滤掉相关性低的内容:
python
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_transformers import EmbeddingsFilter
embeddings_filter = EmbeddingsFilter(
embeddings=embeddings,
similarity_threshold=0.8 # 只保留相似度 > 0.8 的结果
)
compression_retriever = ContextualCompressionRetriever(
base_retriever=vectorstore.as_retriever(),
document_compressor=embeddings_filter
)7.5.7 BM25 混合搜索:向量 + 关键词
向量检索擅长语义匹配,关键词检索擅长精确匹配——二者结合效果最好:
python
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
# 关键词搜索
bm25_retriever = BM25Retriever.from_texts(
texts=[doc.page_content for doc in chunks]
)
bm25_retriever.k = 2
# 向量搜索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
# 组合检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.5, 0.5] # 各占一半
)
results = ensemble_retriever.invoke("LangChain Agent")7.5.8 重排序:用更强模型精挑
先用向量库召回20条候选,再用重排序模型挑出最相关的5条:
python
from langchain_community.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever
compressor = CohereRerank(
cohere_api_key="your-key",
top_n=5 # 返回前5个
)
rerank_retriever = ContextualCompressionRetriever(
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20}),
document_compressor=compressor
)7.6 Agent的"记忆仓库"
7.6.1 向量是什么
你可能注意到了,RAG里一直在说"向量"。那向量到底是什么?
向量就是语义的"坐标"。 意思相近的两句话,它们的向量位置也很近。
比如:
- "今天天气真好" 的向量和 "外面阳光明媚" 的向量很接近
- "今天天气真好" 的向量和 "如何解微积分" 的向量就离得很远
当用户问"退款政策",系统就把这个问题的向量和所有文档的向量做比较,找距离最近的几条。
7.6.2 向量数据库选型
| 数据库 | 适合场景 | 特点 |
|---|---|---|
| ChromaDB | 原型、小项目、学习 | 轻量,Python原生,几行代码就能跑 |
| FAISS | 大规模检索 | Meta开源,速度快,生产级性能 |
| Milvus | 企业级生产环境 | 分布式,支持海量数据,运维成本高 |
本书用ChromaDB。 因为它最简单——不用装Docker、不用配服务器,pip install chromadb 就能跑。
7.7 让Agent"记得准、记得快"
7.7.1 文档切分:太大太小都不好
RAG的质量很大程度上取决于你怎么切分文档。
切太大(比如一段2000字):一段里包含了太多无关信息,检索精度差。
切太小(比如一段50字):上下文太碎,LLM看不懂。
一般建议500-800字一段,并且要有50-100字的重叠——防止关键信息刚好卡在两段的边界上。
python
# 好的切分方式
RecursiveCharacterTextSplitter(
chunk_size=500, # 每段500字
chunk_overlap=50, # 重叠50字
separators=["\n\n", "\n", "。", ","] # 优先按段落切,再按句子
)7.7.2 检索优化:别只靠"相似度"
简单的向量检索有局限性——"最相似"不一定等于"最相关"。
优化手段:
混合检索:向量相似度 + 关键词匹配。用户搜"退款政策",不仅匹配语义相似的文档,也精确匹配包含"退款"关键词的。
重排序:从向量库召回了20条候选,再用一个更强的模型挑出最相关的5条。
元数据过滤:提取文档的属性(日期、类别、来源),先缩小检索范围再匹配。
7.7.3 嵌入模型的选择
把文字转成向量的"翻译官"叫嵌入模型(Embedding Model)。
| 模型 | 特点 |
|---|---|
| OpenAI text-embedding-3-small | 便宜、通用、效果好 |
| BGE 系列 | 中文表现优秀,开源 |
| sentence-transformers | 本地运行,离线可用 |
如果做中文场景,BGE-large-zh 是个好选择;如果追求省事,OpenAI的embedding最简单。
7.8 用 LangGraph 持久化状态
前面讲的记忆机制解决的是"Agent记不记得"的问题,但还有一个问题:对话断了怎么办?
用户聊到一半关掉了网页,再打开时Agent要从头开始吗?这就是 LangGraph 的 Checkpointer 要解决的问题——把 Agent 的状态持久化存下来。
7.8.1 Checkpointer:Agent 的"存档"
python
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# 编译图时传入 checkpointer
app = graph.compile(checkpointer=checkpointer)Checkpointer 让你可以:暂停和恢复执行、维护多轮对话上下文、实现线程隔离。
7.8.2 thread_id:一人一线程,互不干扰
每个对话会话对应一个 thread_id,不同用户的状态完全隔离:
python
# 用户 A 的对话
config_a = {"configurable": {"thread_id": "user_123"}}
result_a = app.invoke({"messages": ["你好"]}, config_a)
# 用户 B 的对话(完全独立,互不影响)
config_b = {"configurable": {"thread_id": "user_456"}}
result_b = app.invoke({"messages": ["你好"]}, config_b)
# 恢复用户 A 的对话(从上次暂停的地方继续)
result_a2 = app.invoke({"messages": ["继续刚才的话题"]}, config_a)不同 thread_id 对应各自独立的状态,存储在共享的 Checkpointer 中,线程之间完全隔离。
7.8.3 三种后端,三种场景
| 后端 | 存储方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| MemorySaver | 内存 | 最快 | 进程重启丢失 | 开发调试 |
| SqliteSaver | SQLite 文件 | 单机持久化 | 不支持并发 | 生产入门 |
| PostgresSaver | PostgreSQL | 高并发、分布式 | 需要额外部署 | 生产环境 |
python
# 开发用
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver() # 进程重启后丢失,仅调试使用
# 单机生产
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("messages.db")
# 分布式生产
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@localhost/db"
)7.8.4 中断与恢复:需要人工审批时
有些操作不能自动执行——大额转账、发送敏感内容——这时候用 NodeInterrupt 暂停 Agent,等人工确认后再恢复:
python
from langgraph.errors import NodeInterrupt
def approval_node(state: AgentState) -> dict:
if state["amount"] > 10000:
# 暂停执行,等待人工审批
raise NodeInterrupt(f"需要审批:金额 {state['amount']}")
return {"approved": True}
# 恢复执行(外部审批通过后)
app.invoke(
None, # 不传新输入,从 checkpoint 恢复
config={"configurable": {"thread_id": "tx_123"}}
)典型使用场景:大额订单需要人工审批、发送敏感内容需要确认、等待外部 API 回调。
7.9 本章小结
这一章我们给Agent装上了"记忆":
- 三种记忆:短期记对话、长期记偏好、知识库记文档。各司其职,互不冲突。
- LangChain Memory 类:BufferMemory、SummaryMemory、滑动窗口、RunnableWithMessageHistory 多会话隔离——让你不用从零写记忆管理。
- Agent 类型:OpenAI Functions、ReAct、Self-Ask 各有适用场景,ReAct 的 Think→Action→Observe 循环是核心。
- RAG原理:回答问题前先去知识库翻资料。问题→向量→检索→拼接→回答,五步走。
- RAG 进阶实战:多种文档加载器(PDF/Word/网页)、FAISS 本地存储、LCEL 管道构建 RAG 链、BM25 混合搜索、Cohere 重排序。
- 向量本质:语义的坐标。意思越近,向量距离越近。
- 检索技巧:文档切500-800字,50字重叠。混合检索+重排序让结果更准。
- LangGraph Checkpointer:MemorySaver/SqliteSaver/PostgresSaver 持久化状态,thread_id 线程隔离,NodeInterrupt 实现人工审批。
✅ 知识点检查
学完这一章,试试回答这几个问题:
- [ ] Agent的三种记忆分别是什么?各解决什么问题?
- [ ] LangChain 的 BufferMemory 和 ConversationSummaryMemory 有什么区别?什么时候用哪个?
- [ ] RunnableWithMessageHistory 如何实现多用户会话隔离?
- [ ] Agent 的四种类型(OpenAI Functions / ReAct / Self-Ask)各适合什么场景?
- [ ] ReAct 的 Think→Action→Observe 循环具体怎么运作?
- [ ] RAG的五个步骤是什么?用你自己的话描述一遍。
- [ ] LCEL 如何用
|管道构建一个 RAG 链? - [ ] BM25 和向量检索各有什么优势?EnsembleRetriever 怎么组合它们?
- [ ] 文档切分为什么要有"重叠"?切太大或太小各有什么问题?
- [ ] LangGraph 的 MemorySaver、SqliteSaver、PostgresSaver 分别适合什么场景?
- [ ] NodeInterrupt 的作用是什么?什么场景下用它?
📚 延伸阅读
- LangChain RAG教程:https://python.langchain.com/docs/tutorials/rag/
- LangGraph Checkpointer 文档:https://langchain-ai.github.io/langgraph/how-tos/persistence/
- ChromaDB官方文档:https://docs.trychroma.com
- FAISS 官方文档:https://github.com/facebookresearch/faiss
- BGE嵌入模型:https://huggingface.co/BAAI/bge-large-zh
- 本书配套源码:关注公众号「图解AI系列」免费领取
🎯 下一章预告
第8章,Agent有脑子、有手脚、有记性了,但还差一样——
"会规划才值钱。让Agent学会拆任务、做计划、错了能改。"

