deerflow-code/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py

"""圆桌功能型 agent 的「调用前自检 + 缺失自动补建」工具。

适用场景
--------
``roundtable-intent`` / ``roundtable-recommender`` / ``roundtable-coordinator``
是圆桌 Step 1 + 推荐 + Step 2 总控链路的**功能型** 内置 agent,任何部署都必须
存在。但仓库根 ``.gitignore`` 把整个 ``.deer-flow/`` 目录排除,新机器
``git clone`` 之后这些目录**不会存在**,直接调对应 endpoint 会因为 agent
解析失败抛 500。

本模块在两个地方被调用:
1. **启动 lifespan**(``app/gateway/app.py``):服务起来就把缺失的目录补齐,
   紧接着的 ``_sync_legacy_agents`` 会把它们 upsert 进 DB,新机器一启动
   ``/api/agents`` 列表里就有这三个 agent,**不必先去点圆桌页**。
2. **四个路由入口**(``intent.init``、``intent.stream``、``recommend.stream``、
   ``multi_agent.init``):防御性兜底——若管理员在启动后手删了目录,下一次请求
   也能自愈。

``_sync_legacy_agents`` 只关心磁盘文件,本模块也只写磁盘文件,不直接动 DB。

为什么 SOUL/config 用资源文件而不是内嵌 Python 字符串
----------------------------------------------------
这三个 agent 的 SOUL 是**经过实测调优、验证过**的版本(intent 含硬规则/反例集、
recommender 含三条铁律/多领域示例),体量大且会持续迭代。早期把 SOUL 内嵌成
Python 三引号字符串,导致「磁盘上的验证版」与「代码里的内嵌版」很容易分叉
(还有 ``\\n`` 之类转义陷阱)。改为把验证版 SOUL.md + config.yaml 作为**源码树里
的资源文件**(``_roundtable_seed_assets/<id>/``,随 git 发布,不在 gitignore 的
``.deer-flow/`` 下),seeder 直接字节级复制过去——调 SOUL 就是改 .md 文件,
单一可信源,走 git review。

> ⚠️ 维护提示:资源文件(``_roundtable_seed_assets/``)是**发布默认值**的唯一可信源。
> 线上调优后,请把改动写回对应的资源 .md,否则下次新机器补建出来的还是旧版。
> ``_ensure_one`` 只在目录缺失时复制,**不覆盖**已存在目录(尊重运维现场手改)。

为什么 ``roundtable-coordinator`` 是单例而不是每次会话新建
------------------------------------------------------------
历史实现每次 Step 2 init 都用 ``roundtable-coordinator-<timestamp36>`` 这种
带时间戳的临时 id 创建新 agent,把"本次可调度席位列表"写进动态拼接的 SOUL。
长期跑下来 ``.deer-flow/agents/`` 累积了几十上百个完全相同的 coordinator
壳,管理页非常杂乱。

改造后 coordinator 是 **单例**(id 固定为 ``roundtable-coordinator``):
- SOUL 是**通用协调规则**,不含本次席位列表;
- 本次圆桌可调度的席位清单由 ``multi_agent.init`` 在创建好 coordinator
  thread 后,以 human 消息的形式 append 到 thread 历史,让 leader run 自然
  读到。这样 SOUL 长期稳定,席位列表随会话动态注入。
"""

from __future__ import annotations

import logging
from collections.abc import Iterable
from pathlib import Path

from deerflow.config.paths import get_paths

logger = logging.getLogger(__name__)


INTENT_AGENT_ID = "roundtable-intent"
RECOMMENDER_AGENT_ID = "roundtable-recommender"
COORDINATOR_AGENT_ID = "roundtable-coordinator"
# Step 3「结果绘制」用的方案可视化总结智能体:读各席位交付 + 共识,画自包含图表 HTML。
# 与上面三个一样是内置单例,种子化一次常驻磁盘+DB,绝不每次调用重建。
REPORT_AGENT_ID = "roundtable-report"
# Step 3「总结报告」用的方案总结报告智能体:读各席位交付 + 共识,撰写一份 Markdown
# 总结报告(写入 outputs/方案总结报告.md),并按业务链总控编排提示组织。与 report 并存
# 的另一内置单例(report 画 HTML 看板,summary 写 Markdown 报告 + 支持问答/增量改)。
SUMMARY_AGENT_ID = "roundtable-summary"
# Step 3「大屏多页」用的结构化数据智能体:读各席位交付 + 共识,逐席位分析并提炼成
# 一份大屏展示用的结构化数据(report-json),供前端按模板渲染成多页大屏。**不调任何工具、
# 不写文件、不输出 HTML**——只在对话里输出一个严格合法的 ```report-json 代码块。与 report
# (画 HTML 看板)/summary(写 Markdown)并存的第三个 Step3 内置单例,把「大屏抽取」从
# 旧版「复用 report agent + 简单粗暴字符串抽取」改为专职智能体 + 健壮抽取(见
# roundtable_orchestrator.report_json)。
DASHBOARD_AGENT_ID = "roundtable-dashboard"
# Step 3「结构化抽取」用的智能体：读「方案总结报告」/ 各席位交付，分析层级结构 → 产出 flow-json
# （也负责按指令改图 / 校验修复）。供 /api/flow-extract 调用；以后可在它身上配置「入库技能」，
# 让它分析出结构化结果后直接入库。与 report/summary 并存的另一内置单例。
STRUCTURE_AGENT_ID = "roundtable-structure"

# 验证版 SOUL.md + config.yaml 的源码资源目录(随 git 发布)。每个子目录名即 agent id,
# 内含该 agent 的 config.yaml 与 SOUL.md,是 seeder 落盘时字节级复制的来源。
_ASSETS_DIR = Path(__file__).parent / "_roundtable_seed_assets"

# 这几个功能型/内置 agent 任何部署都必须存在;顺序无关紧要(逐个独立补建)。
# 前三个支撑 Step 1 澄清 / 推荐 / Step 2 总控;roundtable-report 支撑 Step 3 结果绘制
# (HTML 看板);roundtable-summary 支撑 Step 3 总结报告(Markdown + 问答/增量改);
# roundtable-dashboard 支撑 Step 3 大屏多页结构化数据(report-json + 问答/增量改)。
_FUNCTIONAL_AGENT_IDS: tuple[str, ...] = (
    INTENT_AGENT_ID,
    RECOMMENDER_AGENT_ID,
    COORDINATOR_AGENT_ID,
    REPORT_AGENT_ID,
    SUMMARY_AGENT_ID,
    DASHBOARD_AGENT_ID,
    STRUCTURE_AGENT_ID,
)

# seeder 会复制资源目录里的这些文件(存在即复制,缺失则跳过)。
_SEED_FILES: tuple[str, ...] = ("config.yaml", "SOUL.md")


def _ensure_one(agent_id: str) -> bool:
    """Copy the verified SOUL.md + config.yaml for one functional agent into
    ``.deer-flow/agents/<id>/`` if that directory is missing.

    Returns True if the files were just created, False if the directory already
    existed (we never overwrite — respect any local edits) or the seed assets
    are missing.
    """
    agent_dir = get_paths().agent_dir(agent_id)
    if agent_dir.exists():
        return False

    src_dir = _ASSETS_DIR / agent_id
    if not src_dir.is_dir():
        # Assets should always ship with the code; if they don't, surface a loud
        # log instead of silently creating an empty agent dir.
        logger.error(
            "Missing seed assets for roundtable agent '%s' (expected at %s)",
            agent_id,
            src_dir,
        )
        return False

    agent_dir.mkdir(parents=True, exist_ok=True)
    for fname in _SEED_FILES:
        src = src_dir / fname
        if src.exists():
            # 字节级复制,保留验证版文件原样(换行/编码),与资源文件 hash 对齐。
            (agent_dir / fname).write_bytes(src.read_bytes())
    return True


def ensure_roundtable_functional_agents() -> list[str]:
    """Seed the three roundtable functional agents (``roundtable-intent`` /
    ``roundtable-recommender`` / ``roundtable-coordinator``) if their on-disk
    directories are missing. Idempotent — fast no-op when all exist.

    Called from the startup lifespan and from the entry of intent/init,
    intent/stream, recommend/stream, and multi_agent/init so the routes
    self-heal on fresh internal deployments where ``.deer-flow/`` is gitignored
    and the agent directories were never shipped.

    Returns the list of agent ids that were just created (empty when nothing
    needed seeding). Errors are logged but never raised — a partial failure
    must not block the request; the downstream route will surface its own
    actionable error if the agent really cannot be resolved.
    """
    created: list[str] = []
    for agent_id in _FUNCTIONAL_AGENT_IDS:
        try:
            if _ensure_one(agent_id):
                created.append(agent_id)
        except Exception:
            logger.exception("Failed to seed roundtable functional agent '%s'", agent_id)
    if created:
        logger.info("Seeded missing roundtable functional agents: %s", ", ".join(created))
    return created


def required_agent_ids() -> Iterable[str]:
    """Stable ids of agents this seeder owns — exposed for tests/diagnostics."""
    return tuple(_FUNCTIONAL_AGENT_IDS)