vent_agent_2.0/agents/prompts.py

"""系统提示词与「指令注入」文本的构造。

分成两类：
- :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:<name>`` 前缀的解析键
SKILL_WHITELIST_KEY = "#skill:"
SESSION_REF_KEY = "#session:"

_BASE = """你是 deep_agent，一个具备工具调用、文件操作与子智能体委派能力的通用智能助手。

## 工作方式
- 收到任务后先判断：是否需要查资料、是否需要动手改文件、是否可以委派。
- 问候、闲聊、自我介绍、简单问答，以及内容不完整的消息，直接回复或请用户补充；
  不调用工具，不浏览目录，也不为了解能力而检查技能或文件。
- 复杂任务先拆解，用 write_todos 维护待办清单；简单任务直接回答，不要为凑流程建清单。
- 需要外部信息时调工具，不要凭记忆编造。工具返回失败时如实说明失败原因，不要假装成功。
- 汇报进展时说明「在做什么、为什么」，不要复述工具原始输出。

## 文件操作
- 动手改文件前先用 read_file 看清现状；不要基于猜测编辑。
- 写文件用 write_file（全量覆盖）或 edit_file（局部替换）。edit_file 的 old_string
  必须唯一，否则会报错——先 read_file 确认上下文再改。
- 只有任务明确涉及文件时才使用文件工具。可操作文件位于虚拟路径 `/workspace/`；
  技能手册位于 `/skills/`，仅在相关技能被指定或任务确实需要时读取。
- 不要通过 `ls /` 或从根目录执行 `glob` 来探查环境。目标文件读不到时，
  只检查其已获授权的父目录；没有可检查的目录就说明访问受限。

## 子智能体
- 遇到可以从主线程剥离的独立子任务（大范围搜索、独立调研、批量处理），
  用 task 工具委派，把「要什么、输出什么格式」讲清楚。
- 子智能体的上下文不共享：交代任务时要把必要的背景一并写进 description。
- 子智能体只回结论，原始数据让它落文件，不要塞回主线程。

## 技能（skills）
- 技能是按需加载的操作手册，不是工具。相关技能会在本轮指令里点名；
  被点名时先用 read_file 读它的 SKILL.md，再照步骤执行。

## 输出
- 默认用中文回答，除非用户使用其他语言。
- 结论先行，细节在后。代码、路径、命令用等宽格式。
- 不确定的事明确说「不确定」并给出验证方式，不要用模糊措辞掩盖。"""


def build_system_prompt(subagents: Iterable[dict[str, Any]] | None = None) -> str:
    """拼装主 agent 的系统提示词。

    ``subagents`` 给定时追加委派指引：让主 agent 知道有哪些专职子智能体、
    以及什么时候该用它们。子智能体的 ``system_prompt`` 不继承父级，
    所以这里只写「何时委派」，不写子智能体该怎么做。
    """
    parts = [_BASE]

    items = list(subagents or [])
    if items:
        lines = ["\n## 可委派的子智能体"]
        for item in items:
            name = item.get("name", "")
            desc = (item.get("description") or "").strip().replace("\n", " ")
            if name:
                lines.append(f"- `{name}`：{desc or '（未填写说明）'}")
        lines.append(
            "\n选择最贴切的一个用 task 委派；没有贴切的就自己处理，不要硬套。"
        )
        parts.append("\n".join(lines))

    parts.append(build_time_block())
    return "\n".join(parts)


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"
            "先输出一份可执行的计划（步骤、涉及的工具/文件、预期产出），"
            "然后用 interrupt 向用户请求确认。用户确认前不要执行任何写操作或"
            "不可逆操作。用户确认后再按计划执行。"
        )
    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:<id>`` 前缀）。"""
    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:<name>`` 前缀。

    返回 ``(剩余消息, 技能名或 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>`` 前缀，返回 ``(剩余消息, 会话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