Mở Đầu: Khi NPC Trở Nên Thông Minh Hơn
Sau 3 năm triển khai AI cho các studio game tại Trung Quốc và Đông Nam Á, tôi đã chứng kiến sự chuyển đổi từ NPC với kịch bản cứng nhắc sang những nhân vật có thể nhớ, phản hồi và thích nghi theo ngữ cảnh thực. Bài viết này là bản tổng hợp những gì tôi đã học được từ việc xây dựng hệ thống ký ức đa cuộc hội thoại cho NPC game — một trong những thách thức kỹ thuật thú vị nhất mà tôi từng đối mặt.
Nếu bạn đang tìm kiếm một giải pháp API AI với độ trễ dưới 50ms, chi phí tiết kiệm đến 85% so với các nhà cung cấp phương Tây, và hỗ trợ thanh toán qua WeChat/Alipay, hãy đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.
Tại Sao Hệ Thống Ký Ức Quan Trọng?
Trong một game RPG open-world hiện đại, người chơi có thể tương tác với hàng trăm NPC qua nhiều phiên chơi khác nhau. Không ai muốn NPC hỏi "Ngươi là ai?" mỗi lần gặp mặt. Hệ thống ký ức đa cuộc hội thoại giải quyết bài toán này bằng cách:
- Duy trì ngữ cảnh: NPC nhớ những gì đã xảy ra trước đó
- Tạo personality chain: Mỗi NPC có cá tính riêng được xây dựng qua thời gian
- Context-aware responses: Phản hồi phù hợp với lịch sử tương tác
- Emotional continuity: Trạng thái cảm xúc được duy trì qua các cuộc trò chuyện
Kiến Trúc Hệ Thống Tổng Quan
"""
Hệ thống Ký Ức Đa Cuộc Hội Thoại cho Game NPC
Kiến trúc 3 tầng với Memory Management
Tầng 1: Working Memory (Short-term) - Ngữ cảnh cuộc hội thoại hiện tại
Tầng 2: Episodic Memory (Medium-term) - Các sự kiện gần đây
Tầng 3: Semantic Memory (Long-term) - Kiến thức và cá tính NPC
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum
from datetime import datetime, timedelta
import json
import asyncio
from collections import deque
import hashlib
class MemoryTier(Enum):
WORKING = "working" # 5-10 phút gần nhất
EPISODIC = "episodic" # 1-24 giờ gần đây
SEMANTIC = "semantic" # Toàn bộ lịch sử
@dataclass
class MemoryEntry:
content: str
timestamp: datetime
tier: MemoryTier
importance: float # 0.0 - 1.0
speaker: str # "player" hoặc "npc"
emotional_tags: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict:
return {
"content": self.content,
"timestamp": self.timestamp.isoformat(),
"tier": self.tier.value,
"importance": self.importance,
"speaker": self.speaker,
"emotional_tags": self.emotional_tags,
"metadata": self.metadata
}
class NPCConsciousness:
"""
Lớp quản lý ý thức NPC - Tích hợp với HolySheep AI API
"""
def __init__(
self,
npc_id: str,
personality: Dict[str, Any],
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
):
self.npc_id = npc_id
self.personality = personality
self.base_url = base_url
self.api_key = api_key
# Ba tầng ký ức
self.working_memory: deque = deque(maxlen=50) # ~10 phút
self.episodic_memory: List[MemoryEntry] = []
self.semantic_memory: Dict[str, Any] = {
"identity": personality.get("identity", {}),
"knowledge": personality.get("knowledge", []),
"relationships": {},
"beliefs": personality.get("beliefs", [])
}
# Trạng thái cảm xúc hiện tại
self.emotional_state = {
"mood": "neutral",
"trust_level": 0.5,
"affection": 0.0,
"memory_of_player": {}
}
# Cấu hình memory pruning
self.episodic_retention_days = 7
self.importance_threshold = 0.3
def add_memory(
self,
content: str,
speaker: str,
emotional_tags: List[str] = None,
importance: float = 0.5
) -> MemoryEntry:
"""Thêm ký ức mới vào hệ thống"""
# Xác định tầng ký ức dựa trên tầm quan trọng
if importance >= 0.7:
tier = MemoryTier.SEMANTIC
elif importance >= 0.4:
tier = MemoryTier.EPISODIC
else:
tier = MemoryTier.WORKING
entry = MemoryEntry(
content=content,
timestamp=datetime.now(),
tier=tier,
importance=importance,
speaker=speaker,
emotional_tags=emotional_tags or [],
metadata={"npc_id": self.npc_id}
)
# Thêm vào tầng phù hợp
if tier == MemoryTier.WORKING:
self.working_memory.append(entry)
elif tier == MemoryTier.EPISODIC:
self.episodic_memory.append(entry)
else:
self._consolidate_to_semantic(entry)
return entry
def _consolidate_to_semantic(self, entry: MemoryEntry):
"""Hợp nhất ký ức quan trọng vào semantic memory"""
if entry.speaker == "player":
player_id = entry.metadata.get("player_id", "unknown")
if player_id not in self.semantic_memory["relationships"]:
self.semantic_memory["relationships"][player_id] = []
self.semantic_memory["relationships"][player_id].append({
"content": entry.content,
"timestamp": entry.timestamp.isoformat(),
"importance": entry.importance
})
def get_context_window(self, max_tokens: int = 4000) -> str:
"""Tạo cửa sổ ngữ cảnh cho LLM - tối ưu cho token limit"""
context_parts = []
remaining_tokens = max_tokens
# 1. Semantic memory (cố định - luôn bao gồm)
semantic_context = self._format_semantic_memory()
context_parts.append(semantic_context)
remaining_tokens -= self._estimate_tokens(semantic_context)
# 2. Emotional state
emotional_context = f"[Trạng thái cảm xúc hiện tại: {self.emotional_state['mood']}, "
emotional_context += f"Mức tin tưởng: {self.emotional_state['trust_level']:.0%}]"
context_parts.append(emotional_context)
remaining_tokens -= self._estimate_tokens(emotional_context)
# 3. Episodic memory (ưu tiên cao)
for entry in reversed(self.episodic_memory[-10:]):
entry_text = self._format_entry(entry)
if self._estimate_tokens(entry_text) <= remaining_tokens:
context_parts.append(entry_text)
remaining_tokens -= self._estimate_tokens(entry_text)
# 4. Working memory (ngữ cảnh gần nhất)
for entry in reversed(self.working_memory):
entry_text = self._format_entry(entry)
if self._estimate_tokens(entry_text) <= remaining_tokens:
context_parts.append(entry_text)
remaining_tokens -= self._estimate_tokens(entry_text)
return "\n".join(reversed(context_parts))
def _format_semantic_memory(self) -> str:
"""Format semantic memory thành prompt"""
return f"""[NHẬN DẠNG NPC]
Tên: {self.semantic_memory['identity'].get('name', 'Unknown')}
Cá tính: {self.semantic_memory['identity'].get('personality', 'Normal')}
Quan điểm: {', '.join(self.semantic_memory.get('beliefs', []))}
Kiến thức: {', '.join(self.semantic_memory.get('knowledge', []))}"""
def _format_entry(self, entry: MemoryEntry) -> str:
"""Format memory entry thành text"""
emotion_str = f" [{', '.join(entry.emotional_tags)}]" if entry.emotional_tags else ""
return f"[{entry.tier.value.upper()}] {entry.speaker}: {entry.content}{emotion_str}"
def _estimate_tokens(self, text: str) -> int:
"""Ước tính token - giả định 1 token ≈ 4 ký tự cho tiếng Việt/Trung"""
return len(text) // 3
async def generate_response(
self,
player_input: str,
game_context: Dict[str, Any] = None
) -> str:
"""Gọi HolySheep AI API để tạo phản hồi NPC"""
import aiohttp
context_window = self.get_context_window()
system_prompt = f"""Bạn là một NPC trong game với hệ thống ký ức đa cuộc hội thoại.
{context_window}
Nguyên tắc phản hồi:
1. Tham khảo ký ức liên quan khi trả lời
2. Duy trì tính cách và giọng nói nhất quán
3. Nếu nhắc đến sự kiện trong quá khứ, hãy thể hiện sự nhớ
4. Điều chỉnh thái độ dựa trên mức tin tưởng hiện tại"""
if game_context:
system_prompt += f"\n\n[Bối cảnh game hiện tại]\n{game_context}"
async with aiohttp.ClientSession() as session:
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": player_input}
],
"temperature": 0.7,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as response:
if response.status == 200:
data = await response.json()
return data["choices"][0]["message"]["content"]
else:
error = await response.text()
raise Exception(f"API Error: {response.status} - {error}")
def prune_old_memories(self):
"""Dọn dẹp ký ức cũ - chạy định kỳ"""
cutoff = datetime.now() - timedelta(days=self.episodic_retention_days)
self.episodic_memory = [
e for e in self.episodic_memory
if e.timestamp > cutoff or e.importance >= self.importance_threshold
]
Quản Lý Đồng Thời Với Session Pool
Trong một game MMO với hàng nghìn người chơi online cùng lúc, việc quản lý session cho NPC là thách thức thực sự. Tôi đã phát triển một session pool với connection pooling và rate limiting để đảm bảo hiệu suất.
"""
Session Pool cho Multi-NPC Conversation System
Hỗ trợ hàng nghìn concurrent conversations
"""
import asyncio
from typing import Dict, Optional, List
from dataclasses import dataclass
import time
from collections import defaultdict
import threading
@dataclass
class ConversationSession:
session_id: str
npc_id: str
player_id: str
consciousness: NPCConsciousness
created_at: float
last_access: float
message_count: int = 0
total_tokens: int = 0
class NPCSessionPool:
"""
Pool quản lý tất cả NPC sessions với:
- Connection pooling
- Rate limiting per NPC
- Automatic session cleanup
- Token budget management
"""
def __init__(
self,
api_key: str,
max_concurrent_per_npc: int = 50,
session_timeout: int = 1800, # 30 phút
max_tokens_per_npc_per_minute: int = 50000
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Session storage
self._sessions: Dict[str, ConversationSession] = {}
self._npc_sessions: Dict[str, set] = defaultdict(set)
self._player_sessions: Dict[str, set] = defaultdict(set)
# Rate limiting
self._npc_tokens: Dict[str, List[float]] = defaultdict(list)
self.max_concurrent_per_npc = max_concurrent_per_npc
self.session_timeout = session_timeout
self.max_tokens_per_npc_per_minute = max_tokens_per_npc_per_minute
# Lock cho thread safety
self._lock = threading.RLock()
# Personality templates
self._personality_templates = self._load_templates()
def _load_templates(self) -> Dict:
return {
"merchant": {
"identity": {"name": "Thương Nhân", "personality": "friendly, greedy"},
"knowledge": ["物品价值", "交易技巧", "稀有道具"],
"beliefs": ["Tiền là vua", "Giao dịch công bằng là chuyện nhỏ"]
},
"guard": {
"identity": {"name": "Vệ Sĩ", "personality": "serious, protective"},
"knowledge": ["守卫技巧", "危险识别", "法律规则"],
"beliefs": ["Trật tự là trên hết", "Bảo vệ người vô tội"]
},
"quest_giver": {
"identity": {"name": "Người Giao Nhiệm Vụ", "personality": "wise, mysterious"},
"knowledge": ["古代传说", "任务信息", "地图知识"],
"beliefs": ["Mỗi anh hùng đều có số phận"]
}
}
def get_or_create_session(
self,
session_id: str,
npc_id: str,
player_id: str,
personality_type: str = "merchant"
) -> ConversationSession:
"""Lấy session hiện có hoặc tạo mới"""
with self._lock:
# Kiểm tra session tồn tại
if session_id in self._sessions:
session = self._sessions[session_id]
session.last_access = time.time()
return session
# Kiểm tra giới hạn concurrent
if len(self._npc_sessions[npc_id]) >= self.max_concurrent_per_npc:
self._evict_oldest_session(npc_id)
# Tạo session mới
consciousness = NPCConsciousness(
npc_id=npc_id,
personality=self._personality_templates.get(
personality_type,
self._personality_templates["merchant"]
),
api_key=self.api_key
)
session = ConversationSession(
session_id=session_id,
npc_id=npc_id,
player_id=player_id,
consciousness=consciousness,
created_at=time.time(),
last_access=time.time()
)
self._sessions[session_id] = session
self._npc_sessions[npc_id].add(session_id)
self._player_sessions[player_id].add(session_id)
return session
def _evict_oldest_session(self, npc_id: str):
"""Xóa session cũ nhất của NPC để nhường chỗ"""
oldest = None
oldest_time = float('inf')
for sid in self._npc_sessions[npc_id]:
session = self._sessions.get(sid)
if session and session.last_access < oldest_time:
oldest = sid
oldest_time = session.last_access
if oldest:
self.close_session(oldest)
def close_session(self, session_id: str):
"""Đóng và dọn dẹp session"""
with self._lock:
if session_id in self._sessions:
session = self._sessions[session_id]
self._npc_sessions[session.npc_id].discard(session_id)
self._player_sessions[session.player_id].discard(session_id)
del self._sessions[session_id]
async def process_message(
self,
session_id: str,
player_message: str,
game_context: Dict = None
) -> Dict[str, any]:
"""Xử lý tin nhắn của người chơi - có rate limiting"""
session = self.get_or_create_session(
session_id=session_id,
npc_id=session_id.split("_")[0],
player_id=session_id.split("_")[1]
)
# Rate limiting check
if not self._check_rate_limit(session.npc_id):
return {
"success": False,
"error": "Rate limit exceeded",
"retry_after": 60
}
# Đánh giá tầm quan trọng của tin nhắn
importance = self._assess_importance(player_message)
# Thêm vào memory
session.consciousness.add_memory(
content=player_message,
speaker="player",
importance=