feat(skills): add progressive skill loading with disclosure_level=0 (U5)

When disclosure_level=0, system prompt only injects skill name + description
(summary mode). SkillDetailTool is injected into the tool set, allowing the
LLM to load full instructions on-demand via skill_detail(query). This reduces
context window consumption when many skills are registered.
This commit is contained in:
chiguyong 2026-06-24 21:49:00 +08:00
parent dfd188b1a4
commit fa152e24ac
4 changed files with 497 additions and 3 deletions

View File

@ -211,6 +211,14 @@ class RequestPreprocessor:
skill_prompt = build_skill_system_prompt(skill_config)
execution_mode = _resolve_execution_mode(skill_config)
# U5: 渐进式技能加载 — disclosure_level == 0 时注入 skill_detail 工具
merged_tools = list(skill_tools)
if getattr(skill_config, "disclosure_level", 1) == 0:
from agentkit.skills.skill_detail import SkillDetailTool
if not any(t.name == "skill_detail" for t in merged_tools):
merged_tools.append(SkillDetailTool(skill_registry=registry))
return SkillRoutingResult(
clean_content=clean_content,
matched=True,
@ -222,7 +230,7 @@ class RequestPreprocessor:
agent_name=skill_name,
model=model,
system_prompt=skill_prompt,
tools=skill_tools,
tools=merged_tools,
execution_mode=execution_mode,
)

View File

@ -103,8 +103,37 @@ def build_skill_system_prompt(skill_config) -> str | None:
v7: skill_config.preconditions 非空在基础 prompt 后追加
## Activation Preconditions 段落(软检查,见 KTD1
U5: 渐进式技能加载 disclosure_level == 0 只注入 skill 名称 + 描述
概要模式LLM 可调用 skill_detail 工具按需加载完整 instructions
disclosure_level >= 1默认时行为不变全量加载向后兼容
"""
if not skill_config or not skill_config.prompt:
if not skill_config:
return None
# U5: Level 0 — 概要模式,只注入 name + description
disclosure_level = getattr(skill_config, "disclosure_level", 1)
if disclosure_level == 0:
name = getattr(skill_config, "name", "unknown")
description = getattr(skill_config, "description", "")
summary = f"## Skill: {name}\n{description}" if description else f"## Skill: {name}"
# 安全守卫:前置条件即使在概要模式下也必须注入
preconditions = getattr(skill_config, "preconditions", None)
if preconditions:
lines = ["## Activation Preconditions", "Before executing this skill, verify:"]
lines.extend(f"- {p}" for p in preconditions)
lines.append(
"If any precondition is not met, refuse to execute or ask the user for clarification."
)
return f"{summary}\n\n" + "\n".join(lines)
# 提示 LLM 可通过 skill_detail 工具加载完整 instructions
return (
f"{summary}\n\n"
"*(Call the skill_detail tool with this skill name to load full instructions)*"
)
# Level 1+: 全量加载(现有行为)
if not skill_config.prompt:
return None
prompt_parts = []
for key in ("identity", "context", "instructions", "constraints", "output_format"):
@ -183,6 +212,13 @@ async def resolve_skill_routing(
if tool.name not in seen_names:
seen_names.add(tool.name)
merged_tools.append(tool)
# U5: 渐进式技能加载 — disclosure_level == 0 时注入 skill_detail 工具
if getattr(result.skill_config, "disclosure_level", 1) == 0 and skill_registry is not None:
from agentkit.skills.skill_detail import SkillDetailTool
if "skill_detail" not in seen_names:
merged_tools.append(SkillDetailTool(skill_registry=skill_registry))
seen_names.add("skill_detail")
result.tools = merged_tools
result.model = (

View File

@ -0,0 +1,143 @@
"""SkillDetailTool — 渐进式技能加载工具 (U5)
skill_config.disclosure_level == 0 system prompt 只注入 skill 名称 + 描述
LLM 可调用此工具按需加载完整 skill instructionsidentity/context/instructions/
constraints/output_format从本地 SkillRegistry 检索
tools/skill_search.py SkillSearchTool 区别
- SkillSearchTool 搜索 npx 外部市场安装前发现
- SkillDetailTool 加载本地已注册 skill 的完整内容执行前加载
"""
from __future__ import annotations
import logging
from typing import Any
from agentkit.tools.base import Tool
logger = logging.getLogger(__name__)
class SkillDetailTool(Tool):
"""按需加载本地已注册 skill 的完整 instructions。
用于渐进式技能加载disclosure_level=0system prompt 只含 skill 概要
LLM 决定执行该 skill 时调用此工具获取完整 instructions
Usage::
from agentkit.skills.skill_detail import SkillDetailTool
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="research")
"""
def __init__(
self,
skill_registry: Any,
name: str = "skill_detail",
description: str = (
"Load full instructions for a registered skill by name or keyword. "
"Use this when the system prompt only shows a skill summary (name + description) "
"and you need the complete instructions to execute the skill. "
"Returns the skill's identity, context, instructions, constraints, and output format."
),
input_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
version: str = "1.0.0",
tags: list[str] | None = None,
):
schema = input_schema or {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": (
"Skill name or keyword to search for. "
"Returns the full instructions of the best-matching registered skill."
),
},
},
"required": ["query"],
}
super().__init__(
name=name,
description=description,
input_schema=schema,
output_schema=output_schema,
version=version,
tags=tags or ["skill", "search", "meta"],
)
self._registry = skill_registry
async def execute(self, **kwargs) -> dict:
"""执行 skill 详情加载。
Args:
query: skill 名称或关键词
Returns:
包含 ``name````description````full_instructions`` 的字典
无匹配时返回 ``message`` 提示
"""
query = str(kwargs.get("query", "")).strip()
if not query:
return {
"error": "query parameter is required",
"results": [],
}
# 精确名称匹配优先
try:
skill = self._registry.get(query)
return self._format_skill_full(skill)
except Exception:
pass # 非精确匹配,降级到关键词搜索
# 关键词搜索:匹配 skill 名称和描述
matches: list[Any] = []
query_lower = query.lower()
for skill in self._registry.list_skills():
name = skill.name.lower()
desc = (skill.config.description or "").lower()
if query_lower in name or query_lower in desc:
matches.append(skill)
if not matches:
return {
"query": query,
"count": 0,
"results": [],
"message": f"No skills matched '{query}'.",
}
# 返回最佳匹配(第一个)
return self._format_skill_full(matches[0])
@staticmethod
def _format_skill_full(skill: Any) -> dict[str, Any]:
"""格式化 skill 的完整 instructions 供 LLM 使用。"""
config = skill.config
prompt_parts: list[str] = []
for key in ("identity", "context", "instructions", "constraints", "output_format"):
val = (config.prompt or {}).get(key)
if val:
prompt_parts.append(f"### {key.title()}\n{val}")
# v7: 注入激活前置条件(安全守卫,即使在渐进加载模式下也不应省略)
preconditions = getattr(config, "preconditions", None)
if preconditions:
lines = ["### Activation Preconditions", "Before executing this skill, verify:"]
lines.extend(f"- {p}" for p in preconditions)
lines.append(
"If any precondition is not met, refuse to execute or ask the user for clarification."
)
prompt_parts.append("\n".join(lines))
return {
"name": config.name,
"description": config.description,
"version": config.version,
"full_instructions": "\n\n".join(prompt_parts) if prompt_parts else "",
"tools": [t.name for t in (skill.tools or [])],
}

View File

@ -2,15 +2,19 @@
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from agentkit.chat.skill_routing import (
ExecutionMode,
SkillRoutingResult,
build_skill_system_prompt,
)
from agentkit.experts.config import ExpertConfig, ExpertTemplate
from agentkit.experts.registry import ExpertTemplateRegistry
from agentkit.experts.router import ExpertTeamRouter
from agentkit.skills.base import Skill, SkillConfig
from agentkit.skills.registry import SkillRegistry
from agentkit.skills.skill_detail import SkillDetailTool
# ---------------------------------------------------------------------------
@ -118,3 +122,306 @@ class TestSkillRoutingResult:
execution_mode=ExecutionMode.DIRECT_CHAT,
)
assert result.execution_mode == ExecutionMode.DIRECT_CHAT
# ---------------------------------------------------------------------------
# U5: Progressive Skill Loading Tests
# ---------------------------------------------------------------------------
def _make_skill_config(
name: str = "research",
description: str = "Research skill for gathering information",
disclosure_level: int = 1,
preconditions: list[str] | None = None,
) -> SkillConfig:
"""Create a SkillConfig for testing progressive loading."""
return SkillConfig(
name=name,
agent_type="skill",
description=description,
task_mode="llm_generate",
execution_mode="react",
prompt={
"identity": f"You are a {name} expert.",
"context": "Use this skill when research is needed.",
"instructions": "Step 1: gather data. Step 2: analyze.",
"constraints": "Do not hallucinate.",
"output_format": "Markdown report.",
},
disclosure_level=disclosure_level,
preconditions=preconditions,
)
def _make_skill_registry_with_skills(*configs: SkillConfig) -> SkillRegistry:
"""Create a SkillRegistry with the given skill configs registered."""
registry = SkillRegistry()
for config in configs:
registry.register(Skill(config=config, tools=[]))
return registry
class TestProgressiveSkillLoading:
"""U5: 渐进式技能加载 — disclosure_level 控制概要 vs 全量注入。"""
def test_level_0_summary_only_name_and_description(self) -> None:
"""disclosure_level=0 时system_prompt 只含 skill name + description。"""
config = _make_skill_config(
name="research",
description="Research skill for gathering information",
disclosure_level=0,
)
prompt = build_skill_system_prompt(config)
assert prompt is not None
assert "## Skill: research" in prompt
assert "Research skill for gathering information" in prompt
# Should NOT contain full prompt sections
assert "You are a research expert" not in prompt
assert "Step 1: gather data" not in prompt
assert "Do not hallucinate" not in prompt
# Should hint at skill_detail tool
assert "skill_detail" in prompt
def test_level_0_with_preconditions_includes_guard(self) -> None:
"""disclosure_level=0 时,前置条件仍然注入(安全守卫)。"""
config = _make_skill_config(
name="dangerous_op",
description="A dangerous operation skill",
disclosure_level=0,
preconditions=["User must confirm the operation", "API key must be set"],
)
prompt = build_skill_system_prompt(config)
assert prompt is not None
assert "## Skill: dangerous_op" in prompt
assert "## Activation Preconditions" in prompt
assert "User must confirm the operation" in prompt
assert "API key must be set" in prompt
def test_level_1_full_prompt_backward_compatible(self) -> None:
"""disclosure_level=1默认行为与现状一致全量加载"""
config = _make_skill_config(
name="research",
description="Research skill",
disclosure_level=1,
)
prompt = build_skill_system_prompt(config)
assert prompt is not None
assert "You are a research expert." in prompt
assert "Step 1: gather data. Step 2: analyze." in prompt
assert "Do not hallucinate." in prompt
assert "Markdown report." in prompt
# Should NOT contain the skill_detail hint
assert "skill_detail" not in prompt
def test_level_0_no_description_still_works(self) -> None:
"""disclosure_level=0 且无 description 时,仍能生成有效 prompt。"""
config = _make_skill_config(
name="minimal",
description="",
disclosure_level=0,
)
prompt = build_skill_system_prompt(config)
assert prompt is not None
assert "## Skill: minimal" in prompt
assert "skill_detail" in prompt
class TestSkillDetailTool:
"""U5: SkillDetailTool — 按需加载完整 skill instructions。"""
@pytest.mark.asyncio
async def test_exact_name_match_returns_full_instructions(self) -> None:
"""精确名称匹配 → 返回完整 instructions。"""
config = _make_skill_config(name="research", disclosure_level=0)
registry = _make_skill_registry_with_skills(config)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="research")
assert result["name"] == "research"
assert result["description"] == "Research skill for gathering information"
assert "You are a research expert." in result["full_instructions"]
assert "Step 1: gather data" in result["full_instructions"]
assert "Do not hallucinate." in result["full_instructions"]
@pytest.mark.asyncio
async def test_keyword_search_returns_matching_skill(self) -> None:
"""关键词搜索 → 返回匹配的 skill 完整内容。"""
config1 = _make_skill_config(name="research", description="Research skill")
config2 = _make_skill_config(
name="code_review",
description="Review code quality",
)
registry = _make_skill_registry_with_skills(config1, config2)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="review")
assert result["name"] == "code_review"
assert "Review code quality" in result["description"]
@pytest.mark.asyncio
async def test_no_match_returns_message(self) -> None:
"""无匹配 → 返回 'No skills matched' 提示。"""
config = _make_skill_config(name="research")
registry = _make_skill_registry_with_skills(config)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="nonexistent_skill")
assert result["count"] == 0
assert result["results"] == []
assert "No skills matched" in result["message"]
@pytest.mark.asyncio
async def test_empty_query_returns_error(self) -> None:
"""空 query → 返回错误。"""
config = _make_skill_config(name="research")
registry = _make_skill_registry_with_skills(config)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="")
assert "error" in result
assert "query parameter is required" in result["error"]
@pytest.mark.asyncio
async def test_includes_preconditions_in_full_instructions(self) -> None:
"""完整 instructions 中包含前置条件(安全守卫)。"""
config = _make_skill_config(
name="dangerous_op",
preconditions=["User must confirm", "API key set"],
disclosure_level=0,
)
registry = _make_skill_registry_with_skills(config)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="dangerous_op")
assert "### Activation Preconditions" in result["full_instructions"]
assert "User must confirm" in result["full_instructions"]
assert "API key set" in result["full_instructions"]
@pytest.mark.asyncio
async def test_case_insensitive_keyword_search(self) -> None:
"""关键词搜索大小写不敏感。"""
config = _make_skill_config(name="Research", description="Research skill")
registry = _make_skill_registry_with_skills(config)
tool = SkillDetailTool(skill_registry=registry)
result = await tool.execute(query="RESEARCH")
assert result["name"] == "Research"
class TestResolveSkillRoutingProgressive:
"""U5: resolve_skill_routing 在 disclosure_level=0 时注入 skill_detail 工具。"""
@pytest.mark.asyncio
async def test_level_0_adds_skill_detail_tool(self) -> None:
"""disclosure_level=0 时routing 结果包含 skill_detail 工具。"""
from agentkit.chat.skill_routing import resolve_skill_routing
config = _make_skill_config(name="research", disclosure_level=0)
registry = _make_skill_registry_with_skills(config)
result = await resolve_skill_routing(
content="@skill:research analyze the data",
skill_registry=registry,
default_tools=[],
default_system_prompt="default",
)
assert result.matched is True
assert result.skill_name == "research"
tool_names = [t.name for t in result.tools]
assert "skill_detail" in tool_names
@pytest.mark.asyncio
async def test_level_1_does_not_add_skill_detail_tool(self) -> None:
"""disclosure_level=1默认不注入 skill_detail 工具(向后兼容)。"""
from agentkit.chat.skill_routing import resolve_skill_routing
config = _make_skill_config(name="research", disclosure_level=1)
registry = _make_skill_registry_with_skills(config)
result = await resolve_skill_routing(
content="@skill:research analyze the data",
skill_registry=registry,
default_tools=[],
default_system_prompt="default",
)
assert result.matched is True
tool_names = [t.name for t in result.tools]
assert "skill_detail" not in tool_names
@pytest.mark.asyncio
async def test_level_0_system_prompt_is_summary(self) -> None:
"""disclosure_level=0 时system_prompt 是概要模式。"""
from agentkit.chat.skill_routing import resolve_skill_routing
config = _make_skill_config(name="research", disclosure_level=0)
registry = _make_skill_registry_with_skills(config)
result = await resolve_skill_routing(
content="@skill:research analyze the data",
skill_registry=registry,
default_tools=[],
default_system_prompt="default",
)
assert result.system_prompt is not None
assert "## Skill: research" in result.system_prompt
# Full instructions should NOT be in the system prompt
assert "You are a research expert." not in result.system_prompt
class TestRequestPreprocessorProgressive:
"""U5: RequestPreprocessor._resolve_explicit_skill 在 disclosure_level=0 时注入 skill_detail。"""
@pytest.mark.asyncio
async def test_preprocessor_level_0_adds_skill_detail(self) -> None:
"""RequestPreprocessor 在 disclosure_level=0 时注入 skill_detail 工具。"""
from agentkit.chat.request_preprocessor import RequestPreprocessor
config = _make_skill_config(name="research", disclosure_level=0)
registry = _make_skill_registry_with_skills(config)
preprocessor = RequestPreprocessor()
result = await preprocessor.preprocess(
content="@skill:research analyze the data",
skill_registry=registry,
default_tools=[],
default_system_prompt="default",
)
assert result.matched is True
assert result.skill_name == "research"
tool_names = [t.name for t in result.tools]
assert "skill_detail" in tool_names
@pytest.mark.asyncio
async def test_preprocessor_level_1_no_skill_detail(self) -> None:
"""RequestPreprocessor 在 disclosure_level=1 时不注入 skill_detail向后兼容"""
from agentkit.chat.request_preprocessor import RequestPreprocessor
config = _make_skill_config(name="research", disclosure_level=1)
registry = _make_skill_registry_with_skills(config)
preprocessor = RequestPreprocessor()
result = await preprocessor.preprocess(
content="@skill:research analyze the data",
skill_registry=registry,
default_tools=[],
default_system_prompt="default",
)
assert result.matched is True
tool_names = [t.name for t in result.tools]
assert "skill_detail" not in tool_names