构建一个动态生成的 AI 选择你自己的冒险游戏
作者:frumu
探索实时、AI 生成叙事和视觉效果背后的技术。
如果冒险游戏的每次游玩体验都截然不同呢?
想象一个“选择你自己的冒险”(CYOA)游戏,其中的故事不仅仅是分支,而是根据**你的**决策实时**创建**,为你量身定制。这不仅仅是选择预先编写好的脚本中的路径 A 或路径 B;这是游戏世界通过人工智能的创造力动态地做出反应和演变,我们称之为 **vibe coding**。
在《硅谷之梦》中,你可以输入任何你想要的操作——你不受限于多项选择。AI 会解释你的自由格式输入,让每次冒险都真正属于你。
我们相信这代表了游戏开发新运动的开始——一个玩家主体性真正驱动叙事,并与 AI 实时共同创作的运动。我们最新的项目探索了这一前沿领域,利用一系列现代技术构建了一个 100% 动态的 AI CYOA 游戏。本文深入探讨了将这一愿景变为现实的技术旅程、挑战和突破,旨在吸引所有对 AI、讲故事和互动娱乐交叉领域着迷的人。
核心技术:基础
构建一个完全动态的 AI 游戏需要一个健壮且可扩展的基础。以下是关键技术及其作用的分解:
- Python: 核心编程语言,因其丰富的库、强大的社区支持以及与 AI 服务和 Web 框架出色的集成能力而被选中。
- Django: 一个高级 Python Web 框架,提供应用程序结构。我们利用其 ORM 进行数据库交互(使用 PostgreSQL),其模板引擎用于渲染用户界面,内置安全功能(如 CSRF 保护),以及用于轻松数据管理的管理界面。
- Celery 和 Redis: 对于异步处理耗时的 AI 生成任务至关重要。Celery 作为分布式任务队列,允许 Web 服务器快速响应用户请求,同时将繁重的工作(对 LLM 和图像生成器的 API 调用)卸载到后台工作进程。Redis 充当快速的内存消息代理。
- Asyncio: 虽然 Celery 处理主要的异步处理,但 Python 的 `asyncio` 库在一些 AI 交互逻辑中(例如 `ai_generators.py` 类)被探索和使用,以在与外部 API 通信时有效地管理并发 I/O 操作,从而在适用情况下防止任务工作器内部的阻塞。
- Cloudflare R2: 提供与 S3 兼容、经济高效的对象存储。所有动态生成的图像都使用 boto3 库上传到这里,R2 直接将其提供给用户,从而减轻应用程序服务器的负载。
AI 核心:动态叙事生成
核心游戏循环围绕着 LLM 生成故事。叙事不再是预先写好的文本,而是根据玩家的选择和不断变化的游戏状态展开。这涉及以下几个步骤:
- 上下文收集: 在调用 LLM 之前,应用程序会收集当前的游戏上下文。这包括玩家属性(生命值、恐惧值等)、物品栏、故事标记(如 `door_unlocked=true`)、最近事件和玩家行动的历史记录,以及玩家的最新选择。
- LLM 交互: 我们主要使用 Google Gemini 系列模型(例如 1.5 Flash)进行叙事生成。精心组装的上下文会发送到 Gemini API。
- 结构化输出解析: LLM 被指示(通过提示)不仅返回叙述文本,还返回结构化数据:2-3 个相关的玩家选择、一个适合图像生成的提示,以及指示游戏状态变化的特定标签(例如,`<STAT_CHANGE stat="hp" change="-1" />`、`<SET_FLAG>found_key=true</SET_FLAG>`、`<CHARACTER id="NPC01" name="神秘陌生人">...</CHARACTER>`)。然后,应用程序代码会解析这些结构化输出,以更新数据库中的游戏状态。
- 与 OpenRouter 的灵活性: 我们集成了 OpenRouter 以提供灵活性。这允许尝试不同的 LLM 进行叙事生成或专门任务(如图像提示优化),而无需锁定在单个提供商,通过统一的 API 访问模型。
挑战:叙事提示工程
最大的挑战之一是 提示工程。为叙事 LLM 制作完美的提示是一种迭代艺术。提示需要指示 AI:
- 根据历史和玩家行动保持严格的叙事连贯性。
- 始终如一地遵循所选的故事主题(例如,奇幻、科幻)。
- 生成相关且引人入胜的选择。
- 准确反映叙事中的玩家统计数据和游戏标志。
- 在适当的时候引入具有独特 ID 和描述的新角色。
- 可靠地输出应用程序可以解析的结构化数据(选择、图像提示、状态更改标签)。
- 正确处理角色引用(在叙事中使用“卫队长”等名称,但在图像提示中使用“GCK01”等特定 ID)。
实现这一点需要进行多次修订,尝试不同的措辞,在提示中提供清晰的示例,并改进 `api_wrappers.py`(后来重构为 `ai_generators.py`)中的解析逻辑,以处理 AI 输出的变化。
例如,从 LLM 的响应中解析结构化标签涉及在 Celery 任务中使用正则表达式:
# game_app/tasks.py (simplified parsing snippet inside the task)
import re
import logging # Added for logger usage example
logger = logging.getLogger(__name__)
# ... inside generate_narrative_and_image_task ...
response_text = """
NARRATIVE:
You cautiously enter the dusty crypt. A faint green glow emanates from a pedestal in the center.
<CHARACTER id=\"GHOUL01\" name=\"Crypt Ghoul\">A shambling ghoul notices you!</CHARACTER>
<STAT_CHANGE stat=\"fear\" change=\"+2\"/>
CHOICES:
1. Attack the ghoul!
2. Sneak past the ghoul towards the pedestal.
3. Flee the crypt!
IMAGE_PROMPT:
Dark crypt interior, glowing green pedestal, shambling crypt ghoul (GHOUL01) lunging, fantasy art style.
<SET_FLAG>entered_crypt=true</SET_FLAG>
""" # Example response text
state_changes = {}
choices = []
narrative = "Error: Could not parse narrative."
image_prompt = "Error state."
# Example: Parsing STAT_CHANGE tags
stat_change_tags = re.findall(
r"<STAT_CHANGE\\s+stat=[\"']([^\"']+)[\"']\\s+change=[\"']([+-]?\\d+)[\"']\\s*/>",
response_text, re.IGNORECASE
)
if stat_change_tags:
stat_updates = {}
for stat, change in stat_change_tags:
try:
stat_updates[stat.lower()] = int(change)
except ValueError:
logger.warning(f"Invalid stat change value '{change}' for '{stat}'")
if stat_updates:
state_changes['stat_updates'] = stat_updates
logger.info(f"Parsed STAT_CHANGE: {stat_updates}")
# Example: Parsing CHOICES block
choices_block_match = re.search(
r"CHOICES:\\s*(.*?)(?:IMAGE_PROMPT:|NARRATIVE:|$)",
response_text, re.IGNORECASE | re.DOTALL
)
if choices_block_match:
choices_text = choices_block_match.group(1).strip()
parsed_choices = re.findall(r"^\\d+\\.\\s*(.*)", choices_text, re.MULTILINE)
choices = [choice.strip() for choice in parsed_choices if choice.strip()]
logger.info(f"Parsed CHOICES: {choices}")
# Example: Parsing NARRATIVE block
narrative_match = re.search(
r"NARRATIVE:\\s*(.*?)(?:CHOICES:|IMAGE_PROMPT:|$)",
response_text, re.IGNORECASE | re.DOTALL
)
if narrative_match:
raw_narrative = narrative_match.group(1).strip()
narrative = re.sub(r"<(?:STAT_CHANGE|SET_FLAG|CHARACTER)[^>]*>", "", raw_narrative, flags=re.IGNORECASE).strip()
logger.info(f"Parsed NARRATIVE (cleaned): {narrative[:100]}...")
# Example: Parsing IMAGE_PROMPT block
image_prompt_match = re.search(
r"IMAGE_PROMPT:\\s*(.*?)(?:NARRATIVE:|CHOICES:|$)",
response_text, re.IGNORECASE | re.DOTALL
)
if image_prompt_match:
raw_image_prompt = image_prompt_match.group(1).strip()
image_prompt = re.sub(r"<(?:STAT_CHANGE|SET_FLAG|CHARACTER)[^>]*>", "", raw_image_prompt, flags=re.IGNORECASE).strip()
logger.info(f"Parsed IMAGE_PROMPT (cleaned): {image_prompt}")
# Similar regex patterns can be used for SET_FLAG, CHARACTER, etc.
# ... rest of the parsing logic to populate state_changes['flags'], state_changes['characters'] ...
冒险可视化:AI 图像生成
动态的故事值得拥有动态的视觉效果。我们实施了一个多阶段的流程来生成独特的插图:
- 叙事到提示生成: 主要 LLM(例如 Gemini)生成的叙事文本通常包含丰富的细节,但并非总是图像模型的理想输入。我们有时会使用一个辅助的、专门的 LLM(例如通过 OpenRouter 访问的 Mistral 模型)来分析叙事并提取关键视觉元素、角色描述(使用其 ID)、场景细节和情绪,从而为图像生成器创建更有效的提示。
- 图像模型交互: 我们利用多种服务来实现灵活性和控制:
- Replicate: 为 Flux.1 Schnell 等强大、快速的模型提供简单的 API 访问,用于标准图像生成。对于我们的浪漫主题,我们利用了令人惊叹的全新 HiDream-I1 模型,它能产生令人惊叹的逼真和富有表现力的结果。
- RunPod 和 ComfyUI: 为了更好地控制图像生成管道,我们使用 RunPod 托管运行 ComfyUI 的持久 GPU 实例。我们的 Django 应用程序与 ComfyUI API 交互,使我们能够执行涉及 SDXL 及其变体等模型的复杂工作流,应用特定的 LoRA,并微调参数。这需要设置 ComfyUI 工作流 JSON 并管理 RunPod 实例。
- 存储: 生成的图像上传到 Cloudflare R2 以实现高效存储和交付。
这个流程确保每张图片都反映了玩家独特旅程中那一刻的特定角色、环境和动作。
响应式、可扩展且用户友好
《硅谷之梦》旨在提供流畅现代的用户体验。界面响应迅速,在桌面和移动设备上都能很好地运行。当你提交一个动作时,你会看到实时反馈和加载指示器,同时 AI 正在生成你的下一个场景——没有页面重载或不舒服的等待。
在幕后,后端是为可扩展性而构建的。得益于 Celery、Redis 和 Docker 等技术,《硅谷之梦》可以同时处理许多玩家,每个玩家都有自己独特的冒险,而不会出现速度减慢或瓶颈。
最大的技术障碍是处理外部 AI API 调用固有的缓慢特性。典型的叙事或图像生成可能需要几秒钟,这对于同步 Web 请求来说太长了。这就是 Celery 变得不可或缺的原因。
问题
如果没有后台任务,用户点击选择会导致 Django 视图挂起,等待 Gemini 和 Replicate/RunPod 响应。这将阻塞 Web 服务器进程,使其无法处理其他用户,并导致超时和糟糕的用户体验。
解决方案 - Celery 任务
我们将核心 AI 逻辑重构为定义在 `game_app/tasks.py` 中的 Celery 任务。当用户做出选择时,Django 视图 (`game_app/views.py`) 现在只做最少的工作:它验证输入,可能更新回合计数器或记录历史,然后立即使用 `task.delay(...)` 将 `generate_narrative_and_image_task` 排入队列。然后,视图立即向用户的浏览器返回一个“处理中”状态。
在 `game_app/views.py` 中将任务入队:
# game_app/views.py (simplified POST handler in game_view)
from .tasks import generate_narrative_and_image_task
from django.http import JsonResponse, HttpRequest, HttpResponse
from django.contrib.auth.decorators import login_required
# from .models import GameState, PlayerStats # Assuming these are imported
import json
import logging # Added for logging
logger = logging.getLogger(__name__)
@login_required
def game_view(request: HttpRequest, game_id: int) -> HttpResponse:
# ... (Assume GET request handling exists here) ...
if request.method == 'POST':
try:
user = request.user
# Fetch game_state and player_stats securely for the logged-in user
# game_state = GameState.objects.get(id=game_id, user=user)
# player_stats = PlayerStats.objects.get(game_state=game_state) # Example fetch
# Placeholder data for demonstration
game_state = type('obj', (object,), {'id': game_id, 'save': lambda: None})()
player_stats = type('obj', (object,), {'game_state_id': game_id})()
current_context = {'turn': 1, 'history': []} # Example context
data = json.loads(request.body)
player_choice = data.get('choice', 'look around') # Default choice
# --- Prepare context ---
# Example: Add choice to history
current_context['history'].append(f"Turn {current_context['turn']}: Player chose '{player_choice}'")
# Example: Increment turn (might happen in task or here)
current_context['turn'] += 1
# --- Record history entry (if applicable before task) ---
# History.objects.create(game_state=game_state, turn=current_context['turn']-1, action=player_choice)
# Ensure game state is saved before enqueuing if needed
game_state.save() # Saves the current state before task starts
logger.info(f"Enqueuing task for game {game_id}, turn {current_context['turn']}")
generate_narrative_and_image_task.delay( # Enqueue the task!
game_id=game_state.id,
user_id=user.id,
player_choice=player_choice,
current_context=current_context, # Pass the updated context
player_stats_id=player_stats.game_state_id # Pass ID for task to fetch
)
# Return immediate response
return JsonResponse({
'status': 'processing',
'message': 'Generating next step...',
'game_id': game_state.id
})
except json.JSONDecodeError:
logger.error("Invalid JSON received in POST request.")
return JsonResponse({'status': 'error', 'error': 'Invalid request format.'}, status=400)
# except GameState.DoesNotExist:
# logger.error(f"Game state {game_id} not found for user {user.id}")
# return JsonResponse({'status': 'error', 'error': 'Game not found.'}, status=404)
except Exception as e:
logger.exception(f"Error in game_view POST for game {game_id}: {e}")
return JsonResponse({'status': 'error', 'error': f'Internal server error.'}, status=500)
else:
# Handle other methods or return error
return JsonResponse({'status': 'error', 'error': 'Method not allowed'}, status=405)
用户反馈 - 轮询
为了在后台任务完成时通知用户,我们实现了一个轮询机制。前端 JavaScript (`static/js/game.js`) 在收到“处理中”状态后,开始定期调用一个专门的状态端点 (`/game/status/<game_id>/`)。此端点检查数据库中 `GameState` 记录的当前状态(Celery 任务在完成后会更新此记录)。
`static/js/game.js` 中的前端轮询逻辑:
// static/js/game.js (simplified polling logic)
let pollingIntervalId = null;
const POLLING_INTERVAL_MS = 3000; // Poll every 3 seconds
let currentGamId = null; // Store the current game ID globally or pass it around
// --- UI Update Functions (Placeholders) ---
function updateGameDisplay(data) {
console.log("Updating display with data:", data);
// Example: Update narrative text, image, choices, stats
// document.getElementById('narrative-text').textContent = data.narrative;
// document.getElementById('game-image').src = data.image_url;
// updateChoices(data.choices);
// updateStats(data.stats);
}
function displayGameOver(data) {
console.log("Game Over:", data);
// Example: Show game over message
// document.getElementById('game-area').innerHTML = `<h2>Game Over!</h2><p>${data.narrative}</p>`;
}
function displayError(errorMessage) {
console.error("Error:", errorMessage);
// Example: Show error message to the user
// document.getElementById('error-message').textContent = `Error: ${errorMessage}`;
// document.getElementById('error-message').style.display = 'block';
}
function showLoading() {
console.log("Showing loading indicator...");
// Example: Display a spinner or loading text
// document.getElementById('loading-indicator').style.display = 'block';
// disableChoiceButtons();
}
function hideLoading() {
console.log("Hiding loading indicator...");
// Example: Hide spinner or loading text
// document.getElementById('loading-indicator').style.display = 'none';
// enableChoiceButtons();
}
// --- End UI Update Functions ---
function startPolling(gameId) {
if (!gameId) {
console.error("Cannot start polling without gameId.");
return;
}
currentGamId = gameId; // Store the game ID
stopPolling(); // Clear any existing interval first
console.log(`Starting polling for game ${currentGamId}`);
showLoading(); // Show loading indicator immediately
pollingIntervalId = setInterval(() => {
pollGameStatus(currentGamId);
}, POLLING_INTERVAL_MS);
// Initial poll after a short delay to get status quickly
setTimeout(() => pollGameStatus(currentGamId), 1500);
}
function stopPolling() {
if (pollingIntervalId) {
console.log(`Stopping polling for game ${currentGamId}`);
clearInterval(pollingIntervalId);
pollingIntervalId = null;
}
// Don't hide loading here, wait for final status update
}
async function pollGameStatus(gameId) {
if (!gameId) {
console.warn("pollGameStatus called without gameId.");
stopPolling();
return;
}
console.log(`Polling status for game ${gameId}...`);
const statusUrl = `/game/status/${gameId}/`; // Ensure URL is correct
try {
const response = await fetch(statusUrl, {
method: 'GET',
headers: {
'Accept': 'application/json',
// Add CSRF token header if needed for GET status endpoint, though often not required
}
});
if (!response.ok) {
console.error(`Polling error: ${response.status} ${response.statusText}`);
displayError(`Failed to get game status (${response.status}). Please try refreshing.`);
stopPolling();
hideLoading();
return;
}
const data = await response.json();
console.log("Poll response:", data);
// Handle different statuses
switch (data.status) {
case 'processing':
// Still processing, continue polling. Optionally update loading text.
console.log("Status: Still processing...");
// document.getElementById('loading-indicator').textContent = 'Still generating...';
break;
case 'ready':
stopPolling();
updateGameDisplay(data); // Update UI with final data
hideLoading();
break;
case 'game_over':
stopPolling();
displayGameOver(data);
hideLoading();
break;
case 'error':
stopPolling();
displayError(data.message || data.error || 'An unknown error occurred.');
hideLoading();
break;
default:
// Unknown status
console.warn(`Unknown game status received: ${data.status}`);
stopPolling();
displayError(`Received unknown game status: ${data.status}`);
hideLoading();
break;
}
} catch (error) {
console.error("Error during fetch/polling:", error);
displayError('Network error while checking game status.');
stopPolling();
hideLoading();
}
}
// Example of how to initiate polling after submitting a choice
// Assume submitChoice() is an async function handling the POST request
async function submitChoice(choiceValue) {
// ... (code to make the POST request to game_view) ...
// const response = await fetch(...)
// const result = await response.json();
// if (result.status === 'processing' && result.game_id) {
// startPolling(result.game_id);
// } else if (result.status === 'error') {
// displayError(result.error);
// }
}
更新用户界面
一旦轮询端点返回“准备就绪”状态(或“游戏结束”或“错误”),JavaScript 就会停止轮询,并使用随附的数据(新叙事、图像 URL、选择、统计数据)动态更新用户界面,而无需完全重新加载页面。
开发洞察:研究与 AI 辅助
构建像这样复杂的项目需要大量的研究和复杂的编码。我们在整个开发过程中大量依赖 AI 工具:
- Gemini 2.5 Pro 深度研究: 广泛用于探索不同的 AI 模型、API、架构模式(如异步任务队列)以及将生成式 AI 集成到 Web 应用程序中的最佳实践。
- Roo Coder(由 Gemini 2.5 Pro 提供支持): 我们的 AI 编码助手 Roo 在编写、调试和重构代码方面发挥了关键作用,特别是与各种 AI API 交互、解析响应以及管理不同组件(Django 视图、Celery 任务、JavaScript)之间游戏状态的复杂逻辑。
应用程序架构图
以下图表说明了用户交互的流程,突出显示了异步处理:
注意: 以下图表以 SVG 格式渲染,以实现最大兼容性。
(如果您更喜欢编辑或查看图表源文件,请访问 MermaidChart.com。)
独特性:没有两款游戏是相同的
《硅谷之梦》的每一次游戏体验都是独一无二的,这得益于动态叙事和图像生成。玩家自主性、AI 驱动的故事情节和程序化视觉效果的结合意味着没有两场冒险是完全相同的。
感谢您的阅读!体验这款真正动态的 AI 冒险游戏:https://silicon.frumu.ai/