Buffer MemoryとWindow Memory

バッファメモリ：すべてを保持する

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({}))

チェーンへのバッファメモリの統合

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'
)

無制限バッファの問題

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ターンを保持する

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によるウィンドウメモリの実装

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
)

トークン単位とターン単位のウィンドウ

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')

ウィンドウサイズと一貫性の測定

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

システムプロンプトとウィンドウメモリの組み合わせ

# 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への永続化

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

本番環境でのメモリ健全性の監視

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

クイックチェック

レッスンのまとめ