缓冲区记忆与窗口记忆
实现 ConversationBufferMemory 和 ConversationBufferWindowMemory，将最近 N 轮对话保留在上下文中，并衡量窗口大小如何影响连贯性和成本。
缓冲区记忆与窗口记忆 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
缓冲记忆：保留全部内容
缓冲记忆是最简单的策略：将每一轮中的每条消息都存储在列表中，并在每次应用程序接口调用时加入完整历史记录。它可以保留完整上下文，让模型引用任何时刻说过的内容。缺点是上下文会线性增长且没有上限。对于单次完成任务这类短会话，缓冲记忆非常合适，也最容易实现。
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(
return_messages=True, # return Message objects, not a string
memory_key='history' # key to inject into prompt template
)
# Add messages manually
memory.chat_memory.add_user_message('What is a neural network?')
memory.chat_memory.add_ai_message('A neural network is a system of layers...')
# Load what will be injected
print(memory.load_memory_variables({}))将缓冲记忆集成到链中
要将缓冲记忆与 LCEL 配合使用，可以通过 RunnableWithMessageHistory 接入，或者使用 ConversationChain 这一旧版方案。记忆对象负责保存历史记录，链则使用提示词模板中的 MessagesPlaceholder 将其注入。每次调用后，记忆都会自动追加新的一轮对话。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
store = {}
def get_history(session_id: str):
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
chain = (
ChatPromptTemplate.from_messages([
('system', 'You are helpful.'),
MessagesPlaceholder('history'),
('human', '{input}')
])
| ChatOpenAI(model='gpt-4o-mini')
| StrOutputParser()
)
with_memory = RunnableWithMessageHistory(
chain, get_history,
input_messages_key='input',
history_messages_key='history'
)无限缓冲的问题
缓冲记忆在对话变得过长并超出模型的上下文窗口之前都能正常工作。GPT-4o-mini 的上下文容量为 128K，普通聊天可能持续 200–400 轮后才会溢出。但更实际的问题是，即使只有 50 轮，每次请求也会发送超过 5 万个令牌，成本相当可观。因此，您需要一种限制历史记录大小的策略。
import tiktoken
enc = tiktoken.encoding_for_model('gpt-4o-mini')
def count_history_tokens(messages: list) -> int:
total = 0
for msg in messages:
total += len(enc.encode(msg.content))
total += 4 # per-message overhead
return total
# Check how big the history has grown
history = store.get('session-1')
if history:
token_count = count_history_tokens(history.messages)
print(f'History size: {len(history.messages)} messages, {token_count} tokens')窗口记忆：保留最近 N 轮
窗口记忆只保留最近 K 轮对话，丢弃较早的消息。无论对话持续多久，它都会将上下文限制在 K * avg_tokens_per_turn 以内。代价是非常早期的上下文会丢失——模型可能忘记用户在许多轮之前说过的内容。对于大多数通用聊天机器人，保留 5–10 轮通常可以在可接受的成本下提供良好的连贯性。
from langchain.memory import ConversationBufferWindowMemory
# Keep last 5 turns (10 messages: 5 user + 5 assistant)
memory = ConversationBufferWindowMemory(
k=5, # number of TURNS to keep (each turn = user + AI)
return_messages=True,
memory_key='history'
)
# After 10 turns, only turns 6-10 will be in context
# Turns 1-5 are silently dropped使用 InMemoryChatMessageHistory 实现窗口记忆
LangChain 的 InMemoryChatMessageHistory 会保留所有消息，但您可以在将历史记录注入提示词之前对其进行裁剪。一种常见模式是：为记录日志而保存完整历史，但只将最近 N 条消息传给 LLM。对 history.messages 使用 Python 切片表示法即可获取最近的窗口。
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import RunnableLambda
WINDOW_SIZE = 10 # last 10 messages (5 turns)
def get_windowed_history(session_id: str):
full_history = store.get(session_id, InMemoryChatMessageHistory())
store[session_id] = full_history
return full_history
# In the chain, trim before injecting
def trim_history(messages):
return messages[-WINDOW_SIZE:] if len(messages) > WINDOW_SIZE else messages
# Use in prompt with trimming
from langchain_core.messages import trim_messages
trimmer = trim_messages(
max_tokens=2000,
strategy='last',
token_counter=ChatOpenAI(model='gpt-4o-mini'),
include_system=True
)基于令牌的窗口与基于轮次的窗口
窗口大小可以按轮次（最近 K 组人类/AI 对话）或按令牌（历史记录中的最近 T 个令牌）指定。基于令牌的窗口更可靠，因为每轮对话的长度不同——包含代码的一轮可能比只有“是”的一轮长 10 倍。LangChain 的 trim_messages() 工具支持这两种策略，并可以在裁剪历史记录时保留系统提示词。
from langchain_core.messages import trim_messages, SystemMessage, HumanMessage, AIMessage
# Token-based trimming — keep last 1000 tokens of conversation
trimmer = trim_messages(
max_tokens=1000,
strategy='last', # keep most recent messages
token_counter=len, # approximate: count characters / 4
include_system=True, # always include the system prompt
allow_partial=False, # don't split a message in half
start_on='human' # start window on a human message
)
trimmed = trimmer.invoke(all_messages)
print(f'Trimmed to {len(trimmed)} messages')衡量连贯性与窗口大小的关系
选择合适的窗口大小，需要衡量窗口缩小时对话连贯性如何下降。运行一组测试对话，让第 N 轮分别引用第 N-5、N-10 和 N-20 轮的信息。使用 5、10 和 20 轮的窗口进行测试，检查模型是否仍能正确回答。测试结果可以告诉您特定使用场景所需的最小窗口大小。
def test_reference_at_distance(chain_with_memory, distances=[5, 10, 20]):
results = {}
for distance in distances:
session_id = f'test-dist-{distance}'
# Fill with filler turns
for i in range(distance):
chain_with_memory.invoke(
{'input': f'Turn {i}: filler message'},
config={'configurable': {'session_id': session_id}}
)
# Ask about something said at the start
first_msg = 'Recall that the user said the magic word is AZURE.'
response = chain_with_memory.invoke(
{'input': 'What was the magic word?'},
config={'configurable': {'session_id': session_id}}
)
results[distance] = 'AZURE' in response.upper()
return results将系统提示词与窗口记忆结合
使用窗口记忆时，务必保留系统提示词——它定义了 AI 的角色和规则。如果窗口滑过第一轮后不再包含系统提示词，模型可能会失去原有角色。可以在裁剪函数中使用 include_system=True 选项，或者始终在提示词模板中的窗口历史记录之前插入系统消息。
# Always put system prompt BEFORE the windowed history
prompt = ChatPromptTemplate.from_messages([
('system', 'You are a Python tutor. Always explain with code examples.'),
MessagesPlaceholder('history'), # windowed history injected here
('human', '{input}'),
])
# The system prompt is never trimmed — only the history window is managed
# This ensures the model's persona is always present regardless of window缓冲记忆与窗口记忆的选择指南
在以下情况下使用缓冲记忆：对话短且有明确上限（一次性任务、表单向导），完整上下文至关重要（法律分析、代码审查），并且不必担心令牌预算。在以下情况下使用窗口记忆：对话可能任意延长（客户支持、通用助手），近期上下文比远期上下文更重要，并且需要可预测且有上限的令牌成本。
# Decision matrix in code
def choose_memory_strategy(
expected_turns: int,
max_context_tokens: int = 128000,
avg_tokens_per_turn: int = 200
) -> str:
buffer_tokens = expected_turns * avg_tokens_per_turn
if buffer_tokens < max_context_tokens * 0.5:
return 'buffer' # Safe to keep everything
elif expected_turns <= 20:
return 'window_10' # Keep last 10 turns
else:
return 'summary' # Need summarization for long convos
print(choose_memory_strategy(5)) # 'buffer'
print(choose_memory_strategy(50)) # 'window_10'
print(choose_memory_strategy(200)) # 'summary'将窗口持久化到 Redis
在生产环境中，将完整对话历史存储在 Redis 中（以便快速检索），并在读取时将其裁剪到窗口大小。Redis 配合 TTL 可以确保旧会话自动过期。可以使用有序集合，或使用带有 LRANGE 的列表，高效获取最近 N 条消息，而无需加载完整历史记录。
from langchain_community.chat_message_histories import RedisChatMessageHistory
def get_windowed_redis_history(session_id: str, window: int = 10):
# RedisChatMessageHistory stores all messages
history = RedisChatMessageHistory(
session_id=session_id,
url='redis://localhost:6379',
ttl=3600 # 1 hour TTL
)
# Trim to window size in-memory before use
all_msgs = history.messages
if len(all_msgs) > window * 2: # window turns = window*2 messages
history.messages = all_msgs[-(window * 2):]
return history监控生产环境中的记忆健康状况
在生产环境中跟踪以下指标，以发现与记忆相关的问题：每个请求的平均历史令牌数，用于检测会话是否变得过大；上下文窗口利用率（超过 80% 时发出警告）；存储中的会话数量，用于检测会话存储中的记忆泄漏；以及从 Redis 重新加载会话时的缓存命中率。当平均历史令牌数超过可配置的阈值时发出警报。
import time
from langchain_core.callbacks import BaseCallbackHandler
class MemoryMonitorCallback(BaseCallbackHandler):
def on_chain_start(self, serialized, inputs, **kwargs):
history = inputs.get('history', [])
token_estimate = sum(len(m.content.split()) * 1.3 for m in history)
if token_estimate > 50000:
print(f'WARNING: Large history {token_estimate:.0f} estimated tokens')
def on_chain_end(self, outputs, **kwargs):
# Log usage for dashboards
pass快速检查
测试您对缓冲记忆和窗口记忆策略的理解。
课程回顾
本课中您学到了：缓冲记忆保留完整历史，但会无限增长，因此只适合有明确上限的短对话；窗口记忆只保留最近 K 轮，以牺牲远期上下文为代价，实现可预测且有上限的成本；由于消息长度各不相同，使用 trim_messages() 进行基于令牌的裁剪比基于轮次的窗口更可靠。接下来我们将探索适用于没有明确终点的长对话的摘要记忆。
常见问题解答
「缓冲区记忆与窗口记忆」课时是免费的吗？
是的 — 「缓冲区记忆与窗口记忆」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「缓冲区记忆与窗口记忆」这节课中我会学到什么？
实现 ConversationBufferMemory 和 ConversationBufferWindowMemory，将最近 N 轮对话保留在上下文中，并衡量窗口大小如何影响连贯性和成本。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「缓冲区记忆与窗口记忆」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。