LangChain内存管理机制深度解析：从短时对话到长时记忆的架构哲学

from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# 初始化一个简单的对话缓冲内存
memory = ConversationBufferMemory()
llm = ChatOpenAI(model="gpt-4")
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行对话
conversation.predict(input="你好，我叫小明。")
# 内存中现在保存了：Human: 你好，我叫小明。\nAI: [模型的回复]
conversation.predict(input="我的名字是什么？")
# AI能够正确回答“小明”，因为它从内存中读取了历史。

from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document

# 1. 准备一个支持检索的向量数据库作为记忆后端
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings, collection_name="long_term_memory")

# 2. 创建一些“记忆”文档并存储
memories = [
    Document(page_content="小明最喜欢的颜色是蓝色。", metadata={"speaker": "human"}),
    Document(page_content="小明的生日是7月15日。", metadata={"speaker": "human"}),
]
vectorstore.add_documents(memories)

# 3. 创建基于向量检索的记忆组件
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
memory = VectorStoreRetrieverMemory(retriever=retriever)

# 4. 在对话中使用
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 提示词模板，通过 `{history}` 和 `{input}` 注入记忆和当前输入
prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=`你是一个贴心的助手，以下是你已知的关于用户的背景信息：\n{history}\n\n当前对话：\nHuman: {input}\nAI:`
)

llm = ChatOpenAI()
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

# 当用户提问时，记忆组件会自动检索相关背景信息
response = chain.run(input="我喜欢的颜色是什么？")
print(response) # 模型应能回答“蓝色”，因为它从向量库中检索到了相关记忆。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# 定义状态结构，其中包含对话消息列表作为“内存”
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add] # 关键：这是一个累积的消息列表，构成了对话内存

# 定义节点函数
def call_model(state: AgentState):
    llm = ChatOpenAI(model="gpt-4")
    messages = state['messages']
    response = llm.invoke(messages)
    return {"messages": [response]}

# 构建图
graph_builder = StateGraph(AgentState)
graph_builder.add_node("model", call_model)
graph_builder.set_entry_point("model")
graph_builder.add_edge("model", END)

# 使用内存检查点保存器
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# 首次运行，并保存一个检查点（checkpoint_id 类似会话ID）
config = {"configurable": {"thread_id": "thread_abc123"}}
initial_state = {"messages": [HumanMessage(content="你好，请记住我的名字是Alice。")]}
result = graph.invoke(initial_state, config)
print(result['messages'][-1].content)

# 模拟在另一个进程或时间点，从同一个检查点恢复状态并继续对话
new_state = {"messages": [HumanMessage(content="我的名字是什么？")]}
# 注意：这里我们传入相同的 thread_id，LangGraph会加载之前保存的消息历史
continued_result = graph.invoke(new_state, config)
print(continued_result['messages'][-1].content) # 模型应能回答“Alice”