480 lines
20 KiB
Python
480 lines
20 KiB
Python
"""skill_view and skill_list — read-only tools for agent skill introspection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import PurePosixPath
|
|
from typing import TYPE_CHECKING
|
|
|
|
from langchain.tools import ToolRuntime, tool
|
|
|
|
if TYPE_CHECKING:
|
|
from deerflow.skills.storage import SkillStorage
|
|
from deerflow.skills.types import Skill
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_ROUNDTABLE_COORDINATOR_AGENT_ID = "roundtable-coordinator"
|
|
COORDINATOR_AGENT_IDS = frozenset(
|
|
{_ROUNDTABLE_COORDINATOR_AGENT_ID, "task-analysis-coordinator"}
|
|
)
|
|
COORDINATOR_ONLY_SKILL_NAMES = frozenset({"agent_orchestration", "agent-orchestration"})
|
|
|
|
|
|
class SkillPathUnavailableError(PermissionError):
|
|
"""Raised when a model tool targets an unknown or offline skill path."""
|
|
|
|
|
|
def require_enabled_skill_path(
|
|
path: str,
|
|
*,
|
|
container_path: str = "/mnt/skills",
|
|
storage: SkillStorage | None = None,
|
|
) -> Skill:
|
|
"""Return the discovered enabled skill containing ``path`` or deny access.
|
|
|
|
Matching uses the discovered directory instead of assuming it equals the
|
|
frontmatter name. Root/category-wide paths and unregistered files fail
|
|
closed; models discover skills through ``skill_list``/``search_skills``.
|
|
"""
|
|
normalized_container = container_path.replace("\\", "/").rstrip("/") or "/"
|
|
normalized_path = path.replace("\\", "/").rstrip("/") or "/"
|
|
if normalized_path != normalized_container and not normalized_path.startswith(f"{normalized_container}/"):
|
|
raise SkillPathUnavailableError("Path is outside the skills mount")
|
|
if ".." in PurePosixPath(normalized_path).parts:
|
|
raise SkillPathUnavailableError("Skill path traversal is not allowed")
|
|
|
|
try:
|
|
from deerflow.config.extensions_config import ExtensionsConfig
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
|
|
# SkillStorage intentionally degrades open on extensions-config errors;
|
|
# model-facing file access must instead fail closed.
|
|
extensions = ExtensionsConfig.from_file()
|
|
catalog = (storage or get_or_new_skill_storage()).load_skills(enabled_only=False)
|
|
except Exception as exc:
|
|
raise SkillPathUnavailableError("Skill availability could not be verified") from exc
|
|
|
|
matches: list[tuple[int, Skill]] = []
|
|
for skill in catalog:
|
|
skill_root = skill.get_container_path(normalized_container).replace("\\", "/").rstrip("/")
|
|
if normalized_path == skill_root or normalized_path.startswith(f"{skill_root}/"):
|
|
matches.append((len(skill_root), skill))
|
|
if not matches:
|
|
raise SkillPathUnavailableError("Skill path is unavailable")
|
|
|
|
skill = max(matches, key=lambda item: item[0])[1]
|
|
category = skill.category.value if hasattr(skill.category, "value") else str(skill.category)
|
|
if not extensions.is_skill_enabled(skill.name, category):
|
|
raise SkillPathUnavailableError("Skill path is unavailable")
|
|
return skill
|
|
|
|
|
|
def can_access_coordinator_only_skill(agent_id: str | None, skill_name: str) -> bool:
|
|
"""Return whether this runtime identity may discover/read a reserved orchestration skill."""
|
|
normalized = (skill_name or "").strip().lower()
|
|
return normalized not in COORDINATOR_ONLY_SKILL_NAMES or agent_id in COORDINATOR_AGENT_IDS
|
|
|
|
|
|
def is_coordinator_only_skill_path(path: str) -> bool:
|
|
"""Recognize reserved skill directories in virtual/container paths."""
|
|
parts = {part.strip().lower() for part in (path or "").replace("\\", "/").split("/") if part}
|
|
return bool(parts & COORDINATOR_ONLY_SKILL_NAMES)
|
|
|
|
|
|
def _resolve_active_agent_id(runtime: ToolRuntime | None) -> str | None:
|
|
"""Extract the agent_id under which the current tool call is running.
|
|
|
|
Custom agents pass their id via either the ``context`` dict or the
|
|
``configurable`` section of the LangGraph runtime config. The default
|
|
lead agent runs without an explicit agent_id — callers should treat
|
|
a None return as "no per-agent skills filter applies".
|
|
"""
|
|
if runtime is None:
|
|
return None
|
|
try:
|
|
if getattr(runtime, "context", None):
|
|
ctx_id = runtime.context.get("agent_id") or runtime.context.get("agent_name")
|
|
if ctx_id:
|
|
return str(ctx_id)
|
|
cfg = getattr(runtime, "config", None)
|
|
if isinstance(cfg, dict):
|
|
configurable = cfg.get("configurable", {})
|
|
cid = configurable.get("agent_id") or configurable.get("agent_name")
|
|
if cid:
|
|
return str(cid)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _load_agent_skill_whitelist(agent_id: str | None, extra_skills: list[str] | None = None) -> set[str] | None:
|
|
"""Return the explicit skill name set declared in an agent's config.yaml.
|
|
|
|
Semantics:
|
|
- ``None`` returned → no agent_id, or the agent's on-disk config is
|
|
missing entirely — caller should fall back to "all skills visible"
|
|
(default lead agent behavior).
|
|
- ``set()`` (empty) → custom agent with ``skills: []`` declared **or**
|
|
with no ``skills`` key at all — visibility is intentionally empty.
|
|
This mirrors the prompt layer (``agents_config.AgentConfig.skills``:
|
|
``None``/``[]`` both mean "no skills" for custom agents), so an agent
|
|
whose config omits ``skills`` can no longer enumerate every skill
|
|
through this tool.
|
|
- Non-empty set → only these names are visible to this agent.
|
|
|
|
Exception — the deep-research collector (``deep-research-collector``) has no
|
|
whitelist: it sees every skill like the default lead agent, matching the
|
|
runtime allowlist in ``lead_agent.agent``. Its on-disk ``skills:`` entry is
|
|
ignored.
|
|
|
|
``extra_skills`` only widens the **deep-research collector** allowlist
|
|
for one run (report-structure 检索来源). Other agents ignore it.
|
|
"""
|
|
if not agent_id:
|
|
return None
|
|
try:
|
|
from deerflow.agents.deep_research.retrieval import COLLECTOR_AGENT_ID
|
|
from deerflow.config.agents_config import load_agent_config
|
|
|
|
cfg = load_agent_config(agent_id)
|
|
if cfg is None:
|
|
return None
|
|
if agent_id == COLLECTOR_AGENT_ID:
|
|
allowed = None # collector: all skills, like the default lead agent
|
|
else:
|
|
allowed = {
|
|
name
|
|
for name in (cfg.skills or [])
|
|
if can_access_coordinator_only_skill(agent_id, name)
|
|
}
|
|
except Exception:
|
|
logger.debug("Failed to load agent config for skill_list filter", exc_info=True)
|
|
return None
|
|
|
|
if extra_skills:
|
|
from deerflow.agents.deep_research.retrieval import merge_collector_skill_allowlist
|
|
|
|
merged = merge_collector_skill_allowlist(agent_id, None if allowed is None else list(allowed), extra_skills)
|
|
return set(merged or []) if merged is not None else None
|
|
return allowed
|
|
|
|
|
|
def _resolve_extra_skills(runtime: ToolRuntime | None) -> list[str]:
|
|
"""Extra skill names from this run's context (collector 检索来源)."""
|
|
if runtime is None:
|
|
return []
|
|
try:
|
|
from deerflow.agents.deep_research.retrieval import runtime_extra_skills
|
|
|
|
if getattr(runtime, "context", None):
|
|
extras = runtime_extra_skills(runtime.context)
|
|
if extras:
|
|
return extras
|
|
cfg = getattr(runtime, "config", None)
|
|
if isinstance(cfg, dict):
|
|
return runtime_extra_skills(cfg.get("configurable", {}) or {})
|
|
except Exception:
|
|
return []
|
|
return []
|
|
|
|
|
|
def _resolve_effective_user_id() -> str | None:
|
|
"""当前请求/线程绑定的用户 id;解析失败时返回 None。"""
|
|
try:
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
return get_effective_user_id()
|
|
except Exception:
|
|
logger.debug("Failed to resolve effective user id for skill_list", exc_info=True)
|
|
return None
|
|
|
|
|
|
def _load_owned_skill_names(user_id: str | None) -> set[str] | None:
|
|
"""返回 ``owner_user_id == user_id`` 的自定义技能名集合。
|
|
|
|
用于默认 lead agent 的可见性过滤:只算"用户自己拥有的"自定义技能,不含
|
|
其他用户发布的。返回 ``None`` 表示 ``SkillStore`` 不可用(未配置数据库
|
|
等)—— 调用方应降级为"放行全部自定义技能",避免离线无库部署下技能凭空
|
|
消失。
|
|
"""
|
|
if not user_id:
|
|
return None
|
|
try:
|
|
from deerflow.persistence.engine import get_session_factory, run_db_blocking
|
|
from deerflow.persistence.skills import make_skill_store
|
|
|
|
factory = get_session_factory()
|
|
if factory is None:
|
|
return None
|
|
async def read_visible():
|
|
store = make_skill_store(get_session_factory())
|
|
return await store.list_visible(user_id)
|
|
|
|
records = run_db_blocking(read_visible())
|
|
return {r["name"] for r in records if r.get("owner_user_id") == user_id}
|
|
except Exception:
|
|
logger.debug("Failed to load owned skill names for skill_list filter", exc_info=True)
|
|
return None
|
|
|
|
|
|
def skill_view_impl(name: str, agent_id: str | None, extra_skills: list[str] | None = None) -> str:
|
|
"""Read a skill's SKILL.md, honoring the agent's config.yaml whitelist.
|
|
|
|
Custom agents may only read skills on their whitelist — same filter as
|
|
``skill_list``, so a restricted agent cannot bypass its allowlist by
|
|
guessing skill names. Replies "not found" rather than "forbidden" to
|
|
avoid leaking which skills exist.
|
|
"""
|
|
if not can_access_coordinator_only_skill(agent_id, name):
|
|
return f"Skill '{name}' not found."
|
|
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
from deerflow.skills.usage import bump_view
|
|
|
|
allowed = _load_agent_skill_whitelist(agent_id, extra_skills)
|
|
if allowed is not None and name not in allowed:
|
|
return f"Skill '{name}' not found."
|
|
|
|
storage = get_or_new_skill_storage()
|
|
try:
|
|
# A warm catalogue already knows each skill's file. Read that one file.
|
|
# A cold catalogue checks the canonical directory before walking every
|
|
# SKILL.md, which is only needed for a directory whose name differs.
|
|
from deerflow.agents.lead_agent.prompt import _load_named_enabled_skill, get_warm_enabled_skills
|
|
|
|
cached = get_warm_enabled_skills(storage)
|
|
if cached is not None:
|
|
skill = next((item for item in cached if item.name == name), None)
|
|
else:
|
|
skill = _load_named_enabled_skill(name)
|
|
if skill is None:
|
|
skill = next((item for item in storage.load_skills(enabled_only=True) if item.name == name), None)
|
|
if skill is None:
|
|
return f"Skill '{name}' not found."
|
|
content = skill.skill_file.read_text(encoding="utf-8")
|
|
except Exception as e:
|
|
logger.debug("skill_view failed for %r: %s", name, e)
|
|
return f"Error reading skill '{name}': {e}"
|
|
|
|
try:
|
|
bump_view(name)
|
|
except Exception:
|
|
pass
|
|
|
|
return content
|
|
|
|
|
|
@tool("skill_view")
|
|
def skill_view_tool(name: str, runtime: ToolRuntime) -> str:
|
|
"""Read the full SKILL.md content of a skill by name.
|
|
|
|
Args:
|
|
name: The skill name (hyphen-case, e.g. 'my-skill').
|
|
"""
|
|
return skill_view_impl(name, _resolve_active_agent_id(runtime), _resolve_extra_skills(runtime))
|
|
|
|
|
|
def _resolve_skills_container_base_path() -> str:
|
|
"""容器内技能挂载根路径(用于回传 location,供 agent read_file)。"""
|
|
try:
|
|
from deerflow.config import get_app_config
|
|
|
|
return get_app_config().skills.container_path
|
|
except Exception:
|
|
return "/mnt/skills"
|
|
|
|
|
|
def _score_skill(query_terms: list[str], haystack: str) -> int:
|
|
"""对单个技能按关键词命中计分(纯关键词、离线、确定性)。"""
|
|
score = 0
|
|
for term in query_terms:
|
|
if term and term in haystack:
|
|
score += 1
|
|
return score
|
|
|
|
|
|
def search_skills_impl(query: str, agent_id: str | None, top_n: int = 5, extra_skills: list[str] | None = None) -> str:
|
|
"""按关键词检索可用技能,返回 top-N 的 name/description/category/location。
|
|
|
|
与 ``skill_list`` 共享可见性过滤(自定义 agent 走白名单;默认 lead agent 只
|
|
暴露未归档且内置或属于当前用户的技能),并只检索**已启用**的技能 —— 它是
|
|
"技能压缩"开启后,agent 调出被压缩(非常驻)技能的主要入口。匹配字段为技能
|
|
名与描述。
|
|
"""
|
|
import re
|
|
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
from deerflow.skills.types import SkillCategory
|
|
from deerflow.skills.usage import STATE_ARCHIVED, all_entries
|
|
|
|
query = (query or "").strip()
|
|
if not query:
|
|
return "search_skills: empty query. Provide a short keyword query describing the task."
|
|
|
|
terms = [t for t in re.findall(r"\w+", query.lower()) if t]
|
|
# 整串也作为一个匹配项,兼顾中文等不被 \w+ 切分出有意义 token 的查询。
|
|
whole = query.lower()
|
|
if whole and whole not in terms:
|
|
terms.append(whole)
|
|
|
|
allowed = _load_agent_skill_whitelist(agent_id, extra_skills)
|
|
owned: set[str] | None = None
|
|
if allowed is None:
|
|
owned = _load_owned_skill_names(_resolve_effective_user_id())
|
|
|
|
try:
|
|
top_n = max(1, min(int(top_n or 5), 50))
|
|
except (TypeError, ValueError):
|
|
top_n = 5
|
|
|
|
try:
|
|
storage = get_or_new_skill_storage()
|
|
skills = _skills_for_catalogue(storage)
|
|
usage = all_entries()
|
|
base_path = _resolve_skills_container_base_path()
|
|
scored: list[tuple[int, dict]] = []
|
|
for skill in skills:
|
|
if not can_access_coordinator_only_skill(agent_id, skill.name):
|
|
continue
|
|
entry = usage.get(skill.name, {})
|
|
if allowed is not None:
|
|
if skill.name not in allowed:
|
|
continue
|
|
else:
|
|
if entry.get("state") == STATE_ARCHIVED:
|
|
continue
|
|
if skill.category == SkillCategory.CUSTOM and owned is not None and skill.name not in owned:
|
|
continue
|
|
haystack = f"{skill.name} {skill.description}".lower()
|
|
score = _score_skill(terms, haystack)
|
|
if score <= 0:
|
|
continue
|
|
scored.append((
|
|
score,
|
|
{
|
|
"name": skill.name,
|
|
"description": skill.description,
|
|
"category": skill.category.value if hasattr(skill.category, "value") else str(skill.category),
|
|
"location": skill.get_container_file_path(base_path),
|
|
},
|
|
))
|
|
scored.sort(key=lambda x: x[0], reverse=True)
|
|
results = [item for _, item in scored[:top_n]]
|
|
if not results:
|
|
return f"No skills matched '{query}'. Try a broader keyword, or call skill_list to browse everything available."
|
|
return json.dumps(results, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
logger.debug("search_skills failed: %s", e)
|
|
return f"Error searching skills: {e}"
|
|
|
|
|
|
@tool("search_skills")
|
|
def search_skills_tool(query: str, runtime: ToolRuntime) -> str:
|
|
"""Search available skills by keyword and return the best matches.
|
|
|
|
Use this to discover skills that are not already described in full in your
|
|
system prompt. Returns a JSON array of matches, each with `name`,
|
|
`description`, `category`, and `location` — then `read_file` the `location`
|
|
of the most relevant match and follow its SKILL.md instructions.
|
|
|
|
Args:
|
|
query: A short keyword query describing the task (e.g. 'pdf table extract').
|
|
"""
|
|
return search_skills_impl(query, _resolve_active_agent_id(runtime), extra_skills=_resolve_extra_skills(runtime))
|
|
|
|
|
|
@tool("skill_list")
|
|
def skill_list_tool(runtime: ToolRuntime) -> str:
|
|
"""List skills available to this agent, with metadata and usage stats.
|
|
Does NOT read SKILL.md content.
|
|
|
|
For a **custom agent**, the result is filtered to skills explicitly
|
|
declared in the agent's ``.deer-flow/agents/{agent_id}/config.yaml``
|
|
under the ``skills:`` field — the agent only sees what it was wired to
|
|
use.
|
|
|
|
For the **default lead agent** (no agent_id in runtime), the result is
|
|
filtered to non-archived skills that are either built-in (public) or
|
|
owned by the current user. Other users' published custom skills and
|
|
archived skills are excluded.
|
|
|
|
Returns a JSON array with fields: name, description, category, enabled,
|
|
state, use_count, view_count, patch_count, last_activity_at, pinned,
|
|
source.
|
|
"""
|
|
return skill_list_impl(_resolve_active_agent_id(runtime), _resolve_extra_skills(runtime))
|
|
|
|
|
|
def _latest_activity(entry: dict) -> str | None:
|
|
candidates = [entry.get("last_used_at"), entry.get("last_viewed_at"), entry.get("last_patched_at")]
|
|
valid = [ts for ts in candidates if ts]
|
|
if valid:
|
|
return max(valid)
|
|
return entry.get("created_at")
|
|
|
|
|
|
def _skills_for_catalogue(storage):
|
|
"""Use the warm prompt catalogue, and parse SKILL.md files only when it is cold."""
|
|
from deerflow.agents.lead_agent.prompt import get_warm_enabled_skills
|
|
|
|
cached = get_warm_enabled_skills(storage)
|
|
if cached is not None:
|
|
return cached
|
|
return storage.load_skills(enabled_only=True)
|
|
|
|
|
|
def skill_list_impl(agent_id: str | None, extra_skills: list[str] | None = None) -> str:
|
|
"""List skills visible to ``agent_id`` (None = default lead agent)."""
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
from deerflow.skills.types import SkillCategory
|
|
from deerflow.skills.usage import STATE_ARCHIVED, all_entries
|
|
|
|
allowed = _load_agent_skill_whitelist(agent_id, extra_skills)
|
|
|
|
# 默认 lead agent 没有 per-agent 白名单 —— 改为只暴露"未归档"且"内置或属于
|
|
# 当前用户"的技能。owned 为 None 表示 SkillStore 不可用,降级放行全部自定义
|
|
# 技能(见 _load_owned_skill_names)。
|
|
owned: set[str] | None = None
|
|
if allowed is None:
|
|
owned = _load_owned_skill_names(_resolve_effective_user_id())
|
|
|
|
try:
|
|
storage = get_or_new_skill_storage()
|
|
skills = _skills_for_catalogue(storage)
|
|
usage = all_entries()
|
|
result = []
|
|
for skill in skills:
|
|
if not can_access_coordinator_only_skill(agent_id, skill.name):
|
|
continue
|
|
entry = usage.get(skill.name, {})
|
|
if allowed is not None:
|
|
# 自定义 agent:严格按 config.yaml 的 skills 白名单过滤。
|
|
if skill.name not in allowed:
|
|
continue
|
|
else:
|
|
# 默认 agent:跳过已归档技能。
|
|
if entry.get("state") == STATE_ARCHIVED:
|
|
continue
|
|
# 自定义技能仅当属于当前用户时可见;内置 public 技能始终可见。
|
|
if skill.category == SkillCategory.CUSTOM and owned is not None and skill.name not in owned:
|
|
continue
|
|
result.append({
|
|
"name": skill.name,
|
|
"description": skill.description,
|
|
"category": skill.category.value if hasattr(skill.category, "value") else str(skill.category),
|
|
"enabled": skill.enabled,
|
|
"state": entry.get("state", "active"),
|
|
"use_count": entry.get("use_count", 0),
|
|
"view_count": entry.get("view_count", 0),
|
|
"patch_count": entry.get("patch_count", 0),
|
|
"last_activity_at": _latest_activity(entry),
|
|
"pinned": entry.get("pinned", False),
|
|
"source": entry.get("source", "upload"),
|
|
})
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
logger.debug("skill_list failed: %s", e)
|
|
return f"Error listing skills: {e}"
|