无状态 LLM 为什么需要外部记忆

LLM 默认没有记忆

实践中的失忆问题

# Naive stateless approach — model forgets everything
from openai import OpenAI

client = OpenAI()

def chat_stateless(user_message: str) -> str:
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[  # Only the new message — no history!
            {'role': 'user', 'content': user_message}
        ]
    )
    return response.choices[0].message.content

chat_stateless('My name is Alice.')   # Model: 'Hello Alice!'
chat_stateless('What is my name?')    # Model: 'I do not know your name.'

简单但不理想的做法：填入全部历史记录

messages = []  # grows with every turn

def chat_with_full_history(user_message: str) -> str:
    messages.append({'role': 'user', 'content': user_message})
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages  # all history every time
    )
    reply = response.choices[0].message.content
    messages.append({'role': 'assistant', 'content': reply})
    return reply

# Works at first, but messages list grows unboundedly
# After 100 turns: could be 50,000+ tokens per request

记忆策略的设计空间

对话历史的令牌成本

import tiktoken

def estimate_conversation_cost(
    turns: int,
    tokens_per_turn: int,
    price_per_1k_input: float = 0.00015  # gpt-4o-mini
) -> float:
    total_input_tokens = sum(
        (i + 1) * tokens_per_turn  # each turn sends all previous turns
        for i in range(turns)
    )
    total_output_tokens = turns * tokens_per_turn
    cost = (total_input_tokens / 1000) * price_per_1k_input
    print(f'{turns} turns: {total_input_tokens:,} input tokens, ${cost:.4f}')
    return cost

estimate_conversation_cost(50, 200)

将对话历史存储在哪里

# Three storage options for conversation history

# 1. In-process dict (development only — lost on restart)
from langchain_core.chat_history import InMemoryChatMessageHistory

# 2. Redis (production — fast, TTL-based expiry)
from langchain_community.chat_message_histories import RedisChatMessageHistory
history = RedisChatMessageHistory(session_id='user-123', url='redis://localhost:6379')

# 3. PostgreSQL (production — durable, queryable)
from langchain_community.chat_message_histories import PostgresChatMessageHistory
history = PostgresChatMessageHistory(
    session_id='user-123',
    connection_string='postgresql://user:pass@localhost/db'
)

会话 ID：区分不同用户对话

import uuid
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}  # session_id -> history (use Redis in production)

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# Create a new session for each user conversation
def new_session() -> str:
    return str(uuid.uuid4())

alice_session = new_session()
bob_session = new_session()
# Alice and Bob's histories are completely independent

RunnableWithMessageHistory 包装器

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ('system', 'You are a helpful assistant.'),
    MessagesPlaceholder(variable_name='history'),  # history injected here
    ('human', '{input}'),
])

chain = prompt | ChatOpenAI(model='gpt-4o-mini') | StrOutputParser()

chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key='input',
    history_messages_key='history'
)

调用启用记忆的链

session_id = 'user-alice-session-1'

# First message
response1 = chain_with_memory.invoke(
    {'input': 'My name is Alice and I love Python.'},
    config={'configurable': {'session_id': session_id}}
)
print(response1)  # 'Hello Alice! Nice to meet you.'

# Second message — model now knows Alice's name and interest
response2 = chain_with_memory.invoke(
    {'input': 'What is my name and what do I love?'},
    config={'configurable': {'session_id': session_id}}
)
print(response2)  # 'Your name is Alice and you love Python!'

应避免的记忆故障模式

# Mistake 1: Shared history for all users
global_history = InMemoryChatMessageHistory()  # BAD!

# Fix: per-session history
store = {}  # keyed by session_id

# Mistake 2: No TTL on Redis history
# BAD: history = RedisChatMessageHistory(session_id=sid, url=url)
# Good: set TTL to 24 hours
history = RedisChatMessageHistory(
    session_id=session_id,
    url='redis://localhost',
    ttl=86400  # 24 hours in seconds
)

选择记忆策略

快速检查

课程回顾