vent_agent_2.0/agents/agent_factory.py

"""主 agent 的组装。

一个进程只编译一个 agent（见 agent_registry 的单例与失效重建）。组装时
刻意把「能用框架的」全部交给框架，只在两处自己动手：

1. ``DynamicTimeMiddleware`` —— 框架没有「每轮刷新系统提示词里的时间」的能力
2. 权限规则 —— 必须自己声明，否则默认对文件系统是全开的

``create_deep_agent`` 的参数只做加法，去不掉内置工具；要裁剪内置能力得改
HarnessProfile（本项目用不到）。
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import config
from agents import agent_memory, model_factory, prompts
from agents.middleware import DynamicTimeMiddleware
from obs import get_logger
from tools import registry as tool_registry
from tools.demo_tools import PLAN_APPROVAL_TOOL

logger = get_logger(__name__)

# 沙箱目录：agent 唯一可写的虚拟根
WORKSPACE_DIR = "workspace"
SKILLS_DIR = "skills"
# 框架摘要中间件把被淘汰的历史落盘到这里的 `<thread_id>.md`，
# 并把路径写进摘要正文，agent 之后可以 read_file 把它们找回来
CONVERSATION_HISTORY_DIR = "conversation_history"


def _ensure_dirs() -> None:
    """确保沙箱、技能、历史落盘目录存在——backend 与技能扫描都要求目录真实存在。"""
    for name in (WORKSPACE_DIR, SKILLS_DIR, CONVERSATION_HISTORY_DIR):
        (config.ROOT / name).mkdir(parents=True, exist_ok=True)


def _build_backend() -> Any:
    """文件系统后端：单一真实目录，虚拟路径锚定在项目根。

    ``virtual_mode=True`` 是硬要求——它为 False 时后端直接按真实绝对路径
    访问磁盘，agent 传 ``/etc/passwd`` 就能读到。开启后 ``/skills/x`` 被解析
    为 ``<项目根>/skills/x``，且禁止 ``..`` 穿越。
    """
    from deepagents.backends import FilesystemBackend

    _ensure_dirs()
    return FilesystemBackend(root_dir=str(config.ROOT), virtual_mode=True)


def build_permissions() -> list[Any]:
    """文件访问权限规则。

    **顺序即语义**：框架逐条匹配、首条命中即返回（见 1.5 的实测记录
    ``tests/probe_w3a_fsglob.py``）。所以具体的 allow 必须排在宽泛的 deny
    之前——反过来写的话 ``deny /**`` 会把后面的 allow 全部短路，agent
    连技能文件都读不到。

    规则集的设计目标是「最小可用」：
    - 可写：仅 ``/workspace/``（agent 的落盘沙箱）
    - 可读：``/workspace/``、``/skills/``（技能手册）、
      ``/conversation_history/``（被压缩淘汰的历史——不放开读的话，
      摘要里那段「原文在此路径」对 agent 就是空头支票）
    - 其余一切（项目源码、.env、data/、系统路径）读写皆拒
    """
    from deepagents import FilesystemPermission

    workspace = f"/{WORKSPACE_DIR}"
    skills = f"/{SKILLS_DIR}"
    history = f"/{CONVERSATION_HISTORY_DIR}"
    return [
        FilesystemPermission(
            operations=["read", "write"],
            paths=[workspace, f"{workspace}/**"],
            mode="allow",
        ),
        FilesystemPermission(
            operations=["read"],
            paths=[skills, f"{skills}/**"],
            mode="allow",
        ),
        FilesystemPermission(
            operations=["read"],
            paths=[history, f"{history}/**"],
            mode="allow",
        ),
        # 兜底拒绝必须排在最后：它命中一切未被上面放行的路径
        FilesystemPermission(operations=["read", "write"], paths=["/**"], mode="deny"),
    ]


def _apply_compaction_setting() -> None:
    """按配置决定是否保留框架的摘要中间件。

    这里刻意**不自己构造 SummarizationMiddleware**。原因有二：

    1. ``create_agent`` 会校验中间件 ``name`` 唯一，而框架已经放了一个
       同名的摘要中间件；再传一个直接抛 "Please remove duplicate
       middleware instances"。1.5 时代的「同名原位替换」在 0.6.12 已不成立。
    2. 更重要的是框架那个版本更强：它会把被淘汰的历史**追加落盘**到
       ``/conversation_history/{thread_id}.md``，并把该路径嵌进摘要正文，
       agent 事后能自己 read_file 找回来。自己塞一个 langchain 原生的
       摘要中间件会把这份能力丢掉——被压缩的内容就真的没了。
    """
    if config.CTX_COMPACT_ENABLED:
        return

    from deepagents import HarnessProfile, register_harness_profile

    # 注册表按「provider 或具体模型」为键，且是叠加合并的
    register_harness_profile(
        config.MODEL,
        HarnessProfile(excluded_middleware=frozenset({"SummarizationMiddleware"})),
    )
    logger.warning(
        "自动压缩已关闭（CTX_COMPACT_ENABLED=0）："
        "被淘汰的历史不再落盘到 /%s/，长会话会在超出模型窗口后报错",
        CONVERSATION_HISTORY_DIR,
    )


def _build_middleware() -> list[Any]:
    """用户中间件栈。

    只放框架没有的能力（动态时间）。压缩交给框架内置的摘要中间件处理，
    见 :func:`_apply_compaction_setting`。
    """
    _apply_compaction_setting()
    return [DynamicTimeMiddleware()]


def _build_interrupt_on() -> dict[str, Any]:
    """需要人工审批的工具集合。

    默认只有计划审批工具。``HITL_TOOLS`` 可追加（逗号分隔的工具名），
    例如 ``HITL_TOOLS=calculate`` 会要求每次计算都经用户点头——
    用来演示与自测中断闭环。
    """
    names = {PLAN_APPROVAL_TOOL}
    extra = config.get_str("HITL_TOOLS", "")
    if extra:
        names.update(n.strip() for n in extra.split(",") if n.strip())

    available = set(tool_registry.get_tool_names())
    missing = names - available
    if missing:
        logger.warning("interrupt_on 里配置了未注册的工具，已忽略：%s", sorted(missing))
    return {name: True for name in sorted(names & available)}


def _build_subagents() -> list[dict[str, Any]]:
    """子智能体定义。

    骨架给一个示例子智能体，演示「主 agent 只拿结论、原始材料留在子线程」
    的用法。子智能体的 ``system_prompt`` **不继承**父级，必须自己写全。

    ``tools`` 必须是 ``BaseTool`` 列表（不是裸函数名，也不能是裸函数）——
    这里用 ``registry.get_tools_by_name`` 转换，由注册表保证名字有效。
    """
    return [
        {
            "name": "demo-searcher",
            "description": (
                "资料检索专员。当任务需要在演示知识库里做多轮检索、比对多个来源、"
                "或需要把检索结果整理成摘要时委派给它。"
                "它只回结论与来源，不回原始文本。"
            ),
            "system_prompt": (
                "你是资料检索专员。职责是在演示知识库里检索并把结果整理成结论。\n\n"
                "工作方式：\n"
                "- 用 search_demo_knowledge 检索，可换多组关键词多轮检索。\n"
                "- 需要数值计算时用 calculate，不要心算。\n"
                "- 检索不到就如实说「未检索到」，不要用常识补写。\n\n"
                "输出格式：\n"
                "- 先给一段不超过 5 句的结论。\n"
                "- 再列来源编号与标题。\n"
                "- 不要把检索到的原文整段复制回来，主线程不需要原始数据。"
            ),
            "tools": tool_registry.get_tools_by_name(
                ["search_demo_knowledge", "calculate", "get_current_time"]
            ),
        }
    ]


def build_agent() -> Any:
    """编译主 agent。由 agent_registry 调用，不直接对外暴露。"""
    from deepagents import create_deep_agent

    tool_registry.load_builtin_tools()
    tools = tool_registry.get_all_tools()
    subagents = _build_subagents()

    agent = create_deep_agent(
        model=model_factory.get_model(),
        tools=tools,
        system_prompt=prompts.build_system_prompt(subagents),
        middleware=_build_middleware(),
        subagents=subagents,
        # 技能目录：agent 启动时只读各技能的 frontmatter，按需才读全文
        skills=[f"/{SKILLS_DIR}/"],
        backend=_build_backend(),
        permissions=build_permissions(),
        interrupt_on=_build_interrupt_on(),
        name="deep-agent",
        checkpointer=agent_memory.get_checkpointer(),
    )

    logger.info(
        "agent 编译完成 tools=%d subagents=%d skills=%s checkpointer=%s",
        len(tools),
        len(subagents),
        f"/{SKILLS_DIR}/",
        agent_memory.saver_kind(),
    )
    return agent


def describe() -> dict[str, Any]:
    """组装配置摘要，供 /health 与调试使用。"""
    return {
        "name": "deep-agent",
        "model": model_factory.describe(),
        "tools": tool_registry.describe(),
        "interrupt_on": sorted(_build_interrupt_on().keys()),
        "subagents": [s["name"] for s in _build_subagents()],
        "skills_path": f"/{SKILLS_DIR}/",
        "workspace_path": f"/{WORKSPACE_DIR}/",
        "sandbox_root": str(Path(config.ROOT)),
        "compaction": {
            "enabled": config.CTX_COMPACT_ENABLED,
            "trigger_fraction": config.FRAMEWORK_COMPACT_TRIGGER,
            "trigger_tokens": int(
                model_factory.context_window() * config.FRAMEWORK_COMPACT_TRIGGER
            ),
            "history_offload_path": f"/{CONVERSATION_HISTORY_DIR}/",
            "lever": "CONTEXT_LIMIT",
        },
        "checkpointer": agent_memory.saver_kind(),
    }