前段时间分享了一款 AI 实时叙事 galgame《CITIZEN / ZERO》的第一篇开发日志收到了不少朋友的私信都在问实时叙事到底是怎么实现的、AI 响应和剧情分支怎么衔接、以及一个准大三学生是怎么把大模型塞进 galgame 引擎里的。这篇开发日志第二篇我来详细拆解一下当前版本的核心实现思路与工程架构。内容包括实时叙事系统的整体流程设计AI 生成内容与 galgame 演出框架的衔接方式角色状态管理与长期记忆的实现性能优化、token 消耗控制、流式输出的落地当前踩过的坑和排查思路本文适合对 AI 游戏开发、LLM 应用落地、视觉小说引擎二次开发感兴趣的开发者阅读。即使你没有做过 galgame只要你有 Python 或 TypeScript 基础也能从中获取一套可以复用的 AI 实时叙事方案。1. AI 实时叙事 galgame 到底是什么1.1 传统 galgame 的叙事方式传统 galgame视觉小说的叙事是预编写 分支树结构。剧本家在开发阶段把每条线路、每个选项、每个结局全部写好玩家在关键节点做出选择游戏引擎根据选择跳转到对应分支。这种方式的优点是剧情质量可控文笔可以反复打磨。演出与文案严格绑定情感节奏容易设计。技术实现简单本质是一个带状态机的内容播放器。缺点是玩家的自由度是预设的无法跳出剧本作者的框架。重复游玩时新鲜感下降。制作成本高一条完整线路需要数万句文案支撑。1.2 引入 AI 之后的实时叙事《CITIZEN / ZERO》要做的是把“剧本编写”这件事从开发阶段移到玩家游玩过程中。游戏不再读取写死的对话文件而是根据当前剧情状态、角色设定、玩家历史行为和实时输入动态生成下一段剧情文本。实时叙事 galgame 的核心特征剧情不是完全预写的而是由 AI 模型在运行时生成。玩家的输入不仅触发分支还直接影响 AI 的角色扮演策略。角色拥有“记忆”能够跨对话引用之前发生的事情。演出系统立绘、背景、BGM、特效响应 AI 生成的情感标签和场景标签。这样做的最大价值是每个玩家的体验都是独特的。同一个角色在不同玩家的档里会形成不同的性格倾向和关系状态。1.3 与 AI Agent 游戏的区别需要说明的是实时叙事 galgame 不完全等同于 AI Agent 游戏。AI Agent 游戏通常让 AI 控制角色在沙盒环境中自主行动比如让 AI 角色自己决定去哪、做什么然后由游戏系统反馈结果。而《CITIZEN / ZERO》的实时叙事把重点放在**“对话 情感 剧情推进”**上。AI 不直接操作游戏世界而是负责生成符合角色人设和剧情状态的文本内容游戏系统再把这些内容转译为可视化演出。2. 整体技术架构与核心流程2.1 系统组成当前版本的技术栈如下游戏客户端Web 技术栈TypeScript Phaser 3 Vue 3叙事引擎Python 编写的事件驱动服务AI 模型层支持接入本地部署的大模型与云端 API通过统一接口调用存储层SQLite 存储剧情节点与玩家行为记录JSON 文件存储角色卡设定通信协议WebSocket HTTP架构上遵循“前后端分离”原则客户端只负责渲染与输入采集所有叙事决策都在服务端完成。这样做的原因会在后面展开。2.2 实时叙事流程一次完整的实时叙事交互流程如下玩家在客户端输入对话或选择动作。客户端把玩家输入、当前场景 ID、当前角色状态打包发送到叙事引擎。叙事引擎从“叙事上下文管理器”中获取该场景的历史对话摘要与角色状态。引擎构建提示词调用大模型生成一段响应文本。模型输出除了对白外还包含情感标签、场景切换指令、角色心理状态变化。叙事引擎解析模型输出更新角色状态数据库生成演出指令。客户端收到文本与演出指令驱动立绘表情、背景、特效、BGM 完成演出。这里最关键的一环是第 5 步模型输出不能只是纯文本必须带有结构化控制信息。否则游戏引擎不知道当前角色是什么表情、是否要切换背景、剧情是否推进。3. 环境准备与项目结构3.1 开发环境说明下面是本文示例使用的基础环境版本需要根据你的项目实际情况调整操作系统Windows 11 / Ubuntu 22.04 Node.js18客户端构建 Python3.10叙事引擎 数据库SQLite 3 模型服务本地 OpenAI 兼容接口 / 云端大模型 API之所以使用 OpenAI 兼容接口是因为目前大多数本地推理框架如 Xinference、Ollama 等和云端模型服务都提供该协议接入成本最低。你完全可以把模型层替换成任意支持该协议的模型服务。3.2 项目结构citizen-zero/ ├── client/ # 游戏客户端Web │ ├── src/ │ │ ├── scenes/ # Phaser 游戏场景 │ │ ├── components/ # Vue UI 组件 │ │ ├── services/ # 网络通信服务 │ │ └── types/ # 类型定义 │ └── package.json ├── server/ # Python 叙事引擎 │ ├── app/ │ │ ├── main.py # FastAPI 入口 │ │ ├── narrative.py # 叙事核心逻辑 │ │ ├── prompt_builder.py# 提示词构建 │ │ ├── game_state.py # 游戏状态管理 │ │ ├── memory.py # 长期记忆模块 │ │ ├── model_client.py # 大模型统一客户端 │ │ └── parsers.py # 模型输出解析 │ ├── data/ │ │ ├── characters/ # 角色卡 JSON │ │ └── scenes/ # 场景配置 │ └── requirements.txt └── README.md4. 核心模块设计与代码实现4.1 角色卡设计角色卡是 AI 扮演角色的“人格基础”。它不像传统的 galgame 人设文档只给美术参考而是直接作为提示词的一部分输入模型。CITIZEN / ZERO中的角色卡设计如下{ id: mira, name: 米拉, age: 19, personality: 冷静理性但在熟悉的人面前会流露出温柔一面, speaking_style: 简洁少用语气词偏向科技感表达, background: 就职于城市数据分析中心负责追踪异常信号..., relationships: { player: 30, unknown_entity: -20 }, secrets: [曾参与过一项未公开的实验...] }这里需要特别注意relationships字段。它记录了角色对玩家和其他重要对象的好感度。这个值不是摆设而是会被写入提示词的动态部分直接影响 AI 生成文本时的情感倾向。4.2 提示词构建器提示词构建是实时叙事中最核心的部分。一个好的提示词需要把角色设定、当前场景、历史摘要、玩家输入、输出格式要求全部组织起来。下面是一个简化但可用的提示词模板# 文件路径server/app/prompt_builder.py system_prompt 你是一位视觉小说剧本家负责扮演角色 {character_name}。 ## 角色设定 {character_card} ## 当前场景 {scene_description} ## 场景内历史摘要 {history_summary} ## 情感状态 {emotional_state} ## 输出要求 请根据以上信息生成角色对玩家输入的一段回应。 你必须输出严格的 JSON 格式字段如下 {{ dialogue: 角色说的话不超过 80 字, emotion: happy|sad|angry|surprised|neutral|anxious, action: 角色动作描述用于演出如她低头看着终端屏幕, scene_change: 无变化时为空字符串需要切换场景时填写目标场景ID, relationship_delta: {{ target: player, value: 1 }} }} ## 注意事项 1. dialogue 必须符合角色设定不能跳出人设。 2. 尽量延续当前场景的情感基调。 3. 不要替玩家做决定不要描述玩家的动作。 def build_prompt(character_card, scene, history, player_input): user_content f玩家说{player_input} return system_prompt.format( character_namecharacter_card[name], character_cardcharacter_card, scene_descriptionscene, history_summaryhistory, emotional_statehistory.get(emotional_state, 平静) ), user_content4.3 模型客户端封装为了兼容本地模型和云端 API这里封装了一个统一客户端。底层使用 HTTP 请求访问 OpenAI 兼容接口。# 文件路径server/app/model_client.py import httpx import json class ModelClient: def __init__(self, base_url, api_key, model_name): base_url: 模型服务地址例如 http://localhost:8000/v1 api_key: 认证密钥本地模型可以填空字符串 model_name: 模型名称例如 qwen2.5-7b-instruct self.base_url base_url.rstrip(/) self.api_key api_key self.model_name model_name async def chat(self, messages, temperature0.8, max_tokens512): url f{self.base_url}/chat/completions headers {Authorization: fBearer {self.api_key}} payload { model: self.model_name, messages: messages, temperature: temperature, max_tokens: max_tokens, stream: False, } async with httpx.AsyncClient(timeout60) as client: resp await client.post(url, jsonpayload, headersheaders) resp.raise_for_status() data resp.json() return data[choices][0][message][content] async def chat_stream(self, messages, temperature0.8): 流式版本用于生成大段剧情文本时逐字返回降低玩家等待感。 url f{self.base_url}/chat/completions headers {Authorization: fBearer {self.api_key}} payload { model: self.model_name, messages: messages, temperature: temperature, stream: True, } async with httpx.AsyncClient(timeout60) as client: async with client.stream(POST, url, jsonpayload, headersheaders) as resp: async for line in resp.aiter_lines(): if not line.startswith(data:): continue data_str line[5:].strip() if data_str [DONE]: break try: chunk json.loads(data_str) delta chunk[choices][0].get(delta, {}) content delta.get(content, ) if content: yield content except json.JSONDecodeError: continue4.4 模型输出解析与容错模型输出 JSON 是一个理想状态实际运行中模型经常输出多余的说明文字、Markdown 代码块或者直接把 JSON 截断。因此解析层必须足够健壮。我的做法是优先尝试直接解析 JSON失败后进入清洗流程。# 文件路径server/app/parsers.py import json import re def extract_json(raw_text: str) - dict: 尝试从模型输出中提取 JSON 对象。 策略先去 Markdown 代码块标记再截取第一个 { 到最后一个 }。 # 去掉 json 和 包裹 cleaned re.sub(r(?:json)?, , raw_text).strip() try: return json.loads(cleaned) except json.JSONDecodeError: pass # 截取大括号区间 start cleaned.find({) end cleaned.rfind(}) if start ! -1 and end ! -1 and end start: try: return json.loads(cleaned[start:end 1]) except json.JSONDecodeError: pass # 降级方案日志记录并返回默认结构 print(f[WARN] 模型输出解析失败原始内容: {raw_text}) return { dialogue: raw_text[:80], emotion: neutral, action: , scene_change: , relationship_delta: {target: player, value: 0}, }这里关键的工程思想是降级策略。实时叙事不能因为一次解析失败就中断游戏流程所以当解析失败时直接把原始文本当作对白展示表情保持默认好感度不变化。这样玩家体验不会中断问题只会在后台日志中记录。4.5 游戏状态管理器游戏状态管理器负责维护当前剧情节点的全部动态信息当前场景 ID角色好感度剧情进度关键标记临时事件状态# 文件路径server/app/game_state.py import sqlite3 import json from datetime import datetime class GameStateManager: def __init__(self, db_pathcitizen_zero.db): self.db_path db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS game_state ( player_id TEXT PRIMARY KEY, scene_id TEXT NOT NULL, character_relations TEXT NOT NULL, flags TEXT NOT NULL, history TEXT NOT NULL, updated_at TEXT NOT NULL ) ) conn.execute( CREATE TABLE IF NOT EXISTS chat_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, player_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, metadata TEXT, created_at TEXT NOT NULL ) ) def load_state(self, player_id: str) - dict: with sqlite3.connect(self.db_path) as conn: row conn.execute( SELECT scene_id, character_relations, flags, history FROM game_state WHERE player_id ?, (player_id,) ).fetchone() if row is None: return self._create_default_state(player_id) return { scene_id: row[0], character_relations: json.loads(row[1]), flags: json.loads(row[2]), history: json.loads(row[3]), } def save_state(self, player_id: str, state: dict): with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO game_state (player_id, scene_id, character_relations, flags, history, updated_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(player_id) DO UPDATE SET scene_id excluded.scene_id, character_relations excluded.character_relations, flags excluded.flags, history excluded.history, updated_at excluded.updated_at , ( player_id, state[scene_id], json.dumps(state[character_relations], ensure_asciiFalse), json.dumps(state[flags], ensure_asciiFalse), json.dumps(state[history], ensure_asciiFalse), datetime.utcnow().isoformat(), ) )4.6 长期记忆与历史摘要LLM 的上下文窗口是有限的。如果每一轮对话都把全部历史塞进提示词很快就会撑爆上下文窗口而且模型对过长的上下文关注度会下降。当前方案采用“滚动摘要 关键事件持久化”策略每次对话结束后把本轮对话追加到临时缓冲区。当缓冲区达到一定长度调用模型生成一段历史摘要。摘要与关键事件标记一起存入数据库。后续请求只携带摘要而非完整历史。# 文件路径server/app/memory.py class MemoryManager: def __init__(self, model_client, max_buffer10): self.model_client model_client self.max_buffer max_buffer self.buffer [] async def add_message(self, role, content): self.buffer.append({role: role, content: content}) if len(self.buffer) self.max_buffer: await self.compress() async def compress(self): 调用模型对缓冲区内容生成摘要。 这是一个异步压缩操作避免阻塞主叙事流程。 messages [ {role: system, content: 你将一段对话历史压缩为 150 字以内的剧情摘要保留关键事件与情感转折。}, {role: user, content: json.dumps(self.buffer, ensure_asciiFalse)} ] summary await self.model_client.chat(messages, temperature0.3, max_tokens256) self.buffer [{role: system, content: f历史摘要{summary}}] return summary4.7 叙事引擎主流程把所有模块串联起来的是叙事核心逻辑。下面是一个简化版的事件循环# 文件路径server/app/narrative.py from fastapi import WebSocket import json class NarrativeEngine: def __init__(self, state_manager, memory_manager, prompt_builder, model_client): self.state_manager state_manager self.memory_manager memory_manager self.prompt_builder prompt_builder self.model_client model_client async def process_input(self, player_id: str, player_input: str): state self.state_manager.load_state(player_id) scene_id state[scene_id] scene_config load_scene_config(scene_id) # 伪代码实际从场景配置读取 # 1. 构建提示词 system_prompt, user_content self.prompt_builder.build( character_cardload_character_card(scene_config[character_id]), scenescene_config, historystate[history], player_inputplayer_input, ) # 2. 调用模型生成结果 messages [ {role: system, content: system_prompt}, {role: user, content: user_content}, ] raw_output await self.model_client.chat(messages) # 3. 解析模型输出 parsed extract_json(raw_output) # 4. 更新角色好感度 delta parsed.get(relationship_delta, {}) target delta.get(target, player) value delta.get(value, 0) relations state[character_relations] relations[target] relations.get(target, 0) value # 5. 处理场景切换 if parsed.get(scene_change): state[scene_id] parsed[scene_change] # 6. 保存状态与聊天记录 state[character_relations] relations state[history][last_summary] parsed.get(dialogue, ) self.state_manager.save_state(player_id, state) # 7. 记录到记忆缓冲区 await self.memory_manager.add_message(assistant, parsed.get(dialogue, )) # 8. 返回客户端演出指令 return { dialogue: parsed.get(dialogue, ), emotion: parsed.get(emotion, neutral), action: parsed.get(action, ), scene_change: parsed.get(scene_change, ), relations: relations, }5. 演出层设计与流式输出5.1 结构化演出指令当叙事引擎返回结果后客户端需要把它翻译成可视化的演出。这里的核心是让 AI 的输出“可驱动引擎”而非“展示给玩家看”。客户端接收到的 JSON 结构{ dialogue: 这个信号源的地点……是西区废弃的数据塔。, emotion: surprised, action: 她快速调出全息地图标记出一个红色光点, scene_change: scene_data_tower, relations: { player: 32 } }客户端根据emotion字段切换立绘表情根据action字段触发角色动画根据scene_change判断是否需要场景转场。5.2 立绘表情映射在客户端中需要一个映射表把情感标签与角色立绘资源对应起来// 文件路径client/src/services/expressionMap.ts export const expressionMap: Recordstring, string { happy: expression_happy, sad: expression_sad, angry: expression_angry, surprised: expression_surprised, neutral: expression_neutral, anxious: expression_anxious, }; export function getExpression(emotion: string): string { return expressionMap[emotion] || expressionMap[neutral]; }5.3 流式输出与打字机效果纯文本生成通常需要 2 到 5 秒如果等完整结果返回再显示玩家会明显感受到卡顿。解决方案是文本流式传输 打字机效果。客户端通过 WebSocket 接收流式文本// 文件路径client/src/services/wsClient.ts export class NarrationClient { private socket: WebSocket | null null; private messageCallback: ((text: string) void) | null null; connect(url: string): void { this.socket new WebSocket(url); this.socket.onmessage (event) { const data JSON.parse(event.data); if (data.type token this.messageCallback) { this.messageCallback(data.content); } }; } sendPlayerInput(input: string): void { this.socket?.send(JSON.stringify({ type: player_input, content: input, })); } onToken(callback: (text: string) void): void { this.messageCallback callback; } }服务端在流式生成时把 model_client 的异步生成器转发到 WebSocketasync def narrative_ws(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() payload json.loads(data) player_input payload[content] # 流式生成 messages build_prompt_messages(player_input) async for chunk in model_client.chat_stream(messages): await websocket.send_json({ type: token, content: chunk, }) # 流式结束后发送完整事件 await websocket.send_json({ type: event, event: await engine.post_process(player_input), })这样玩家在文本生成过程中就能看到文字逐字出现对话体验与传统 galgame 的打字机效果一致同时消除了等待焦虑。6. 性能优化与成本控制6.1 上下文窗口管理实时叙事游戏运行时间越长历史数据越多。如果不做上下文管理token 消耗会随着游戏时长线性增长最终导致两个问题模型响应速度变慢。API 费用快速上升。目前采用三层上下文策略第一层必带角色卡 场景配置 即时对话轮次 第二层摘要历史剧情摘要由模型定期生成 第三层按需关键事件完整记录只在相关剧情出现时注入关键事件按需注入是降本增效的关键。比如角色 A 在第 2 章提过一个关键道具第 10 章再次提及前通过关键词匹配注入相关历史片段而不是全部历史。6.2 token 预算控制在模型调用前计算当前消息的预估 token 数超过阈值时触发压缩def estimate_tokens(text: str) - int: # 中文约 1.5 字符/token英文约 4 字符/token # 这里用简化的估算方式实际可用 tiktoken 精确计算 return int(len(text) / 1.5) def ensure_budget(messages, max_tokens3000): total sum(estimate_tokens(m[content]) for m in messages) while total max_tokens and len(messages) 2: # 移除最老的中间轮次保留 system 和最后两轮 messages.pop(1) total sum(estimate_tokens(m[content]) for m in messages) return messages6.3 重复请求缓存对于常见的玩家输入比如“打招呼”“沉默”“继续”可以启用缓存。相同输入且角色状态未变化时返回缓存结果避免重复调用模型。7. 常见问题与排查思路在开发实时叙事 galgame 过程中遇到的问题远不止“写代码”这么简单。下面整理几个高频问题与排查方案。问题现象常见原因解决思路模型输出 JSON 解析失败模型输出了 Markdown 代码块或多余说明文字使用清洗函数提取 JSON 区间设置降级默认值角色出现“跳出人设”的发言提示词中角色卡信息不足或上下文丢失严格固定角色卡每次请求都注入人格设定对话响应速度超过 5 秒上下文过长、模型推理慢或网络延迟压缩历史摘要、限制 max_tokens、开启流式输出玩家输入导致剧情失控提示词未限制 AI 替玩家做决定在 System Prompt 中明确写“不要替玩家做决定”WebSocket 连接中断服务端超时时间过短设置合理的超时时间加入心跳检测与重连机制好感度增长过快或过慢relationship_delta 数值设计不合理在提示词中约束单次变化范围为 -3 到 3生成文本重复或陷入循环temperature 设置过低或历史摘要覆盖不足适当提高 temperature增加随机性约束7.1 模型输出偏离 JSON 格式的排查清单查看模型原始输出日志确认是格式问题还是内容截断。如果是格式问题检查 System Prompt 中 JSON 示例是否清晰。尝试降低 temperature 到 0.5 以下模型会更倾向于输出规范格式。检查 max_tokens 是否过短输出被截断会导致 JSON 不完整。在解析逻辑最后设置降级方案保证游戏不会崩溃。7.2 角色记忆丢失的排查思路角色记忆丢失通常表现为玩家在第 5 章提到第 1 章发生的事情角色完全没有反应。排查思路检查历史摘要是否真的保存到数据库。检查提示词构建时是否把历史摘要注入。查看 memory_manager 的 compress 触发条件确认缓冲区内是否积累了足够多消息。考虑增加关键词索引在特定剧情节点强制注入关键事件。8. 最佳实践与工程建议8.1 设计上以“可控随机”为目标实时叙事不能完全“自由”。完全自由的 AI 叙事很容易让剧情走向失控玩家会觉得内容零散缺乏主线。建议实现一套“目标引导机制”每个场景定义一个或多个“叙事目标”。把当前目标写进 System Prompt。当模型输出推进了目标时给予正向关系度奖励。当模型输出偏离目标时在下一次提示词中做出方向性纠正。8.2 建立完整的日志体系实时叙事游戏比传统游戏更需要日志。因为每次生成结果都不可完全复现必须记录下模型输入输出、token 消耗、响应耗时、解析结果。推荐日志字段时间戳 玩家 ID 场景 ID 模型请求 token 数 模型响应 token 数 响应耗时 模型输出原始文本 解析后 JSON 是否降级 好感度变化有了这些日志才能定位问题、优化提示词、控制成本。8.3 提示词版本管理提示词是实时叙事游戏的“核心玩法代码”。它和普通代码一样需要版本管理。建议把提示词模板放入 Git 仓库每次修改都记录变更原因。上线前使用固定测试案例回归测试比如“打招呼” - 角色回应是否符合人设“询问关键剧情” - 是否透露正确信息“做出极端选择” - 剧情是否有合理反馈8.4 关于 AI 生成内容的合规与边界实时叙事游戏涉及 AI 生成内容在开发阶段就需要注意内容安全在设计阶段确定目标受众与内容分级。在 System Prompt 中明确内容禁区设定负面提示词。对模型输出做基础过滤拦截明显违规内容。保留玩家举报与内容回溯机制。这些问题无论是对玩家体验还是对项目长期维护都至关重要。9. 下一步开发计划当前版本的实时叙事流程已经可以跑通从玩家输入到 AI 生成文本再到演出展示全链路响应时间控制在 3 秒左右基本达到了可玩状态。接下来重点做三件事完善记忆系统。当前摘要式记忆还有信息丢失的问题计划加入实体关系抽取让角色对关键人物和事件的记忆更精准。增强演出表现力。目前 AI 只能控制情感标签和场景切换计划增加镜头语言标签、音效指令、屏幕震动等更丰富的演出控制。剧情目标系统。为每个章节设计明确的叙事目标用目标引导模型生成方向保证玩家自由度和主线推进的平衡。实时叙事 galgame 是一个很适合 AI 应用开发者练手的项目方向。它不要求你有 AAA 游戏开发经验但能让你把一个 LLM 真正“用起来”涉及提示词工程、流式处理、状态管理、异步架构、数据存储、前端渲染等多个环节是一个完整度很高的全栈 AI 应用实践。如果你正在学大模型应用开发或者对游戏开发感兴趣不妨从这个方向切入试试。如果你也在做 AI 游戏或实时叙事相关项目欢迎在评论区交流你的实现思路。下一篇日志我会重点写记忆系统的深度设计方案包括实体抽取与剧情关键事件索引的实现细节。 AI实时叙事Galgame开发实战：大模型驱动剧情生成的架构与工程实现
前段时间分享了一款 AI 实时叙事 galgame《CITIZEN / ZERO》的第一篇开发日志收到了不少朋友的私信都在问实时叙事到底是怎么实现的、AI 响应和剧情分支怎么衔接、以及一个准大三学生是怎么把大模型塞进 galgame 引擎里的。这篇开发日志第二篇我来详细拆解一下当前版本的核心实现思路与工程架构。内容包括实时叙事系统的整体流程设计AI 生成内容与 galgame 演出框架的衔接方式角色状态管理与长期记忆的实现性能优化、token 消耗控制、流式输出的落地当前踩过的坑和排查思路本文适合对 AI 游戏开发、LLM 应用落地、视觉小说引擎二次开发感兴趣的开发者阅读。即使你没有做过 galgame只要你有 Python 或 TypeScript 基础也能从中获取一套可以复用的 AI 实时叙事方案。1. AI 实时叙事 galgame 到底是什么1.1 传统 galgame 的叙事方式传统 galgame视觉小说的叙事是预编写 分支树结构。剧本家在开发阶段把每条线路、每个选项、每个结局全部写好玩家在关键节点做出选择游戏引擎根据选择跳转到对应分支。这种方式的优点是剧情质量可控文笔可以反复打磨。演出与文案严格绑定情感节奏容易设计。技术实现简单本质是一个带状态机的内容播放器。缺点是玩家的自由度是预设的无法跳出剧本作者的框架。重复游玩时新鲜感下降。制作成本高一条完整线路需要数万句文案支撑。1.2 引入 AI 之后的实时叙事《CITIZEN / ZERO》要做的是把“剧本编写”这件事从开发阶段移到玩家游玩过程中。游戏不再读取写死的对话文件而是根据当前剧情状态、角色设定、玩家历史行为和实时输入动态生成下一段剧情文本。实时叙事 galgame 的核心特征剧情不是完全预写的而是由 AI 模型在运行时生成。玩家的输入不仅触发分支还直接影响 AI 的角色扮演策略。角色拥有“记忆”能够跨对话引用之前发生的事情。演出系统立绘、背景、BGM、特效响应 AI 生成的情感标签和场景标签。这样做的最大价值是每个玩家的体验都是独特的。同一个角色在不同玩家的档里会形成不同的性格倾向和关系状态。1.3 与 AI Agent 游戏的区别需要说明的是实时叙事 galgame 不完全等同于 AI Agent 游戏。AI Agent 游戏通常让 AI 控制角色在沙盒环境中自主行动比如让 AI 角色自己决定去哪、做什么然后由游戏系统反馈结果。而《CITIZEN / ZERO》的实时叙事把重点放在**“对话 情感 剧情推进”**上。AI 不直接操作游戏世界而是负责生成符合角色人设和剧情状态的文本内容游戏系统再把这些内容转译为可视化演出。2. 整体技术架构与核心流程2.1 系统组成当前版本的技术栈如下游戏客户端Web 技术栈TypeScript Phaser 3 Vue 3叙事引擎Python 编写的事件驱动服务AI 模型层支持接入本地部署的大模型与云端 API通过统一接口调用存储层SQLite 存储剧情节点与玩家行为记录JSON 文件存储角色卡设定通信协议WebSocket HTTP架构上遵循“前后端分离”原则客户端只负责渲染与输入采集所有叙事决策都在服务端完成。这样做的原因会在后面展开。2.2 实时叙事流程一次完整的实时叙事交互流程如下玩家在客户端输入对话或选择动作。客户端把玩家输入、当前场景 ID、当前角色状态打包发送到叙事引擎。叙事引擎从“叙事上下文管理器”中获取该场景的历史对话摘要与角色状态。引擎构建提示词调用大模型生成一段响应文本。模型输出除了对白外还包含情感标签、场景切换指令、角色心理状态变化。叙事引擎解析模型输出更新角色状态数据库生成演出指令。客户端收到文本与演出指令驱动立绘表情、背景、特效、BGM 完成演出。这里最关键的一环是第 5 步模型输出不能只是纯文本必须带有结构化控制信息。否则游戏引擎不知道当前角色是什么表情、是否要切换背景、剧情是否推进。3. 环境准备与项目结构3.1 开发环境说明下面是本文示例使用的基础环境版本需要根据你的项目实际情况调整操作系统Windows 11 / Ubuntu 22.04 Node.js18客户端构建 Python3.10叙事引擎 数据库SQLite 3 模型服务本地 OpenAI 兼容接口 / 云端大模型 API之所以使用 OpenAI 兼容接口是因为目前大多数本地推理框架如 Xinference、Ollama 等和云端模型服务都提供该协议接入成本最低。你完全可以把模型层替换成任意支持该协议的模型服务。3.2 项目结构citizen-zero/ ├── client/ # 游戏客户端Web │ ├── src/ │ │ ├── scenes/ # Phaser 游戏场景 │ │ ├── components/ # Vue UI 组件 │ │ ├── services/ # 网络通信服务 │ │ └── types/ # 类型定义 │ └── package.json ├── server/ # Python 叙事引擎 │ ├── app/ │ │ ├── main.py # FastAPI 入口 │ │ ├── narrative.py # 叙事核心逻辑 │ │ ├── prompt_builder.py# 提示词构建 │ │ ├── game_state.py # 游戏状态管理 │ │ ├── memory.py # 长期记忆模块 │ │ ├── model_client.py # 大模型统一客户端 │ │ └── parsers.py # 模型输出解析 │ ├── data/ │ │ ├── characters/ # 角色卡 JSON │ │ └── scenes/ # 场景配置 │ └── requirements.txt └── README.md4. 核心模块设计与代码实现4.1 角色卡设计角色卡是 AI 扮演角色的“人格基础”。它不像传统的 galgame 人设文档只给美术参考而是直接作为提示词的一部分输入模型。CITIZEN / ZERO中的角色卡设计如下{ id: mira, name: 米拉, age: 19, personality: 冷静理性但在熟悉的人面前会流露出温柔一面, speaking_style: 简洁少用语气词偏向科技感表达, background: 就职于城市数据分析中心负责追踪异常信号..., relationships: { player: 30, unknown_entity: -20 }, secrets: [曾参与过一项未公开的实验...] }这里需要特别注意relationships字段。它记录了角色对玩家和其他重要对象的好感度。这个值不是摆设而是会被写入提示词的动态部分直接影响 AI 生成文本时的情感倾向。4.2 提示词构建器提示词构建是实时叙事中最核心的部分。一个好的提示词需要把角色设定、当前场景、历史摘要、玩家输入、输出格式要求全部组织起来。下面是一个简化但可用的提示词模板# 文件路径server/app/prompt_builder.py system_prompt 你是一位视觉小说剧本家负责扮演角色 {character_name}。 ## 角色设定 {character_card} ## 当前场景 {scene_description} ## 场景内历史摘要 {history_summary} ## 情感状态 {emotional_state} ## 输出要求 请根据以上信息生成角色对玩家输入的一段回应。 你必须输出严格的 JSON 格式字段如下 {{ dialogue: 角色说的话不超过 80 字, emotion: happy|sad|angry|surprised|neutral|anxious, action: 角色动作描述用于演出如她低头看着终端屏幕, scene_change: 无变化时为空字符串需要切换场景时填写目标场景ID, relationship_delta: {{ target: player, value: 1 }} }} ## 注意事项 1. dialogue 必须符合角色设定不能跳出人设。 2. 尽量延续当前场景的情感基调。 3. 不要替玩家做决定不要描述玩家的动作。 def build_prompt(character_card, scene, history, player_input): user_content f玩家说{player_input} return system_prompt.format( character_namecharacter_card[name], character_cardcharacter_card, scene_descriptionscene, history_summaryhistory, emotional_statehistory.get(emotional_state, 平静) ), user_content4.3 模型客户端封装为了兼容本地模型和云端 API这里封装了一个统一客户端。底层使用 HTTP 请求访问 OpenAI 兼容接口。# 文件路径server/app/model_client.py import httpx import json class ModelClient: def __init__(self, base_url, api_key, model_name): base_url: 模型服务地址例如 http://localhost:8000/v1 api_key: 认证密钥本地模型可以填空字符串 model_name: 模型名称例如 qwen2.5-7b-instruct self.base_url base_url.rstrip(/) self.api_key api_key self.model_name model_name async def chat(self, messages, temperature0.8, max_tokens512): url f{self.base_url}/chat/completions headers {Authorization: fBearer {self.api_key}} payload { model: self.model_name, messages: messages, temperature: temperature, max_tokens: max_tokens, stream: False, } async with httpx.AsyncClient(timeout60) as client: resp await client.post(url, jsonpayload, headersheaders) resp.raise_for_status() data resp.json() return data[choices][0][message][content] async def chat_stream(self, messages, temperature0.8): 流式版本用于生成大段剧情文本时逐字返回降低玩家等待感。 url f{self.base_url}/chat/completions headers {Authorization: fBearer {self.api_key}} payload { model: self.model_name, messages: messages, temperature: temperature, stream: True, } async with httpx.AsyncClient(timeout60) as client: async with client.stream(POST, url, jsonpayload, headersheaders) as resp: async for line in resp.aiter_lines(): if not line.startswith(data:): continue data_str line[5:].strip() if data_str [DONE]: break try: chunk json.loads(data_str) delta chunk[choices][0].get(delta, {}) content delta.get(content, ) if content: yield content except json.JSONDecodeError: continue4.4 模型输出解析与容错模型输出 JSON 是一个理想状态实际运行中模型经常输出多余的说明文字、Markdown 代码块或者直接把 JSON 截断。因此解析层必须足够健壮。我的做法是优先尝试直接解析 JSON失败后进入清洗流程。# 文件路径server/app/parsers.py import json import re def extract_json(raw_text: str) - dict: 尝试从模型输出中提取 JSON 对象。 策略先去 Markdown 代码块标记再截取第一个 { 到最后一个 }。 # 去掉 json 和 包裹 cleaned re.sub(r(?:json)?, , raw_text).strip() try: return json.loads(cleaned) except json.JSONDecodeError: pass # 截取大括号区间 start cleaned.find({) end cleaned.rfind(}) if start ! -1 and end ! -1 and end start: try: return json.loads(cleaned[start:end 1]) except json.JSONDecodeError: pass # 降级方案日志记录并返回默认结构 print(f[WARN] 模型输出解析失败原始内容: {raw_text}) return { dialogue: raw_text[:80], emotion: neutral, action: , scene_change: , relationship_delta: {target: player, value: 0}, }这里关键的工程思想是降级策略。实时叙事不能因为一次解析失败就中断游戏流程所以当解析失败时直接把原始文本当作对白展示表情保持默认好感度不变化。这样玩家体验不会中断问题只会在后台日志中记录。4.5 游戏状态管理器游戏状态管理器负责维护当前剧情节点的全部动态信息当前场景 ID角色好感度剧情进度关键标记临时事件状态# 文件路径server/app/game_state.py import sqlite3 import json from datetime import datetime class GameStateManager: def __init__(self, db_pathcitizen_zero.db): self.db_path db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS game_state ( player_id TEXT PRIMARY KEY, scene_id TEXT NOT NULL, character_relations TEXT NOT NULL, flags TEXT NOT NULL, history TEXT NOT NULL, updated_at TEXT NOT NULL ) ) conn.execute( CREATE TABLE IF NOT EXISTS chat_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, player_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, metadata TEXT, created_at TEXT NOT NULL ) ) def load_state(self, player_id: str) - dict: with sqlite3.connect(self.db_path) as conn: row conn.execute( SELECT scene_id, character_relations, flags, history FROM game_state WHERE player_id ?, (player_id,) ).fetchone() if row is None: return self._create_default_state(player_id) return { scene_id: row[0], character_relations: json.loads(row[1]), flags: json.loads(row[2]), history: json.loads(row[3]), } def save_state(self, player_id: str, state: dict): with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO game_state (player_id, scene_id, character_relations, flags, history, updated_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(player_id) DO UPDATE SET scene_id excluded.scene_id, character_relations excluded.character_relations, flags excluded.flags, history excluded.history, updated_at excluded.updated_at , ( player_id, state[scene_id], json.dumps(state[character_relations], ensure_asciiFalse), json.dumps(state[flags], ensure_asciiFalse), json.dumps(state[history], ensure_asciiFalse), datetime.utcnow().isoformat(), ) )4.6 长期记忆与历史摘要LLM 的上下文窗口是有限的。如果每一轮对话都把全部历史塞进提示词很快就会撑爆上下文窗口而且模型对过长的上下文关注度会下降。当前方案采用“滚动摘要 关键事件持久化”策略每次对话结束后把本轮对话追加到临时缓冲区。当缓冲区达到一定长度调用模型生成一段历史摘要。摘要与关键事件标记一起存入数据库。后续请求只携带摘要而非完整历史。# 文件路径server/app/memory.py class MemoryManager: def __init__(self, model_client, max_buffer10): self.model_client model_client self.max_buffer max_buffer self.buffer [] async def add_message(self, role, content): self.buffer.append({role: role, content: content}) if len(self.buffer) self.max_buffer: await self.compress() async def compress(self): 调用模型对缓冲区内容生成摘要。 这是一个异步压缩操作避免阻塞主叙事流程。 messages [ {role: system, content: 你将一段对话历史压缩为 150 字以内的剧情摘要保留关键事件与情感转折。}, {role: user, content: json.dumps(self.buffer, ensure_asciiFalse)} ] summary await self.model_client.chat(messages, temperature0.3, max_tokens256) self.buffer [{role: system, content: f历史摘要{summary}}] return summary4.7 叙事引擎主流程把所有模块串联起来的是叙事核心逻辑。下面是一个简化版的事件循环# 文件路径server/app/narrative.py from fastapi import WebSocket import json class NarrativeEngine: def __init__(self, state_manager, memory_manager, prompt_builder, model_client): self.state_manager state_manager self.memory_manager memory_manager self.prompt_builder prompt_builder self.model_client model_client async def process_input(self, player_id: str, player_input: str): state self.state_manager.load_state(player_id) scene_id state[scene_id] scene_config load_scene_config(scene_id) # 伪代码实际从场景配置读取 # 1. 构建提示词 system_prompt, user_content self.prompt_builder.build( character_cardload_character_card(scene_config[character_id]), scenescene_config, historystate[history], player_inputplayer_input, ) # 2. 调用模型生成结果 messages [ {role: system, content: system_prompt}, {role: user, content: user_content}, ] raw_output await self.model_client.chat(messages) # 3. 解析模型输出 parsed extract_json(raw_output) # 4. 更新角色好感度 delta parsed.get(relationship_delta, {}) target delta.get(target, player) value delta.get(value, 0) relations state[character_relations] relations[target] relations.get(target, 0) value # 5. 处理场景切换 if parsed.get(scene_change): state[scene_id] parsed[scene_change] # 6. 保存状态与聊天记录 state[character_relations] relations state[history][last_summary] parsed.get(dialogue, ) self.state_manager.save_state(player_id, state) # 7. 记录到记忆缓冲区 await self.memory_manager.add_message(assistant, parsed.get(dialogue, )) # 8. 返回客户端演出指令 return { dialogue: parsed.get(dialogue, ), emotion: parsed.get(emotion, neutral), action: parsed.get(action, ), scene_change: parsed.get(scene_change, ), relations: relations, }5. 演出层设计与流式输出5.1 结构化演出指令当叙事引擎返回结果后客户端需要把它翻译成可视化的演出。这里的核心是让 AI 的输出“可驱动引擎”而非“展示给玩家看”。客户端接收到的 JSON 结构{ dialogue: 这个信号源的地点……是西区废弃的数据塔。, emotion: surprised, action: 她快速调出全息地图标记出一个红色光点, scene_change: scene_data_tower, relations: { player: 32 } }客户端根据emotion字段切换立绘表情根据action字段触发角色动画根据scene_change判断是否需要场景转场。5.2 立绘表情映射在客户端中需要一个映射表把情感标签与角色立绘资源对应起来// 文件路径client/src/services/expressionMap.ts export const expressionMap: Recordstring, string { happy: expression_happy, sad: expression_sad, angry: expression_angry, surprised: expression_surprised, neutral: expression_neutral, anxious: expression_anxious, }; export function getExpression(emotion: string): string { return expressionMap[emotion] || expressionMap[neutral]; }5.3 流式输出与打字机效果纯文本生成通常需要 2 到 5 秒如果等完整结果返回再显示玩家会明显感受到卡顿。解决方案是文本流式传输 打字机效果。客户端通过 WebSocket 接收流式文本// 文件路径client/src/services/wsClient.ts export class NarrationClient { private socket: WebSocket | null null; private messageCallback: ((text: string) void) | null null; connect(url: string): void { this.socket new WebSocket(url); this.socket.onmessage (event) { const data JSON.parse(event.data); if (data.type token this.messageCallback) { this.messageCallback(data.content); } }; } sendPlayerInput(input: string): void { this.socket?.send(JSON.stringify({ type: player_input, content: input, })); } onToken(callback: (text: string) void): void { this.messageCallback callback; } }服务端在流式生成时把 model_client 的异步生成器转发到 WebSocketasync def narrative_ws(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() payload json.loads(data) player_input payload[content] # 流式生成 messages build_prompt_messages(player_input) async for chunk in model_client.chat_stream(messages): await websocket.send_json({ type: token, content: chunk, }) # 流式结束后发送完整事件 await websocket.send_json({ type: event, event: await engine.post_process(player_input), })这样玩家在文本生成过程中就能看到文字逐字出现对话体验与传统 galgame 的打字机效果一致同时消除了等待焦虑。6. 性能优化与成本控制6.1 上下文窗口管理实时叙事游戏运行时间越长历史数据越多。如果不做上下文管理token 消耗会随着游戏时长线性增长最终导致两个问题模型响应速度变慢。API 费用快速上升。目前采用三层上下文策略第一层必带角色卡 场景配置 即时对话轮次 第二层摘要历史剧情摘要由模型定期生成 第三层按需关键事件完整记录只在相关剧情出现时注入关键事件按需注入是降本增效的关键。比如角色 A 在第 2 章提过一个关键道具第 10 章再次提及前通过关键词匹配注入相关历史片段而不是全部历史。6.2 token 预算控制在模型调用前计算当前消息的预估 token 数超过阈值时触发压缩def estimate_tokens(text: str) - int: # 中文约 1.5 字符/token英文约 4 字符/token # 这里用简化的估算方式实际可用 tiktoken 精确计算 return int(len(text) / 1.5) def ensure_budget(messages, max_tokens3000): total sum(estimate_tokens(m[content]) for m in messages) while total max_tokens and len(messages) 2: # 移除最老的中间轮次保留 system 和最后两轮 messages.pop(1) total sum(estimate_tokens(m[content]) for m in messages) return messages6.3 重复请求缓存对于常见的玩家输入比如“打招呼”“沉默”“继续”可以启用缓存。相同输入且角色状态未变化时返回缓存结果避免重复调用模型。7. 常见问题与排查思路在开发实时叙事 galgame 过程中遇到的问题远不止“写代码”这么简单。下面整理几个高频问题与排查方案。问题现象常见原因解决思路模型输出 JSON 解析失败模型输出了 Markdown 代码块或多余说明文字使用清洗函数提取 JSON 区间设置降级默认值角色出现“跳出人设”的发言提示词中角色卡信息不足或上下文丢失严格固定角色卡每次请求都注入人格设定对话响应速度超过 5 秒上下文过长、模型推理慢或网络延迟压缩历史摘要、限制 max_tokens、开启流式输出玩家输入导致剧情失控提示词未限制 AI 替玩家做决定在 System Prompt 中明确写“不要替玩家做决定”WebSocket 连接中断服务端超时时间过短设置合理的超时时间加入心跳检测与重连机制好感度增长过快或过慢relationship_delta 数值设计不合理在提示词中约束单次变化范围为 -3 到 3生成文本重复或陷入循环temperature 设置过低或历史摘要覆盖不足适当提高 temperature增加随机性约束7.1 模型输出偏离 JSON 格式的排查清单查看模型原始输出日志确认是格式问题还是内容截断。如果是格式问题检查 System Prompt 中 JSON 示例是否清晰。尝试降低 temperature 到 0.5 以下模型会更倾向于输出规范格式。检查 max_tokens 是否过短输出被截断会导致 JSON 不完整。在解析逻辑最后设置降级方案保证游戏不会崩溃。7.2 角色记忆丢失的排查思路角色记忆丢失通常表现为玩家在第 5 章提到第 1 章发生的事情角色完全没有反应。排查思路检查历史摘要是否真的保存到数据库。检查提示词构建时是否把历史摘要注入。查看 memory_manager 的 compress 触发条件确认缓冲区内是否积累了足够多消息。考虑增加关键词索引在特定剧情节点强制注入关键事件。8. 最佳实践与工程建议8.1 设计上以“可控随机”为目标实时叙事不能完全“自由”。完全自由的 AI 叙事很容易让剧情走向失控玩家会觉得内容零散缺乏主线。建议实现一套“目标引导机制”每个场景定义一个或多个“叙事目标”。把当前目标写进 System Prompt。当模型输出推进了目标时给予正向关系度奖励。当模型输出偏离目标时在下一次提示词中做出方向性纠正。8.2 建立完整的日志体系实时叙事游戏比传统游戏更需要日志。因为每次生成结果都不可完全复现必须记录下模型输入输出、token 消耗、响应耗时、解析结果。推荐日志字段时间戳 玩家 ID 场景 ID 模型请求 token 数 模型响应 token 数 响应耗时 模型输出原始文本 解析后 JSON 是否降级 好感度变化有了这些日志才能定位问题、优化提示词、控制成本。8.3 提示词版本管理提示词是实时叙事游戏的“核心玩法代码”。它和普通代码一样需要版本管理。建议把提示词模板放入 Git 仓库每次修改都记录变更原因。上线前使用固定测试案例回归测试比如“打招呼” - 角色回应是否符合人设“询问关键剧情” - 是否透露正确信息“做出极端选择” - 剧情是否有合理反馈8.4 关于 AI 生成内容的合规与边界实时叙事游戏涉及 AI 生成内容在开发阶段就需要注意内容安全在设计阶段确定目标受众与内容分级。在 System Prompt 中明确内容禁区设定负面提示词。对模型输出做基础过滤拦截明显违规内容。保留玩家举报与内容回溯机制。这些问题无论是对玩家体验还是对项目长期维护都至关重要。9. 下一步开发计划当前版本的实时叙事流程已经可以跑通从玩家输入到 AI 生成文本再到演出展示全链路响应时间控制在 3 秒左右基本达到了可玩状态。接下来重点做三件事完善记忆系统。当前摘要式记忆还有信息丢失的问题计划加入实体关系抽取让角色对关键人物和事件的记忆更精准。增强演出表现力。目前 AI 只能控制情感标签和场景切换计划增加镜头语言标签、音效指令、屏幕震动等更丰富的演出控制。剧情目标系统。为每个章节设计明确的叙事目标用目标引导模型生成方向保证玩家自由度和主线推进的平衡。实时叙事 galgame 是一个很适合 AI 应用开发者练手的项目方向。它不要求你有 AAA 游戏开发经验但能让你把一个 LLM 真正“用起来”涉及提示词工程、流式处理、状态管理、异步架构、数据存储、前端渲染等多个环节是一个完整度很高的全栈 AI 应用实践。如果你正在学大模型应用开发或者对游戏开发感兴趣不妨从这个方向切入试试。如果你也在做 AI 游戏或实时叙事相关项目欢迎在评论区交流你的实现思路。下一篇日志我会重点写记忆系统的深度设计方案包括实体抽取与剧情关键事件索引的实现细节。