Vấn đề thực tế: Khi Context bị tràn bộ nhớ
Tuần trước, một khách hàng của tôi gặp lỗi nghiêm trọng khi triển khai chatbot hỗ trợ khách hàng bằng Claude API. Họ gọi API liên tục trong một phiên trò chuyện dài, và khi cuộc hội thoại đến turn thứ 47, server trả về lỗi kinh hoàng:
anthropic.RateLimitError: error - type: "rate_limit_error"
message: "Context window exceeded. Maximum tokens: 200000"
code: 422
http_status: 422
retryable: false
Toàn bộ session bị mất. 47 câu hỏi của khách hàng, tất cả context quan trọng, đều tan thành mây khói. Tôi đã phải ngồi lại 3 tiếng để phân tích và viết lại toàn bộ hệ thống context management từ đầu.
Bài viết này là tổng hợp kinh nghiệm thực chiến của tôi trong việc quản lý multi-turn conversation với Claude API, giúp bạn tránh những sai lầm mà tôi đã mất thời gian học hỏi.
Tại sao Context Management lại quan trọng?
Claude có giới hạn context window. Với các model mới nhất, con số này là 200K tokens, nhưng nếu bạn không quản lý tốt, bạn sẽ gặp vấn đề nghiêm trọng:
- Token budget cạn kiệt giữa chừng cuộc hội thoại
- Chi phí API tăng vọt do gửi lặp lại lịch sử
- Response quality giảm khi context quá dài và nhiễu
- Timeout và connection errors khi payload quá lớn
Với tỷ giá hiện tại qua
HolySheheep AI, Claude Sonnet 4.5 có giá $15/1M tokens. Mỗi lần gửi full conversation history không cần thiết, bạn đang đốt tiền một cách lãng phí.
Chiến lược 1: Sliding Window Context
Đây là chiến lược phổ biến nhất và hiệu quả nhất. Thay vì gửi toàn bộ lịch sử, bạn chỉ giữ lại N messages gần nhất.
import anthropic
import os
class SlidingWindowContext:
def __init__(self, max_messages=20, max_tokens=180000):
self.client = anthropic.Anthropic(
api_key=os.environ.get("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
self.messages = []
self.max_messages = max_messages
self.max_tokens = max_tokens
def add_message(self, role, content):
self.messages.append({
"role": role,
"content": content
})
self._trim_context()
def _trim_context(self):
if len(self.messages) > self.max_messages:
self.messages = self.messages[-self.max_messages:]
def chat(self, user_input):
self.add_message("user", user_input)
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=self.messages
)
assistant_response = response.content[0].text
self.add_message("assistant", assistant_response)
return assistant_response
Sử dụng
context = SlidingWindowContext(max_messages=20)
response = context.chat("Xin chào, tôi cần hỗ trợ về API")
print(response)
Ưu điểm: Đơn giản, dễ implement, kiểm soát được token usage.
Nhược điểm: Có thể mất context quan trọng nếu conversation cần refer đến thông tin cũ.
Chiến lược 2: Semantic Chunking
Với các use case phức tạp hơn, sliding window đơn giản không đủ. Tôi đã phát triển một hệ thống semantic chunking cho phép Claude hiểu "chủ đề" của từng phần conversation.
from dataclasses import dataclass
from typing import List, Optional
import anthropic
import os
@dataclass
class ConversationChunk:
topic: str
messages: List[dict]
summary: str
importance: int # 1-5
class SemanticContextManager:
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.chunks: List[ConversationChunk] = []
self.current_chunk: Optional[ConversationChunk] = None
def _summarize_chunk(self, messages: List[dict]) -> str:
combined = "\n".join([f"{m['role']}: {m['content']}" for m in messages])
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[{
"role": "user",
"content": f"Tóm tắt ngắn gọn cuộc hội thoại sau trong 2-3 câu:\n{combined}"
}]
)
return response.content[0].text
def add_message(self, role: str, content: str, topic: Optional[str] = None):
if topic and self.current_chunk and self.current_chunk.topic != topic:
self._finalize_current_chunk()
self.current_chunk = ConversationChunk(
topic=topic,
messages=[],
summary="",
importance=3
)
elif not self.current_chunk:
self.current_chunk = ConversationChunk(
topic=topic or "General",
messages=[],
summary="",
importance=3
)
self.current_chunk.messages.append({"role": role, "content": content})
if len(self.current_chunk.messages) >= 10:
self._finalize_current_chunk()
def _finalize_current_chunk(self):
if self.current_chunk and self.current_chunk.messages:
self.current_chunk.summary = self._summarize_chunk(
self.current_chunk.messages
)
self.chunks.append(self.current_chunk)
self.current_chunk = None
def build_context_for_api(self) -> List[dict]:
result = []
# Thêm summaries của các chunk cũ
for chunk in self.chunks[-3:]:
result.append({
"role": "system",
"content": f"[Context: {chunk.topic}] {chunk.summary}"
})
# Thêm messages của current chunk
if
Tài nguyên liên quan
Bài viết liên quan