diff --git a/offline-backend-20260512/backend/app/gateway/routers/_builtin_agent_tag_seed.py b/offline-backend-20260512/backend/app/gateway/routers/_builtin_agent_tag_seed.py index 22cb6588..62445e43 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/_builtin_agent_tag_seed.py +++ b/offline-backend-20260512/backend/app/gateway/routers/_builtin_agent_tag_seed.py @@ -1,19 +1,8 @@ -"""内置智能体「用途」默认标签的种子化。 +"""Seed default purpose tags for built-in agents. -把**功能型内置 agent**(roundtable-* / ai-writing-* / workflow-planner)按用途默认打上标签,使管理员在 -「内置智能体」筛选里一眼能按用途归类。没有明确用途的内置 agent(各种测试/用户残留的 -user_id 为空的 agent)**不打标签**。 - -用途映射(可继续补充): -- **多智能体规划**:roundtable-intent(任务理解)/ roundtable-recommender(智能体推荐)/ - roundtable-coordinator(总控协调)/ roundtable-report / roundtable-summary / - roundtable-execution-plan / roundtable-risk-review / roundtable-solution-design -- **AI写作**:ai-writing-researcher / ai-writing-outliner / ai-writing-writer(作家)/ - ai-writing-editor -- **工作流编排**:workflow-planner（只做候选工作流规划，不参与圆桌会商或正式执行） - -幂等:标签缺失则建,assignment 缺失则补;**只做并集追加**,绝不删用户/管理员后加的 -其它标签。在启动 lifespan 里 `_sync_legacy_agents` 之后调用(那时 agent 行已入库)。 +Known built-in agents receive their specific purpose tag. Other built-in +agents receive the fallback purpose tag so newly added built-ins are not +omitted. Existing user or administrator tags are never removed. """ from __future__ import annotations @@ -22,48 +11,56 @@ import logging logger = logging.getLogger(__name__) -# 标签按用途分组(tag_type 固定为 "agent",与 target_type 对齐)。id 用稳定 slug, -# 便于幂等;name 是展示名;color 走标签库的命名色。 _PURPOSE_TAGS: dict[str, dict[str, str]] = { "多智能体规划": {"id": "builtin-purpose-multi-agent-planning", "color": "blue"}, "AI写作": {"id": "builtin-purpose-ai-writing", "color": "green"}, + "模板填充": {"id": "builtin-purpose-template-filling", "color": "orange"}, + "任务研判": {"id": "builtin-purpose-task-analysis", "color": "purple"}, + "岗位会商": {"id": "builtin-purpose-position-roundtable", "color": "cyan"}, + "深度研究": {"id": "builtin-purpose-deep-research", "color": "indigo"}, + "强制检索": {"id": "builtin-purpose-forced-research", "color": "red"}, "工作流编排": {"id": "builtin-purpose-workflow-planning", "color": "indigo"}, + "其他内置功能": {"id": "builtin-purpose-other", "color": "gray"}, } -# 内置 agent id → 用途名(必须是 _PURPOSE_TAGS 的 key)。未列出的内置 agent 不打标签。 +# Every value must be a key in _PURPOSE_TAGS. _AGENT_PURPOSE: dict[str, str] = { - # 多智能体规划(圆桌全链路) + "agentfx-analyst": "任务研判", + "position-roundtable-intent": "岗位会商", "roundtable-intent": "多智能体规划", "roundtable-recommender": "多智能体规划", "roundtable-coordinator": "多智能体规划", "roundtable-report": "多智能体规划", "roundtable-summary": "多智能体规划", "position-action-planner": "多智能体规划", + "roundtable-dashboard": "多智能体规划", + "roundtable-structure": "多智能体规划", "roundtable-execution-plan": "多智能体规划", "roundtable-risk-review": "多智能体规划", "roundtable-solution-design": "多智能体规划", - # 8BF 独立会商第一步的内置意图智能体(roundtable-intent 的同源复制) "eightbf-roundtable-intent": "多智能体规划", - # AI 写作(杂志社协作写作四阶段) "ai-writing-researcher": "AI写作", "ai-writing-outliner": "AI写作", "ai-writing-writer": "AI写作", "ai-writing-editor": "AI写作", - # Workflow Studio（候选流程总控，和圆桌总控隔离） + "template-json-builder": "模板填充", + "deep-research-collector": "深度研究", + "forced-research-responder": "强制检索", "workflow-planner": "工作流编排", } _TARGET_TYPE = "agent" +_FALLBACK_PURPOSE = "其他内置功能" async def ensure_builtin_agent_purpose_tags(agent_store, tag_store) -> int: - """确保内置功能型 agent 带上默认「用途」标签。返回新增 assignment 数。 + """Ensure built-in agents have a default purpose tag. - 错误只记录不抛——种子化失败绝不能阻塞服务启动。 + Returns the number of newly created tag assignments. Failures are logged + instead of blocking application startup. """ try: - # 1) 确保用途标签存在,建立 name -> id 映射。 - existing = {t.get("name"): t for t in await tag_store.list_tags(_TARGET_TYPE)} + existing = {tag.get("name"): tag for tag in await tag_store.list_tags(_TARGET_TYPE)} name_to_id: dict[str, str] = {} for name, meta in _PURPOSE_TAGS.items(): found = existing.get(name) @@ -76,25 +73,26 @@ async def ensure_builtin_agent_purpose_tags(agent_store, tag_store) -> int: ) name_to_id[name] = created["id"] except ValueError: - # 并发/已存在(按 name 唯一)——重查一次取其 id。 - again = {t.get("name"): t for t in await tag_store.list_tags(_TARGET_TYPE)} + again = {tag.get("name"): tag for tag in await tag_store.list_tags(_TARGET_TYPE)} if name in again: name_to_id[name] = again[name]["id"] - # 2) 给映射里**实际存在**的内置 agent 追加用途标签(并集,不覆盖已有标签)。 + builtin_ids = { + agent["id"] + for agent in await agent_store.list_all() + if agent.get("user_id") is None + } added = 0 - for agent_id, purpose in _AGENT_PURPOSE.items(): + for agent_id in builtin_ids: + purpose = _AGENT_PURPOSE.get(agent_id, _FALLBACK_PURPOSE) tag_id = name_to_id.get(purpose) if not tag_id: continue - agent = await agent_store.get_any(agent_id) - if agent is None: - continue # 该部署未装此内置 agent,跳过 current = await tag_store.get_target_tags(_TARGET_TYPE, agent_id) - cur_ids = [t["id"] for t in current] - if tag_id in cur_ids: - continue # 已打过该用途标签 - await tag_store.set_target_tags(_TARGET_TYPE, agent_id, [*cur_ids, tag_id]) + current_ids = [tag["id"] for tag in current] + if tag_id in current_ids: + continue + await tag_store.set_target_tags(_TARGET_TYPE, agent_id, [*current_ids, tag_id]) added += 1 if added: logger.info("Seeded %d built-in agent purpose tag assignment(s)", added)