大模型的临时会话
临时会话指的是在一次对话会话（Session）期间，大模型能够记住之前交流过的内容，从而理解上下文、进行连贯对话的能力。会话结束后，这些记忆通常会被丢弃。
核心机制
1. 上下文窗口（Context Window）
- 大模型本身是无状态的，每次调用都是独立的。
- 所谓的“记忆”，实际上是把历史对话记录拼接在每次请求中，一起发送给模型。
- 模型根据上下文窗口内的全部内容进行推理和生成。
- 上下文窗口大小有限（如 128K、200K、1M tokens），超出部分会被截断或丢失。
2. 消息列表（Message List）
- 每次对话由一组消息组成，每条消息包含角色（
system/user/assistant）和内容。 - 新消息不断追加到列表末尾，形成对话历史。
- 下次请求时，将整个消息列表（或截取最近的一部分）传给模型。
3. Checkpoint / State（状态持久化）
- 在 Agent 或复杂工作流中，临时会话的本质是状态管理：每一步执行都会更新状态，下一次调用从最新状态继续。
- 状态不仅包含消息列表，还包含中间变量、工具调用结果、流程控制信息等。
- 通过将状态保存在内存或外部存储中，实现会话的持续和恢复。
临时会话的典型实现方式
1. 手动维护消息列表（最基础）
messages=[{"role":"system","content":"你是一个友好的助手。"}]whileTrue:user_input=input("你: ")messages.append({"role":"user","content":user_input})response=model.invoke(messages)messages.append({"role":"assistant","content":response})print("助手:",response)2. 使用RunnableWithMessageHistory（LangChain 风格）
- 自动管理消息历史的存储和注入。
- 通过
session_id区分不同会话。 - 支持多种历史存储后端（内存、Redis、文件等）。
fromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistory store={}defget_history(session_id):ifsession_idnotinstore:store[session_id]=InMemoryChatMessageHistory()returnstore[session_id]conversation=RunnableWithMessageHistory(chain,get_history,input_messages_key="input",history_messages_key="chat_history")result=conversation.invoke({"input":"你好"},{"configurable":{"session_id":"user_001"}})3. LangGraph 的 State 管理
- 图（Graph）的执行过程中，State 是全局共享的。
- 每次节点更新 State，下一次调用从上次的 State 继续。
- 结合 Checkpointer 可实现会话持久化和恢复。
fromlanggraph.graphimportStateGraph,MessagesStatefromlanggraph.checkpoint.memoryimportMemorySaver graph=builder.compile(checkpointer=MemorySaver())# 第一次调用result=graph.invoke({"messages":[HumanMessage(content="你好")]},{"configurable":{"thread_id":"session_001"}})# 第二次调用（同一 thread_id，自动恢复历史）result=graph.invoke({"messages":[HumanMessage(content="我刚才说了什么？")]},{"configurable":{"thread_id":"session_001"}})临时会话 vs 长期记忆
| 对比维度 | 临时会话 | 长期记忆 |
|---|---|---|
| 作用范围 | 单次会话内（同一session_id/thread_id） | 跨会话、跨用户 |
| 存储内容 | 对话历史、当前状态、中间变量 | 用户画像、知识图谱、持久化偏好 |
| 生命周期 | 会话结束或程序重启后可能丢失 | 永久保存 |
| 实现方式 | 消息列表、State、Checkpointer | 外部存储（数据库、向量库） |
| 访问方式 | 自动注入（框架管理） | 需手动读写 |
临时会话的局限性
- 上下文窗口限制：对话过长时会丢失早期信息。
- Token 成本：每次请求都携带全部历史，成本线性增长。
- 信息遗忘：模型对长文本中间部分的内容“注意力”会减弱（Lost-in-the-Middle 问题）。
应对策略
- 消息裁剪：只保留最近 N 轮对话（
ConversationBufferWindowMemory）。 - 摘要压缩：使用 LLM 对旧消息生成摘要（
ConversationSummaryMemory）。 - 向量记忆：将历史消息向量化，按需检索相关片段（
VectorStoreRetrieverMemory）。 - 状态快照：使用 Checkpointer 保存完整状态，支持时间旅行和故障恢复。
应对策略代码示例
下面分别给出四种策略的代码示例，帮助你直观理解每种方式如何实现。
1. 消息裁剪（ConversationBufferWindowMemory）
适用框架：LangChain
原理：只保留最近 K 轮对话，超出部分自动丢弃。
fromlangchain.memoryimportConversationBufferWindowMemoryfromlangchain.chainsimportConversationChainfromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-3.5-turbo")# 只保留最近 2 轮对话（即 2 次 user + 2 次 assistant）memory=ConversationBufferWindowMemory(k=2)conversation=ConversationChain(llm=llm,memory=memory,verbose=True)conversation.predict(input="我叫小明，我喜欢编程。")conversation.predict(input="你还记得我叫什么吗？")# 此时 memory 中只包含最近两轮消息，之前的“我叫小明”可能被丢弃（取决于 k 值）print(memory.buffer)# 查看当前存储的对话历史2. 摘要压缩（ConversationSummaryMemory）
适用框架：LangChain
原理：每次调用 LLM 将历史对话压缩成一段摘要，保留核心信息，节省 Token。
fromlangchain.memoryimportConversationSummaryMemoryfromlangchain.chainsimportConversationChainfromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-3.5-turbo")# 使用 LLM 生成摘要memory=ConversationSummaryMemory(llm=llm)conversation=ConversationChain(llm=llm,memory=memory,verbose=True)conversation.predict(input="我叫小明，我喜欢编程。")conversation.predict(input="我还喜欢打篮球。")conversation.predict(input="你还记得我的爱好吗？")# 查看当前存储的摘要print(memory.load_memory_variables({})["history"])# 输出可能类似："Human 说他的名字是小明，喜欢编程和打篮球。"3. 向量记忆（VectorStoreRetrieverMemory）
适用框架：LangChain
原理：将历史消息向量化存入向量库，查询时只检索最相关的片段，适合超长对话。
fromlangchain.memoryimportVectorStoreRetrieverMemoryfromlangchain.vectorstoresimportChromafromlangchain.embeddingsimportOpenAIEmbeddingsfromlangchain.chainsimportConversationChainfromlangchain_openaiimportChatOpenAI# 初始化向量库embedding=OpenAIEmbeddings()vectorstore=Chroma(collection_name="memory",embedding_function=embedding,persist_directory="./chroma_memory")# 创建检索器（每次检索返回最相似的 2 条记忆）retriever=vectorstore.as_retriever(search_kwargs={"k":2})# 创建向量记忆memory=VectorStoreRetrieverMemory(retriever=retriever)# 保存记忆（每次对话后调用）memory.save_context({"input":"我叫小明，我喜欢编程。"},{"output":"好的，我记住了。小明喜欢编程。"})memory.save_context({"input":"我还喜欢打篮球。"},{"output":"记住了，小明还喜欢打篮球。"})# 查询相关记忆result=memory.load_memory_variables({"input":"小明有什么爱好？"})print(result["history"])# 输出会包含“小明喜欢编程”和“小明喜欢打篮球”的相关片段4. 状态快照（Checkpointer）
适用框架：LangGraph
原理：在图的每个节点执行后，自动保存整个 State 的快照。支持恢复、回退和会话持久化。
4.1 内存版（MemorySaver）
fromlanggraph.graphimportStateGraph,MessagesStatefromlanggraph.checkpoint.memoryimportMemorySaverfromlangchain_core.messagesimportHumanMessage# 构建一个简单图builder=StateGraph(MessagesState)builder.add_node("chat",lambdastate:{"messages":[llm.invoke(state["messages"])]})builder.set_entry_point("chat")graph=builder.compile(checkpointer=MemorySaver())# 第一次调用，保存状态config={"configurable":{"thread_id":"user_001"}}result=graph.invoke({"messages":[HumanMessage(content="我叫小明")]},config=config)# 第二次调用，从同一 thread_id 恢复状态result=graph.invoke({"messages":[HumanMessage(content="我叫什么名字？")]},config=config)print(result["messages"][-1].content)# 模型会回答“小明”4.2 持久化版（SqliteSaver）
importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver# 创建 SQLite 存储conn=sqlite3.connect("checkpoints.db",check_same_thread=False)checkpointer=SqliteSaver(conn)# 编译图时传入 checkpointergraph=builder.compile(checkpointer=checkpointer)# 之后的使用与 MemorySaver 完全一致，只是状态会持久化到磁盘# 程序重启后，相同 thread_id 仍能恢复历史策略对比总结
| 策略 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 消息裁剪 | 对话轮次可控，不关心早期内容 | 实现简单，Token 可控 | 可能丢失重要早期信息 |
| 摘要压缩 | 长对话但需保留核心信息 | 节省 Token，保留语义摘要 | 摘要可能丢失细节，需额外 LLM 调用 |
| 向量记忆 | 超长对话或知识库场景 | 可扩展至海量历史，按需检索 | 实现复杂，有额外检索延迟 |
| 状态快照 | Agent 工作流、故障恢复 | 保存完整状态，支持时间旅行 | 存储开销大，需持久化支持 |
根据你的实际场景，可以组合使用这些策略，例如：短对话用消息裁剪，长对话用摘要+向量记忆，关键工作流用状态快照。
