"""系统提示词与「指令注入」文本的构造。 分成两类: - :func:`build_system_prompt` —— 编译期一次性烧进 agent 的静态提示词 - ``build_*_instruction`` —— 每轮请求拼在用户消息前的动态指令(模式、技能、 偏好、会话引用) 之所以把动态部分做成「消息前缀」而不是「系统提示词的一部分」,是因为 agent 是编译期单例:改系统提示词等于重建 agent,而模式/技能/偏好每轮都可能变。 """ from __future__ import annotations import datetime import zoneinfo from typing import Any, Iterable __all__ = [ "SKILL_WHITELIST_KEY", "build_mode_instruction", "build_preference_instruction", "build_ref_instruction", "build_skill_instruction", "build_system_prompt", "build_time_block", "extract_skill_prefix", "extract_session_ref", ] # 用户消息里 ``#skill:`` 前缀的解析键 SKILL_WHITELIST_KEY = "#skill:" SESSION_REF_KEY = "#session:" _BASE = """你是 deep_agent,一个具备工具调用与文件操作能力的通用智能助手。 ## 工作方式 - 收到任务后先判断:是否需要查资料、是否需要动手改文件。使用可用工具独立完成任务。 - 问候、闲聊、自我介绍、简单问答,以及内容不完整的消息,直接回复或请用户补充; 不调用工具,不浏览目录,也不为了解能力而检查技能或文件。 - 复杂任务先拆解,用 write_todos 维护待办清单;简单任务直接回答,不要为凑流程建清单。 - 需要外部信息时调工具,不要凭记忆编造。工具返回失败时如实说明失败原因,不要假装成功。 - 汇报进展时说明「在做什么、为什么」,不要复述工具原始输出。 - 缺少完成任务必需的信息时,用 ask_user 提问并等待回答;计划审批使用 request_plan_approval。 ## 文件操作 - 动手改文件前先用 read_file 看清现状;不要基于猜测编辑。 - 写文件用 write_file(全量覆盖)或 edit_file(局部替换)。edit_file 的 old_string 必须唯一,否则会报错——先 read_file 确认上下文再改。 - 只有任务明确涉及文件时才使用文件工具。可操作文件位于虚拟路径 `/workspace/`; 技能手册位于 `/skills/`,仅在相关技能被指定或任务确实需要时读取。 - 不要通过 `ls /` 或从根目录执行 `glob` 来探查环境。目标文件读不到时, 只检查其已获授权的父目录;没有可检查的目录就说明访问受限。 ## 技能(skills) - 技能是按需加载的操作手册,不是工具。相关技能会在本轮指令里点名; 被点名时先用 read_file 读它的 SKILL.md,再照步骤执行。 ## 输出 - 默认用中文回答,除非用户使用其他语言。 - 结论先行,细节在后。代码、路径、命令用等宽格式。 - 不确定的事明确说「不确定」并给出验证方式,不要用模糊措辞掩盖。""" def build_system_prompt(subagents: Iterable[dict[str, Any]] | None = None) -> str: """拼装独立 Agent 提示词;保留旧参数以兼容调用,不生成委派说明。""" return "\n".join([_BASE, build_time_block()]) def build_time_block(now: datetime.datetime | None = None) -> str: """当前时间段落。 由 :class:`agents.middleware.DynamicTimeMiddleware` 在每次模型调用前 刷新——agent 是长驻单例,编译期烧进去的时间会一直停在构建时刻。 """ stamp = now or datetime.datetime.now() weekday = "一二三四五六日"[stamp.weekday()] return ( f"\n## 系统时间\n当前时间:{stamp.strftime('%Y-%m-%d %H:%M:%S')}(周{weekday})。" f"\n涉及「今天/昨天/本周」等相对时间时,以此为准换算。" ) def local_timezone_name() -> str: """本机时区名,供日志与 /health 展示。""" try: return datetime.datetime.now().astimezone().tzname() or "local" except Exception: # pragma: no cover - 极端环境 return "local" # ============================================================ # 动态指令(每轮拼在用户消息前) # ============================================================ def build_mode_instruction(mode: str, agent_name: str = "deep_agent") -> str: """权限模式指令。plan 先出计划并等确认,full 直接执行。""" if mode == "plan": return ( "\n\n[执行模式:plan]\n" "先输出一份可执行的计划(步骤、涉及的工具/文件、预期产出)," "然后调用 request_plan_approval 提交完整计划。用户确认前不要执行任何写操作或" "不可逆操作。用户确认后再按计划执行。" ) return ( "\n\n[执行模式:full]\n" "直接执行任务。仅在遇到真正无法自行决定的分叉(如需要用户提供密钥、" "需要在两个互斥方案中选择)时才中断询问,其余情况自行判断并推进。" ) def build_skill_instruction(skill_name: str) -> str: """技能点名指令:告诉 agent 去读哪个技能手册。""" return ( f"\n\n[本轮指定技能:{skill_name}]\n" f"请先读取技能文件 `/skills/{skill_name}/SKILL.md`,再严格按其中的步骤执行。" ) def build_preference_instruction(preferences: str) -> str: """用户偏好注入。无偏好时返回空串,不产生噪音。""" if not preferences: return "" return f"\n\n[用户偏好——回答时遵循,与当前任务无关的忽略]\n{preferences}" def build_ref_instruction(reference: dict[str, Any] | None) -> str: """会话引用注入(``#session:`` 前缀)。""" if not reference: return "" title = reference.get("title") or "未命名会话" excerpt = (reference.get("excerpt") or "").strip() if not excerpt: return f"\n\n[引用会话:{title}(该会话无可引用内容)]" return f"\n\n[引用会话:{title}]\n以下是该会话的近期内容,供参考:\n{excerpt}" # ============================================================ # 用户消息前缀解析 # ============================================================ def extract_skill_prefix(message: str) -> tuple[str, str | None]: """从用户消息里剥出 ``#skill:`` 前缀。 返回 ``(剩余消息, 技能名或 None)``。前缀只能出现在消息最前面;出现在 中间时按普通文本处理,避免用户正文里的 ``#skill:`` 被误当指令。 """ text = message or "" if not text.startswith(SKILL_WHITELIST_KEY): return text, None rest = text[len(SKILL_WHITELIST_KEY):] parts = rest.split(maxsplit=1) if not parts: return text, None name = parts[0].strip() remainder = parts[1] if len(parts) > 1 else "" if not name or "/" in name or "\\" in name or name.startswith("."): return text, None return remainder, name def extract_session_ref(message: str) -> tuple[str, str | None]: """从用户消息里剥出 ``#session:`` 前缀,返回 ``(剩余消息, 会话ID)``。""" text = message or "" if not text.startswith(SESSION_REF_KEY): return text, None rest = text[len(SESSION_REF_KEY):] parts = rest.split(maxsplit=1) if not parts: return text, None ref_id = parts[0].strip() remainder = parts[1] if len(parts) > 1 else "" if not ref_id: return text, None return remainder, ref_id