AI_RDKS100P_Robot/brain/planner.py

"""
planner.py - AI Agent 通用 Planner 核心模块

该模块负责根据世界状态、用户意图、可用能力生成标准化的 Plan 对象。

职责边界：
    - 接收输入：world_snapshot, user_intent, available_tools, domain_rules, planner_mode
    - 输出：标准 Plan 对象
    - 不执行 capability
    - 不依赖 ROS2
    - 不包含场景硬编码逻辑

Planner 支持三种模式：
    - rule：纯规则生成
    - llm：LLM 生成（调用 llm_client）
    - hybrid：优先规则，复杂情况调用 LLM

使用方式：
    from planner import Planner, PlannerConfig
    from tool_protocol import RiskLevel

    config = PlannerConfig(default_risk_level=RiskLevel.LOW)
    planner = Planner(config)
    plan = planner.generate_plan(
        world_snapshot={"temperature": 35},
        user_intent="降温",
        available_tools=["adjust_fan", "speak"],
    )
    
    # 使用 LLM 模式
    plan = planner.generate_plan(
        world_snapshot={"temperature": 35},
        user_intent="降温",
        available_tools=["adjust_fan", "speak"],
        planner_mode="llm",
    )
"""

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from tool_protocol import (
    Plan,
    PlanStep,
    RiskLevel,
    PlanStatus,
    StepStatus,
    ToolCallType,
)


# =============================================================================
# 通用意图映射层
# =============================================================================

# 通用意图关键词到 action 名称的映射
# 用途：当用户意图包含某个关键词时，映射到对应的 action
# 可扩展：后续可通过配置文件或知识库动态加载
INTENT_TO_ACTION: dict[str, str] = {
    # 动作类
    "喂": "feed",
    "喂食": "feed",
    "喂养": "feed",
    "打开风扇": "adjust_fan",
    "关闭风扇": "adjust_fan",
    "调整风扇": "adjust_fan",
    "降温": "adjust_fan",
    "升温": "adjust_fan",
    "开灯": "control_light",
    "关灯": "control_light",
    "调整灯光": "control_light",
    "打开": "turn_on",
    "关闭": "turn_off",
    "启动": "turn_on",
    "停止": "turn_off",
    "调节": "adjust",
    "移动": "move",
    "前往": "move",
    "走到": "move",
    # 查询类
    "查询": "query",
    "查看": "query",
    "检查": "inspect",
    "巡检": "inspect",
    "监测": "monitor",
    "监控": "monitor",
    # 通信类
    "播报": "speak",
    "报告": "speak",
    "说": "speak",
    "告诉": "speak",
    "通知": "notify",
    "提醒": "remind",
    "警告": "warn",
    "询问": "ask",
}

# action 到 ToolCallType 的默认映射
ACTION_TO_TOOL_CALL_TYPE: dict[str, ToolCallType] = {
    "feed": ToolCallType.EXECUTE,
    "adjust_fan": ToolCallType.EXECUTE,
    "control_light": ToolCallType.EXECUTE,
    "turn_on": ToolCallType.EXECUTE,
    "turn_off": ToolCallType.EXECUTE,
    "adjust": ToolCallType.EXECUTE,
    "move": ToolCallType.EXECUTE,
    "query": ToolCallType.QUERY_WORLD,
    "inspect": ToolCallType.QUERY_WORLD,
    "monitor": ToolCallType.QUERY_WORLD,
    "speak": ToolCallType.SPEAK,
    "notify": ToolCallType.SPEAK,
    "remind": ToolCallType.SPEAK,
    "warn": ToolCallType.SPEAK,
    "ask": ToolCallType.ASK_USER,
}

# 高风险关键词列表（通用）
HIGH_RISK_KEYWORDS: list[str] = [
    "紧急", "危险", "停止", "关机", "断电", "全部", "清空",
    "emergency", "danger", "stop", "shutdown", "all", "clear",
]

# 中风险关键词列表（通用）
MEDIUM_RISK_KEYWORDS: list[str] = [
    "调整", "改变", "修改", "设置", "配置",
    "adjust", "change", "modify", "set", "configure",
]


# =============================================================================
# Planner 配置
# =============================================================================

@dataclass
class PlannerConfig:
    """Planner 配置类
    
    用于配置 Planner 的行为参数。
    
    属性：
        default_risk_level: 默认风险等级
        require_confirmation_on_medium_risk: 中风险是否需要确认
        require_confirmation_on_high_risk: 高风险是否需要确认
        max_plan_steps: 单个 Plan 最大步骤数
        default_source: 默认计划来源
    
    示例：
        >>> config = PlannerConfig(
        ...     default_risk_level=RiskLevel.LOW,
        ...     require_confirmation_on_medium_risk=True,
        ...     require_confirmation_on_high_risk=True,
        ...     max_plan_steps=10,
        ...     default_source="hybrid"
        ... )
    """
    
    default_risk_level: RiskLevel = RiskLevel.LOW
    require_confirmation_on_medium_risk: bool = True
    require_confirmation_on_high_risk: bool = True
    max_plan_steps: int = 10
    default_source: str = "hybrid"


# =============================================================================
# Planner 核心类
# =============================================================================

class Planner:
    """通用 Planner 类
    
    根据世界状态、用户意图、可用能力生成标准化 Plan 对象。
    
    属性：
        config: Planner 配置
        intent_to_action: 意图到 action 的映射字典
        action_to_tool_type: action 到 ToolCallType 的映射字典
    
    示例：
        >>> planner = Planner()
        >>> plan = planner.generate_plan(
        ...     world_snapshot={"temperature": 30},
        ...     user_intent="降温",
        ...     available_tools=["adjust_fan", "speak"]
        ... )
    """
    
    def __init__(
        self,
        config: PlannerConfig | None = None,
        intent_to_action: dict[str, str] | None = None,
        action_to_tool_type: dict[str, ToolCallType] | None = None,
    ) -> None:
        """初始化 Planner
        
        Args:
            config: Planner 配置，None 时使用默认配置
            intent_to_action: 自定义意图映射，None 时使用内置映射
            action_to_tool_type: 自定义 action 类型映射，None 时使用内置映射
        """
        self.config = config or PlannerConfig()
        self.intent_to_action = intent_to_action or INTENT_TO_ACTION.copy()
        self.action_to_tool_type = action_to_tool_type or ACTION_TO_TOOL_CALL_TYPE.copy()
        self._cached_normalized_snapshot: dict[str, Any] | None = None
        self._cached_snapshot_raw: dict[str, Any] | None = None
    
    def generate_plan(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        available_tools: list[str],
        domain_rules: dict[str, Any] | None = None,
        planner_mode: str = "hybrid",
    ) -> Plan:
        """生成执行计划
        
        根据输入信息生成标准化的 Plan 对象。
        
        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图描述
            available_tools: 可用能力列表
            domain_rules: 领域规则约束，None 表示无约束
            planner_mode: 规划模式，可选 "rule" / "llm" / "hybrid"
        
        Returns:
            标准化的 Plan 对象
        
        Raises:
            ValueError: 当 planner_mode 不合法
        """
        if not user_intent or not user_intent.strip():
            return self._generate_empty_intent_plan(world_snapshot, available_tools)
        
        mode = planner_mode.lower()
        if mode == "rule":
            return self._generate_rule_based_plan(
                world_snapshot, user_intent, available_tools, domain_rules
            )
        elif mode == "llm":
            return self._generate_llm_plan(
                world_snapshot, user_intent, available_tools, domain_rules
            )
        elif mode == "hybrid":
            return self._generate_hybrid_plan(
                world_snapshot, user_intent, available_tools, domain_rules
            )
        else:
            raise ValueError(f"无效的 planner_mode: {planner_mode}，有效值为: rule/llm/hybrid")
    
    def _generate_rule_based_plan(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        available_tools: list[str],
        domain_rules: dict[str, Any] | None,
    ) -> Plan:
        """基于规则生成 Plan
        
        内部方法，使用规则引擎生成计划。

        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图描述
            available_tools: 可用能力列表
            domain_rules: 领域规则约束
        
        Returns:
            Plan 对象
        """
        normalized = self._normalize_world_snapshot(world_snapshot)
        action = self._map_intent_to_action(user_intent)
        context = self._extract_recent_context(normalized)

        requires_confirm, confirm_msg = self._needs_confirmation(
            normalized, user_intent, action, domain_rules
        )
        risk_level = self._assess_risk(user_intent, action, normalized, domain_rules)

        plan_id = f"plan_{uuid.uuid4().hex[:8]}"

        if action is None:
            return self._generate_unknown_intent_plan(
                plan_id, user_intent, normalized, available_tools, risk_level, context
            )

        if not self._validate_tool_available(action, available_tools):
            return self._generate_tool_unavailable_plan(
                plan_id, user_intent, action, normalized, risk_level, context
            )

        return self._generate_executable_plan(
            plan_id, user_intent, action, normalized,
            risk_level, requires_confirm, confirm_msg, context
        )
    
    def _generate_llm_plan(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        available_tools: list[str],
        domain_rules: dict[str, Any] | None,
    ) -> Plan:
        """LLM 规划器实现
        
        内部方法，调用 LLM 生成计划。

        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图描述
            available_tools: 可用能力列表
            domain_rules: 领域规则约束
        
        Returns:
            Plan 对象
        """
        # 导入 LLM 客户端（延迟导入避免循环依赖）
        try:
            from llm_client import LLMPlannerClient
            llm_client = LLMPlannerClient()
        except ImportError:
            # LLM 客户端不可用，返回安全 fallback
            return self._generate_safe_fallback_plan(
                user_intent=user_intent,
                reason="llm_client_unavailable",
            )

        # 提取上下文（使用 normalize 后的结果）
        normalized = self._normalize_world_snapshot(world_snapshot)
        context = self._extract_recent_context(normalized)

        # 构造 Planner 配置
        planner_config = {
            "default_risk_level": self.config.default_risk_level.value,
            "require_confirmation_on_medium_risk": self.config.require_confirmation_on_medium_risk,
            "require_confirmation_on_high_risk": self.config.require_confirmation_on_high_risk,
            "max_plan_steps": self.config.max_plan_steps,
        }

        try:
            # 调用 LLM 生成 Plan JSON
            plan_json = llm_client.generate_plan_json(
                user_intent=user_intent,
                world_snapshot=normalized,
                available_tools=available_tools,
                domain_rules=domain_rules,
                planner_config=planner_config,
            )

            # 使用 Plan.from_dict() 转换
            plan = Plan.from_dict(plan_json)

            return plan

        except Exception as e:
            # LLM 调用失败，返回安全 fallback
            return self._generate_safe_fallback_plan(
                user_intent=user_intent,
                reason=f"llm_error: {e}",
            )
    
    def _generate_hybrid_plan(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        available_tools: list[str],
        domain_rules: dict[str, Any] | None,
    ) -> Plan:
        """混合模式生成 Plan
        
        内部方法，优先使用规则，简单情况直接生成，复杂情况调用 LLM。

        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图描述
            available_tools: 可用能力列表
            domain_rules: 领域规则约束
        
        Returns:
            Plan 对象
        """
        normalized = self._normalize_world_snapshot(world_snapshot)
        action = self._map_intent_to_action(user_intent)
        context = self._extract_recent_context(normalized)

        # 判断是否需要委托给 LLM
        if self._should_delegate_to_llm(
            normalized, user_intent, available_tools, domain_rules, action
        ):
            return self._generate_llm_plan(
                normalized, user_intent, available_tools, domain_rules
            )

        # 规则可处理，直接生成
        requires_confirm, confirm_msg = self._needs_confirmation(
            normalized, user_intent, action, domain_rules
        )
        risk_level = self._assess_risk(user_intent, action, normalized, domain_rules)

        plan_id = f"plan_{uuid.uuid4().hex[:8]}"

        if action is None:
            return self._generate_unknown_intent_plan(
                plan_id, user_intent, normalized, available_tools, risk_level, context
            )

        if not self._validate_tool_available(action, available_tools):
            return self._generate_tool_unavailable_plan(
                plan_id, user_intent, action, normalized, risk_level, context
            )

        return self._generate_executable_plan(
            plan_id, user_intent, action, normalized,
            risk_level, requires_confirm, confirm_msg, context
        )
    
    def _generate_executable_plan(
        self,
        plan_id: str,
        user_intent: str,
        action: str,
        world_snapshot: dict[str, Any],
        risk_level: RiskLevel,
        requires_confirm: bool,
        confirm_msg: str | None,
        context: dict[str, Any],
    ) -> Plan:
        """生成可执行的 Plan
        
        内部方法，生成包含具体执行步骤的 Plan。

        当 requires_confirm == True 时，生成 ASK_USER/SPEAK 询问步骤而非 EXECUTE 执行步骤，
        以确保 Executor 不会在用户确认前误执行风险动作。

        Args:
            plan_id: 计划 ID
            user_intent: 用户意图
            action: 动作名称
            world_snapshot: 世界状态（应传入 normalized 后的结果）
            risk_level: 风险等级
            requires_confirm: 是否需要确认
            confirm_msg: 确认消息
            context: 提取的上下文
        
        Returns:
            可执行的 Plan 对象
        """
        reasoning_parts = [
            f"用户意图: {user_intent}",
            f"匹配动作: {action}",
            f"风险等级: {risk_level.value}",
        ]
        if context:
            reasoning_parts.append(f"相关上下文: {context}")
        if requires_confirm and confirm_msg:
            reasoning_parts.append(f"需要确认: {confirm_msg}")

        plan = Plan(
            plan_id=plan_id,
            goal=user_intent,
            reasoning="; ".join(reasoning_parts),
            risk_level=risk_level,
            requires_confirmation=requires_confirm,
            confirmation_message=confirm_msg,
            status=PlanStatus.CREATED,
            source=self.config.default_source,
        )

        # 当需要确认时，生成询问/告知步骤而非执行步骤
        if requires_confirm:
            step = PlanStep(
                step_id=1,
                action="ask_user",
                tool_call_type=ToolCallType.ASK_USER,
                parameters={
                    "user_intent": user_intent,
                    "context": context,
                    "suggested_action": action,
                    "question": confirm_msg or f"您确定要执行 '{action}' 吗？",
                },
                description=f"询问用户确认: {action}",
                requires_confirmation=True,
                confirmation_message=confirm_msg,
            )
        else:
            # 不需要确认时，生成正常的执行步骤
            step = PlanStep(
                step_id=1,
                action=action,
                tool_call_type=self.action_to_tool_type.get(action, ToolCallType.EXECUTE),
                parameters={"user_intent": user_intent, "context": context},
                description=f"执行动作: {action}",
            )

        plan.add_step(step)

        return plan
    
    def _generate_tool_unavailable_plan(
        self,
        plan_id: str,
        user_intent: str,
        action: str,
        world_snapshot: dict[str, Any],
        risk_level: RiskLevel,
        context: dict[str, Any],
    ) -> Plan:
        """生成工具不可用时的 Plan
        
        内部方法，当请求的能力不在可用列表中时调用。
        
        Args:
            plan_id: 计划 ID
            user_intent: 用户意图
            action: 尝试的动作名称
            world_snapshot: 世界状态
            risk_level: 风险等级
            context: 提取的上下文
        
        Returns:
            提示能力不可用的 Plan 对象
        """
        plan = Plan(
            plan_id=plan_id,
            goal=f"处理不可用能力请求: {user_intent}",
            reasoning=(
                f"用户请求执行动作 '{action}'，但该能力当前不可用。"
                f"用户意图: {user_intent}"
            ),
            risk_level=RiskLevel.LOW,
            requires_confirmation=False,
            status=PlanStatus.CREATED,
            source=self.config.default_source,
        )
        
        step = PlanStep(
            step_id=1,
            action="speak",
            tool_call_type=ToolCallType.SPEAK,
            parameters={
                "message": f"抱歉，当前系统不支持 '{action}' 操作。"
                           f"您的请求 '{user_intent}' 无法完成。",
                "level": "info",
            },
            description=f"通知用户能力不可用: {action}",
        )
        plan.add_step(step)
        
        return plan
    
    def _generate_unknown_intent_plan(
        self,
        plan_id: str,
        user_intent: str,
        world_snapshot: dict[str, Any],
        available_tools: list[str],
        risk_level: RiskLevel,
        context: dict[str, Any],
    ) -> Plan:
        """生成意图不清晰时的 Plan
        
        内部方法，当无法从用户意图中识别出具体动作时调用。
        
        Args:
            plan_id: 计划 ID
            user_intent: 用户意图
            world_snapshot: 世界状态
            available_tools: 可用能力列表
            risk_level: 风险等级
            context: 提取的上下文
        
        Returns:
            需要进一步确认意图的 Plan 对象
        """
        plan = Plan(
            plan_id=plan_id,
            goal=f"澄清用户意图: {user_intent}",
            reasoning=(
                f"无法从用户意图 '{user_intent}' 中识别出具体动作。"
                f"可用能力: {available_tools}"
            ),
            risk_level=RiskLevel.LOW,
            requires_confirmation=False,
            status=PlanStatus.CREATED,
            source=self.config.default_source,
        )
        
        query_step = PlanStep(
            step_id=1,
            action="clarify_intent",
            tool_call_type=ToolCallType.ASK_USER,
            parameters={
                "original_intent": user_intent,
                "available_actions": available_tools,
            },
            description=f"询问用户确认具体意图: {user_intent}",
            requires_confirmation=True,
            confirmation_message=f"您是想要执行以下哪个操作？{', '.join(available_tools[:5])}",
        )
        plan.add_step(query_step)
        
        return plan
    
    def _generate_empty_intent_plan(
        self,
        world_snapshot: dict[str, Any],
        available_tools: list[str],
    ) -> Plan:
        """生成空意图的 Plan
        
        内部方法，当用户意图为空时调用。
        
        Args:
            world_snapshot: 世界状态
            available_tools: 可用能力列表
        
        Returns:
            提示输入意图的 Plan 对象
        """
        plan = Plan(
            plan_id=f"plan_{uuid.uuid4().hex[:8]}",
            goal="等待用户输入",
            reasoning="用户未提供有效意图",
            risk_level=RiskLevel.LOW,
            requires_confirmation=False,
            status=PlanStatus.CREATED,
            source=self.config.default_source,
        )
        
        step = PlanStep(
            step_id=1,
            action="speak",
            tool_call_type=ToolCallType.SPEAK,
            parameters={"message": "您好，请告诉我您想要执行的操作。"},
            description="等待用户输入有效意图",
        )
        plan.add_step(step)
        
        return plan
    
    def _needs_confirmation(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        action: str | None,
        domain_rules: dict[str, Any] | None,
    ) -> tuple[bool, str | None]:
        """判断是否需要用户确认
        
        根据世界状态、用户意图、动作类型和领域规则判断。
        优先使用 _normalize_world_snapshot() 转换后的结果。

        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图
            action: 识别的动作名称
            domain_rules: 领域规则
        
        Returns:
            (是否需要确认, 确认消息)
        """
        if action is None:
            return True, "无法识别您的意图，请确认您想要的操作。"

        normalized = self._normalize_world_snapshot(world_snapshot)
        if self._has_world_conflict(normalized, action, domain_rules):
            return True, f"检测到与当前状态的潜在冲突，请确认是否继续执行 '{action}'？"

        risk_level = self._assess_risk(user_intent, action, normalized, domain_rules)

        if risk_level == RiskLevel.HIGH and self.config.require_confirmation_on_high_risk:
            return True, f"该操作风险较高 '{action}'，是否确认执行？"

        if risk_level == RiskLevel.MEDIUM and self.config.require_confirmation_on_medium_risk:
            return True, f"执行操作 '{action}' 前，请确认。"

        return False, None
    
    def _has_world_conflict(
        self,
        world_snapshot: dict[str, Any],
        action: str,
        domain_rules: dict[str, Any] | None,
    ) -> bool:
        """检查世界状态是否存在冲突
        
        内部方法，检测可能导致执行冲突的状态。
        优先使用 _normalize_world_snapshot() 转换后的结果。

        Args:
            world_snapshot: 世界状态快照
            action: 动作名称
            domain_rules: 领域规则
        
        Returns:
            是否存在冲突
        """
        normalized = self._normalize_world_snapshot(world_snapshot)

        if normalized.get("actuator_status"):
            actuator = normalized.get("actuator_status", {})
            if isinstance(actuator, dict):
                for key in ["unavailable", "error", "offline"]:
                    if key in actuator and actuator[key]:
                        return True

        if normalized.get("recent_actions"):
            recent = normalized.get("recent_actions", [])
            if isinstance(recent, list) and len(recent) > 0:
                last_action = recent[-1] if recent else None
                if last_action and last_action.get("action") == action:
                    return True

        return False
    
    def _assess_risk(
        self,
        user_intent: str,
        action: str | None,
        world_snapshot: dict[str, Any],
        domain_rules: dict[str, Any] | None,
    ) -> RiskLevel:
        """评估风险等级
        
        内部方法，根据多个因素综合评估风险。
        优先使用 _normalize_world_snapshot() 转换后的结果。

        Args:
            user_intent: 用户意图
            action: 动作名称
            world_snapshot: 世界状态
            domain_rules: 领域规则
        
        Returns:
            风险等级
        """
        normalized = self._normalize_world_snapshot(world_snapshot)

        if action is None:
            return RiskLevel.LOW

        for keyword in HIGH_RISK_KEYWORDS:
            if keyword in user_intent:
                return RiskLevel.HIGH

        for keyword in MEDIUM_RISK_KEYWORDS:
            if keyword in user_intent:
                return RiskLevel.MEDIUM

        if domain_rules and "high_risk_actions" in domain_rules:
            if action in domain_rules["high_risk_actions"]:
                return RiskLevel.HIGH

        return self.config.default_risk_level
    
    def _should_delegate_to_llm(
        self,
        world_snapshot: dict[str, Any],
        user_intent: str,
        available_tools: list[str],
        domain_rules: dict[str, Any] | None,
        action: str | None,
    ) -> bool:
        """判断是否应该委托给 LLM
        
        内部方法，根据多个因素判断是否需要 LLM 介入。
        
        委托条件：
        1. action 无法明确映射（意图复杂）
        2. 世界状态存在冲突
        3. 需要结合领域规则做高层判断
        4. 多步骤规划需求
        5. 意图语义模糊但可理解
        
        Args:
            world_snapshot: 世界状态快照
            user_intent: 用户意图描述
            available_tools: 可用能力列表
            domain_rules: 领域规则约束
            action: 识别的动作（None 表示未识别）
        
        Returns:
            是否应该委托给 LLM
        """
        domain_rules = domain_rules or {}
        
        # 条件1：action 无法明确映射
        if action is None:
            # 检查是否有任何关键词匹配
            has_partial_match = False
            for keyword in self.intent_to_action.keys():
                if keyword in user_intent:
                    has_partial_match = True
                    break
            # 有部分匹配但无法确定具体 action，需要 LLM
            if has_partial_match:
                return True
        
        # 条件2：世界状态存在冲突
        if self._has_world_conflict(world_snapshot, action, domain_rules):
            # 检查冲突是否严重到需要 LLM
            if self._is_complex_conflict(world_snapshot, action):
                return True
        
        # 条件3：涉及领域规则限制
        if domain_rules:
            high_risk_actions = domain_rules.get("high_risk_actions", [])
            medium_risk_actions = domain_rules.get("medium_risk_actions", [])
            if action in high_risk_actions or action in medium_risk_actions:
                # 风险动作需要更复杂判断
                return True
        
        # 条件4：多步骤意图（包含连接词）
        multi_step_keywords = ["然后", "接下来", "之后", "再", "和", "以及", "并且", "再然后"]
        if any(kw in user_intent for kw in multi_step_keywords):
            return True
        
        # 条件5：意图包含多个动作词
        action_count = 0
        for keyword in self.intent_to_action.keys():
            if keyword in user_intent:
                action_count += 1
        if action_count > 1:
            return True
        
        # 条件6：意图不完整但包含上下文暗示
        incomplete_indicators = ["它", "那个", "这个", "刚才", "上次", "再"]
        if any(ind in user_intent for ind in incomplete_indicators):
            if action is None:
                return True
        
        return False
    
    def _is_complex_conflict(
        self,
        world_snapshot: dict[str, Any],
        action: str | None,
    ) -> bool:
        """判断是否为复杂冲突
        
        复杂冲突需要 LLM 介入判断。
        优先使用 _normalize_world_snapshot() 转换后的结果。

        Args:
            world_snapshot: 世界状态
            action: 动作名称
        
        Returns:
            是否为复杂冲突
        """
        normalized = self._normalize_world_snapshot(world_snapshot)

        # 检查 recent_actions 中是否有相同动作
        recent_actions = normalized.get("recent_actions", [])
        if isinstance(recent_actions, list) and len(recent_actions) > 0:
            last_action = recent_actions[-1] if recent_actions else {}
            if isinstance(last_action, dict) and last_action.get("action") == action:
                # 相同动作，检查时间间隔
                if "timestamp" in last_action:
                    time_diff = time.time() - last_action.get("timestamp", 0)
                    # 短时间内重复动作视为复杂冲突
                    if time_diff < 600:  # 10分钟内
                        return True

        # 检查是否有多个警告
        warnings = normalized.get("warnings", [])
        if isinstance(warnings, list) and len(warnings) > 1:
            return True

        return False
    
    def _generate_safe_fallback_plan(
        self,
        user_intent: str,
        reason: str = "",
    ) -> Plan:
        """生成安全的 fallback Plan
        
        当 LLM 调用失败时，返回一个安全的 fallback plan。
        优先使用 ASK_USER 或 SPEAK，不直接执行风险动作。
        
        Args:
            user_intent: 用户原始意图
            reason: 回退原因
        
        Returns:
            安全的 Plan 对象
        """
        plan = Plan(
            plan_id=f"plan_fallback_{uuid.uuid4().hex[:8]}",
            goal=f"无法处理的请求: {user_intent}",
            reasoning=f"LLM 处理失败或响应无效 ({reason})，返回安全 fallback",
            risk_level=RiskLevel.LOW,
            requires_confirmation=False,
            confirmation_message=None,
            status=PlanStatus.CREATED,
            source="llm_fallback",
            metadata={"fallback": True, "reason": reason},
        )
        
        step = PlanStep(
            step_id=1,
            action="ask_user",
            tool_call_type=ToolCallType.ASK_USER,
            parameters={
                "question": f"抱歉，我无法理解或处理您的请求: {user_intent}。请重新描述您的需求。",
            },
            preconditions={},
            fallback=None,
            status=StepStatus.PENDING,
            description="询问用户澄清意图",
            requires_confirmation=False,
            confirmation_message=None,
            metadata={"fallback_reason": reason},
        )
        plan.add_step(step)
        
        return plan
    
    def _validate_tool_available(self, tool_name: str, available_tools: list[str]) -> bool:
        """验证工具是否可用
        
        内部方法，检查请求的动作是否在可用能力列表中。
        
        Args:
            tool_name: 工具/动作名称
            available_tools: 可用能力列表
        
        Returns:
            是否可用
        """
        if not available_tools:
            return False
        return tool_name in available_tools
    
    def _normalize_world_snapshot(self, world_snapshot: dict[str, Any]) -> dict[str, Any]:
        """将真实 /world/snapshot 结构统一转换为 Planner 扁平上下文

        真实 world_snapshot 结构可能包含嵌套的 environment、system、entities 等字段，
        此方法将其转换为扁平的、易于访问的格式。

        对齐真实结构：
        - environment: temperature, humidity, light_level, warnings, errors
        - system: device_status, actuator_status, mode
        - entities: 与 actuator/device/recent action 相关的信息

        Args:
            world_snapshot: 原始世界状态快照

        Returns:
            扁平的 normalized dict
        """
        # 如果快照没有变化，使用缓存
        if self._cached_snapshot_raw is world_snapshot and self._cached_normalized_snapshot is not None:
            return self._cached_normalized_snapshot

        normalized: dict[str, Any] = {
            "temperature": None,
            "humidity": None,
            "light_level": None,
            "warnings": [],
            "errors": [],
            "device_status": {},
            "actuator_status": {},
            "mode": None,
            "recent_actions": [],
            "raw_entities": {},
        }

        # 处理 environment 字段（常见于 /world/snapshot）
        environment = world_snapshot.get("environment", {})
        if isinstance(environment, dict):
            normalized["temperature"] = environment.get("temperature") or environment.get("temperature_ambient")
            normalized["humidity"] = environment.get("humidity") or environment.get("humidity_ambient")
            normalized["light_level"] = environment.get("light_level") or environment.get("illuminance")
            normalized["warnings"] = environment.get("warnings", [])
            normalized["errors"] = environment.get("errors", [])

        # 处理 system 字段
        system = world_snapshot.get("system", {})
        if isinstance(system, dict):
            normalized["device_status"] = system.get("device_status", {})
            normalized["actuator_status"] = system.get("actuator_status", {})
            normalized["mode"] = system.get("mode") or system.get("operating_mode")

        # 直接字段（已经是扁平格式）
        for key in ["temperature", "humidity", "light_level", "warnings", "errors",
                    "device_status", "actuator_status", "mode"]:
            if key not in environment and key in world_snapshot:
                if normalized.get(key) is None or normalized.get(key) == [] or normalized.get(key) == {}:
                    normalized[key] = world_snapshot.get(key)

        # 处理 entities 字段（提取与 actuator/device 相关信息）
        entities = world_snapshot.get("entities", {})
        if isinstance(entities, dict):
            actuators = entities.get("actuator", [])
            devices = entities.get("device", [])
            if isinstance(actuators, list):
                for act in actuators:
                    if isinstance(act, dict):
                        device_id = act.get("entity_id") or act.get("device_id", "unknown")
                        normalized["raw_entities"][f"actuator_{device_id}"] = act
            if isinstance(devices, list):
                for dev in devices:
                    if isinstance(dev, dict):
                        device_id = dev.get("entity_id") or dev.get("device_id", "unknown")
                        normalized["raw_entities"][f"device_{device_id}"] = dev

        # 处理 recent_actions（可能在顶层或 recent_context 下）
        recent_context = world_snapshot.get("recent_context", {}) or world_snapshot.get("recent_actions", [])
        if isinstance(recent_context, dict):
            normalized["recent_actions"] = recent_context.get("recent_actions", [])
        elif isinstance(recent_context, list):
            normalized["recent_actions"] = recent_context

        # 如果顶层直接有 recent_actions
        if not normalized["recent_actions"] and "recent_actions" in world_snapshot:
            normalized["recent_actions"] = world_snapshot.get("recent_actions", [])

        # 缓存结果
        self._cached_snapshot_raw = world_snapshot
        self._cached_normalized_snapshot = normalized

        return normalized

    def _extract_recent_context(self, world_snapshot: dict[str, Any]) -> dict[str, Any]:
        """提取与当前决策相关的上下文
        
        内部方法，从世界状态中提取关键信息。
        优先使用 _normalize_world_snapshot() 转换后的结果。
        
        Args:
            world_snapshot: 世界状态快照
        
        Returns:
            相关的上下文字典
        """
        normalized = self._normalize_world_snapshot(world_snapshot)

        context_keys = [
            "temperature",
            "humidity",
            "light_level",
            "device_status",
            "actuator_status",
            "recent_actions",
            "warnings",
            "errors",
        ]

        context = {}
        for key in context_keys:
            if key in normalized and normalized[key] is not None:
                context[key] = normalized[key]

        if normalized.get("raw_entities"):
            context["raw_entities"] = normalized["raw_entities"]

        return context
    
    def _map_intent_to_action(self, user_intent: str) -> str | None:
        """将用户意图映射到 action
        
        内部方法，通过关键词匹配查找对应的 action。
        按关键词长度从长到短排序后再匹配，避免短关键词优先匹配导致误判。

        Args:
            user_intent: 用户意图描述
        
        Returns:
            匹配到的 action 名称，未匹配到返回 None
        """
        if not user_intent:
            return None

        intent_lower = user_intent.lower()

        # 按关键词长度从长到短排序，避免"喂"比"喂食"先匹配
        for keyword in sorted(self.intent_to_action.keys(), key=len, reverse=True):
            if keyword in intent_lower or keyword in user_intent:
                return self.intent_to_action[keyword]

        return None
    
    def register_intent_mapping(self, keyword: str, action: str) -> None:
        """注册新的意图映射
        
        公共方法，运行时添加新的意图到 action 的映射。
        
        Args:
            keyword: 意图关键词
            action: 对应的 action 名称
        """
        self.intent_to_action[keyword] = action
    
    def register_tool_type(self, action: str, tool_type: ToolCallType) -> None:
        """注册 action 到 ToolCallType 的映射
        
        公共方法，运行时添加新的映射。
        
        Args:
            action: action 名称
            tool_type: 对应的 ToolCallType
        """
        self.action_to_tool_type[action] = tool_type


# =============================================================================
# 主程序入口（测试示例）
# =============================================================================

if __name__ == "__main__":
    import json
    
    print("=" * 70)
    print("Planner 测试演示")
    print("=" * 70)
    
    # 初始化 Planner
    planner = Planner()
    
    # -------------------------------------------------------------------------
    # 构造测试数据
    # -------------------------------------------------------------------------
    world_snapshot = {
        "temperature": 32.5,
        "humidity": 70,
        "actuator_status": {
            "fan": "running",
            "light": "on",
        },
        "recent_actions": [
            {"action": "feed", "timestamp": time.time() - 300},
        ],
        "warnings": [],
        "errors": [],
    }
    
    available_tools = [
        "feed",
        "adjust_fan",
        "speak",
        "query",
        "move",
        "inspect",
    ]
    
    # -------------------------------------------------------------------------
    # 场景 1：直接可执行（工具存在 + 风险低）
    # -------------------------------------------------------------------------
    print("\n[场景 1] 直接可执行")
    print("-" * 40)
    
    user_intent_1 = "打开风扇降温"
    plan_1 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_1,
        available_tools=available_tools,
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_1}")
    print(f"Plan ID: {plan_1.plan_id}")
    print(f"目标: {plan_1.goal}")
    print(f"风险等级: {plan_1.risk_level.value}")
    print(f"需要确认: {plan_1.requires_confirmation}")
    print(f"步骤数: {len(plan_1.steps)}")
    if plan_1.steps:
        step = plan_1.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
    
    # -------------------------------------------------------------------------
    # 场景 2：需要确认（风险中高或世界冲突）
    # -------------------------------------------------------------------------
    print("\n[场景 2] 需要确认")
    print("-" * 40)
    
    world_snapshot_conflict = {
        "temperature": 32.5,
        "actuator_status": {
            "fan": "error",
        },
        "recent_actions": [
            {"action": "adjust_fan", "timestamp": time.time() - 60},
        ],
    }
    
    user_intent_2 = "再次降温"
    plan_2 = planner.generate_plan(
        world_snapshot=world_snapshot_conflict,
        user_intent=user_intent_2,
        available_tools=available_tools,
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_2}")
    print(f"Plan ID: {plan_2.plan_id}")
    print(f"需要确认: {plan_2.requires_confirmation}")
    print(f"确认消息: {plan_2.confirmation_message}")
    print(f"步骤数: {len(plan_2.steps)}")
    if plan_2.steps:
        step = plan_2.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
        print(f"     需要确认: {step.requires_confirmation}")
    
    # -------------------------------------------------------------------------
    # 场景 3：工具不存在
    # -------------------------------------------------------------------------
    print("\n[场景 3] 工具不存在")
    print("-" * 40)
    
    user_intent_3 = "打开空调"
    available_tools_3 = ["fan", "light", "speak"]
    
    plan_3 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_3,
        available_tools=available_tools_3,
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_3}")
    print(f"可用工具: {available_tools_3}")
    print(f"Plan ID: {plan_3.plan_id}")
    print(f"目标: {plan_3.goal}")
    print(f"步骤数: {len(plan_3.steps)}")
    if plan_3.steps:
        step = plan_3.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
        print(f"     消息: {step.parameters.get('message', '')}")
    
    # -------------------------------------------------------------------------
    # 场景 4：LLM 模式（真实实现）
    # -------------------------------------------------------------------------
    print("\n[场景 4] LLM 模式")
    print("-" * 40)
    
    user_intent_4 = "打开风扇降温"
    plan_4 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_4,
        available_tools=available_tools,
        planner_mode="llm",
    )
    
    print(f"用户意图: {user_intent_4}")
    print(f"Plan ID: {plan_4.plan_id}")
    print(f"来源: {plan_4.source}")
    print(f"目标: {plan_4.goal}")
    print(f"风险等级: {plan_4.risk_level.value}")
    print(f"步骤数: {len(plan_4.steps)}")
    if plan_4.steps:
        step = plan_4.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
    
    # -------------------------------------------------------------------------
    # 场景 5：LLM 模式 - 喂食询问
    # -------------------------------------------------------------------------
    print("\n[场景 5] LLM 模式 - 喂食询问")
    print("-" * 40)
    
    user_intent_5 = "再喂一次"
    world_snapshot_recent_feed = {
        "temperature": 28,
        "last_feed_time": "2026-04-10T10:00:00",
        "recent_actions": [
            {"action": "feed", "timestamp": time.time() - 300},
        ],
    }
    plan_5 = planner.generate_plan(
        world_snapshot=world_snapshot_recent_feed,
        user_intent=user_intent_5,
        available_tools=["feed", "speak", "query"],
        planner_mode="llm",
    )
    
    print(f"用户意图: {user_intent_5}")
    print(f"Plan ID: {plan_5.plan_id}")
    print(f"来源: {plan_5.source}")
    print(f"目标: {plan_5.goal}")
    print(f"风险等级: {plan_5.risk_level.value}")
    print(f"步骤数: {len(plan_5.steps)}")
    if plan_5.steps:
        step = plan_5.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
        if step.tool_call_type == ToolCallType.ASK_USER:
            print(f"     询问: {step.parameters.get('question', 'N/A')}")
    
    # -------------------------------------------------------------------------
    # 场景 6：Hybrid 模式 - 简单场景走规则
    # -------------------------------------------------------------------------
    print("\n[场景 6] Hybrid 模式 - 简单场景")
    print("-" * 40)
    
    user_intent_6 = "查询温度"
    plan_6 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_6,
        available_tools=available_tools,
        planner_mode="hybrid",
    )
    
    print(f"用户意图: {user_intent_6}")
    print(f"Plan ID: {plan_6.plan_id}")
    print(f"来源: {plan_6.source}")
    print(f"风险等级: {plan_6.risk_level.value}")
    
    # -------------------------------------------------------------------------
    # 场景 7：Hybrid 模式 - 复杂场景走 LLM
    # -------------------------------------------------------------------------
    print("\n[场景 7] Hybrid 模式 - 复杂场景")
    print("-" * 40)
    
    user_intent_7 = "先降温然后喂食"  # 多步骤意图
    plan_7 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_7,
        available_tools=available_tools,
        planner_mode="hybrid",
    )
    
    print(f"用户意图: {user_intent_7}")
    print(f"Plan ID: {plan_7.plan_id}")
    print(f"来源: {plan_7.source}")
    print(f"目标: {plan_7.goal}")
    print(f"风险等级: {plan_7.risk_level.value}")
    print(f"步骤数: {len(plan_7.steps)}")
    for step in plan_7.steps:
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
    
    # -------------------------------------------------------------------------
    # 场景 8：意图不清晰 - 走规则未知意图
    # -------------------------------------------------------------------------
    print("\n[场景 8] 意图不清晰")
    print("-" * 40)
    
    user_intent_8 = "随便弄一下"
    plan_8 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_8,
        available_tools=available_tools,
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_8}")
    print(f"Plan ID: {plan_8.plan_id}")
    print(f"目标: {plan_8.goal}")
    print(f"步骤数: {len(plan_8.steps)}")
    if plan_8.steps:
        step = plan_8.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
    
    # -------------------------------------------------------------------------
    # 场景 9：工具不存在
    # -------------------------------------------------------------------------
    print("\n[场景 9] 工具不存在")
    print("-" * 40)
    
    user_intent_9 = "打开空调"
    available_tools_9 = ["fan", "light", "speak"]
    
    plan_9 = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_9,
        available_tools=available_tools_9,
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_9}")
    print(f"可用工具: {available_tools_9}")
    print(f"Plan ID: {plan_9.plan_id}")
    print(f"目标: {plan_9.goal}")
    print(f"步骤数: {len(plan_9.steps)}")
    if plan_9.steps:
        step = plan_9.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
        print(f"     消息: {step.parameters.get('message', '')}")
    
    # -------------------------------------------------------------------------
    # 场景 10：LLM 工具不存在场景
    # -------------------------------------------------------------------------
    print("\n[场景 10] LLM 模式 - 工具不存在")
    print("-" * 40)
    
    user_intent_10 = "打开空调"
    plan_10 = planner.generate_plan(
        world_snapshot={"temperature": 30},
        user_intent=user_intent_10,
        available_tools=["adjust_fan", "speak"],
        planner_mode="llm",
    )
    
    print(f"用户意图: {user_intent_10}")
    print(f"Plan ID: {plan_10.plan_id}")
    print(f"来源: {plan_10.source}")
    print(f"目标: {plan_10.goal}")
    print(f"步骤数: {len(plan_10.steps)}")
    if plan_10.steps:
        step = plan_10.steps[0]
        print(f"  → Step {step.step_id}: {step.action} ({step.tool_call_type.value})")
    
    # -------------------------------------------------------------------------
    # 输出完整 JSON
    # -------------------------------------------------------------------------
    print("\n[完整 JSON 输出]")
    print("-" * 40)
    
    print("\n场景 1 Plan JSON (rule 模式):")
    print(json.dumps(plan_1.to_dict(), indent=2, ensure_ascii=False))
    
    print("\n场景 4 Plan JSON (llm 模式):")
    print(json.dumps(plan_4.to_dict(), indent=2, ensure_ascii=False))
    
    # -------------------------------------------------------------------------
    # 测试自定义意图映射
    # -------------------------------------------------------------------------
    print("\n[自定义意图映射测试]")
    print("-" * 40)
    
    planner.register_intent_mapping("打扫", "clean")
    planner.register_tool_type("clean", ToolCallType.EXECUTE)
    
    user_intent_custom = "帮我打扫一下"
    plan_custom = planner.generate_plan(
        world_snapshot=world_snapshot,
        user_intent=user_intent_custom,
        available_tools=["clean", "speak"],
        planner_mode="rule",
    )
    
    print(f"用户意图: {user_intent_custom}")
    print(f"识别动作: {plan_custom.steps[0].action if plan_custom.steps else 'None'}")
    
    print("\n" + "=" * 70)
    print("Planner 测试演示完成")
    print("=" * 70)