一、存在问题

原来prompt的实现方式本质上是一个静态提示词加简单状态拼接的架构。在旧的story_engine.py中,系统提示词虽然也尝试动态注入当前节点和已解锁节点信息,但其核心逻辑非常简单:每次调用LLM时,仅仅将当前节点ID、已解锁节点列表、角色属性、背包物品等基础状态拼接到一个相对固定的系统提示词模板中,然后让LLM自由发挥生成剧情和状态更新。

这种旧架构的核心问题在于提示词与游戏逻辑严重耦合。AI在生成JSON响应时,需要自己记住所有节点之间的连接关系、每个支线任务的触发条件、NPC的好感度阈值、物品的有效ID等大量固定规则。这意味着AI既要充当故事创作者,又要充当游戏规则执行者,极易出错。举例来说,当玩家在森林入口与老猎人对话时,旧架构的AI必须自己判断“老猎人的好感度阈值40会讲述森林传说”,自己决定是否触发支线任务side_001,自己记录任务进度。但AI的上下文窗口有限,记忆不可靠,常常出现节点ID拼写错误、解锁条件判断失误、任务奖励应用不一致等问题。

例如下面图片中经过迷雾小径以后应该会有三个岔路口,但大模型擅自改成了两个。

二、解决方案

在和ai讨论后,ai建议实现较长的prompt使用分层prompt+动态注入的架构

三、实现

1.game_data.py

# ==================== NPC数据 ====================
NPCS = {
    "老猎人": {
        "id": "npc_001",
        "name": "老猎人",
        "portrait": "🏹",
        "location": "node_entrance",
        "initial_affinity": 40,
        "likes": ["野果", "猎物", "治疗药水"],
        "dislikes": ["暗影水晶"],
        "dialogue": {
            "greeting": "年轻人,你也来寻找生命之树吗?这片森林最近越来越危险了...",
            "quest": "我的儿子哈尔在森林里失踪了,你能帮我找到他吗?",
            "farewell": "小心那些暗影生物,它们最近变得异常狂暴。"
        },
        "quest_chain": "side_001",
        "affinity_thresholds": {
            30: "开始讲述森林的传说",
            50: "透露进入森林深处的秘密路径",
            70: "告诉生命之树的真实位置",
            90: "传授猎人秘技(永久+3力量)"
        }
    },
    "艾莉西亚": {
        "id": "npc_002",
        "name": "艾莉西亚",
        "portrait": "🧝",
        "location": "node_elf_village",
        "initial_affinity": 50,
        "likes": ["月光碎片", "精灵信物", "净化之水"],
        "dislikes": ["暗影水晶", "黑暗之心"],
        "dialogue": {
            "greeting": "欢迎来到精灵村落,陌生人。我能感受到你身上有特别的气息...",
            "trust": "暗影力量正在侵蚀生命之树,我们精灵族的力量在减弱...",
            "farewell": "月光会指引你的道路。"
        },
        "quest_chain": "side_003",
        "affinity_thresholds": {
            40: "开始信任你并讲述精灵族的困境",
            60: "请求帮助寻找被盗的月光竖琴",
            80: "揭示生命之树被污染的真相",
            95: "愿意牺牲自己来净化生命之树"
        }
    },
    "暗影": {
        "id": "npc_003",
        "name": "暗影",
        "portrait": "🌑",
        "location": "node_shadow_swamp",
        "initial_affinity": 0,
        "likes": ["暗影水晶", "黑暗之心"],
        "dislikes": ["净化之水", "生命之叶"],
        "dialogue": {
            "greeting": "...(发出低沉的吼声)",
            "trust": "我曾...是...守护者...黑暗...吞噬了我...",
            "farewell": "小心...领主...他...比我...更强大..."
        },
        "quest_chain": "side_004",
        "affinity_thresholds": {
            10: "不再攻击你",
            30: "开始与你交流",
            50: "透露暗影领主的弱点",
            70: "恢复部分记忆,帮助对抗暗影领主"
        }
    },
    "盗贼头目": {
        "id": "npc_004",
        "name": "盗贼头目",
        "portrait": "🗡️",
        "location": "node_bandit_camp",
        "initial_affinity": -10,
        "likes": ["金币", "盗贼匕首", "净化之水"],
        "dislikes": ["精灵信物"],
        "dialogue": {
            "greeting": "站住!把身上的东西都交出来!...不,等等,我的头...很痛...",
            "quest": "帮帮我...我被什么东西控制了...",
            "farewell": "谢谢你让我重获自由...我会报答你的。"
        },
        "quest_chain": "side_005",
        "affinity_thresholds": {
            0: "停止攻击你",
            30: "开始清醒,讲述被控制的经历",
            60: "请求帮助解除暗影控制",
            80: "恢复本性,加入对抗暗影领主的战斗"
        }
    },
    "旅行的商人": {
        "id": "npc_005",
        "name": "旅行的商人",
        "portrait": "🧳",
        "location": "node_crossroad",
        "initial_affinity": 30,
        "likes": ["金币", "稀有物品"],
        "dislikes": [],
        "dialogue": {
            "greeting": "哦!又一个冒险者!三条路都通向不同的命运,选哪条?我可以给你一些建议...当然不是免费的。",
            "hint": "精灵们很友善但警惕,暗影沼泽有危险但也有机遇,盗贼嘛...他们曾经是好人。",
            "farewell": "记住,选择没有对错,只有不同的故事。"
        },
        "shop_items": [
            {"name": "治疗药水", "price": 50, "effect": {"hp": 30}},
            {"name": "力量药剂", "price": 80, "effect": {"strength": 5}},
            {"name": "解毒草", "price": 30, "effect": {"hp": 10, "remove_poison": True}},
            {"name": "地图碎片", "price": 100, "effect": "解锁隐藏地点"}
        ]
    },
    "森林守护者": {
        "id": "npc_006",
        "name": "森林守护者",
        "portrait": "🌳",
        "location": "node_tree_of_life",
        "appear_condition": "完成side_006守护者试炼,且善良值≥70",
        "initial_affinity": 60,
        "likes": ["生命之叶", "守护符文"],
        "dislikes": ["黑暗之心"],
        "dialogue": {
            "greeting": "终于...有人听到了我的呼唤...",
            "truth": "暗影领主并非天生邪恶...他是我的另一半...被人类伤害后堕落的...",
            "farewell": "记住,真正的力量来自于守护,而非征服。"
        },
        "affinity_thresholds": {
            80: "揭示隐藏结局条件",
            100: "授予'森林守护者'称号"
        }
    }
}

# ==================== 支线任务 ====================
SIDE_QUESTS = {
    "side_001": {
        "id": "side_001",
        "name": "猎人的请求",
        "location": "node_entrance",
        "trigger_npc": "老猎人",
        "description": "老猎人的儿子在森林中失踪了,他请求你帮忙寻找",
        "objectives": [
            {"id": "sobj_001", "description": "在迷雾小径找到猎人的踪迹", "completed": False},
            {"id": "sobj_002", "description": "击败绑架猎人之子的野兽", "completed": False, "need_strength": 18},
            {"id": "sobj_003", "description": "将猎人之子带回森林入口", "completed": False}
        ],
        "rewards": {"exp": 300, "gold": 150, "item": "猎人护符", "attribute_bonus": {"strength": 2}}
    },
    "side_002": {
        "id": "side_002",
        "name": "古老的秘密",
        "location": "node_mist_path",
        "trigger_condition": "在迷雾小径探索古老石碑",
        "description": "石碑上记载着远古守护者的秘密,需要集齐三块符文碎片才能解读",
        "objectives": [
            {"id": "sobj_004", "description": "在精灵村落找到第一块符文碎片", "completed": False},
            {"id": "sobj_005", "description": "在暗影沼泽找到第二块符文碎片", "completed": False},
            {"id": "sobj_006", "description": "在盗贼营地找到第三块符文碎片", "completed": False}
        ],
        "rewards": {"exp": 500, "item": "远古符文", "attribute_bonus": {"intelligence": 5, "mp": 20}},
        "unlock_ending": "ancient_guardian_ending"
    },
    "side_003": {
        "id": "side_003",
        "name": "精灵公主的烦恼",
        "location": "node_elf_village",
        "trigger_npc": "艾莉西亚",
        "min_affinity": 30,
        "description": "精灵族的神器'月光竖琴'被暗影生物偷走了",
        "objectives": [
            {"id": "sobj_007", "description": "前往暗影沼泽追踪暗影生物", "completed": False},
            {"id": "sobj_008", "description": "击败暗影守卫取回竖琴", "completed": False, "need_agility": 15},
            {"id": "sobj_009", "description": "将竖琴交还给艾莉西亚", "completed": False}
        ],
        "rewards": {"exp": 400, "item": "月光碎片", "relationship_change": {"艾莉西亚": 20}}
    },
    "side_004": {
        "id": "side_004",
        "name": "净化之泉",
        "location": "node_shadow_swamp",
        "trigger_condition": "探索次数>=3",
        "description": "沼泽深处有一口被污染的净化之泉,恢复它可以削弱暗影力量",
        "objectives": [
            {"id": "sobj_010", "description": "收集3株净化草", "completed": False},
            {"id": "sobj_011", "description": "击败守护泉水的暗影触手", "completed": False, "need_strength": 20},
            {"id": "sobj_012", "description": "使用净化草净化泉水", "completed": False}
        ],
        "rewards": {"exp": 350, "item": "净化之水", "attribute_bonus": {"hp": 20}},
        "world_effect": "暗影沼泽危险度降低,暗影领主最终战难度降低"
    },
    "side_005": {
        "id": "side_005",
        "name": "盗贼的救赎",
        "location": "node_bandit_camp",
        "trigger_npc": "盗贼头目",
        "description": "盗贼们并非自愿为恶,他们被暗影力量控制了心智",
        "objectives": [
            {"id": "sobj_013", "description": "与盗贼头目对话了解真相", "completed": False},
            {"id": "sobj_014", "description": "使用净化之水解除控制(需完成side_004)", "completed": False},
            {"id": "sobj_015", "description": "帮助盗贼们重建营地", "completed": False}
        ],
        "rewards": {"exp": 400, "gold": 300, "item": "盗贼匕首", "attribute_bonus": {"agility": 3, "luck": 2}}
    },
    "side_006": {
        "id": "side_006",
        "name": "守护者的试炼",
        "location": "node_tree_of_life",
        "trigger_condition": "到达生命之树",
        "description": "想获得生命之树的力量,必须先通过守护者的试炼",
        "objectives": [
            {"id": "sobj_016", "description": "通过智慧的试炼", "completed": False, "need_intelligence": 25},
            {"id": "sobj_017", "description": "通过勇气的试炼", "completed": False, "need_strength": 20},
            {"id": "sobj_018", "description": "通过仁爱的试炼", "completed": False, "need_relationship": "any_npc_affinity_60"}
        ],
        "rewards": {"exp": 600, "item": "生命之叶", "attribute_bonus": {"hp": 30, "mp": 20}}
    }
}

# ==================== 主线任务 ====================
MAIN_QUESTS = {
    "main_001": {
        "id": "main_001",
        "name": "进入迷雾森林",
        "description": "探索森林入口,找到进入森林深处的路径",
        "objectives": [
            {"id": "obj_001", "description": "与老猎人对话"},
            {"id": "obj_002", "description": "查看古老石碑"},
            {"id": "obj_003", "description": "选择一条路径进入"}
        ],
        "rewards": {"exp": 100, "gold": 50},
        "next_quest": "main_002"
    },
    "main_002": {
        "id": "main_002",
        "name": "寻找精灵族",
        "description": "深入森林,寻找传说中的精灵族",
        "objectives": [
            {"id": "obj_004", "description": "通过迷雾小径"},
            {"id": "obj_005", "description": "找到精灵族线索"},
            {"id": "obj_006", "description": "到达精灵村落"}
        ],
        "rewards": {"exp": 200, "gold": 100, "item": "精灵信物"},
        "next_quest": "main_003"
    },
    "main_003": {
        "id": "main_003",
        "name": "生命之树的秘密",
        "description": "揭开生命之树被侵蚀的真相",
        "objectives": [
            {"id": "obj_007", "description": "与精灵公主艾莉西亚对话"},
            {"id": "obj_008", "description": "收集三块守护符文"},
            {"id": "obj_009", "description": "前往生命之树"}
        ],
        "rewards": {"exp": 500, "gold": 300, "item": "守护符文"},
        "next_quest": "main_004"
    },
    "main_004": {
        "id": "main_004",
        "name": "最终之战",
        "description": "在生命之树前做出最终选择",
        "objectives": [
            {"id": "obj_010", "description": "面对暗影领主"},
            {"id": "obj_011", "description": "做出最终决定"}
        ],
        "rewards": {"exp": 1000, "gold": 500},
        "next_quest": None
    }
}

# ==================== 结局数据 ====================
ENDINGS = {
    "ending_good_001": {
        "id": "ending_good_001",
        "name": "生命之树的守护者",
        "type": "main",
        "condition": "善良值≥60,选择守护生命之树",
        "description": "你成功净化了暗影力量,成为了生命之树新的守护者。",
        "ending_text": "你将手放在生命之树的树干上,感受着它的脉动。纯净的光芒从你身上扩散,暗影消散,森林重获新生。\n\n艾莉西亚微笑着说:\"你做到了。生命之树认可了你。\"\n\n从此,你成为了迷雾森林的守护者,与精灵们一起守护这片神圣的土地。"
    },
    "ending_dark_002": {
        "id": "ending_dark_002",
        "name": "暗影之王",
        "type": "main",
        "condition": "邪恶值≥60,选择吸收暗影力量",
        "description": "你吸收了暗影力量,成为了新的暗影领主。",
        "ending_text": "黑暗的力量涌入你的体内,你感到前所未有的强大。生命之树在你的触摸下枯萎,但新的力量在你心中萌芽。\n\n暗影领主在你面前消散,他的力量成为了你的力量。森林中的生物俯首称臣,你成为了新的暗影之王。"
    },
    "ending_sacrifice_003": {
        "id": "ending_sacrifice_003",
        "name": "牺牲",
        "type": "main",
        "condition": "善良值≥40,hp≥50,选择牺牲自己",
        "description": "你选择用自己的生命力来净化生命之树。",
        "ending_text": "你闭上眼睛,将所有生命力注入生命之树。你的身体开始发光,化作漫天的光点。\n\n\"不!\"艾莉西亚哭喊着,但你已经无法回头。\n\n生命之树绽放出前所未有的光芒,暗影被彻底驱散。你成为了森林的传说。"
    }
}

# ==================== 节点连接规则 ====================
NODE_CONNECTIONS = {
    "node_entrance": ["node_mist_path"],
    "node_mist_path": ["node_crossroad"],
    "node_crossroad": ["node_elf_village", "node_shadow_swamp", "node_bandit_camp"],
    "node_elf_village": [],  # 需要支线解锁 node_elf_trial
    "node_elf_trial": ["node_tree_of_life"],
    "node_shadow_swamp": [],  # 需要支线解锁 node_shadow_lair
    "node_shadow_lair": ["node_tree_of_life"],
    "node_bandit_camp": [],  # 需要支线
    "node_tree_of_life": []
}

# 节点解锁条件
NODE_UNLOCK_CONDITIONS = {
    "node_elf_trial": {"quest": "side_003", "min_affinity": 50, "npc": "艾莉西亚"},
    "node_shadow_lair": {"quest": "side_004", "item": "暗影水晶"},
    "node_tree_of_life": {"quest": "main_003"}
}

# 各节点的NPC
NODE_NPCS = {
    "node_entrance": ["老猎人"],
    "node_crossroad": ["旅行的商人"],
    "node_elf_village": ["艾莉西亚"],
    "node_shadow_swamp": ["暗影"],
    "node_bandit_camp": ["盗贼头目"],
    "node_tree_of_life": ["森林守护者"]
}

# 各节点的物品
NODE_ITEMS = {
    "node_entrance": ["古老石碑"],
    "node_mist_path": ["符文碎片(1/3)", "猎人的踪迹"],
    "node_elf_village": ["月光碎片", "符文碎片(2/3)"],
    "node_elf_trial": ["勇气之证"],
    "node_shadow_swamp": ["符文碎片(2/3)", "暗影水晶"],
    "node_shadow_lair": ["黑暗之心"],
    "node_bandit_camp": ["符文碎片(2/3)", "盗贼的宝藏"],
    "node_tree_of_life": ["生命之叶"]
}

game_data.py是整个游戏系统的静态数据层,它集中定义了游戏世界中的所有固定规则和内容,包括NPC数据、支线任务、主线任务、结局条件、节点拓扑结构等六大核心模块。

NPC数据模块为每个角色定义了唯一标识、头像图标、所处位置、初始好感度、喜好与厌恶物品、预设对话模板,以及好感度阈值映射表——后者决定了当玩家与NPC的好感度达到不同数值时,NPC会透露何种层级的秘密或提供何种特殊奖励。支线任务模块定义了每个任务的触发节点、触发NPC或触发条件、任务目标的详细描述以及完成后的奖励内容,包括经验值、金币、物品、属性加成和好感度变化。主线任务模块则定义了四个线性推进的任务阶段,每个阶段包含多个目标、奖励以及指向下一任务的引用。节点连接规则模块以邻接表的形式定义了节点之间的连通关系,而节点解锁条件模块则补充了那些需要满足特定任务或属性要求才能进入的节点的解锁逻辑。各节点的NPC映射表和物品清单则为PromptManager提供了快速查询当前场景存在哪些可交互对象的依据。

整个文件作为单一事实来源,可以集中修改任意规则,而PromptManager、QuestManager等上层模块通过读取这些数据结构来动态生成提示词和执行任务逻辑,实现了数据与行为的分离。

2.prompt_manager.py

# backend/prompt_manager.py
from typing import Dict, List
from .game_data import NPCS, NODE_NPCS, NODE_CONNECTIONS, SIDE_QUESTS, MAIN_QUESTS


class PromptManager:
    def __init__(self):
        self.core_rules = self._get_core_rules()
    
    def _get_core_rules(self) -> str:
        """精简的核心规则"""
        return """# 迷雾森林文字冒险游戏引擎

## 输出格式(必须严格遵守JSON)
{
  "story_text": "剧情文本(150-300字,生动描写,推动剧情)",
  "state_update": {
    "current_node": "当前节点ID",
    "unlock_nodes": ["最多1个新节点"],
    "discoveries": [{"type":"item","name":"物品名","description":"描述"}],
    "attribute_changes": {"hp":-5,"strength":2},
    "task_update": {"main":"新任务描述"},
    "relationship_update": {"NPC名":5},
    "flags": ["剧情标志"],
    "events": ["事件描述"]
  }
}

## 核心规则
1. 一次只能解锁1个节点,必须是当前节点的邻居
2. 不能越级解锁,必须按顺序推进
3. 玩家没真正到达某地时,不能切换current_node
4. 新物品通过discoveries添加
5. 禁止输出JSON以外的任何内容

## 有效节点ID
node_entrance, node_mist_path, node_crossroad, node_elf_village, 
node_elf_trial, node_shadow_swamp, node_shadow_lair, node_bandit_camp, 
node_tree_of_life

## 节点连接关系
- node_entrance → node_mist_path
- node_mist_path → node_crossroad
- node_crossroad → node_elf_village / node_shadow_swamp / node_bandit_camp
- node_elf_village → (需完成支线后解锁 node_elf_trial)
- node_shadow_swamp → (需完成支线后解锁 node_shadow_lair)
- node_bandit_camp → (需完成支线)
- node_elf_trial / node_shadow_lair → node_tree_of_life

## NPC好感度影响
- 好感度越高,NPC透露信息越多
- 部分支线需要好感度达到阈值才能触发
- 在对话中体现好感度差异
"""
    
    def build_system_prompt(self, state: Dict) -> str:
        """构建完整的系统提示词"""
        current_node = state.get('current_node', 'node_entrance')
        unlocked_nodes = state.get('unlocked_nodes', [])
        
        return f"""
{self.core_rules}

【当前游戏状态】
- 当前节点: {current_node}
- 已解锁节点: {', '.join(unlocked_nodes)}
- 当前任务: {state.get('task_progress', {}).get('main', '探索迷雾森林')}

{self._get_npcs_prompt(state, current_node)}
{self._get_quests_prompt(state, current_node)}
{self._get_available_actions_prompt(current_node)}
"""
    
    def _get_npcs_prompt(self, state: Dict, current_node: str) -> str:
        """获取当前节点的NPC信息"""
        npc_names = NODE_NPCS.get(current_node, [])
        if not npc_names:
            return "【当前区域】无特殊NPC"
        
        result = "【当前区域的NPC】\n"
        for npc_name in npc_names:
            npc = NPCS.get(npc_name, {})
            affinity = state.get('relationships', {}).get(npc_name, npc.get('initial_affinity', 0))
            result += f"- {npc.get('portrait', '👤')} {npc_name} (好感度:{affinity})\n"
            result += f"  可以: 与{npc_name}对话\n"
            
            # 根据好感度显示不同提示
            for threshold, hint in npc.get('affinity_thresholds', {}).items():
                if affinity >= int(threshold):
                    result += f"  💡 {hint}\n"
                    break
        return result
    
    def _get_quests_prompt(self, state: Dict, current_node: str) -> str:
        """获取当前相关的任务信息"""
        result = "【当前任务】\n"
        
        # 主线任务
        main_quest = state.get('task_progress', {}).get('main', '')
        if main_quest:
            result += f"📜 主线: {main_quest}\n"
        
        # 可触发的支线
        triggerable = []
        for qid, quest in SIDE_QUESTS.items():
            if quest.get('location') == current_node:
                trigger_npc = quest.get('trigger_npc')
                if trigger_npc:
                    triggerable.append(f"• 可与{trigger_npc}对话接取任务「{quest['name']}」")
                else:
                    triggerable.append(f"• 探索可触发任务「{quest['name']}」")
        
        if triggerable:
            result += "\n【可接取的任务】\n" + "\n".join(triggerable)
        
        return result
    
    def _get_available_actions_prompt(self, current_node: str) -> str:
        """获取当前节点可用的行动建议"""
        actions_map = {
            "node_entrance": ["与老猎人对话", "查看古老石碑", "进入迷雾小径"],
            "node_mist_path": ["探索迷雾小径", "寻找猎人的踪迹", "继续前进"],
            "node_crossroad": ["向左转(前往精灵村落)", "直行(前往暗影沼泽)", "向右转(前往盗贼营地)", "与旅行的商人对话"],
            "node_elf_village": ["与艾莉西亚对话", "探索精灵村落", "查看符文碎片"],
            "node_shadow_swamp": ["探索沼泽", "寻找净化草", "与暗影互动"],
            "node_bandit_camp": ["与盗贼头目对话", "探索营地", "寻找符文碎片"],
            "node_tree_of_life": ["接受守护者试炼", "与森林守护者对话", "做出最终选择"]
        }
        
        actions = actions_map.get(current_node, ["探索周围", "检查状态", "使用物品"])
        return "【可能的行动】\n" + "\n".join([f"• {a}" for a in actions])

PromptManager的核心职责是根据当前游戏状态,从预定义的NPC数据、任务数据、节点数据中提取出“此刻真正需要告诉AI的信息”,以清晰的结构化文本注入提示词,避免AI从庞大的规则集中自行推导。

遵循“按需查询、动态组装”的原则。在`build_system_prompt`方法中,它接收当前游戏状态作为输入,首先通过`NODE_NPCS`映射表反查出当前节点存在哪些NPC,再根据状态中存储的好感度数值,遍历该NPC的`affinity_thresholds`字典,找到第一个满足条件的阈值并将对应的提示信息注入提示词。对于任务信息,它遍历`SIDE_QUESTS`筛选出`location`字段匹配当前节点的任务,区分NPC触发和探索触发两种类型后分别格式化。最后通过`actions_map`映射表为AI提供当前节点可执行的行动选项。

整个过程实现了游戏逻辑与AI推理的解耦,AI无需记忆任何静态规则,所有NPC位置、好感度阈值、任务触发条件均由PromptManager从`game_data.py`中实时查询并转换为结构化文本。

3.quest_manager.py

# backend/quest_manager.py
from typing import Dict, List
from .game_data import SIDE_QUESTS, MAIN_QUESTS, NPCS


class QuestManager:
    def __init__(self):
        # 存储玩家活跃任务: {user_id: {quest_id: progress}}
        self.active_quests = {}
        self.completed_quests = {}
    
    def _get_user_quests(self, user_id: str) -> Dict:
        if user_id not in self.active_quests:
            self.active_quests[user_id] = {}
        return self.active_quests[user_id]
    
    def check_trigger(self, state: Dict, action: str, current_node: str, 
                      target_npc: str = None) -> Dict:
        """检查是否触发新支线"""
        user_id = state.get('user_id', 'anonymous')
        user_quests = self._get_user_quests(user_id)
        updates = {"new_quests": [], "quest_progress": []}
        
        for qid, quest in SIDE_QUESTS.items():
            # 跳过已接取或已完成的任务
            if qid in user_quests or qid in self.completed_quests.get(user_id, {}):
                continue
            
            # 检查触发条件
            if quest.get('location') != current_node:
                continue
            
            trigger_npc = quest.get('trigger_npc')
            if trigger_npc and target_npc != trigger_npc:
                continue
            
            # 检查好感度要求
            min_affinity = quest.get('min_affinity', 0)
            current_affinity = state.get('relationships', {}).get(trigger_npc, 0)
            if current_affinity < min_affinity:
                continue
            
            # 触发任务
            user_quests[qid] = {
                "name": quest['name'],
                "objectives": {obj['id']: False for obj in quest['objectives']}
            }
            updates["new_quests"].append({
                "id": qid,
                "name": quest['name'],
                "description": quest['description'],
                "objectives": quest['objectives']
            })
        
        return updates
    
    def check_progress(self, state: Dict, action: str, current_node: str) -> Dict:
        """检查是否推进任务进度"""
        user_id = state.get('user_id', 'anonymous')
        user_quests = self._get_user_quests(user_id)
        updates = {"completed": [], "progress": [], "rewards": []}
        
        for qid, progress in user_quests.items():
            quest = SIDE_QUESTS[qid]
            all_complete = True
            
            for obj in quest['objectives']:
                obj_id = obj['id']
                if not progress['objectives'].get(obj_id):
                    # 检查是否完成此目标
                    if self._check_objective(state, action, current_node, obj):
                        progress['objectives'][obj_id] = True
                        updates["progress"].append({
                            "quest": quest['name'],
                            "objective": obj['description']
                        })
                    else:
                        all_complete = False
            
            # 任务完成
            if all_complete and all(progress['objectives'].values()):
                # 记录奖励
                rewards = quest.get('rewards', {})
                updates["rewards"].append({
                    "quest": quest['name'],
                    "exp": rewards.get('exp', 0),
                    "gold": rewards.get('gold', 0),
                    "item": rewards.get('item'),
                    "attribute_bonus": rewards.get('attribute_bonus', {}),
                    "relationship_change": rewards.get('relationship_change', {})
                })
                updates["completed"].append(quest['name'])
                # 标记完成
                if user_id not in self.completed_quests:
                    self.completed_quests[user_id] = {}
                self.completed_quests[user_id][qid] = True
                del user_quests[qid]
        
        return updates
    
    def _check_objective(self, state: Dict, action: str, current_node: str, 
                         objective: Dict) -> bool:
        """检查是否完成单个目标"""
        obj_desc = objective['description'].lower()
        
        # 根据目标类型检查
        if "找到" in obj_desc and "踪迹" in obj_desc:
            return current_node == "node_mist_path"
        
        if "击败" in obj_desc:
            return "攻击" in action or "战斗" in action
        
        if "对话" in obj_desc:
            target = objective.get('target_npc', '')
            return target in action if target else True
        
        if "收集" in obj_desc:
            # 简单模拟:探索可能获得物品
            return "探索" in action and "收集" in action
        
        if "使用" in obj_desc:
            return "使用" in action
        
        return False
    
    def apply_rewards(self, state: Dict, rewards: Dict) -> Dict:
        """应用任务奖励"""
        changes = {"attribute_changes": {}, "relationship_changes": {}, "new_items": []}
        
        # 属性奖励
        for attr, value in rewards.get('attribute_bonus', {}).items():
            changes["attribute_changes"][attr] = value
        
        # 好感度奖励
        for npc, value in rewards.get('relationship_change', {}).items():
            changes["relationship_changes"][npc] = value
        
        # 物品奖励
        if rewards.get('item'):
            changes["new_items"].append(rewards['item'])
        
        # 金币奖励
        if rewards.get('gold'):
            changes["attribute_changes"]['money'] = rewards['gold']
        
        return changes
    
    def get_active_quests_for_prompt(self, user_id: str) -> str:
        """获取活跃任务用于Prompt"""
        user_quests = self._get_user_quests(user_id)
        if not user_quests:
            return ""
        
        result = "\n【进行中的支线任务】\n"
        for qid, progress in user_quests.items():
            quest = SIDE_QUESTS[qid]
            result += f"\n📜 {quest['name']}\n"
            result += f"   {quest['description']}\n"
            for obj in quest['objectives']:
                done = "✓" if progress['objectives'].get(obj['id']) else "○"
                result += f"   {done} {obj['description']}\n"
        
        return result

QuestManager通过三个核心方法将任务系统的控制权从AI完全剥离。`check_trigger`方法遍历SIDE_QUESTS,根据当前节点、目标NPC和好感度阈值进行多层过滤,符合条件的任务被初始化到active_quests字典中,任务进度以目标ID到布尔值的映射存储。`check_progress`方法在每个回合遍历活跃任务,调用`_check_objective`通过关键词匹配判定玩家行动是否达成某个目标,当所有目标均为True时触发完成逻辑,提取rewards字段并移入completed_quests。`apply_rewards`将奖励拆解为属性变化、好感度变化和新物品三类返回。`get_active_quests_for_prompt`将进行中任务格式化为带进度标记的文本注入提示词,使AI知晓任务状态但无需理解任务ID和奖励规则,实现了叙事自由与规则确定性的分离。

4.dynamic_provider.py

# backend/dynamic_provider.py
from typing import Dict, List
from .game_data import NPCS, NODE_NPCS, NODE_ITEMS, NODE_CONNECTIONS


class DynamicContentProvider:
    
    def get_current_context(self, state: Dict) -> str:
        """获取当前上下文"""
        attrs = state.get('attributes', {})
        inventory = state.get('inventory', [])
        
        return f"""
- 当前节点: {state.get('current_node', 'node_entrance')}
- 已解锁节点: {', '.join(state.get('unlocked_nodes', []))}
- 生命值: {attrs.get('hp', 100)}/100
- 力量: {attrs.get('strength', 0)} | 智力: {attrs.get('intelligence', 0)}
- 金币: {attrs.get('money', 0)}
- 背包: {', '.join(inventory[:5]) if inventory else '空'}
"""
    
    def get_node_info(self, current_node: str) -> str:
        """获取节点详细信息"""
        items = NODE_ITEMS.get(current_node, [])
        neighbors = NODE_CONNECTIONS.get(current_node, [])
        
        result = ""
        if items:
            result += f"\n【此地的物品】\n" + "\n".join([f"• {item}" for item in items])
        
        if neighbors:
            result += f"\n【可前往的地点】\n" + "\n".join([f"• {neighbor}" for neighbor in neighbors])
        
        return result
    
    def get_main_quest_progress(self, state: Dict) -> str:
        """获取主线任务进度"""
        task_progress = state.get('task_progress', {})
        main_quest = task_progress.get('main', '')
        
        if not main_quest:
            return ""
        
        return f"\n【主线任务】{main_quest}"
    
    def check_node_unlock(self, current_node: str, unlocked_nodes: List[str], 
                          completed_quests: List[str], attributes: Dict,
                          relationships: Dict) -> Dict:
        """检查是否有节点可以解锁(用于规则验证)"""
        # 这个函数给AI参考,不强制
        possible_unlocks = []
        
        # 从十字路口解锁三个分支
        if current_node == "node_crossroad":
            for branch in ["node_elf_village", "node_shadow_swamp", "node_bandit_camp"]:
                if branch not in unlocked_nodes:
                    possible_unlocks.append(branch)
        
        # 精灵试炼场需要完成支线
        if current_node == "node_elf_village" and "side_003" in completed_quests:
            if "node_elf_trial" not in unlocked_nodes:
                possible_unlocks.append("node_elf_trial")
        
        # 暗影巢穴需要完成支线
        if current_node == "node_shadow_swamp" and "side_004" in completed_quests:
            if "node_shadow_lair" not in unlocked_nodes:
                possible_unlocks.append("node_shadow_lair")
        
        return {"possible_unlocks": possible_unlocks}

DynamicContentProvider负责生成用户提示词中的动态上下文,让AI清楚了解玩家当前的环境状态和任务进度。

`get_current_context`方法从状态中提取当前节点、已解锁节点列表、生命值、力量智力属性、金币数量以及背包前五项物品,格式化为结构化的状态快照注入提示词。`get_node_info`方法通过查询NODE_ITEMS和NODE_CONNECTIONS两个映射表,分别获取当前节点可交互的物品列表和可前往的邻居节点列表,使AI知道玩家在此地能发现什么以及能走向哪里。`get_main_quest_progress`方法从task_progress中提取主线任务描述并格式化返回。`check_node_unlock`方法则提供节点解锁的参考规则——它根据当前节点和已完成支线任务,判断是否有新分支节点可以解锁,返回possible_unlocks列表供AI在生成state_update时参考。

通过这些方法,让AI在生成剧情时能够清楚知道玩家当前所处的环境、已经完成了哪些目标、还差哪些目标没有完成。AI无需记忆任何静态数据映射关系,只需读取注入的结构化信息即可生成符合当前场景的剧情。

5.story_engine.py

旧版本的`_build_dynamic_system_prompt`方法硬编码了允许节点列表和解锁条件,`_build_user_prompt`方法手动拼接属性面板,任务进度完全依赖AI在JSON中返回的`task_update`字段,没有任何独立的任务检测机制。

新版本主要修改如下:

5.1. __init__

新增了三个模块的实例化:`self.prompt_manager`、`self.quest_manager`、`self.dynamic_provider`

5.2. _build_dynamic_system_prompt

从原来的几十行硬编码逻辑和解锁条件的实现,简化为直接调用`self.prompt_manager.build_system_prompt(state)`,将所有NPC信息、好感度阈值、可接取任务等动态内容的注入工作委托给PromptManager。

5.3. _build_user_prompt

增加了对 `get_active_quests_for_prompt` 的调用,将活跃支线任务注入提示词;同时用 `dynamic_provider.get_current_context` 和 `get_main_quest_progress` 替代了手动拼接属性面板

    def _build_user_prompt(self, state: StoryState, template: Dict, history_text: str) -> str:
        """改进的用户提示词"""
        last_action = self._get_last_user_action(state)
        
        # 获取活跃任务
        user_id = state.get('user_id', 'anonymous')
        active_quests = self.quest_manager.get_active_quests_for_prompt(user_id)
        
        return f"""【玩家行动】{last_action}

{self.dynamic_provider.get_current_context(state)}

{self.dynamic_provider.get_main_quest_progress(state)}
{active_quests}
{self.dynamic_provider.get_node_info(state.get('current_node', 'node_entrance'))}

【剧情回顾】
{history_text[-500:] if history_text else "游戏刚开始"}

请根据玩家行动续写剧情,推动任务进度,并在合适时触发支线任务。输出JSON。
"""

5.4. narrative_director

在LLM响应解析和状态更新之间,新版本新增了支线触发检查、进度检查和奖励应用三段逻辑。

先调用`quest_manager.check_trigger`,传入当前状态、玩家行动、当前节点和目标NPC,检测是否有新支线任务满足触发条件,将检测到的任务通过events字段告知玩家;再调用`quest_manager.check_progress`,遍历活跃任务并通过`_check_objective`的关键词匹配判定玩家行动是否达成了某个目标,完成的任务提取rewards字段返回;最后调用`apply_rewards`将奖励拆解为属性变化、好感度变化和新物品三类,合并到state_update中。这意味着任务系统的控制权从AI的task_update字段转移到了QuestManager的确定性代码中,AI只需通过events字段告知发生了什么事件,QuestManager独立判断是否触发任务、是否完成目标、如何应用奖励。

    def narrative_director(self, state: StoryState) -> StoryState:
        """使用 LLM 生成故事文本和结构化状态更新"""
        template = self.story_templates.get(state["story_id"], self.story_templates["fantasy_001"])
        
        current_node = state.get("current_node", "node_entrance")
        unlocked_nodes = state.get("unlocked_nodes", ["node_entrance"])
        
        history_text = self._compress_history(state.get("messages", []), state.get("summary", ""))
        
        dynamic_system_prompt = self._build_dynamic_system_prompt(template, current_node, unlocked_nodes, state)
        user_prompt = self._build_user_prompt(state, template, history_text)
        
        llm_messages = [
            {"role": "system", "content": dynamic_system_prompt},
            {"role": "user", "content": user_prompt}
        ]
        
        raw_response = self.llm.generate(llm_messages, temperature=0.7)
        story_text, state_update = self._parse_llm_response(raw_response, state)
        
        # 检查支线触发
        last_action = self._get_last_user_action(state)
        quest_updates = self.quest_manager.check_trigger(
            state, last_action, current_node, self._extract_target_npc(last_action)
        )
        
        # 应用支线触发
        for new_quest in quest_updates.get("new_quests", []):
            if "events" not in state_update:
                state_update["events"] = []
            state_update["events"].append(f"触发支线任务:{new_quest['name']}")
        
        # 检查支线进度
        progress_updates = self.quest_manager.check_progress(state, last_action, current_node)
        for reward in progress_updates.get("rewards", []):
            reward_changes = self.quest_manager.apply_rewards(state, reward)
            for attr, val in reward_changes.get("attribute_changes", {}).items():
                state_update.setdefault("attribute_changes", {})[attr] = state_update.get("attribute_changes", {}).get(attr, 0) + val
            for npc, val in reward_changes.get("relationship_changes", {}).items():
                state_update.setdefault("relationship_update", {})[npc] = state_update.get("relationship_update", {}).get(npc, 0) + val
            for item in reward_changes.get("new_items", []):
                state_update.setdefault("discoveries", []).append({
                    "type": "item", "name": item, "description": f"完成任务获得{item}"
                })
        
        # 应用状态更新
        state = self._apply_state_update(state, state_update)
        
        # 构建最终叙事
        status_bar = self._build_status_bar(state)
        full_narrative = story_text + status_bar
        
        state["messages"].append({"role": "assistant", "content": full_narrative, "timestamp": datetime.now().isoformat()})
        state["current_description"] = story_text
        state["turn_count"] = state.get("turn_count", 0) + 1
        
        if state["turn_count"] % 5 == 0:
            state["summary"] = self._update_summary(state)
        
        return state

6.举例说明

假设玩家开始时处于森林入口节点。首先,DynamicProvider检查到当前节点有NPC"老猎人"且好感度为初始值,PromptManager将这个信息注入系统提示词,明确告诉AI"当前有NPC老猎人,可以与他对话,好感度40时他会讲述森林传说"。当玩家输入"与老猎人对话"后,AI根据提示词生成剧情文本"老猎人向你求助,他的儿子在森林中失踪了",同时在state_update中标记已触发对话。此时QuestManager检测到玩家当前节点为森林入口、且与老猎人进行了对话,这正好符合支线任务"猎人的请求"的触发条件,于是自动将该任务加入玩家的活跃任务列表。此后,当玩家后续执行"探索迷雾小径"、"击败野兽"等行动时,QuestManager在每个回合都会检查这些行动是否匹配任务目标,发现匹配后就标记对应目标为已完成。当所有三个目标都完成后,QuestManager自动触发任务完成逻辑,向state中增加300经验、150金币、猎人护符等奖励,并更新老猎人的好感度,同时将任务从活跃列表中移除。

四、结果

下图是两次游戏过程,可以看出碰见老猎人就一定会触发帮老猎人找儿子的任务,但是根据回答的措辞不同,大模型的回复也有差别,但总体发生总线不变

和一开始出现的问题对比,可以看出,遇到岔路口一定会提示有三条路,和设定一致,解决了原来存在的问题

五、总结

保证机制 作用 实现位置
数据定义 明确NPC和支线的关联关系 game_data.py
动态注入 只告诉AI当前相关的NPC/支线 dynamic_provider.py
状态跟踪 记录任务进度和触发条件 quest_manager.py
规则指导 在Prompt中明确支线处理规则 prompt_manager.py

最终的效果是职责分离:确定性的游戏规则(节点解锁条件、任务触发条件、物品效果、好感度阈值)全部写在game_data.py和各个Manager类中,由Python代码严格执行;而不确定性的创意内容(剧情文本、NPC对话、场景描写)交给AI自由发挥。AI的响应只需要输出简洁的结构化状态更新(例如“解锁某个节点”、“属性变化-5”、“获得某个物品”),所有复杂的逻辑判断和状态一致性保证都由后端的Manager模块完成。这样既保证了游戏规则的一致性,又保留了AI叙事的灵活性。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐