Trong thời đại AI bùng nổ, việc xây dựng một engine kể chuyện tương tác (interactive narrative engine) không còn là đặc quyền của các tập đoàn công nghệ lớn. Bài viết này sẽ hướng dẫn bạn từng bước triển khai hệ thống tương tự AI Dungeon, tích hợp với HolySheep AI — nền tảng API LLM với độ trễ dưới 50ms và chi phí thấp nhất thị trường.
Nghiên cứu điển hình: Startup Game AI tại TP.HCM
Một studio game indie tại TP.HCM chuyên phát triển game nhập vai tiếng Việt đã gặp khó khăn nghiêm trọng khi vận hành engine kể chuyện trên nền tảng cũ. Hệ thống sử dụng API của nhà cung cấp quốc tế với độ trễ trung bình 850ms cho mỗi lượt tương tác — quá chậm khiến người chơi mất hứng thú. Bên cạnh đó, chi phí API hàng tháng lên đến $4,200 cho 2 triệu token, trong khi ngân sách marketing chỉ còn $800.
Sau khi chuyển sang HolySheep AI, đội ngũ đã giảm độ trễ xuống còn 180ms (giảm 79%) và tiết kiệm chi phí API xuống chỉ còn $680/tháng (giảm 84%). Điều này cho phép họ mở rộng hệ thống từ 500 lượt chơi đồng thời lên 5,000 mà không cần tăng ngân sách.
Kiến trúc hệ thống AI Dungeon Engine
Tổng quan kiến trúc
Hệ thống AI Dungeon bao gồm các thành phần chính: Context Manager (quản lý lịch sử hội thoại), Story Engine (xử lý sinh nội dung), State Tracker (theo dõi trạng thái thế giới), và Action Resolver (xử lý hành động người chơi). Dưới đây là kiến trúc mô-đun hoá dễ mở rộng.
# Cấu trúc thư mục dự án
ai-dungeon-engine/
├── src/
│ ├── __init__.py
│ ├── config.py # Cấu hình HolySheep API
│ ├── story_engine.py # Engine sinh nội dung
│ ├── context_manager.py # Quản lý ngữ cảnh
│ ├── state_tracker.py # Theo dõi trạng thái thế giới
│ ├── action_resolver.py # Xử lý hành động
│ └── models.py # Định nghĩa data models
├── tests/
├── requirements.txt
└── main.py # Entry point
requirements.txt
openai>=1.12.0
pydantic>=2.5.0
redis>=5.0.0
fasthtml>=0.1.0Cấu hình kết nối HolySheep AI
Việc cấu hình đúng endpoint và authentication là bước nền tảng quan trọng nhất. HolySheep AI cung cấp endpoint tương thích OpenAI format, giúp việc di chuyển trở nên vô cùng đơn giản.
# src/config.py
import os
from dataclasses import dataclass
@dataclass
class HolySheepConfig:
"""Cấu hình kết nối HolySheep AI - Độ trễ <50ms, giá thấp nhất"""
# Endpoint chuẩn hoá - KHÔNG sử dụng api.openai.com
base_url: str = "https://api.holysheep.ai/v1"
# API Key từ HolySheep Dashboard
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Model mapping theo use case
# DeepSeek V3.2: $0.42/MTok - Tối ưu chi phí cho generation
# Gemini 2.5 Flash: $2.50/MTok - Cân bằng speed/quality
# GPT-4.1: $8/MTok - Chất lượng cao nhất
STORY_MODEL: str = "deepseek-v3.2" # Sinh nội dung chính
ACTION_MODEL: str = "gemini-2.5-flash" # Xử lý hành động nhanh
PLANNING_MODEL: str = "gpt-4.1" # Lập kế hoạch dài hạn
# Timeout và retry settings
request_timeout: int = 30
max_retries: int = 3
retry_delay: float = 1.0
Singleton instance
config = HolySheepConfig()# src/story_engine.py
import httpx
from typing import List, Dict, Optional, AsyncIterator
from .config import config
class HolySheepClient:
"""Client tương thích OpenAI cho HolySheep AI"""
def __init__(self):
self.base_url = config.base_url
self.api_key = config.api_key
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=config.request_timeout
)
async def generate_stream(
self,
messages: List[Dict],
model: str,
temperature: float = 0.8,
max_tokens: int = 500
) -> AsyncIterator[str]:
"""Streaming generation - phản hồi từng chunk cho UX mượt mà"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": True
}
async with self._client.stream("POST", "/chat/completions", json=payload) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
# Parse SSE format
chunk = json.loads(data)
if chunk.get("choices")[0].get("delta", {}).get("content"):
yield chunk["choices"][0]["delta"]["content"]
import json
class StoryEngine:
"""Engine sinh nội dung narrative - trái tim của AI Dungeon"""
SYSTEM_PROMPT = """Bạn là một game master chuyên nghiệp trong game nhập vai.
Nhiệm vụ của bạn:
1. Mô tả cảnh quan, nhân vật, sự kiện một cách sống động
2. Tạo tình huống bất ngờ nhưng logic với cốt truyện
3. Không quá 200 từ cho mỗi đoạn văn
4. Luôn kết thúc bằng lựa chọn cho người chơi
5. Ngôn ngữ: tiếng Việt, giọng văn hấp dẫn"""
def __init__(self):
self.client = HolySheepClient()
self.conversation_history: List[Dict] = []
async def generate_story_segment(
self,
player_action: str,
world_state: Dict
) -> str:
"""Sinh đoạn kể chuyện tiếp theo dựa trên hành động người chơi"""
# Build context với world state
context_msg = f"Trạng thái thế giới hiện tại: {world_state}"
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "system", "content": context_msg},
*self.conversation_history[-10:], # Giữ 10 message gần nhất
{"role": "user", "content": player_action}
]
# Streaming response - chunks được yield liên tục
response_chunks = []
async for chunk in self.client.generate_stream(
messages=messages,
model=config.STORY_MODEL,
temperature=0.8
):
response_chunks.append(chunk)
yield chunk
full_response = "".join(response_chunks)
self.conversation_history.append(
{"role": "user", "content": player_action},
{"role": "assistant", "content": full_response}
)
return full_response
Ví dụ sử dụng trong main.py
async def main():
engine = StoryEngine()
world_state = {
"location": "Rừng sâu với ánh trăng le lói",
"health": 85,
"inventory": ["kiếm cũ", "bình thuốc"],
"enemy_nearby": " sói đói"
}
print("🎮 Chào mừng đến với AI Dungeon!")
print("=" * 50)
while True:
player_input = input("\n👤 Hành động của bạn: ")
if player_input.lower() in ["exit", "quit"]:
break
print("\n📖 Đang viết...", end="", flush=True)
story = ""
async for chunk in engine.generate_story_segment(player_input, world_state):
print(chunk, end="", flush=True)
story += chunk
print("\n" + "-" * 50)Context Manager — Quản lý ngữ cảnh thông minh
Một trong những thách thức lớn nhất của AI Dungeon là duy trì continuity (tính liên tục) của câu chuyện. Context Manager giải quyết bằng cách tóm tắt lịch sử và theo dõi các sự kiện quan trọng.
# src/context_manager.py
from typing import List, Dict, Optional
from dataclasses import dataclass, field
import tiktoken
@dataclass
class ConversationTurn:
"""Một lượt hội thoại"""
player_action: str
narrative: str
timestamp: float
importance: float = 1.0 # Độ quan trọng (0-1)
entities_mentioned: List[str] = field(default_factory=list)
class ContextManager:
"""Quản lý ngữ cảnh với chiến lược tóm tắt thông minh"""
MAX_TOKENS = 128000 # Giới hạn context window
SUMMARY_TRIGGER = 90000 # Trigger summarization khi gần đạt limit
def __init__(self, encoding_model: str = "cl100k_base"):
self.encoding = tiktoken.get_encoding(encoding_model)
self.turns: List[ConversationTurn] = []
self.current_summary: Optional[str] = None
self.key_events: List[Dict] = [] # Sự kiện quan trọng cần nhớ
def add_turn(self, player_action: str, narrative: str, timestamp: float):
"""Thêm một lượt hội thoại mới"""
turn = ConversationTurn(
player_action=player_action,
narrative=narrative,
timestamp=timestamp
)
self.turns.append(turn)
self._update_importance()
def _update_importance(self):
"""Tính độ quan trọng dựa trên từ khoá"""
important_keywords = [
"chiến đấu", "chết", "yêu", "ghét", "phát hiện",
"bí mật", "kho báu", "phép thuật", "nguy hiểm",
"quyết định", "giao kèo", "biến mất"
]
for turn in self.turns:
text = (turn.player_action + turn.narrative).lower()
turn.importance = sum(1 for kw in important_keywords if kw in text) / len(important_keywords)
async def summarize_old_context(self) -> str:
"""Tóm tắt các lượt cũ để tiết kiệm context"""
if not self.turns:
return ""
# Lấy các lượt ít quan trọng để tóm tắt
low_importance_turns = [
t for t in self.turns
if t.importance < 0.3
]
if not low_importance_turns:
return ""
# Gọi HolySheep để tạo summary
old_content = "\n".join([
f"Player: {t.player_action}\nNarrator: {t.narrative}"
for t in low_importance_turns
])
summary_prompt = f"""Tóm tắt đoạn hội thoại sau thành 3-5 câu,
giữ lại thông tin quan trọng về nhân vật, địa điểm, và sự kiện:
{old_content}"""
# Sử dụng Gemini Flash cho summarization nhanh
async with HolySheepClient() as client:
summary = ""
async for chunk in client.generate_stream(
messages=[{"role": "user", "content": summary_prompt}],
model="gemini-2.5-flash"
):
summary += chunk
self.current_summary = summary
return summary
def get_relevant_context(self, current_location: str) -> str:
"""Lấy ngữ cảnh liên quan cho prompt"""
parts = []
if self.current_summary:
parts.append(f"Tổng hợp sự kiện trước: {self.current_summary}")
# Lấy các sự kiện quan trọng gần đây
important_turns = sorted(
self.turns[-20:], # 20 lượt gần nhất
key=lambda t: (t.importance, t.timestamp),
reverse=True
)[:5]
if important_turns:
parts.append("Các sự kiện quan trọng gần đây:")
for turn in important_turns:
parts.append(f"- {turn.narrative[:100]}...")
# Key events liên quan đến location
relevant_events = [
e for e in self.key_events
if e.get("location") == current_location
]
if relevant_events:
parts.append(f"Sự kiện tại {current_location}:")
for event in relevant_events[-3:]:
parts.append(f"- {event['description']}")
return "\n".join(parts) if parts else "Không có ngữ cảnh trước đó."
def count_tokens(self, text: str) -> int:
"""Đếm số tokens trong text"""
return len(self.encoding.encode(text))
Integration với Story Engine
class AdvancedStoryEngine(StoryEngine):
"""Story Engine với Context Management nâng cao"""
def __init__(self):
super().__init__()
self.context = ContextManager()
async def generate_with_context(
self,
player_action: str,
world_state: Dict
) -> str:
"""Generation với context thông minh"""
# Kiểm tra và tóm tắt nếu cần
total_tokens = self.context.count_tokens(
str(self.context.turns)
)
if total_tokens > self.context.SUMMARY_TRIGGER:
await self.context.summarize_old_context()
# Build messages với context
context_str = self.context.get_relevant_context(
world_state.get("location", "unknown")
)
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "system", "content": f"Ngữ cảnh: {context_str}"},
{"role": "system", "content": f"Trạng thái: {world_state}"},
*self.conversation_history[-6:],
{"role": "user", "content": player_action}
]
response = ""
async for chunk in self.client.generate_stream(
messages=messages,
model=config.STORY_MODEL
):
response += chunk
yield chunk
# Cập nhật context
self.context.add_turn(player_action, response, time.time())
self.conversation_history.extend([
{"role": "user", "content": player_action},
{"role": "assistant", "content": response}
])State Tracker — Theo dõi trạng thái thế giới
Để narrative engine hoạt động nhất quán, hệ thống cần theo dõi trạng thái thế giới (world state) — bao gồm vị trí, vật phẩm, quan hệ NPC, và các biến số quan trọng. Dưới đây là implementation chi tiết.
# src/state_tracker.py
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from enum import Enum
import json
class EntityType(Enum):
PLAYER = "player"
NPC = "npc"
ITEM = "item"
LOCATION = "location"
EVENT = "event"
@dataclass
class Entity:
"""Đại diện cho một thực thể trong thế giới game"""
id: str
type: EntityType
name: str
properties: Dict[str, Any] = field(default_factory=dict)
relationships: Dict[str, List[str]] = field(default_factory=dict) # type -> [entity_ids]
@dataclass
class WorldState:
"""Trạng thái toàn bộ thế giới game"""
player: Dict[str, Any]
current_location: str
locations: Dict[str, Dict] = field(default_factory=dict)
entities: Dict[str, Entity] = field(default_factory=dict)
story_flags: Dict[str, bool] = field(default_factory=dict)
inventory: List[str] = field(default_factory=list)
quest_log: List[Dict] = field(default_factory=dict)
def to_dict(self) -> Dict:
return asdict(self)
def summary(self) -> str:
"""Tạo summary ngắn gọn cho prompt"""
return f"""
Vị trí: {self.current_location}
Mạng: {self.player.get('health', 100)}/100
Vật phẩm: {', '.join(self.inventory[:5])}
Nhiệm vụ: {[q['title'] for q in self.quest_log[:2]]}
"""
class StateTracker:
"""Theo dõi và cập nhật trạng thái thế giới"""
def __init__(self):
self.state = WorldState(
Tài nguyên liên quan
Bài viết liên quan