Trong thế giới AI application ngày nay, việc xây dựng chatbot thông minh không chỉ đơn giản là kết nối LLM API. Điều thực sự quan trọng là làm sao để mô hình có thể "nhớ" được ngữ cảnh của cuộc hội thoại trước đó. Bài viết này sẽ đưa bạn đi sâu vào ConversationBufferMemory — một trong những công cụ quản lý bộ nhớ phổ biến nhất của LangChain, kèm theo case study thực tế từ một startup AI tại Hà Nội đã tiết kiệm $3,520 mỗi tháng sau khi tối ưu hóa memory management.

Case Study: Startup AI Việt Nam Giảm 85% Chi Phí API

Bối cảnh kinh doanh: Một startup AI tại Hà Nội chuyên cung cấp giải pháp chatbot chăm sóc khách hàng cho các doanh nghiệp TMĐT đã gặp vấn đề nghiêm trọng với chi phí API. Hệ thống của họ xử lý khoảng 50,000 cuộc hội thoại mỗi ngày, mỗi cuộc hội thoại trung bình 15-20 turn messages.

Điểm đau của nhà cung cấp cũ: Sử dụng OpenAI với chi phí $4,200/tháng, nhưng gặp phải:

Giải pháp HolySheep AI: Sau khi di chuyển sang HolySheep AI, startup này đã:

Kết quả sau 30 ngày:

ConversationBufferMemory Là Gì?

ConversationBufferMemory là một memory class trong LangChain cho phép lưu trữ toàn bộ lịch sử hội thoại dưới dạng một chuỗi messages. Mỗi khi có một message mới được thêm vào, memory sẽ append vào buffer và truyền nguyên vẹn vào context của LLM.

Cơ Chế Hoạt Động

Memory hoạt động theo nguyên tắc FIFO (First-In-First-Out) hoặc có thể được configure để:

Cài Đặt và Sử Dụng Cơ Bản

1. Khởi Tạo LangChain Chain Với HolySheep AI

# Cài đặt thư viện cần thiết
!pip install langchain langchain-community langchain-openai --quiet

import os
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate


CẤU HÌNH HOLYSHEEP AI - KHÔNG DÙNG api.openai.com

os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"


Khởi tạo Chat Model với HolySheep


Giá: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok, Gemini 2.5 Flash $2.50/MTok

llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"],
    request_timeout=30,
    max_retries=3
)


Khởi tạo ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="history",
    return_messages=True,
    output_key="response"
)


Tạo Conversation Chain

template = """Bạn là trợ lý AI thông minh. 
Lịch sử hội thoại:
{history}
Người dùng: {input}
Trợ lý:"""

prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=template
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    prompt=prompt,
    verbose=True
)


Test conversation

response = conversation.predict(input="Xin chào, tôi tên là Minh")
print(response)

Output: Xin chào Minh! Rất vui được gặp bạn. Tôi có thể giúp gì cho bạn hôm nay?


response = conversation.predict(input="Tôi đang làm việc với dữ liệu")
print(response)

Output: Chào bạn Minh! Việc làm việc với dữ liệu nghe rất thú vị. Bạn đang làm việc với loại dữ liệu nào?

Điểm mấu chốt ở đây là base_url phải là https://api.holysheep.ai/v1 — đây là endpoint chính thức của HolySheep AI, không phải api.openai.com. Điều này giúp bạn:

Memory Management Nâng Cao

2. Auto-Truncating Memory Với Token Limit

from langchain.memory import ConversationBufferWindowMemory
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

class SmartTruncatingMemory:
    """
    Memory thông minh tự động truncate khi vượt ngưỡng token
    Giải quyết vấn đề: Context window overflow và escalating token costs
    """
    
    def __init__(self, max_tokens=4000, preserve_recent=4):
        self.max_tokens = max_tokens
        self.preserve_recent = preserve_recent  # Giữ lại N messages gần nhất
        self.messages = []
        self.estimated_token_per_message = 4  # Rough estimate
        
    def add_user_message(self, message):
        self.messages.append(HumanMessage(content=message))
        self._auto_truncate()
        
    def add_ai_message(self, message):
        self.messages.append(AIMessage(content=message))
        self._auto_truncate()
        
    def _auto_truncate(self):
        """Tự động cắt bớt messages cũ nếu vượt ngưỡng"""
        estimated_tokens = sum(
            len(m.content.split()) * self.estimated_token_per_message 
            for m in self.messages
        )
        
        while estimated_tokens > self.max_tokens and len(self.messages) > self.preserve_recent * 2:
            # Xóa message cũ nhất (1 user + 1 AI = 1 turn)
            if self.messages:
                self.messages.pop(0)
            if self.messages:
                self.messages.pop(0)
                
            # Recalculate
            estimated_tokens = sum(
                len(m.content.split()) * self.estimated_token_per_message 
                for m in self.messages
            )
            
    def get_messages(self):
        return self.messages
        
    def clear(self):
        self.messages = []
        
    def get_token_estimate(self):
        return sum(
            len(m.content.split()) * self.estimated_token_per_message 
            for m in self.messages
        )


Sử dụng Smart Memory

smart_memory = SmartTruncatingMemory(max_tokens=4000, preserve_recent=4)


Thêm messages

smart_memory.add_user_message("Tôi muốn tạo một chatbot")
smart_memory.add_ai_message("OK, tôi sẽ giúp bạn. Bạn muốn chatbot cho mục đích gì?")
smart_memory.add_user_message("Chatbot chăm sóc khách hàng")
smart_memory.add_ai_message("Hay! Chatbot CSKH thường cần: FAQ tự động, tích hợp CRM, và escalation logic.")
smart_memory.add_user_message("Tích hợp với hệ thống ERP được không?")
smart_memory.add_ai_message("Hoàn toàn có thể. ERP phổ biến như SAP, Oracle, hoặc giải pháp cloud như Odoo đều có API.")

print(f"Số messages hiện tại: {len(smart_memory.messages)}")
print(f"Estimated tokens: {smart_memory.get_token_estimate()}")

Kết quả: Nếu vượt 4000 tokens, messages cũ sẽ tự động bị truncate

3. Persistent Memory Với Database Storage

import json
import sqlite3
from datetime import datetime
from typing import Optional, List
from langchain.memory import BaseMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage

class SQLiteChatMemory(BaseMemory):
    """
    Memory lưu trữ lịch sử hội thoại vào SQLite database
    Phù hợp cho production systems cần persistence và scalability
    """
    
    def __init__(self, db_path: str, session_id: str, max_history: int = 50):
        self.db_path = db_path
        self.session_id = session_id
        self.max_history = max_history
        self._init_db()
        
    def _init_db(self):
        """Khởi tạo database schema"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS chat_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                metadata TEXT
            )
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_session_timestamp 
            ON chat_history(session_id, timestamp)
        """)
        conn.commit()
        conn.close()
        
    @property
    def memory_variables(self) -> List[str]:
        return ["chat_history"]
    
    def load_memory_variables(self, inputs: dict) -> dict:
        """Load lịch sử từ database"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT role, content FROM chat_history 
            WHERE session_id = ? 
            ORDER BY timestamp DESC 
            LIMIT ?
        """, (self.session_id, self.max_history))
        
        rows = cursor.fetchall()
        conn.close()
        
        # Reverse để có thứ tự chronological
        messages = []
        for role, content in reversed(rows):
            if role == "human":
                messages.append(HumanMessage(content=content))
            else:
                messages.append(AIMessage(content=content))
                
        return {"chat_history": messages}
    
    def save_context(self, inputs: dict, outputs: dict) -> None:
        """Lưu context mới vào database"""
        human_msg = inputs.get("input", "")
        ai_msg = outputs.get("response", "")
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Xóa messages cũ nếu vượt max_history
        cursor.execute("""
            DELETE FROM chat_history 
            WHERE session_id = ? 
            AND id NOT IN (
                SELECT id FROM chat_history 
                WHERE session_id = ? 
                ORDER BY timestamp DESC 
                LIMIT ?
            )
        """, (self.session_id, self.session_id, self.max_history * 2))
        
        # Insert messages mới
        cursor.execute(
            "INSERT INTO chat_history (session_id, role, content) VALUES (?, ?, ?)",
            (self.session_id, "human", human_msg)
        )
        cursor.execute(
            "INSERT INTO chat_history (session_id, role, content) VALUES (?, ?, ?)",
            (self.session_id, "ai", ai_msg)
        )
        
        conn.commit()
        conn.close()
        
    def clear(self) -> None:
        """Xóa toàn bộ lịch sử của session"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "DELETE FROM chat_history WHERE session_id = ?",
            (self.session_id,)
        )
        conn.commit()
        conn.close()


Sử dụng với HolySheep AI

db_memory = SQLiteChatMemory(
    db_path="./chat_history.db",
    session_id="user_12345",
    max_history=50
)


Kết hợp với LangChain

conversation_with_db = ConversationChain(
    llm=llm,
    memory=db_memory,
    verbose=False
)


Hội thoại sẽ được tự động lưu vào SQLite

response = conversation_with_db.predict(
    input="Tôi thích màu xanh dương và không thích màu đỏ"
)

Message này được lưu vào database


response = conversation_with_db.predict(
    input="Gợi ý cho tôi 3 sản phẩm phù hợp"
)

AI sẽ nhớ sở thích màu sắc từ message trước

Chiến Lược Tối Ưu Hóa Chi Phí

4. Dynamic Model Selection Theo Conversation Length

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
import os

class CostOptimizedConversationManager:
    """
    Quản lý hội thoại thông minh với dynamic model selection
    - Short conversation: Gemini 2.5 Flash $2.50/MTok (nhanh, rẻ)
    - Long conversation: DeepSeek V3.2 $0.42/MTok (rẻ nhất)
    - Complex tasks: GPT-4.1 $8/MTok (mạnh nhất)
    """
    
    # Cấu hình HolySheep API
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    API_BASE = "https://api.holysheep.ai/v1"
    
    # Model pricing (2026)
    MODELS = {
        "fast": {"name": "gpt-4.1-mini", "price_per_mtok": 2.50},
        "cheap": {"name": "deepseek-v3.2", "price_per_mtok": 0.42},
        "powerful": {"name": "gpt-4.1", "price_per_mtok": 8.00}
    }
    
    # Thresholds
    SHORT_CONVO_TOKENS = 1000
    MEDIUM_CONVO_TOKENS = 4000
    
    def __init__(self):
        self.conversations = {}  # session_id -> {memory, model, stats}
        os.environ["OPENAI_API_KEY"] = self.API_KEY
        os.environ["OPENAI_API_BASE"] = self.API_BASE
        
    def _create_llm(self, model_name: str) -> ChatOpenAI:
        return ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=self.API_KEY,
            base_url=self.API_BASE,
            request_timeout=30
        )
    
    def _estimate_tokens(self, messages: list) -> int:
        """Ước tính tokens từ messages"""
        total_chars = sum(len(m.content) for m in messages)
        return int(total_chars / 4)  # Rough estimate
        
    def get_or_create_conversation(self, session_id: str) -> dict:
        if session_id not in self.conversations:
            memory = ConversationBufferMemory(
                memory_key="history",
                return_messages=True
            )
            
            # Mặc định dùng model cheap cho new conversations
            llm = self._create_llm(self.MODELS["cheap"]["name"])
            
            self.conversations[session_id] = {
                "memory": memory,
                "llm": llm,
                "current_model": "cheap",
                "stats": {
                    "total_tokens": 0,
                    "total_cost": 0.0,
                    "message_count": 0
                }
            }
        return self.conversations[session_id]
    
    def chat(self, session_id: str, user_input: str) -> tuple[str, dict]:
        """Thực hiện chat với smart model selection"""
        conv_data = self.get_or_create_conversation(session_id)
        memory = conv_data["memory"]
        stats = conv_data["stats"]
        
        # Check memory size để quyết định model
        messages = memory.chat_memory.messages
        estimated_tokens = self._estimate_tokens(messages)
        
        # Auto-upgrade/downgrade model dựa trên token count
        if estimated_tokens > self.MEDIUM_CONVO_TOKENS:
            # Long conversation -> switch to cheapest model
            new_model = "cheap"
        elif estimated_tokens > self.SHORT_CONVO_TOKENS:
            # Medium -> use fast model
            new_model = "fast"
        else:
            new_model = "cheap"
            
        if conv_data["current_model"] != new_model:
            print(f"🔄 Switching model: {conv_data['current_model']} -> {new_model}")
            conv_data["llm"] = self._create_llm(self.MODELS[new_model]["name"])
            conv_data["current_model"] = new_model
            
        # Create chain và predict
        prompt = PromptTemplate(
            input_variables=["history", "input"],
            template="Lịch sử: {history}\nNgười dùng: {input}\nTrợ lý:"
        )
        
        chain = ConversationChain(
            llm=conv_data["llm"],
            memory=memory,
            prompt=prompt
        )
        
        response = chain.predict(input=user_input)
        
        # Update stats
        stats["message_count"] += 1
        input_tokens = self._estimate_tokens([user_input])
        output_tokens = self._estimate_tokens([response])
        total_tokens = input_tokens + output_tokens
        stats["total_tokens"] += total_tokens
        
        model_price = self.MODELS[new_model]["price_per_mtok"]
        cost = (total_tokens / 1_000_000) * model_price
        stats["total_cost"] += cost
        
        return response, stats
    
    def get_cost_summary(self, session_id: str) -> dict:
        """Lấy tổng kết chi phí cho session"""
        if session_id in self.conversations:
            return self.conversations[session_id]["stats"]
        return {}


Demo usage

manager = CostOptimizedConversationManager()


Simulate conversation

session = "customer_001"
responses = []
for i, msg in enumerate([
    "Chào bạn, tôi cần hỗ trợ về sản phẩm",
    "Tôi muốn biết về chính sách đổi trả",
    "Sản phẩm mua được 15 ngày có đổi được không?",
    "Cảm ơn, tôi sẽ liên hệ lại sau"
]):
    response, stats = manager.chat(session, msg)
    responses.append(response)
    print(f"Q{i+1}: {msg[:30]}...")
    print(f"   Stats: {stats['total_tokens']} tokens, ${stats['total_cost']:.4f}\n")

print(f"\n📊 Final Summary: {manager.get_cost_summary(session)}")

Lỗi Thường Gặp Và Cách Khắc Phục

Lỗi 1: Memory Leak Khi Không Clear Context

Mô tả lỗi: Memory buffer không được clear định kỳ, dẫn đến:

Nguyên nhân: ConversationBufferMemory mặc định lưu trữ tất cả messages mà không có giới hạn. Khi ứng dụng chạy liên tục trong nhiều ngày, memory sẽ phình to.

# ❌ CODE SAI - Gây memory leak
memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=memory)


Loop chạy liên tục - memory sẽ phình to không giới hạn

while True:
    user_input = get_user_input()
    response = conversation.predict(input=user_input)
    # BUG: Memory không bao giờ được clear!
    send_to_user(response)


✅ CODE ĐÚNG - Có auto-truncation

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    k=10,  # Chỉ giữ lại 10 messages gần nhất
    return_messages=True,
    ai_prefix="Assistant",
    human_prefix="Human"
)


Hoặc implement custom auto-cleanup

class MemoryWithAutoCleanup:
    def __init__(self, max_messages=20, max_idle_seconds=1800):
        self.memory = ConversationBufferMemory()
        self.max_messages = max_messages
        self.max_idle_seconds = max_idle_seconds
        self.last_cleanup = time.time()
        
    def cleanup_if_needed(self):
        current_time = time.time()
        messages = self.memory.chat_memory.messages
        
        # Cleanup nếu vượt ngưỡng messages HOẶC idle quá lâu
        if len(messages) > self.max_messages or \
           (current_time - self.last_cleanup) > self.max_idle_seconds:
            print(f"🧹 Cleaning up memory: {len(messages)} -> {self.max_messages}")
            # Xóa nửa số messages cũ
            if len(messages) > self.max_messages:
                for _ in range(len(messages) // 2):
                    self.memory.chat_memory.messages.pop(0)
                if len(self.memory.chat_memory.messages) > 1:
                    self.memory.chat_memory.messages.pop(0)
            self.last_cleanup = current_time

Lỗi 2: Sai API Endpoint (Dùng api.openai.com Thay Vì HolySheep)

Mô tả lỗi: Code sử dụng sai base_url, dẫn đến:

# ❌ CODE SAI - Dùng OpenAI endpoint (không được phép!)
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1"  # LỖI NGHIÊM TRỌNG
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key="sk-..."  # OpenAI key - tốn chi phí cao
)


❌ CODE SAI - Generic base_url cũng không tốt

os.environ["OPENAI_API_BASE"] = "https://api.anthropic.com"


✅ CODE ĐÚNG - Sử dụng HolySheep AI endpoint

import os


LUÔN LUÔN sử dụng https://api.holysheep.ai/v1

os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep key
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"  # ✅ ĐÚNG

llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"],  # ✅ HolySheep endpoint
    request_timeout=30,
    max_retries=3
)


Verify connection

try:
    response = llm.invoke("Test connection")
    print(f"✅ Connected to HolySheep AI successfully")
except Exception as e:
    print(f"❌ Connection failed: {e}")

Lỗi 3: Token Count Không Chính Xác Gây Context Overflow

Mô tả lỗi: Ứng dụng crash với lỗi context length khi tổng tokens vượt giới hạn của model. Đặc biệt nguy hiểm khi:

# ❌ CODE SAI - Không kiểm tra token count
memory = ConversationBufferMemory()
chain = ConversationChain(llm=llm, memory=memory)


Thêm messages không giới hạn - sẽ crash khi vượt context

for i in range(100):
    chain.predict(input=f"Message number {i}")


✅ CODE ĐÚNG - Pre-checking token count

from langchain_core.messages import HumanMessage, AIMessage

class TokenAwareConversationBufferMemory:
    """
    Memory có kiểm tra token count trước khi thêm message
    """
    
    # Context limits phổ biến (2026)
    MODEL_CONTEXTS = {
        "gpt-4.1": 128000,
        "gpt-4.1-mini": 128000,
        "gpt-4o": 128000,
        "gpt-4o-mini": 128000,
        "claude-sonnet-4.5": 200000,
        "gemini-2.5-flash": 1000000,
        "deepseek-v3.2": 64000
    }
    
    def __init__(self, model_name: str = "gpt-4.1", max_context_ratio: float = 0.9):
        self.messages = []
        self.model_name = model_name
        self.max_context = self.MODEL_CONTEXTS.get(model_name, 128000)
        self.max_tokens = int(self.max_context * max_context_ratio)  # 90% của limit
        
    def _count_tokens(self, text: str) -> int:
        """Đếm tokens bằng tiktoken hoặc ước lượng"""
        try:
            import tiktoken
            enc = tiktoken.get_encoding("cl100k_base")
            return len(enc.encode(text))
        except:
            # Fallback: rough estimate
            return len(text) // 4
            
    def _count_all_tokens(self) -> int:
        """Đếm tổng tokens của tất cả messages"""
        total = 0
        for msg in self.messages:
            # Cộng thêm overhead cho message format
            total += self._count_tokens(msg.content) + 10
        return total
        
    def add_user_message(self, text: str) -> bool:
        """Thêm user message với token check"""
        message_tokens = self._count_tokens(text)
        current_tokens = self._count_all_tokens()
        
        # Check nếu thêm message này sẽ vượt limit
        if current_tokens + message_tokens > self.max_tokens:
            print(f"⚠️ Would exceed token limit: {current_tokens + message_tokens} > {self.max_tokens}")
            return False
            
        self.messages.append(HumanMessage(content=text))
        self._auto_truncate_if_needed()
        return True
        
    def _auto_truncate_if_needed(self):
        """Tự động truncate nếu cần thiết"""
        while self._count_all_tokens() > self.max_tokens and len(self.messages) > 2:
            # Xóa oldest message pair
            self.messages.pop(0)
            if self.messages:
                self.messages.pop(0)
            print(f"🗑️ Auto-truncated, remaining: {len(self.messages)} messages")
            
    def get_history(self) -> list:
        return self.messages


Sử dụng với token protection

memory = TokenAwareConversationBufferMemory(model_name="gpt-4.1")
print(f"Model: gpt-4.1, Max tokens: {memory.max_tokens:,}")


Safe to add messages with automatic protection

for i in range(50):
    success = memory.add_user_message(f"Đây là message số {i} với nội dung dài để test")
    if not success:
        print(f"Cannot add message {i} - would exceed limit")
        break
    print(f"✅ Added message {i}: {memory._count_all_tokens()} tokens")

Lỗi 4: Memory Not Serializable Trong Multi-Threaded Environment

Mô tả lỗi: Khi sử dụng memory trong ứng dụng web (FastAPI, Flask) hoặc async environment:

# ❌ CODE SAI - Shared mutable state trong async environment
from fastapi import FastAPI
from langchain.memory import ConversationBufferMemory

app = FastAPI()
shared_memory = ConversationBufferMemory()  # ❌ Shared state - race condition!

@app.post("/chat")
async def chat(message: str):
    shared_memory.chat_memory.add_user_message(message)  # ❌ Race condition
    response = conversation.predict(input=message)
    shared_memory.chat_memory.add_ai_message(response)
    return {"response": response}


✅ CODE ĐÚNG - Session-based memory với async-safe storage

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from typing import Dict
from datetime import datetime, timedelta

class AsyncSessionMemory:
    """Memory manager cho async web application"""
    
    def __init__(self, session_timeout_minutes: int = 30):
        self._sessions: Dict[str, list] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._last_access: Dict[str, datetime] = {}
        self._timeout = timedelta(minutes=session_timeout_minutes)
        self._cleanup_lock = asyncio.Lock()
        
    async def get_memory(self,

Tài nguyên liên quan
Bài viết liên quan