vent_agent_2.0/agents/agent_factory.py

"""主 agent 的组装。

每个 Agent 名称缓存一张图（见 agent_registry 的单飞编译与失效重建）。组装时
刻意把「能用框架的」全部交给框架，额外声明：

1. ``DynamicTimeMiddleware`` —— 框架没有「每轮刷新系统提示词里的时间」的能力
2. ``SkillCompletionMiddleware`` —— 提交正式报告后结束探索，失败时继续修正
3. 权限规则 —— 必须自己声明，否则默认对文件系统是全开的

``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, SkillCompletionMiddleware
from obs import get_logger
from tools import registry as tool_registry
from extensions.registry import approval_names, load_extensions
from extensions.services import service_view

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:
    """延迟到每次执行，按服务端身份构造隔离文件后端。"""
    from agents.workspace import backend_for_runtime

    _ensure_dirs()
    return backend_for_runtime


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(), *load_extensions().build_middleware("main")]


def _build_interrupt_on(tool_names=None) -> dict[str, Any]:
    """审批来自工具声明、扩展声明与 HITL_TOOLS；未知工具在启动时拒绝。"""
    return {name: True for name in sorted(approval_names())
            if tool_names is None or name in tool_names}


def _build_subagents() -> list[dict[str, Any]]:
    """从统一扩展目录组装子 Agent、系统提示词与工具审批。"""
    return load_extensions().subagents(approval_names())


def _coordinator_tools(catalog):
    # 专业业务工具经委派使用，主 Agent 保留现有通用和示例工具。
    business_tools = {n for b in catalog.business_agents.values() for n in catalog.agents[b.name].tools}
    shared = {n for a in catalog.agents.values() if a.name not in catalog.business_agents for n in a.tools}
    return [t for t in tool_registry.get_all_tools() if t.name not in business_tools or t.name in shared]


def build_agent(agent_name: str = "main") -> Any:
    """编译主协调 Agent 或可独立执行的业务 Agent。"""
    from deepagents import create_deep_agent

    tool_registry.load_builtin_tools()
    catalog = load_extensions()
    service_view(catalog.services)
    if agent_name == "main":
        tools = _coordinator_tools(catalog)
        subagents = _build_subagents()
        for spec in subagents:
            spec["middleware"] = catalog.build_middleware(spec["name"])
            if spec["name"] in catalog.business_agents:
                from runtime.business import BusinessGateMiddleware
                spec["middleware"].insert(0, BusinessGateMiddleware(spec["name"]))
                if catalog.business(spec['name']).skill_path:
                    spec['middleware'].append(SkillCompletionMiddleware())
        prompt = prompts.build_system_prompt(subagents)
        middleware = _build_middleware()
        result_schema = None
        graph_name = "deep-agent"
    else:
        business = catalog.business(agent_name)
        spec = catalog.agents[agent_name]
        from runtime.business import BusinessGateMiddleware
        _apply_compaction_setting()
        tools = tool_registry.get_tools_by_name((*spec.tools, "ask_user", "request_plan_approval"))
        subagents = []
        prompt = spec.prompt
        middleware = [BusinessGateMiddleware(agent_name), DynamicTimeMiddleware(),
                      *catalog.build_middleware(agent_name)]
        if business.skill_path:
            middleware.append(SkillCompletionMiddleware())
        result_schema = None if business.skill_path else business.result_schema
        graph_name = agent_name

    agent = create_deep_agent(
        model=model_factory.get_model(),
        tools=tools,
        system_prompt=prompt,
        middleware=middleware,
        subagents=subagents,
        # 技能目录：agent 启动时只读各技能的 frontmatter，按需才读全文
        skills=[f"/{SKILLS_DIR}/"],
        backend=_build_backend(),
        permissions=build_permissions(),
        interrupt_on=_build_interrupt_on({t.name for t in tools}),
        response_format=result_schema,
        name=graph_name,
        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",
        "extensions": list(load_extensions().modules),
        "disabled_extensions": list(load_extensions().disabled_modules),
        "services": sorted(load_extensions().services),
        "middleware": [{"name": name, "targets": list(load_extensions().middleware[name].targets)}
                       for name in load_extensions().middleware_order()],
        "model": model_factory.describe(),
        "tools": tool_registry.describe(),
        "interrupt_on": sorted(_build_interrupt_on().keys()),
        "subagents": [s["name"] for s in _build_subagents()],
        "business_agents": [{"name": b.name, "display_name": b.display_name,
                             "tools": list(load_extensions().agents[b.name].tools),
                             "execution_mode": "skill_exploration" if b.skill_path else "structured_agent",
                             "skill_path": b.skill_path,
                             "max_concurrency": b.max_concurrency,
                             "gateway_configured": bool(b.gateway.agent_code),
                             "result_platform_configured": bool(b.result_binding.agent_code) and config.get_bool("RESULT_PLATFORM_ENABLED", False)}
                            for b in load_extensions().business_agents.values()],
        "skills_path": f"/{SKILLS_DIR}/",
        "workspace_path": f"/{WORKSPACE_DIR}/",
        "sandbox_root": str(Path(config.ROOT) / "data" / "workspaces"),
        "workspace_scope": "user/session",
        "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(),
    }