Bài 24: Memory — lưu lịch sử conversation

Sau bài này bạn sẽ:

• Hiểu tại sao LLM không tự nhớ context giữa các lần gọi và app phải tự xử lý.
• Implement chatbot multi-turn bằng RunnableWithMessageHistory — API chuẩn của LangChain 0.3+.
• Biết cách chuyển backend lưu history từ InMemory sang Redis/Postgres mà không đổi logic chain.
• Áp dụng trim_messages để giới hạn token history và tránh chi phí tăng vô kiểm.
• Biết cách implement summary memory khi cần giữ context dài hạn.
• Nhận diện và sửa được 5 pitfall phổ biến khi implement memory.

Khi gọi API của OpenAI, Anthropic hay bất kỳ LLM provider nào, mỗi request là độc lập. Server không lưu gì giữa hai lần gọi. Nếu bạn hỏi:

Turn 1: "FastAPI là gì?"
Turn 2: "Còn nếu dùng async thì sao?"

Ở turn 2, model không biết "async" liên quan tới FastAPI nếu bạn không đưa lại lịch sử từ turn 1 vào request. Đây là hành vi thiết kế, không phải bug — REST API không có session state.

Giải pháp duy nhất: app tự lưu danh sách message và đưa toàn bộ (hoặc một phần giới hạn) danh sách đó vào prompt của mỗi turn mới.

Cơ chế đó chính là "memory" trong LangChain — không phải LLM nhớ, mà app nhớ và re-inject vào mỗi lần gọi.

Chi phí thực tế

Nếu mỗi turn trung bình 200 token và conversation kéo dài 50 turn, turn thứ 50 sẽ gửi ~10 000 token chỉ cho lịch sử, chưa kể system prompt và câu hỏi hiện tại. Với gpt-4o (tính theo input tokens), chi phí tăng tuyến tính theo số turn. Đây là lý do cần chiến lược truncate/summarize.

LangChain biểu diễn lịch sử hội thoại bằng list[BaseMessage]. Các subclass thường dùng:

• HumanMessage — message từ user.
• AIMessage — response từ LLM.
• SystemMessage — system prompt (thường chỉ có 1, ở đầu).
• ToolMessage — kết quả từ tool call (dùng trong agent).

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

history = [
    SystemMessage(content="Bạn là trợ lý kỹ thuật."),
    HumanMessage(content="FastAPI là gì?"),
    AIMessage(content="FastAPI là web framework Python hiệu năng cao..."),
    HumanMessage(content="Còn nếu dùng async thì sao?"),
]

Khi gọi llm.invoke(history), model nhận được toàn bộ ngữ cảnh và trả lời cho turn hiện tại (message cuối cùng trong list).

ChatMessageHistory là wrapper đơn giản quanh list[BaseMessage], thêm các method tiện lợi:

from langchain_community.chat_message_histories import ChatMessageHistory

hist = ChatMessageHistory()
hist.add_user_message("FastAPI là gì?")
hist.add_ai_message("FastAPI là web framework...")
print(hist.messages)  # [HumanMessage(...), AIMessage(...)]

RunnableWithMessageHistory là lớp wrapper LCEL-native (có từ LangChain 0.1, stable trong 0.3). Nó tự động:

1. Load history từ backend theo session_id.
2. Inject history vào input dict trước khi gọi chain.
3. Sau khi chain trả kết quả, lưu message mới (human + AI) vào backend.

Cấu trúc prompt cần có MessagesPlaceholder để nhận danh sách message:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# Prompt: system + placeholder cho history + câu hỏi hiện tại
prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là trợ lý kỹ thuật."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

# Store in-memory: session_id → ChatMessageHistory
store: dict[str, ChatMessageHistory] = {}

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

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

Sử dụng — truyền session_id qua config:

config = {"configurable": {"session_id": "user-123"}}

r1 = chain_with_history.invoke({"input": "FastAPI là gì?"}, config=config)
print(r1.content)

r2 = chain_with_history.invoke({"input": "Còn nếu dùng async thì sao?"}, config=config)
print(r2.content)
# → LLM trả lời với đầy đủ context turn trước

Ở turn 2, chain_with_history tự động đưa cả HumanMessage("FastAPI là gì?") và AIMessage(r1) vào slot history của prompt trước khi gọi LLM.

Tham số quan trọng

• input_messages_key: tên key trong input dict chứa câu hỏi hiện tại. Mặc định "input".
• history_messages_key: tên biến trong prompt MessagesPlaceholder. Mặc định "history".
• output_messages_key: dùng khi chain trả về dict thay vì AIMessage trực tiếp.

ChatMessageHistory (từ langchain_community hoặc langchain_core) lưu message trong một Python dict trong RAM. Phù hợp cho prototype và unit test, không phù hợp cho production vì:

• Mất toàn bộ history khi process restart.
• Không chia sẻ được giữa nhiều process hoặc nhiều worker.

Nếu dùng FastAPI với uvicorn --workers 4, mỗi worker có store riêng — user có thể bị route sang worker khác và mất history. Với InMemory backend, chỉ nên dùng worker đơn hoặc chấp nhận mất history.

Pattern đặt store ở module level (không phải bên trong hàm) để tồn tại qua nhiều request trong cùng một process:

# app/memory_store.py
from langchain_community.chat_message_histories import ChatMessageHistory

_store: dict[str, ChatMessageHistory] = {}

def get_in_memory_history(session_id: str) -> ChatMessageHistory:
    if session_id not in _store:
        _store[session_id] = ChatMessageHistory()
    return _store[session_id]

Thay ChatMessageHistory bằng RedisChatMessageHistory — interface giống nhau, không cần đổi gì trong chain hay RunnableWithMessageHistory:

pip install langchain-community redis

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_redis_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379/0",
        ttl=3600,  # tự xóa sau 1 giờ (giây), tuỳ chọn
    )

# Thay get_in_memory_history bằng get_redis_history
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_redis_history,
    input_messages_key="input",
    history_messages_key="history",
)

Message được serialize dưới dạng JSON và lưu vào Redis key có dạng message_store:{session_id}. Nhiều worker, nhiều process cùng đọc/ghi được vì tất cả chỉ vào một Redis instance.

TTL (Time To Live)

Nên set ttl để tự dọn history cũ. Nếu không set, key tồn tại vĩnh viễn — Redis có thể hết RAM nếu số session lớn. Giá trị hợp lý tuỳ use case: 1–24 giờ cho chatbot support, vài ngày cho assistant cá nhân.

Redis trên production

Dùng Redis Sentinel hoặc Redis Cluster cho high availability. URL format: redis://user:password@host:6379/0 hoặc rediss://host:6380/0 (TLS). Với managed Redis (ElastiCache, Upstash, Redis Cloud), lấy URL từ dashboard.

langchain_community cung cấp nhiều backend, tất cả đều implement BaseChatMessageHistory:

• PostgresChatMessageHistory (pip install psycopg2-binary): lưu vào bảng SQL. Tiện nếu hệ thống đã dùng Postgres.
• SQLChatMessageHistory: generic SQLAlchemy — SQLite, MySQL, Postgres đều được.
• DynamoDBChatMessageHistory: AWS DynamoDB.
• CosmosDBChatMessageHistory: Azure Cosmos DB.
• MongoDBChatMessageHistory: MongoDB.
• FirestoreChatMessageHistory: Google Firestore.
• UpstashRedisChatMessageHistory: Upstash Redis (serverless).

Ví dụ SQLite (dùng trong local dev thay InMemory để history persist qua restart):

from langchain_community.chat_message_histories import SQLChatMessageHistory

def get_sqlite_history(session_id: str) -> SQLChatMessageHistory:
    return SQLChatMessageHistory(
        session_id=session_id,
        connection_string="sqlite:///chat_history.db",
    )

Tự implement backend

Nếu cần backend tùy chỉnh (ví dụ lưu vào Elasticsearch hoặc custom DB nội bộ), subclass BaseChatMessageHistory:

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, messages_from_dict, messages_to_dict

class MyCustomHistory(BaseChatMessageHistory):
    def __init__(self, session_id: str):
        self.session_id = session_id

    @property
    def messages(self) -> list[BaseMessage]:
        # đọc từ custom storage
        raw = my_db.get(self.session_id) or []
        return messages_from_dict(raw)

    def add_messages(self, messages: list[BaseMessage]) -> None:
        # ghi vào custom storage
        existing = messages_to_dict(self.messages)
        existing.extend(messages_to_dict(messages))
        my_db.set(self.session_id, existing)

    def clear(self) -> None:
        my_db.delete(self.session_id)

Nếu không giới hạn, history tăng không ngừng theo số turn và chi phí token tăng theo. Bốn chiến lược phổ biến, mỗi cái có đánh đổi khác nhau:

Chiến lược Vector-based (lưu old messages vào vector DB, retrieve theo query) cũng tồn tại nhưng phức tạp hơn đáng kể và ít dùng trừ khi conversation cực kỳ dài với nhu cầu truy xuất lịch sử ngẫu nhiên.

trim_messages là utility function trong langchain_core (0.2+), thực hiện token-limited strategy:

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI

def trim_history(messages):
    return trim_messages(
        messages,
        max_tokens=4000,           # giới hạn token
        strategy="last",           # giữ messages cuối (bỏ messages đầu)
        token_counter=ChatOpenAI(model="gpt-4o-mini"),  # dùng tiktoken nội bộ
        include_system=True,       # KHÔNG drop SystemMessage dù nằm đầu list
        start_on="human",          # đảm bảo sau trim, message đầu tiên là HumanMessage
    )

Tích hợp vào chain bằng RunnablePassthrough.assign:

from langchain_core.runnables import RunnablePassthrough

# Trim history trước khi đưa vào prompt
chain = (
    RunnablePassthrough.assign(
        history=lambda x: trim_history(x["history"])
    )
    | prompt
    | llm
)

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_redis_history,
    input_messages_key="input",
    history_messages_key="history",
)

Lưu ý quan trọng: trim_messages chỉ trim những gì đưa vào LLM — nó không xóa message khỏi backend. Backend (Redis, Postgres, ...) vẫn lưu đầy đủ. Nếu muốn dọn storage, cần gọi history.clear() hoặc set TTL.

Thay token_counter bằng tiktoken trực tiếp

Dùng ChatOpenAI làm token counter sẽ gọi tiktoken locally (không gọi API), nên không tốn chi phí. Nếu muốn avoid dependency vào OpenAI cho việc đếm token, có thể truyền callable:

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o-mini")

def count_tokens(messages) -> int:
    # đếm tổng token của tất cả message content
    total = 0
    for m in messages:
        total += len(enc.encode(m.content))
    return total

trimmed = trim_messages(
    messages,
    max_tokens=4000,
    strategy="last",
    token_counter=count_tokens,
    include_system=True,
    start_on="human",
)

Thay vì drop hard, summary strategy gọi LLM để tóm tắt các message cũ thành một SystemMessage ngắn, rồi thay thế phần đó trong history.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini")

# Prompt để summarize
summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "Tóm tắt ngắn gọn cuộc hội thoại dưới đây thành tối đa 3 câu, "
               "giữ lại các thông tin kỹ thuật quan trọng."),
    ("human", "{conversation}"),
])
summary_chain = summary_prompt | llm

def maybe_summarize(messages: list, threshold: int = 20) -> list:
    """
    Nếu có hơn `threshold` messages, summarize phần đầu
    và giữ lại 4 messages cuối + summary ở đầu.
    """
    if len(messages) <= threshold:
        return messages

    # Tách: system message (nếu có) + old messages + recent messages
    system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    non_system = [m for m in messages if not isinstance(m, SystemMessage)]

    old = non_system[:-4]   # tất cả trừ 4 message cuối
    recent = non_system[-4:]  # 4 message gần nhất

    # Gọi LLM để summarize
    conversation_text = "\n".join(
        f"{'User' if isinstance(m, HumanMessage) else 'Assistant'}: {m.content}"
        for m in old
    )
    summary_result = summary_chain.invoke({"conversation": conversation_text})
    summary_msg = SystemMessage(content=f"[Tóm tắt hội thoại trước]: {summary_result.content}")

    return system_msgs + [summary_msg] + recent

Hàm này gọi LLM thêm 1 lần mỗi khi summary được trigger, tốn thêm chi phí. Trong production, cân nhắc chỉ summarize theo batch (ví dụ mỗi 50 message thay vì mỗi turn), hoặc chạy background task.

Tích hợp vào chain tương tự trim_history:

chain = (
    RunnablePassthrough.assign(
        history=lambda x: maybe_summarize(x["history"])
    )
    | prompt
    | llm
)

session_id là key phân tách history giữa các conversation. Trong FastAPI endpoint, lấy session_id từ request của client (thường từ header hoặc path param), không tự generate:

from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@app.post("/chat")
async def chat(
    request: ChatRequest,
    x_session_id: str = Header(...),  # client gửi session_id qua header
):
    config = {"configurable": {"session_id": x_session_id}}
    response = await chain_with_history.ainvoke(
        {"input": request.message},
        config=config,
    )
    return {"reply": response.content}

Namespace cho multi-user

Trong hệ thống có nhiều user, dùng compound key để tránh nhầm lẫn:

def get_history_for_user(user_id: str, thread_id: str):
    session_id = f"{user_id}:{thread_id}"
    return RedisChatMessageHistory(session_id=session_id, url=REDIS_URL)

Không bao giờ dùng chung session_id giữa hai user khác nhau. Nếu backend không có access control (Redis thường không), thì isolation phải đảm bảo ở tầng application — session_id phải không guessable hoặc kèm theo authentication check.

LangChain 0.3 đã đánh dấu deprecated (và sẽ remove trong 1.0) các API cũ sau đây. Nếu bạn thấy trong codebase cũ, đây là cách migrate:

Cách nhận biết code đang dùng legacy: import từ langchain.memory hoặc dùng ConversationChain. Khi chạy, LangChain 0.3 sẽ in LangChainDeprecationWarning.

# Legacy — KHÔNG dùng nữa (sẽ remove trong LangChain 1.0)
from langchain.memory import ConversationBufferMemory      # deprecated
from langchain.chains import ConversationChain             # deprecated

# Hiện tại — LCEL + RunnableWithMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

Với chatbot đơn giản, RunnableWithMessageHistory là đủ. Tuy nhiên với agent stateful phức tạp hơn — nhiều bước xử lý, conditional branching, human-in-the-loop — LangGraph cung cấp cơ chế quản lý state toàn diện hơn.

LangGraph có MemorySaver (in-memory) và RedisSaver, SqliteSaver để persist toàn bộ graph state (không chỉ chat history) qua các lần chạy. State trong LangGraph là dict tùy chỉnh — có thể chứa messages, intermediate results, metadata, bất kỳ field nào bạn định nghĩa.

Module 5 của series này đi sâu vào LangGraph. Nếu usecase là chatbot thuần (không có tool, không branching phức tạp), RunnableWithMessageHistory trong bài này là đủ.

1. Thiếu MessagesPlaceholder trong prompt

# Sai — prompt không có slot cho history
prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là trợ lý."),
    ("human", "{input}"),
])
# RunnableWithMessageHistory sẽ load history nhưng không đâu để inject
# → LLM không nhận được context cũ, multi-turn không hoạt động

# Đúng
prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là trợ lý."),
    MessagesPlaceholder(variable_name="history"),  # phải có
    ("human", "{input}"),
])

2. Sai history_messages_key hoặc input_messages_key

# MessagesPlaceholder dùng variable_name="history"
# nhưng khai báo history_messages_key="chat_history" → không khớp → KeyError hoặc empty

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",  # phải khớp với variable_name trong MessagesPlaceholder
)

3. Store in-memory khai báo bên trong hàm get_history

# Sai — store tạo mới mỗi lần hàm được gọi → history luôn rỗng
def get_history(session_id: str):
    store = {}  # ← lỗi: local variable, không persist
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Đúng — store ở module level
_store: dict[str, ChatMessageHistory] = {}

def get_history(session_id: str):
    if session_id not in _store:
        _store[session_id] = ChatMessageHistory()
    return _store[session_id]

4. Không giới hạn history → chi phí tăng không kiểm soát

Conversation 100 turn với trung bình 300 token/turn = 30 000 token chỉ cho history ở turn cuối. Với gpt-4o ($2.50/1M input token), mỗi turn thứ 100 tốn ~$0.075 chỉ cho history. Phải có trim_messages hoặc summary strategy trước khi production.

5. Dùng cùng session_id cho nhiều chain khác nhau

# Nếu chain A và chain B cùng dùng một get_history với session_id "user-123"
# → history của A lẫn với history của B
# → LLM nhận được context không liên quan, trả lời sai

# Cách xử lý: namespace session_id theo chain
def get_history_for_chain_a(session_id):
    return get_redis_history(f"chain_a:{session_id}")

def get_history_for_chain_b(session_id):
    return get_redis_history(f"chain_b:{session_id}")

• LLM API stateless — app phải tự lưu và re-inject list[BaseMessage] vào mỗi turn.
• Pattern chuẩn LangChain 0.3+: RunnableWithMessageHistory + MessagesPlaceholder trong prompt.
• Backend đổi được không cần thay đổi chain: InMemory (prototype) → Redis/Postgres (production).
• trim_messages(strategy="last", max_tokens=N) để giới hạn token trước khi gọi LLM (không xóa storage).
• Summary strategy: gọi LLM summarize old messages → SystemMessage ngắn thay thế.
• Multi-user: dùng compound session_id (user_id:thread_id) để tránh cross-contamination.
• Legacy API (ConversationBufferMemory, ConversationChain): deprecated trong 0.3, remove trong 1.0.
• Agent phức tạp hơn → LangGraph (Module 5).

Bài 25: Tools và Toolkits — gắn function vào agent