代码提交

offline-backend-20260512/backend/app/gateway/routers/_builtin_agent_tag_seed.py

@@ -1,15 +1,21 @@
 """内置智能体「用途」默认标签的种子化。
 
-把**功能型内置 agent**(roundtable-* / ai-writing-*)按用途默认打上标签,使管理员在
+把**所有内置 agent**按用途默认打上标签,使管理员在「内置智能体」筛选里一眼能按用途
-「内置智能体」筛选里一眼能按用途归类。没有明确用途的内置 agent(各种测试/用户残留的
+归类。已知功能型智能体使用具体用途；未列入映射的内置智能体统一标为「其他内置功能」，
-user_id 为空的 agent)**不打标签**。
+确保新增加的内置智能体也不会漏标。
 
 用途映射(可继续补充):
 - **多智能体规划**:roundtable-intent(任务理解)/ roundtable-recommender(智能体推荐)/
   roundtable-coordinator(总控协调)/ roundtable-report / roundtable-summary /
-  roundtable-execution-plan / roundtable-risk-review / roundtable-solution-design
+  position-action-planner / roundtable-dashboard / roundtable-structure
 - **AI写作**:ai-writing-researcher / ai-writing-outliner / ai-writing-writer(作家)/
   ai-writing-editor
+- **模板填充**:template-json-builder
+- **任务研判**:agentfx-analyst
+- **岗位会商**:position-roundtable-intent
+- **深度研究**:deep-research-collector
+- **强制检索**:forced-research-responder
+- **其他内置功能**:未显式列出的所有内置智能体（兜底）
 
 幂等:标签缺失则建,assignment 缺失则补;**只做并集追加**,绝不删用户/管理员后加的
 其它标签。在启动 lifespan 里 `_sync_legacy_agents` 之后调用(那时 agent 行已入库)。
@@ -26,10 +32,21 @@ logger = logging.getLogger(__name__)
 _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-other", "color": "gray"},
 }
 
-# 内置 agent id → 用途名(必须是 _PURPOSE_TAGS 的 key)。未列出的内置 agent 不打标签。
+# 内置 agent id → 用途名(必须是 _PURPOSE_TAGS 的 key)。未列出的内置 agent 由
+# ``_FALLBACK_PURPOSE`` 兜底，确保所有内置智能体均有用途标签。
 _AGENT_PURPOSE: dict[str, str] = {
+    # 任务工作区的研判报告生成。
+    "agentfx-analyst": "任务研判",
+    # 岗位会商在进入业务链前的意图澄清。
+    "position-roundtable-intent": "岗位会商",
     # 多智能体规划(圆桌全链路)
     "roundtable-intent": "多智能体规划",
     "roundtable-recommender": "多智能体规划",
@@ -37,17 +54,22 @@ _AGENT_PURPOSE: dict[str, str] = {
     "roundtable-report": "多智能体规划",
     "roundtable-summary": "多智能体规划",
     "position-action-planner": "多智能体规划",
-    "roundtable-execution-plan": "多智能体规划",
+    "roundtable-dashboard": "多智能体规划",
-    "roundtable-risk-review": "多智能体规划",
+    "roundtable-structure": "多智能体规划",
-    "roundtable-solution-design": "多智能体规划",
     # AI 写作(杂志社协作写作四阶段)
     "ai-writing-researcher": "AI写作",
     "ai-writing-outliner": "AI写作",
     "ai-writing-writer": "AI写作",
     "ai-writing-editor": "AI写作",
+    # 定时任务 JSON → HTML 模板的数据映射。
+    "template-json-builder": "模板填充",
+    # 深度调研管线的材料收集与强制检索输出。
+    "deep-research-collector": "深度研究",
+    "forced-research-responder": "强制检索",
 }
 
 _TARGET_TYPE = "agent"
+_FALLBACK_PURPOSE = "其他内置功能"
 
 
 async def ensure_builtin_agent_purpose_tags(agent_store, tag_store) -> int:
@@ -75,15 +97,20 @@ async def ensure_builtin_agent_purpose_tags(agent_store, tag_store) -> int:
                 if name in again:
                     name_to_id[name] = again[name]["id"]
 
-        # 2) 给映射里**实际存在**的内置 agent 追加用途标签(并集,不覆盖已有标签)。
+        # 2) 给所有实际存在的内置 agent 追加用途标签(并集,不覆盖已有标签)。
+        # 显式映射使用精确用途；日后新增且尚未归类的内置 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: