AI_RDKS100P_Robot/brain/prompt_manager.py

"""
prompt_manager.py - Planner Prompt 构造器

该模块负责构造适合 LLM Planner 的 Prompt，使 LLM 能够生成标准化的 Plan JSON。

职责边界：
    - 构造高质量 Prompt
    - 管理 system instruction
    - 不调用 LLM
    - 不执行 capability

设计原则：
    - Prompt 必须清晰、无歧义
    - 必须明确告诉 LLM 角色和约束
    - 必须提供完整的输出格式规范
    - 必须考虑安全边界

使用方式：
    from prompt_manager import PlannerPromptManager
    
    manager = PlannerPromptManager()
    prompt = manager.build_plan_prompt(
        user_intent="降温",
        world_snapshot={"temperature": 32},
        available_tools=["adjust_fan", "speak"],
        domain_rules={"high_risk_actions": ["turn_off"]},
    )
"""

from __future__ import annotations

import json
from typing import Any


# =============================================================================
# Prompt 模板常量
# =============================================================================

# Planner System Instruction
PLANNER_SYSTEM_INSTRUCTION = """你是一个专业的 AI Agent Planner。

你的职责是根据用户意图和当前世界状态，生成标准化的行动计划（Plan）。

## 核心约束

1. 只生成 Plan，不执行动作
   - 你只输出 Plan JSON，不调用任何能力
   - 实际执行由 Executor 负责

2. 只能使用给定的能力
   - 必须从 available_tools 列表中选择
   - 禁止虚构或使用未提供的能力

3. 必须输出严格 JSON
   - 输出必须是一个可被 Plan.from_dict() 和 PlanStep.from_dict() 解析的标准 JSON 对象
   - 禁止输出 markdown
   - 禁止输出解释文字
   - 禁止输出额外前后缀
   - 禁止缺失必需字段
   - 禁止使用未定义字段替代标准字段

4. 安全优先原则
   - 高风险动作必须设置 requires_confirmation=true
   - 不确定时优先 ask_user 或 speak
   - 禁止直接执行危险动作

5. 世界状态冲突处理
   - 当世界状态显示可能存在冲突时
   - 优先询问用户确认（ask_user）
   - 不要自动执行可能重复或有风险的动作

## 工具类型说明

- execute: 执行具体动作
- ask_user: 询问用户确认
- speak: 语音/文本输出
- query_world: 查询世界状态
- query_knowledge: 查询知识库
- noop: 空操作

## 风险等级

- low: 低风险，可自动执行
- medium: 中风险，建议确认
- high: 高风险，必须确认

## 输出格式要求

输出 JSON 必须包含以下字段：

Plan 顶层字段：
- plan_id: string (自动生成，唯一标识)
- goal: string (计划目标描述)
- reasoning: string (推理过程说明)
- risk_level: "low" | "medium" | "high"
- requires_confirmation: true | false
- confirmation_message: string | null
- steps: array (步骤数组)
- status: "created" | "pending" | "executing" | "completed" | "failed"
- source: string (计划来源，如 "llm"）
- metadata: object (元数据)

每个 step 必须包含以下字段：
- step_id: number (步骤序号，从 1 开始)
- action: string (能力名称)
- tool_call_type: "execute" | "ask_user" | "speak" | "query_world" | "query_knowledge" | "noop"
- parameters: object (执行参数)
- preconditions: object (前置条件)
- fallback: null | object (失败回退)
- status: "pending" | "executing" | "completed" | "failed" | "skipped"
- description: string (步骤描述)
- requires_confirmation: true | false
- confirmation_message: string | null
- metadata: object (步骤元数据)

重要：直接输出 JSON 对象，不要包裹在任何代码块或前后缀中。"""
"""

# 简单场景 Prompt 模板
SIMPLE_SCENARIO_TEMPLATE = """## 用户意图
{user_intent}

## 当前世界状态
{world_snapshot}

## 可用能力
{available_tools}

{domain_rules_section}

{recent_context_section}

请根据以上信息，生成一个标准 Plan JSON。"""


# 无可用能力时的 Prompt
NO_TOOLS_TEMPLATE = """## 用户意图
{user_intent}

## 当前世界状态
{world_snapshot}

## 警告：没有可用能力

你的系统目前没有任何可用能力。
请生成一个 SPEAK plan，告知用户当前无法满足其请求。

请生成一个标准 Plan JSON。"""


# 意图不明确时的 Prompt
UNCLEAR_INTENT_TEMPLATE = """## 用户意图
{user_intent}

## 当前世界状态
{world_snapshot}

## 可用能力
{available_tools}

## 警告：用户意图不明确

用户请求不够明确，系统无法确定具体行动。
请生成一个 ASK_USER plan，询问用户澄清其意图。

请生成一个标准 Plan JSON。"""


# =============================================================================
# Planner Prompt 管理器
# =============================================================================

class PlannerPromptManager:
    """Planner Prompt 管理器
    
    负责构造适合 LLM Planner 的 Prompt。
    
    属性：
        system_instruction: 系统指令
        include_reasoning: 是否在输出中包含推理过程
    
    示例：
        >>> manager = PlannerPromptManager()
        >>> prompt = manager.build_plan_prompt(
        ...     user_intent="降温",
        ...     world_snapshot={"temperature": 32},
        ...     available_tools=["adjust_fan"],
        ... )
    """
    
    def __init__(
        self,
        system_instruction: str | None = None,
        include_reasoning: bool = True,
    ):
        """初始化 Prompt 管理器
        
        Args:
            system_instruction: 自定义系统指令，None 则使用默认指令
            include_reasoning: 是否包含推理过程
        """
        self.system_instruction = system_instruction or PLANNER_SYSTEM_INSTRUCTION
        self.include_reasoning = include_reasoning
    
    def build_plan_prompt(
        self,
        user_intent: str,
        world_snapshot: dict[str, Any],
        available_tools: list[str],
        domain_rules: dict[str, Any] | None = None,
        recent_context: dict[str, Any] | None = None,
    ) -> str:
        """构建 Plan Prompt
        
        主要入口方法。根据输入构建完整的 prompt。

        设计原则：
        - 专注于 prompt 构造，不做复杂的业务判断
        - 复杂场景判断由 planner.py 负责
        - 只保留必要的轻量分支：
          1. 无可用工具
          2. 意图明显不清晰
          3. 默认普通 prompt

        Args:
            user_intent: 用户意图
            world_snapshot: 世界状态快照
            available_tools: 可用能力列表
            domain_rules: 领域规则
            recent_context: 最近上下文（传递给 LLM 参考）
        
        Returns:
            完整的 prompt 字符串
        """
        domain_rules = domain_rules or {}
        recent_context = recent_context or {}

        # 轻量判断分支
        if not available_tools:
            scenario_prompt = self._build_no_tools_prompt(user_intent, world_snapshot)
        elif self._is_unclear_intent_simple(user_intent):
            scenario_prompt = self._build_unclear_intent_prompt(
                user_intent, world_snapshot, available_tools
            )
        else:
            scenario_prompt = self._build_default_prompt(
                user_intent, world_snapshot, available_tools, domain_rules, recent_context
            )

        # 组合完整 prompt
        return f"{self.system_instruction}\n\n{scenario_prompt}"

    def _is_unclear_intent_simple(self, user_intent: str) -> bool:
        """简单判断意图是否不清晰
        
        轻量方法，只做最基础的检查。
        复杂判断由 planner.py 负责。
        
        Args:
            user_intent: 用户意图
        
        Returns:
            意图是否不清晰
        """
        # 意图过短
        if len(user_intent.strip()) < 2:
            return True
        return False
    
    def build_system_instruction(self) -> str:
        """获取系统指令
        
        Returns:
            系统指令字符串
        """
        return self.system_instruction

    def _build_default_prompt(
        self,
        user_intent: str,
        world_snapshot: dict[str, Any],
        available_tools: list[str],
        domain_rules: dict[str, Any],
        recent_context: dict[str, Any],
    ) -> str:
        """构建默认 Prompt
        
        统一处理大多数场景的 prompt 构造。
        
        Args:
            user_intent: 用户意图
            world_snapshot: 世界状态
            available_tools: 可用能力
            domain_rules: 领域规则
            recent_context: 最近上下文
        
        Returns:
            场景 prompt 字符串
        """
        # 格式化世界状态
        if isinstance(world_snapshot, dict):
            world_str = json.dumps(world_snapshot, ensure_ascii=False, indent=2)
        else:
            world_str = str(world_snapshot)

        # 格式化可用工具
        tools_str = ", ".join(f'"{t}"' for t in available_tools)

        # 格式化领域规则
        if domain_rules:
            rules_str = f"\n\n## 领域规则\n{json.dumps(domain_rules, ensure_ascii=False, indent=2)}"
        else:
            rules_str = ""

        # 格式化最近上下文
        if recent_context:
            context_str = f"\n\n## 历史上下文\n{json.dumps(recent_context, ensure_ascii=False, indent=2)}"
        else:
            context_str = ""

        return SIMPLE_SCENARIO_TEMPLATE.format(
            user_intent=user_intent,
            world_snapshot=world_str,
            available_tools=tools_str,
            domain_rules_section=rules_str,
            recent_context_section=context_str,
        )
    
    def _build_no_tools_prompt(
        self,
        user_intent: str,
        world_snapshot: dict[str, Any],
    ) -> str:
        """构建无可用工具时的 Prompt"""
        if isinstance(world_snapshot, dict):
            world_str = json.dumps(world_snapshot, ensure_ascii=False, indent=2)
        else:
            world_str = str(world_snapshot)
        
        return NO_TOOLS_TEMPLATE.format(
            user_intent=user_intent,
            world_snapshot=world_str,
        )
    
    def _build_unclear_intent_prompt(
        self,
        user_intent: str,
        world_snapshot: dict[str, Any],
        available_tools: list[str],
    ) -> str:
        """构建意图不明确时的 Prompt"""
        if isinstance(world_snapshot, dict):
            world_str = json.dumps(world_snapshot, ensure_ascii=False, indent=2)
        else:
            world_str = str(world_snapshot)
        
        tools_str = ", ".join(f'"{t}"' for t in available_tools)
        
        return UNCLEAR_INTENT_TEMPLATE.format(
            user_intent=user_intent,
            world_snapshot=world_str,
            available_tools=tools_str,
        )


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

if __name__ == "__main__":
    print("=" * 70)
    print("Planner Prompt Manager 测试")
    print("=" * 70)
    
    manager = PlannerPromptManager()
    
    # 测试 1：简单场景
    print("\n[测试 1] 简单场景 - 降温")
    print("-" * 40)
    prompt = manager.build_plan_prompt(
        user_intent="打开风扇降温",
        world_snapshot={"temperature": 32, "humidity": 70},
        available_tools=["adjust_fan", "speak", "query"],
        domain_rules={"high_risk_actions": ["turn_off"]},
    )
    print(f"Prompt 长度: {len(prompt)} 字符")
    print(f"包含 system instruction: {'AI Agent Planner' in prompt}")
    print(f"包含 JSON 格式: {'plan_id' in prompt}")
    
    # 测试 2：意图不明确
    print("\n[测试 2] 意图不明确")
    print("-" * 40)
    prompt = manager.build_plan_prompt(
        user_intent="那个事情",
        world_snapshot={"status": "normal"},
        available_tools=["query", "speak"],
    )
    print(f"Prompt 长度: {len(prompt)} 字符")
    print(f"包含 ASK_USER 提示: {'询问用户' in prompt or 'ask_user' in prompt.lower()}")
    
    # 测试 3：无可用工具
    print("\n[测试 3] 无可用工具")
    print("-" * 40)
    prompt = manager.build_plan_prompt(
        user_intent="打开空调",
        world_snapshot={"temperature": 30},
        available_tools=[],
    )
    print(f"Prompt 长度: {len(prompt)} 字符")
    print(f"包含无可用能力提示: {'没有可用能力' in prompt}")
    
    # 测试 4：正常意图（含历史上下文）
    print("\n[测试 4] 正常意图")
    print("-" * 40)
    prompt = manager.build_plan_prompt(
        user_intent="先降温然后喂食",
        world_snapshot={"temperature": 32, "hunger_level": 80},
        available_tools=["adjust_fan", "feed", "speak"],
        domain_rules={"high_risk_actions": ["turn_off"], "medium_risk_actions": ["adjust"]},
        recent_context={"recent_actions": ["adjust_fan"]},
    )
    print(f"Prompt 长度: {len(prompt)} 字符")
    print(f"包含历史上下文: {'历史上下文' in prompt}")

    # 测试 5：打印完整 prompt 示例
    print("\n[测试 5] 完整 Prompt 示例")
    print("-" * 40)
    prompt = manager.build_plan_prompt(
        user_intent="打开风扇降温",
        world_snapshot={"temperature": 32, "humidity": 70},
        available_tools=["adjust_fan", "speak"],
        domain_rules={"high_risk_actions": ["turn_off"]},
    )
    # 只打印前 500 字符
    print(prompt[:500])
    print("...")

    # 测试 6：验证 system instruction 中没有 markdown 代码块模板
    print("\n[测试 6] 验证 JSON 输出约束")
    print("-" * 40)
    system_instr = manager.build_system_instruction()
    print(f"包含 markdown 代码块: {'```' in system_instr}")
    print(f"包含禁止输出 markdown: {'禁止输出 markdown' in system_instr}")
    print(f"包含禁止解释文字: {'禁止输出解释文字' in system_instr}")

    print("\n" + "=" * 70)
    print("Planner Prompt Manager 测试完成")
    print("=" * 70)