""" 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)