feat: 增加 Skill 预加载机制

ARCHITECTURE.md

@@ -105,7 +105,7 @@ Yuxi 是一个面向 RAG、知识图谱和多智能体工作流的知识库平
 - 跨 repository 的身份管理用例只有一个 service 事务 Owner；Department、User 与强制 OperationLog 同一提交。API Key 由独立服务端主密钥和客户端幂等 ID 确定性派生，只保存 hash；原始创建意图使用不可变指纹校验，撤销保留 request-id tombstone，同一请求可恢复响应但不能复活已撤销凭据。
 - 前端 API 调用集中在 `web/src/apis`，组件不要散落拼接普通 HTTP 接口。
 - 智能体能力通过 context、middleware、toolkits、Skills、MCP 和 backends 组合；不要把知识库、沙盒或扩展逻辑硬编码进单个页面或路由。
-- Skill 依赖工具只有在对应 Skill 激活后才对模型开放；基础工具与受 Skill 门控的工具要保持边界。
+- Skill 的依赖工具只有在对应 Skill 被显式预加载或动态激活后才对模型开放；基础工具与受 Skill 门控的工具保持边界。
 - LITE 模式必须允许跳过知识库、图谱和评估等重依赖能力，新增导入、路由和启动逻辑时要尊重该边界。
 - 文件边界只使用三种跨层路径：数据库 `projects/<uuid>`、Viewer 当前 scope 相对 `/foo`、Agent/artifact runtime 绝对 `/home/gem/user-data/...`；宿主 `Path` 由 `yuxi.workspace` 或显式 v0.7.1 storage migration 内部持有，普通 Service/Repository 不得取得。
 - 沙盒虚拟路径由当前 Project Workdir、User Data 与共享 Skills 根共同约束；个人 Skill 保存在 UserWorkspace 的 `agents/skills`，共享与内置 Skill 才投影到只读 `/home/gem/skills`。用户可见路径、对象存储 URL 与宿主机真实路径不能混用。

backend/package/yuxi/agents/context.py

@@ -139,9 +139,6 @@ def _lite_mode_enabled() -> bool:
     return lite_mode_enabled()
 
 
-_LITE_DISABLED_SKILL_SLUGS = frozenset({"knowledge-base"})
-
-
 @dataclass(kw_only=True)
 class BaseContext:
     """
@@ -269,6 +266,17 @@ def update(self, data: dict):
         },
     )
 
+    preload_skills: list[str] = field(
+        default_factory=list,
+        metadata={
+            "name": "预加载 Skills",
+            "options": [],
+            "description": "创建 Agent Graph 时加载完整 Skill 说明，并从首轮开放其依赖工具。默认不预加载。",
+            "type": "list",
+            "kind": "skills",
+        },
+    )
+
     summary_threshold: int = field(
         default=DEFAULT_SUMMARY_THRESHOLD_K,
         metadata={
@@ -498,13 +506,14 @@ async def resolve_agent_resource_options(
             if server.slug in enabled_slugs
         ]
     if "skills" in fields_to_load:
+        from yuxi.agents.skills.runtime import is_skill_allowed_in_runtime_mode
         from yuxi.agents.skills.service import list_accessible_skills
 
         skills = await list_accessible_skills(db, user)
         options["skills"] = [
             _resource_option(skill.slug, skill.name, skill.description)
             for skill in skills
-            if skill.slug and not (_lite_mode_enabled() and skill.slug in _LITE_DISABLED_SKILL_SLUGS)
+            if skill.slug and is_skill_allowed_in_runtime_mode(skill.slug)
         ]
     if "subagents" in fields_to_load:
         from yuxi.repositories.agent_repository import AgentRepository
@@ -530,25 +539,26 @@ async def normalize_agent_context_config(
     normalized = dict(filtered.get("context") or {})
     field_names = {item.name for item in fields(schema)}
     resource_fields = _AGENT_RESOURCE_FIELDS & field_names
-    if not resource_fields:
-        return normalized
-
     fields_to_load = _resource_fields_requiring_available_keys(normalized, resource_fields)
-    if not fields_to_load:
-        return normalized
+    if fields_to_load:
+        resource_options = await resolve_agent_resource_options(fields_to_load, db=db, user=user)
+        available = {
+            field_name: [option["key"] for option in field_options]
+            for field_name, field_options in resource_options.items()
+        }
 
-    resource_options = await resolve_agent_resource_options(fields_to_load, db=db, user=user)
-    available = {
-        field_name: [option["key"] for option in field_options]
-        for field_name, field_options in resource_options.items()
-    }
+        for field_name, available_keys in available.items():
+            current = normalized.get(field_name)
+            if current is None:
+                normalized[field_name] = available_keys
+            else:
+                normalized[field_name] = _normalize_selected_resource_keys(current, available_keys)
 
-    for field_name, available_keys in available.items():
-        current = normalized.get(field_name)
-        if current is None:
-            normalized[field_name] = available_keys
-        else:
-            normalized[field_name] = _normalize_selected_resource_keys(current, available_keys)
+    if "preload_skills" in field_names:
+        normalized["preload_skills"] = _normalize_selected_resource_keys(
+            normalized.get("preload_skills"),
+            normalized.get("skills", []),
+        )
 
     return normalized
 
@@ -569,22 +579,25 @@ async def prepare_agent_runtime_context(
     from yuxi.storage.postgres.manager import pg_manager
 
     resource_fields = _AGENT_RESOURCE_FIELDS
+    context_resource_fields = resource_fields | {"preload_skills"}
     async with pg_manager.get_async_session_context() as db:
         if not str(getattr(context, "model", "") or "").strip():
             setattr(context, "model", (await system_options.get(db))["default_model"])
         user = await UserRepository().get_by_uid_with_db(db, uid)
         if user is None:
-            for field_name in resource_fields:
+            for field_name in context_resource_fields:
                 if hasattr(context, field_name):
                     setattr(context, field_name, [])
             setattr(context, "_visible_knowledge_bases", [])
             setattr(context, "_effective_skill_slugs", [])
             setattr(context, "_runtime_skills", {})
+            setattr(context, "_preloaded_skills", [])
+            setattr(context, "_preloaded_skill_contents", {})
             return context
 
         raw_resources = {
             field_name: getattr(context, field_name, None)
-            for field_name in resource_fields
+            for field_name in context_resource_fields
             if hasattr(context, field_name)
         }
         normalized = await normalize_agent_context_config(
@@ -593,7 +606,7 @@ async def prepare_agent_runtime_context(
             user=user,
             context_schema=schema,
         )
-        for field_name in resource_fields:
+        for field_name in context_resource_fields:
             if hasattr(context, field_name):
                 setattr(context, field_name, normalized.get(field_name, []))
 
@@ -604,9 +617,14 @@ async def prepare_agent_runtime_context(
             from yuxi.agents.backends.knowledge_base_backend import resolve_visible_knowledge_bases_for_context
 
             await resolve_visible_knowledge_bases_for_context(context)
-        skill_scope = await resolve_runtime_skills_for_context(context, db=db, user=user)
+        skill_scope = getattr(context, "_skill_runtime_snapshot", None)
+        if not isinstance(skill_scope, dict):
+            skill_scope = await resolve_runtime_skills_for_context(context, db=db, user=user)
         context.skills = skill_scope["context_skills"]
+        context.preload_skills = skill_scope["context_preload_skills"]
         setattr(context, "_effective_skill_slugs", skill_scope["effective_skills"])
         setattr(context, "_runtime_skills", skill_scope["runtime_skills"])
+        setattr(context, "_preloaded_skills", skill_scope["preloaded_skills"])
+        setattr(context, "_preloaded_skill_contents", skill_scope["preloaded_skill_contents"])
 
     return context

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

@@ -33,12 +33,12 @@ class SkillsState(AgentState):
 
 
 class SkillsMiddleware(AgentMiddleware):
-    """Skills 中间件 - 处理 skills 提示词注入、依赖展开、动态激活
+    """Skills 中间件 - 处理提示词注入、预加载、依赖展开和动态激活
 
     职责：
-    - Skills 提示词注入（直接从数据库加载）
-    - 依赖展开（用户配置 + 动态激活）
-    - 工具/MCP 动态加载
+    - Skills 摘要提示与预加载完整说明注入
+    - 依赖展开（预加载配置 + 动态激活）
+    - 本地/MCP 依赖工具的模型可见性门控
     """
 
     state_schema = SkillsState
@@ -72,14 +72,21 @@ async def awrap_model_call(
             effective_skills = getattr(runtime_context, "_effective_skill_slugs", None)
             if isinstance(effective_skills, list):
                 effective_skills = normalize_string_list(effective_skills)
-                if effective_skills:
-                    skills_meta = self._collect_prompt_metadata(effective_skills, runtime_context)
+                preloaded_skills = self._get_preloaded_skills(runtime_context)
+                preloaded_set = set(preloaded_skills)
+                prompt_sections: list[str] = []
+                lazy_skills = [slug for slug in effective_skills if slug not in preloaded_set]
+                if lazy_skills:
+                    skills_meta = self._collect_prompt_metadata(lazy_skills, runtime_context)
                     if skills_meta:
-                        skills_section = self._build_skills_section(skills_meta)
-                        system_message = append_to_system_message(
-                            getattr(request, "system_message", None), skills_section
-                        )
-                        request = request.override(system_message=system_message)
+                        prompt_sections.append(self._build_skills_section(skills_meta))
+                if preloaded_skills:
+                    prompt_sections.append(self._build_preloaded_skills_section(preloaded_skills, runtime_context))
+                if prompt_sections:
+                    system_message = append_to_system_message(
+                        getattr(request, "system_message", None), "\n\n".join(prompt_sections)
+                    )
+                    request = request.override(system_message=system_message)
 
         state = request.state if isinstance(request.state, dict) else {}
         activated = state.get("activated_skills", []) or []
@@ -88,6 +95,7 @@ async def awrap_model_call(
 
         effective_skills = self._get_effective_skills(runtime_context)
         activated = [slug for slug in normalize_string_list(activated) if slug in effective_skills]
+        activated = _activated_skills_reducer(self._get_preloaded_skills(runtime_context), activated)
 
         deps_bundle = build_dependency_bundle(activated, self._get_runtime_skills(runtime_context))
         activated_tool_names = set(deps_bundle["tools"])
@@ -100,22 +108,34 @@ async def awrap_model_call(
         if gated_tool_names:
             model_tools = [t for t in model_tools if t.name not in gated_tool_names]
 
-        # 追加已激活 Skill 的依赖工具：本地工具确保绑定给模型，MCP 工具按需加载
+        # 追加已激活或预加载 Skill 的依赖工具。
         enabled_tools = []
+        active_mcp_tools = []
         if activated_tool_names:
             enabled_tools = [t for t in get_all_tool_instances() if t.name in activated_tool_names]
         if deps_bundle["mcps"]:
-            enabled_tools.extend(
-                await self._get_mcp_tools_from_context(runtime_context, extra_mcps=deps_bundle["mcps"])
-            )
+            active_mcp_tools = await self._get_mcp_tools_from_context(runtime_context, extra_mcps=deps_bundle["mcps"])
+        active_mcp_tools_by_name = {}
+        for tool in active_mcp_tools:
+            existing = active_mcp_tools_by_name.get(tool.name)
+            if existing is not None and existing is not tool:
+                raise RuntimeError(f"Skill MCP 工具名冲突：{tool.name}")
+            active_mcp_tools_by_name[tool.name] = tool
+        setattr(runtime_context, "_active_skill_mcp_tools", active_mcp_tools_by_name)
 
         existing_tool_names = {t.name for t in model_tools}
         for t in enabled_tools:
-            if t.name not in existing_tool_names:
-                model_tools.append(t)
-                existing_tool_names.add(t.name)
-
-        if gated_tool_names or enabled_tools:
+            if t.name in existing_tool_names:
+                continue
+            model_tools.append(t)
+            existing_tool_names.add(t.name)
+        for t in active_mcp_tools:
+            if t.name in existing_tool_names:
+                raise RuntimeError(f"Skill MCP 工具名冲突：{t.name}")
+            model_tools.append(t)
+            existing_tool_names.add(t.name)
+
+        if gated_tool_names or enabled_tools or active_mcp_tools:
             request = request.override(tools=model_tools)
 
         return await handler(request)
@@ -130,18 +150,6 @@ def _resolve_gated_tool_names(self, runtime_context) -> set[str]:
             gated.update(runtime_skills.get(slug, {}).get("tools", []))
         return gated - base_tool_names
 
-    def _collect_prompt_metadata(self, slugs: list[str], runtime_context) -> list[RuntimeSkill]:
-        """收集指定 slugs 的提示词元数据"""
-        runtime_skills = self._get_runtime_skills(runtime_context)
-        result: list[RuntimeSkill] = []
-        for slug in slugs:
-            item = runtime_skills.get(slug)
-            if not item:
-                logger.debug(f"Skill slug not found in prompt metadata, skip: {slug}")
-                continue
-            result.append(dict(item))
-        return result
-
     async def _get_mcp_tools_from_context(
         self,
         context,
@@ -151,14 +159,11 @@ async def _get_mcp_tools_from_context(
         """从上下文配置中获取 MCP 工具列表"""
         import asyncio
 
-        # MCP 工具（并行加载）
-        mcps = getattr(context, "mcps", None) or []
+        # 显式 MCP 已在 Graph 基础工具中注册，这里只加载 Skill 新增依赖。
+        configured_mcps = set(normalize_string_list(getattr(context, "mcps", None)))
         all_mcp_names: list[str] = []
-        for server_name in mcps:
-            if isinstance(server_name, str):
-                all_mcp_names.append(server_name)
         for server_name in extra_mcps or []:
-            if isinstance(server_name, str):
+            if isinstance(server_name, str) and server_name not in configured_mcps:
                 all_mcp_names.append(server_name)
 
         # 去重
@@ -183,6 +188,18 @@ async def load_mcp_tools(server_name: str) -> list:
 
         return selected_tools
 
+    def _collect_prompt_metadata(self, slugs: list[str], runtime_context) -> list[RuntimeSkill]:
+        """收集指定 slugs 的提示词元数据"""
+        runtime_skills = self._get_runtime_skills(runtime_context)
+        result: list[RuntimeSkill] = []
+        for slug in slugs:
+            item = runtime_skills.get(slug)
+            if not item:
+                logger.debug(f"Skill slug not found in prompt metadata, skip: {slug}")
+                continue
+            result.append(dict(item))
+        return result
+
     def _process_tool_call_result(self, result: Any, request: ToolCallRequest) -> Any:
         """处理工具调用结果，检查并处理 skill 动态激活"""
         if request.tool_call.get("name") != "read_file":
@@ -208,6 +225,7 @@ async def awrap_tool_call(
         handler: Callable[[ToolCallRequest], Any],
     ):
         """包装工具调用，处理 skill 动态激活"""
+        request = self._bind_active_mcp_tool(request)
         result = await handler(request)
         return self._process_tool_call_result(result, request)
 
@@ -217,9 +235,22 @@ def wrap_tool_call(
         handler: Callable[[ToolCallRequest], Any],
     ):
         """同步版本的工具调用包装"""
+        request = self._bind_active_mcp_tool(request)
         result = handler(request)
         return self._process_tool_call_result(result, request)
 
+    def _bind_active_mcp_tool(self, request: ToolCallRequest) -> ToolCallRequest:
+        """为当前模型轮次已开放的动态 MCP 调用绑定真实工具。"""
+
+        if request.tool is not None:
+            return request
+        context = request.runtime.context
+        active_tools = getattr(context, "_active_skill_mcp_tools", {})
+        if not isinstance(active_tools, dict):
+            return request
+        tool = active_tools.get(request.tool_call.get("name"))
+        return request.override(tool=tool) if tool is not None else request
+
     def _extract_skill_slug_from_skill_md_path(self, file_path: Any) -> str | None:
         """从共享投影或个人 UserWorkspace 的 SKILL.md 路径中提取 slug。"""
         if not isinstance(file_path, str):
@@ -245,6 +276,26 @@ def _get_runtime_skills(self, runtime_context) -> dict[str, RuntimeSkill]:
         runtime_skills = getattr(runtime_context, "_runtime_skills", {})
         return runtime_skills if isinstance(runtime_skills, dict) else {}
 
+    def _get_preloaded_skills(self, runtime_context) -> list[str]:
+        selected = getattr(runtime_context, "_preloaded_skills", [])
+        effective = self._get_effective_skills(runtime_context)
+        return [
+            slug for slug in normalize_string_list(selected if isinstance(selected, list) else []) if slug in effective
+        ]
+
+    def _build_preloaded_skills_section(self, slugs: list[str], runtime_context) -> str:
+        """构建已预加载 Skill 的完整系统提示段。"""
+
+        contents = getattr(runtime_context, "_preloaded_skill_contents", {})
+        if not isinstance(contents, dict):
+            contents = {}
+        sections = ["# Preloaded Skills", "The following Skill instructions are already loaded and active."]
+        for slug in slugs:
+            content = contents.get(slug)
+            if isinstance(content, str):
+                sections.append(f'<preloaded_skill slug="{slug}">\n{content}\n</preloaded_skill>')
+        return "\n\n".join(sections)
+
     def _merge_activated_skill_update(self, result: Any, slug: str):
         """合并动态激活的 skill 更新"""
         from langchain_core.messages import ToolMessage