feat(agent): 新增通用子智能体并统一共享配置

backend/package/yuxi/agents/skills/service.py

@@ -22,6 +22,7 @@ from yuxi.agents.mcp.service import get_enabled_mcp_server_slugs
 from yuxi.agents.skills.repository import SkillRepository
 from yuxi.storage.postgres.models_business import Skill, User
 from yuxi.utils.logging_config import logger
+from yuxi.utils.share_config import SHARE_ACCESS_LEVELS, normalize_share_config
 
 SKILL_SLUG_PATTERN = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
 SKILL_NAME_PATTERN = SKILL_SLUG_PATTERN
@@ -58,7 +59,7 @@ TEXT_FILE_EXTENSIONS = {
 
 BUILTIN_SKILL_OPERATOR = "builtin-system"
 SKILL_SOURCE_TYPES = {"builtin", "upload", "remote"}
-ACCESS_LEVELS = {"global", "department", "user"}
+ACCESS_LEVELS = SHARE_ACCESS_LEVELS
 ADMIN_ROLES = {"admin", "superadmin"}
 DEFAULT_SKILL_SHARE_CONFIG = {"access_level": "user", "department_ids": [], "user_uids": []}
 BUILTIN_SKILL_SHARE_CONFIG = {"access_level": "global", "department_ids": [], "user_uids": []}
@@ -103,17 +104,6 @@ def is_builtin_skill(item: Skill | dict) -> bool:
     return source_type == "builtin"
 
 
-def _normalize_department_ids(department_ids: list | None) -> list[int]:
-    normalized = []
-    for department_id in department_ids or []:
-        normalized.append(int(department_id))
-    return normalized
-
-
-def _normalize_user_uids(user_uids: list | None) -> list[str]:
-    return [uid for uid in (str(uid).strip() for uid in user_uids or []) if uid]
-
-
 def get_allowed_skill_access_levels(user: User) -> list[str]:
     if user.role in ADMIN_ROLES:
         return ["global", "department", "user"]
@@ -131,32 +121,16 @@ def normalize_skill_share_config(
     if source_type == "builtin":
         return BUILTIN_SKILL_SHARE_CONFIG.copy()
 
-    config = share_config or DEFAULT_SKILL_SHARE_CONFIG
-    access_level = config.get("access_level") or "user"
-    if access_level not in ACCESS_LEVELS:
-        raise ValueError("无效的 Skill 权限等级")
-    if allowed_access_levels is not None and access_level not in allowed_access_levels:
-        raise ValueError("当前用户无权使用该 Skill 共享范围")
-
-    if access_level == "global":
-        return {"access_level": "global", "department_ids": [], "user_uids": []}
-
-    if access_level == "department":
-        department_ids = _normalize_department_ids(config.get("department_ids"))
-        if operator_department_id is not None:
-            department_ids.append(int(operator_department_id))
-        department_ids = sorted(set(department_ids))
-        if not department_ids:
-            raise ValueError("部门共享至少需要选择一个部门")
-        return {"access_level": "department", "department_ids": department_ids, "user_uids": []}
-
-    user_uids = _normalize_user_uids(config.get("user_uids"))
-    if operator_uid:
-        user_uids.append(str(operator_uid))
-    user_uids = sorted(set(user_uids))
-    if not user_uids:
-        raise ValueError("指定人可访问至少需要选择一个用户")
-    return {"access_level": "user", "department_ids": [], "user_uids": user_uids}
+    return normalize_share_config(
+        share_config,
+        default_config=DEFAULT_SKILL_SHARE_CONFIG,
+        default_access_level="user",
+        invalid_access_level_message="无效的 Skill 权限等级",
+        user_uid=operator_uid,
+        department_id=operator_department_id,
+        allowed_access_levels=allowed_access_levels,
+        unauthorized_access_level_message="当前用户无权使用该 Skill 共享范围",
+    )
 
 
 def user_can_access_skill(user: User, skill: Skill, *, require_enabled: bool = True) -> bool:

backend/package/yuxi/knowledge/manager.py

@@ -7,9 +7,10 @@ from yuxi.knowledge.factory import KnowledgeBaseFactory
 from yuxi.storage.postgres.models_business import User
 from yuxi.utils import logger
 from yuxi.utils.datetime_utils import utc_isoformat
+from yuxi.utils.share_config import SHARE_ACCESS_LEVELS, normalize_share_config
 
 DEFAULT_SHARE_CONFIG = {"access_level": "global", "department_ids": [], "user_uids": []}
-ACCESS_LEVELS = {"global", "department", "user"}
+ACCESS_LEVELS = SHARE_ACCESS_LEVELS
 
 
 class KnowledgeBaseManager:
@@ -160,39 +161,14 @@ class KnowledgeBaseManager:
         user_uid: str | None = None,
         department_id: int | str | None = None,
     ) -> dict:
-        config = share_config or {}
-        access_level = config.get("access_level") or "global"
-        if access_level not in ACCESS_LEVELS:
-            raise ValueError("无效的知识库权限等级")
-
-        if access_level == "global":
-            return DEFAULT_SHARE_CONFIG.copy()
-
-        if access_level == "department":
-            department_ids = self._normalize_department_ids(config.get("department_ids"))
-            if department_id is not None:
-                department_ids.append(int(department_id))
-            department_ids = sorted(set(department_ids))
-            if not department_ids:
-                raise ValueError("部门共享至少需要选择一个部门")
-            return {"access_level": "department", "department_ids": department_ids, "user_uids": []}
-
-        user_uids = self._normalize_user_uids(config.get("user_uids"))
-        if user_uid:
-            user_uids.append(str(user_uid))
-        user_uids = sorted(set(user_uids))
-        if not user_uids:
-            raise ValueError("指定人可访问至少需要选择一个用户")
-        return {"access_level": "user", "department_ids": [], "user_uids": user_uids}
-
-    def _normalize_department_ids(self, department_ids: list | None) -> list[int]:
-        normalized = []
-        for department_id in department_ids or []:
-            normalized.append(int(department_id))
-        return normalized
-
-    def _normalize_user_uids(self, user_uids: list | None) -> list[str]:
-        return [uid for uid in (str(uid).strip() for uid in user_uids or []) if uid]
+        return normalize_share_config(
+            share_config,
+            default_config=DEFAULT_SHARE_CONFIG,
+            default_access_level="global",
+            invalid_access_level_message="无效的知识库权限等级",
+            user_uid=user_uid,
+            department_id=department_id,
+        )
 
     async def get_databases(self) -> dict:
         """获取所有数据库信息"""

backend/package/yuxi/repositories/agent_repository.py

@@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from yuxi.storage.postgres.models_business import Agent, User
 from yuxi.utils.datetime_utils import utc_now_naive
+from yuxi.utils.share_config import SHARE_ACCESS_LEVELS, normalize_share_config
 
 DEFAULT_AGENT_SLUG = "default-chatbot"
 DEFAULT_AGENT_NAME = "智能助手"
@@ -17,6 +18,12 @@ SUB_AGENT_BACKEND_ID = "SubAgentBackend"
 DEFAULT_AGENT_DESCRIPTION = "基础的对话机器人，可以回答问题，可在配置中启用需要的工具。"
 DEFAULT_SHARE_CONFIG = {"access_level": "global", "department_ids": [], "user_uids": []}
 
+GENERAL_PURPOSE_AGENT_SLUG = "general-purpose"
+GENERAL_PURPOSE_AGENT_NAME = "通用任务"
+GENERAL_PURPOSE_AGENT_DESCRIPTION = (
+    "面向没有专用角色约束的一般任务，使用默认运行配置独立完成分析、整理、写作或文件处理。"
+)
+
 WEB_SEARCH_AGENT_SLUG = "web-search"
 WEB_SEARCH_AGENT_NAME = "网页检索"
 WEB_SEARCH_AGENT_DESCRIPTION = "围绕检索目标持续搜索网页，返回带引用来源的摘要资料。"
@@ -92,7 +99,7 @@ FACT_VERIFIER_SYSTEM_PROMPT = """你是「事实核查员」子智能体，专
 - 明确标注无法查证或来源相互冲突的论断。
 - 不要编造来源或链接。"""
 
-ACCESS_LEVELS = {"global", "department", "user"}
+ACCESS_LEVELS = SHARE_ACCESS_LEVELS
 ADMIN_ROLES = {"admin", "superadmin"}
 
 
@@ -107,14 +114,6 @@ def resolve_agent_is_subagent(backend_id: str, is_subagent: bool | None = None)
     return expected
 
 
-def _normalize_department_ids(department_ids: list | None) -> list[int]:
-    return [int(department_id) for department_id in department_ids or []]
-
-
-def _normalize_user_uids(user_uids: list | None) -> list[str]:
-    return [uid for uid in (str(uid).strip() for uid in user_uids or []) if uid]
-
-
 def normalize_agent_share_config(
     share_config: dict | None,
     *,
@@ -127,30 +126,14 @@ def normalize_agent_share_config(
             raise ValueError("私有智能体必须绑定创建用户")
         return {"access_level": "user", "department_ids": [], "user_uids": [str(user_uid)]}
 
-    config = share_config or {}
-    access_level = config.get("access_level") or "global"
-    if access_level not in ACCESS_LEVELS:
-        raise ValueError("无效的智能体权限等级")
-
-    if access_level == "global":
-        return DEFAULT_SHARE_CONFIG.copy()
-
-    if access_level == "department":
-        department_ids = _normalize_department_ids(config.get("department_ids"))
-        if department_id is not None:
-            department_ids.append(int(department_id))
-        department_ids = sorted(set(department_ids))
-        if not department_ids:
-            raise ValueError("部门共享至少需要选择一个部门")
-        return {"access_level": "department", "department_ids": department_ids, "user_uids": []}
-
-    user_uids = _normalize_user_uids(config.get("user_uids"))
-    if user_uid:
-        user_uids.append(str(user_uid))
-    user_uids = sorted(set(user_uids))
-    if not user_uids:
-        raise ValueError("指定人可访问至少需要选择一个用户")
-    return {"access_level": "user", "department_ids": [], "user_uids": user_uids}
+    return normalize_share_config(
+        share_config,
+        default_config=DEFAULT_SHARE_CONFIG,
+        default_access_level="global",
+        invalid_access_level_message="无效的智能体权限等级",
+        user_uid=user_uid,
+        department_id=department_id,
+    )
 
 
 def user_can_access_agent(user: User, agent: Agent) -> bool:
@@ -261,6 +244,17 @@ class AgentRepository:
         await self.db.refresh(agent)
         return agent
 
+    async def ensure_general_purpose_subagent(self, *, created_by: str | None = None) -> Agent:
+        return await self._ensure_builtin_agent(
+            slug=GENERAL_PURPOSE_AGENT_SLUG,
+            backend_id=SUB_AGENT_BACKEND_ID,
+            name=GENERAL_PURPOSE_AGENT_NAME,
+            description=GENERAL_PURPOSE_AGENT_DESCRIPTION,
+            config_context={},
+            is_subagent=True,
+            created_by=created_by,
+        )
+
     async def _ensure_builtin_agent(
         self,
         *,

backend/package/yuxi/utils/share_config.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from collections.abc import Collection
+
+SHARE_ACCESS_LEVELS = {"global", "department", "user"}
+EMPTY_SHARE_CONFIG = {"access_level": "global", "department_ids": [], "user_uids": []}
+
+
+def _normalize_department_ids(department_ids: list | None) -> list[int]:
+    return [int(department_id) for department_id in department_ids or []]
+
+
+def _normalize_user_uids(user_uids: list | None) -> list[str]:
+    return [uid for uid in (str(uid).strip() for uid in user_uids or []) if uid]
+
+
+def normalize_share_config(
+    share_config: dict | None,
+    *,
+    default_config: dict | None,
+    default_access_level: str,
+    invalid_access_level_message: str,
+    user_uid: str | None = None,
+    department_id: int | str | None = None,
+    allowed_access_levels: Collection[str] | None = None,
+    unauthorized_access_level_message: str | None = None,
+) -> dict:
+    config = share_config or default_config or {}
+    access_level = config.get("access_level") or default_access_level
+    if access_level not in SHARE_ACCESS_LEVELS:
+        raise ValueError(invalid_access_level_message)
+    if allowed_access_levels is not None and access_level not in allowed_access_levels:
+        raise ValueError(unauthorized_access_level_message or invalid_access_level_message)
+
+    if access_level == "global":
+        return EMPTY_SHARE_CONFIG.copy()
+
+    if access_level == "department":
+        department_ids = _normalize_department_ids(config.get("department_ids"))
+        if department_id is not None:
+            department_ids.append(int(department_id))
+        department_ids = sorted(set(department_ids))
+        if not department_ids:
+            raise ValueError("部门共享至少需要选择一个部门")
+        return {"access_level": "department", "department_ids": department_ids, "user_uids": []}
+
+    user_uids = _normalize_user_uids(config.get("user_uids"))
+    if user_uid:
+        user_uids.append(str(user_uid))
+    user_uids = sorted(set(user_uids))
+    if not user_uids:
+        raise ValueError("指定人至少需要选择一个用户")
+    return {"access_level": "user", "department_ids": [], "user_uids": user_uids}

backend/server/utils/lifespan.py

@@ -47,6 +47,7 @@ async def lifespan(app: FastAPI):
         async with pg_manager.get_async_session_context() as session:
             repository = AgentRepository(session)
             await repository.ensure_default_agent()
+            await repository.ensure_general_purpose_subagent()
             await repository.ensure_web_search_subagent()
             await repository.ensure_deep_research_agents()
     except Exception as e:

backend/test/unit/repositories/test_agent_repository.py

@@ -9,6 +9,10 @@ from yuxi.repositories.agent_repository import (
     AgentRepository,
     DEFAULT_AGENT_DESCRIPTION,
     DEFAULT_SHARE_CONFIG,
+    GENERAL_PURPOSE_AGENT_DESCRIPTION,
+    GENERAL_PURPOSE_AGENT_NAME,
+    GENERAL_PURPOSE_AGENT_SLUG,
+    SUB_AGENT_BACKEND_ID,
     user_can_access_agent,
     user_can_manage_agent,
 )
@@ -69,6 +73,51 @@ async def test_ensure_default_agent_backfills_missing_description(monkeypatch):
     db.refresh.assert_awaited_once_with(agent)
 
 
+@pytest.mark.asyncio
+async def test_ensure_general_purpose_subagent_creates_empty_config_subagent(monkeypatch):
+    db = FakeDb()
+    repo = AgentRepository(db)
+
+    async def get_by_slug(_slug):
+        return None
+
+    monkeypatch.setattr(repo, "get_by_slug", get_by_slug)
+
+    agent = await repo.ensure_general_purpose_subagent(created_by="system")
+
+    assert agent.slug == GENERAL_PURPOSE_AGENT_SLUG
+    assert agent.name == GENERAL_PURPOSE_AGENT_NAME
+    assert agent.description == GENERAL_PURPOSE_AGENT_DESCRIPTION
+    assert agent.backend_id == SUB_AGENT_BACKEND_ID
+    assert agent.is_subagent is True
+    assert agent.is_default is False
+    assert agent.config_json == {"context": {}}
+    assert agent.share_config == DEFAULT_SHARE_CONFIG
+    assert agent.created_by == "system"
+    assert db.added is agent
+    db.commit.assert_awaited_once()
+    db.refresh.assert_awaited_once_with(agent)
+
+
+@pytest.mark.asyncio
+async def test_ensure_general_purpose_subagent_is_idempotent(monkeypatch):
+    db = FakeDb()
+    repo = AgentRepository(db)
+    existing = SimpleNamespace(slug=GENERAL_PURPOSE_AGENT_SLUG, config_json={"context": {"model": "custom:model"}})
+
+    async def get_by_slug(_slug):
+        return existing
+
+    monkeypatch.setattr(repo, "get_by_slug", get_by_slug)
+
+    agent = await repo.ensure_general_purpose_subagent()
+
+    assert agent is existing
+    assert db.added is None
+    db.commit.assert_not_awaited()
+    db.refresh.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_create_agent_for_normal_user_forces_private_share(monkeypatch):
     db = FakeDb()

backend/test/unit/utils/test_share_config.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import pytest
+
+from yuxi.utils.share_config import normalize_share_config
+
+
+def test_normalize_share_config_defaults_to_global() -> None:
+    result = normalize_share_config(
+        None,
+        default_config={"access_level": "global", "department_ids": [], "user_uids": []},
+        default_access_level="global",
+        invalid_access_level_message="无效的权限等级",
+    )
+
+    assert result == {"access_level": "global", "department_ids": [], "user_uids": []}
+
+
+def test_normalize_share_config_department_adds_actor_department_and_deduplicates() -> None:
+    result = normalize_share_config(
+        {"access_level": "department", "department_ids": ["2", 1], "user_uids": ["ignored"]},
+        default_config={"access_level": "global", "department_ids": [], "user_uids": []},
+        default_access_level="global",
+        invalid_access_level_message="无效的权限等级",
+        department_id="2",
+    )
+
+    assert result == {"access_level": "department", "department_ids": [1, 2], "user_uids": []}
+
+
+def test_normalize_share_config_user_adds_actor_and_deduplicates() -> None:
+    result = normalize_share_config(
+        {"access_level": "user", "department_ids": [1], "user_uids": [" other ", "owner", ""]},
+        default_config={"access_level": "user", "department_ids": [], "user_uids": []},
+        default_access_level="user",
+        invalid_access_level_message="无效的权限等级",
+        user_uid="owner",
+    )
+
+    assert result == {"access_level": "user", "department_ids": [], "user_uids": ["other", "owner"]}
+
+
+def test_normalize_share_config_rejects_disallowed_access_level() -> None:
+    with pytest.raises(ValueError, match="无权使用该共享范围"):
+        normalize_share_config(
+            {"access_level": "global", "department_ids": [], "user_uids": []},
+            default_config={"access_level": "user", "department_ids": [], "user_uids": []},
+            default_access_level="user",
+            invalid_access_level_message="无效的权限等级",
+            allowed_access_levels={"user"},
+            unauthorized_access_level_message="无权使用该共享范围",
+        )