nlu_engine.py

"""

nlu_engine.py - StoryWeaver 自然语言理解引擎



职责：

1. 解析用户自然语言输入，提取结构化意图

2. 将玩家"乱七八糟的输入"映射到具体的动作类型

3. 封装意图识别的 Prompt 与 API 调用



设计思路：

- 使用 Qwen API 进行意图识别，利用 LLM 的语义理解能力

- Prompt 设计中明确列出所有可能的意图类型和示例

- 低温度 (0.2) 确保输出的 JSON 格式稳定可靠

- 提供降级机制：如果 API 调用失败，使用关键词匹配兜底



输入/输出示例（来自需求文档）：

  Input: "我想攻击那个哥布林"

  Output: {"intent": "ATTACK", "target": "哥布林", "details": null}

"""

import re
import logging
from typing import Optional

from demo_rules import build_scene_actions
from utils import safe_json_call, DEFAULT_MODEL
from state_manager import GameState

logger = logging.getLogger("StoryWeaver")


# ============================================================
# 意图识别 Prompt 模板
#
# 设计思路：
# - System Prompt 提供完整的意图类型列表和示例
# - 注入当前可用的行动上下文（当前场景的NPC、物品等）
# - 要求严格输出 JSON 格式
# - 低温度确保稳定性
# ============================================================

NLU_SYSTEM_PROMPT_TEMPLATE = """你是一个 RPG 游戏的自然语言理解模块（NLU）。你的任务是将玩家的自然语言输入解析为结构化的 JSON 意图数据。



【当前游戏上下文】

{context}



【支持的意图类型】

以下是所有合法的意图类型及其说明和示例：



| 意图 (intent) | 说明 | 示例输入 |

|:--|:--|:--|

| ATTACK | 攻击目标 | "攻击哥布林"、"打那个怪物"、"我要和它战斗" |

| TALK | 与NPC对话 | "和村长说话"、"找铁匠聊聊"、"我想打听消息" |

| MOVE | 移动到某地 | "去森林"、"回村庄"、"我要离开这里" |

| EXPLORE | 探索/观察环境 | "看看周围"、"仔细搜索"、"调查这个地方" |

| USE_ITEM | 使用物品 | "喝治疗药水"、"使用火把"、"吃面包" |

| TRADE | 交易（买/卖） | "买一把剑"、"卖掉这个"、"看看有什么卖的" |

| EQUIP | 装备物品 | "装备铁剑"、"穿上皮甲" |

| REST | 休息恢复 | "休息一下"、"在旅店过夜"、"睡觉" |

| QUEST | 接受/查看任务 | "接受任务"、"查看任务"、"任务完成了" |

| SKILL | 使用技能 | "施放火球术"、"使用隐身技能" |

| PICKUP | 拾取物品 | "捡起来"、"拿走那个东西" |

| FLEE | 逃跑 | "快跑"、"逃离这里"、"我要撤退" |

| CUSTOM | 其他自由行动 | "给NPC唱首歌"、"在墙上涂鸦" |



【当前场景中可交互的对象】

{interactables}



【输出格式要求】

请严格输出以下 JSON 格式（不要输出任何其他文字）：

{{

    "intent": "意图类型（从上表中选择）",

    "target": "行动目标（NPC名称、物品名称、地点名称等，如果没有明确目标则为 null）",

    "details": "补充细节（如 '用剑攻击'、'询问关于森林的事情' 等，如果没有额外细节则为 null）"

}}



【解析规则】

1. 如果玩家输入模糊（如"我不知道该干什么"），意图设为 EXPLORE。

2. 如果玩家输入包含多个动作，提取最主要的一个。

3. target 应尽量匹配当前场景中实际存在的对象。

4. 如果输入完全无法理解，设 intent 为 CUSTOM。

"""


class NLUEngine:
    """

    自然语言理解引擎



    核心能力：将玩家自由文本输入映射到结构化意图。



    工作流程：

    1. 收集当前场景上下文（NPC、物品、可达地点等）

    2. 构造 Prompt 并调用 Qwen API

    3. 解析返回的 JSON 意图

    4. 如果 API 失败，使用关键词匹配降级



    为什么用 LLM 而不是规则匹配：

    - 玩家输入千变万化，规则难以覆盖

    - LLM 能理解同义词、口语化表达、上下文隐含意图

    - 例如："我饿了" → 可能是 USE_ITEM（吃东西）或 MOVE（去旅店）

    """

    def __init__(self, game_state: GameState, model: str = DEFAULT_MODEL):
        self.game_state = game_state
        self.model = model

    def parse_intent(self, user_input: str) -> dict:
        """

        解析用户输入，返回结构化意图。



        Args:

            user_input: 玩家的原始文本输入



        Returns:

            {

                "intent": "ATTACK",

                "target": "哥布林",

                "details": "用剑攻击",

                "raw_input": "我想用剑攻击那个哥布林"

            }

        """
        if not user_input or not user_input.strip():
            return {
                "intent": "EXPLORE",
                "target": None,
                "details": "玩家沉默不语",
                "raw_input": "",
                "parser_source": "empty_input",
            }

        user_input = user_input.strip()
        logger.info(f"NLU 解析输入: '{user_input}'")

        # 尝试 LLM 解析
        result = self._llm_parse(user_input)

        # 如果 LLM 解析失败，使用关键词降级
        if result is None:
            logger.warning("LLM 解析失败，使用关键词降级")
            result = self._keyword_fallback(user_input)

        result = self._apply_intent_postprocessing(result, user_input)

        # 附加原始输入
        result["raw_input"] = user_input

        logger.info(f"NLU 解析结果: {result}")
        return result

    def _llm_parse(self, user_input: str) -> Optional[dict]:
        """

        使用 Qwen API 进行意图识别。

        低温度 (0.2) 确保 JSON 输出稳定。

        """
        context = self._build_context()
        interactables = self._build_interactables()

        system_prompt = NLU_SYSTEM_PROMPT_TEMPLATE.format(
            context=context,
            interactables=interactables,
        )

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input},
        ]

        result = safe_json_call(
            messages,
            model=self.model,
            temperature=0.2,
            max_tokens=300,
            max_retries=2,
        )

        if result and isinstance(result, dict) and "intent" in result:
            # 验证意图类型合法
            valid_intents = {
                "ATTACK", "TALK", "MOVE", "EXPLORE", "USE_ITEM",
                "TRADE", "EQUIP", "REST", "QUEST", "SKILL",
                "PICKUP", "FLEE", "CUSTOM",
            }
            if result["intent"] not in valid_intents:
                result["intent"] = "CUSTOM"
            result.setdefault("parser_source", "llm")
            return result

        return None

    def _keyword_fallback(self, user_input: str) -> dict:
        """

        关键词匹配降级方案。



        设计思路：

        - 当 API 不可用时的兜底策略

        - 使用正则匹配常见中文关键词

        - 覆盖最常见的意图类型

        - 无法匹配时默认为 EXPLORE

        """
        text = user_input.lower()

        # 关键词 → 意图映射（按优先级排序）
        keyword_rules = [
            # 攻击相关
            (r"攻击|打|杀|战斗|砍|刺|射|揍", "ATTACK"),
            # 逃跑相关
            (r"逃|跑|撤退|逃离|闪", "FLEE"),
            # 对话相关
            (r"说话|对话|交谈|聊|打听|询问|问", "TALK"),
            # 移动相关
            (r"去|前往|移动|走|回|离开|进入", "MOVE"),
            # 物品使用
            (r"使用|喝|吃|用|服用", "USE_ITEM"),
            # 交易
            (r"买|卖|交易|购买|出售|商店", "TRADE"),
            # 装备
            (r"装备|穿|戴|换装", "EQUIP"),
            # 休息
            (r"休息|睡|过夜|恢复|歇", "REST"),
            # 任务
            (r"任务|接受|完成|查看任务", "QUEST"),
            # 技能
            (r"施放|技能|魔法|法术|释放", "SKILL"),
            # 拾取
            (r"捡|拾|拿|拿走|拾取|收集", "PICKUP"),
            # 探索
            (r"看|观察|搜索|调查|探索|检查|四周", "EXPLORE"),
        ]

        detected_intent = "CUSTOM"
        for pattern, intent in keyword_rules:
            if re.search(pattern, text):
                detected_intent = intent
                break

        # 尝试提取目标
        target = self._extract_target_from_text(user_input)

        return {
            "intent": detected_intent,
            "target": target,
            "details": None,
            "parser_source": "keyword_fallback",
        }

    def _extract_target_from_text(self, text: str) -> Optional[str]:
        """

        从文本中提取可能的目标对象。

        尝试匹配当前场景中的 NPC、物品、地点名称。

        """
        # 检查 NPC 名称
        for npc_name in self.game_state.world.npcs:
            if npc_name in text:
                return npc_name

        # 检查物品名称（背包 + 当前场景）
        for item in self.game_state.player.inventory:
            if item in text:
                return item

        # 检查地点名称
        current_loc = self.game_state.world.locations.get(self.game_state.player.location)
        if current_loc:
            for loc_name in current_loc.connected_to:
                if loc_name in text:
                    return loc_name

        # 检查物品注册表
        for item_name in self.game_state.world.item_registry:
            if item_name in text:
                return item_name

        return None

    def _apply_intent_postprocessing(self, result: dict, user_input: str) -> dict:
        """Apply narrow intent corrections for high-confidence mixed phrases."""
        normalized = dict(result)
        intent = str(normalized.get("intent", "")).upper()
        if intent == "MOVE" and self._looks_like_trade_request(user_input, normalized.get("target")):
            inferred_trade_target = self._infer_trade_target(user_input, normalized.get("target"))
            target_text = str(normalized.get("target") or "")
            target_location = self.game_state.world.locations.get(target_text)
            # 目标是商店地点且玩家尚未到达时，优先保持 MOVE，避免生成“未到店先扣钱”的错误交易。
            if (
                target_location is not None
                and target_location.shop_available
                and target_text != self.game_state.player.location
            ):
                normalized["intent_correction"] = "preserve_move_for_shop_travel"
            elif inferred_trade_target is not None:
                normalized["intent"] = "TRADE"
                normalized["target"] = inferred_trade_target
                normalized["intent_correction"] = "move_to_trade_with_structured_target"
        if intent == "TRADE" and not isinstance(normalized.get("target"), dict):
            target_text = str(normalized.get("target") or "")
            target_location = self.game_state.world.locations.get(target_text)
            if (
                target_location is not None
                and target_location.shop_available
                and target_text != self.game_state.player.location
            ):
                normalized["intent"] = "MOVE"
                normalized["intent_correction"] = "trade_to_move_for_shop_travel"
                return normalized
            inferred_trade_target = self._infer_trade_target(user_input, normalized.get("target"))
            if inferred_trade_target is not None:
                normalized["target"] = inferred_trade_target
                normalized["intent_correction"] = "trade_target_inferred_from_text"
        if intent in {"ATTACK", "COMBAT"}:
            target = normalized.get("target")
            if not isinstance(target, str) or not target.strip() or target in {"怪物", "敌人", "它", "那个怪物"}:
                inferred_target = self._infer_attack_target()
                if inferred_target:
                    normalized["target"] = inferred_target
                    normalized["intent_correction"] = "attack_target_inferred_from_scene"
        return normalized

    def _looks_like_trade_request(self, user_input: str, target: Optional[str]) -> bool:
        trade_pattern = r"买|卖|交易|购买|出售|看看有什么卖的|买点"
        if not re.search(trade_pattern, user_input):
            return False

        target_text = str(target or "")
        if target_text:
            npc = self.game_state.world.npcs.get(target_text)
            if npc and npc.can_trade:
                return True

            location = self.game_state.world.locations.get(target_text)
            if location and location.shop_available:
                return True

        shop_hint_pattern = r"商店|杂货铺|旅店|铁匠铺"
        return bool(re.search(shop_hint_pattern, user_input))

    def _infer_attack_target(self) -> Optional[str]:
        """Infer a concrete ATTACK target from deterministic scene actions first."""
        try:
            scene_actions = build_scene_actions(self.game_state, self.game_state.player.location)
        except Exception:
            scene_actions = []
        for action in scene_actions:
            if str(action.get("action_type", "")).upper() != "ATTACK":
                continue
            target = action.get("target")
            if isinstance(target, str) and target.strip():
                return target

        current_loc = self.game_state.world.locations.get(self.game_state.player.location)
        if current_loc and current_loc.enemies:
            return str(current_loc.enemies[0])
        return None

    def _infer_trade_target(self, user_input: str, target: object) -> Optional[dict]:
        """Infer structured trade target for rule-based TRADE handling."""
        text_blob = f"{user_input} {target if isinstance(target, str) else ''}"

        merchant_name: Optional[str] = None
        for npc in self.game_state.world.npcs.values():
            if not npc.can_trade or npc.location != self.game_state.player.location:
                continue
            if npc.name in text_blob or (npc.occupation and npc.occupation in text_blob):
                merchant_name = npc.name
                break

        if merchant_name is None:
            for npc in self.game_state.world.npcs.values():
                if npc.can_trade and npc.location == self.game_state.player.location:
                    merchant_name = npc.name
                    break
        if merchant_name is None:
            return None

        merchant = self.game_state.world.npcs.get(merchant_name)
        if merchant is None:
            return None

        item_name: Optional[str] = None
        for candidate in merchant.shop_inventory:
            if candidate in text_blob:
                item_name = candidate
                break
        if item_name is None and isinstance(target, str) and target in merchant.shop_inventory:
            item_name = target
        if item_name is None:
            return None

        return {"merchant": merchant_name, "item": item_name, "confirm": False}

    def _build_context(self) -> str:
        """构建当前场景的简要上下文描述"""
        gs = self.game_state
        return (
            f"场景: {gs.world.current_scene}\n"
            f"时间: 第{gs.world.day_count}天 {gs.world.time_of_day}\n"
            f"玩家位置: {gs.player.location}\n"
            f"玩家 HP: {gs.player.hp}/{gs.player.max_hp}\n"
            f"玩家背包: {', '.join(gs.player.inventory) if gs.player.inventory else '空'}"
        )

    def _build_interactables(self) -> str:
        """构建当前场景中可交互对象的列表"""
        gs = self.game_state
        lines = []

        # 当前场景的 NPC
        current_npcs = [
            npc for npc in gs.world.npcs.values()
            if npc.location == gs.player.location and npc.is_alive
        ]
        if current_npcs:
            npc_names = [f"{npc.name}（{npc.occupation}）" for npc in current_npcs]
            lines.append(f"NPC: {', '.join(npc_names)}")

        # 可前往的地点
        loc = gs.world.locations.get(gs.player.location)
        if loc and loc.connected_to:
            lines.append(f"可前往: {', '.join(loc.connected_to)}")

        # 场景中的敌人
        if loc and loc.enemies:
            lines.append(f"可能的敌人: {', '.join(loc.enemies)}")

        # 背包物品
        if gs.player.inventory:
            lines.append(f"背包物品: {', '.join(gs.player.inventory)}")

        # 技能
        if gs.player.skills:
            lines.append(f"可用技能: {', '.join(gs.player.skills)}")

        return "\n".join(lines) if lines else "当前场景中没有特别的可交互对象"