nearxos
0375b20bb4
- Added configuration options for requiring human approval before applying LLM-generated MCP patches. - Updated Docker setup to include skills directory. - Integrated skills management into the backend, allowing for procedural guides and skill matching. - Refactored database initialization to apply Alembic migrations. - Enhanced task approval process to handle MCP patch applications with optional approval. - Introduced new schemas for skills and updated existing APIs to support skills functionality. This commit lays the groundwork for improved agent capabilities and better management of MCP development processes.
265 lines
7.8 KiB
Python
265 lines
7.8 KiB
Python
"""Agent skills — procedural guides loaded from skills/*/SKILL.md on disk.
|
|
|
|
Skills complement troubleshooting rules: rules route devices and supply short hints;
|
|
skills inject deeper procedural knowledge into triage and reasoning prompts.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import yaml
|
|
|
|
from app.config import settings
|
|
from app.services.troubleshooting_rules import _text_has_term
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_skill_cache: dict[str, dict] = {}
|
|
_skill_dir_mtime: float = -1.0
|
|
|
|
|
|
@dataclass
|
|
class MatchedSkill:
|
|
id: str
|
|
name: str
|
|
description: str
|
|
body: str
|
|
priority: int
|
|
score: int
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"priority": self.priority,
|
|
"score": self.score,
|
|
"body_chars": len(self.body),
|
|
}
|
|
|
|
|
|
def skills_dir() -> str:
|
|
return os.environ.get("SKILLS_PATH", settings.skills_path)
|
|
|
|
|
|
def invalidate_skills_cache() -> None:
|
|
global _skill_cache, _skill_dir_mtime
|
|
_skill_cache = {}
|
|
_skill_dir_mtime = -1.0
|
|
|
|
|
|
def _parse_skill_file(path: str) -> dict | None:
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
raw = fh.read()
|
|
except OSError:
|
|
return None
|
|
|
|
meta: dict = {}
|
|
body = raw.strip()
|
|
if raw.startswith("---"):
|
|
parts = raw.split("---", 2)
|
|
if len(parts) >= 3:
|
|
try:
|
|
parsed = yaml.safe_load(parts[1])
|
|
meta = parsed if isinstance(parsed, dict) else {}
|
|
except yaml.YAMLError as exc:
|
|
logger.warning("invalid skill frontmatter %s: %s", path, exc)
|
|
meta = {}
|
|
body = parts[2].strip()
|
|
|
|
folder = os.path.basename(os.path.dirname(path))
|
|
skill_id = str(meta.get("id") or folder)
|
|
return {
|
|
"id": skill_id,
|
|
"name": str(meta.get("name") or skill_id),
|
|
"description": str(meta.get("description") or ""),
|
|
"priority": int(meta.get("priority") or 0),
|
|
"match": meta.get("match") or {},
|
|
"rule_ids": list(meta.get("rule_ids") or []),
|
|
"body": body,
|
|
"path": path,
|
|
}
|
|
|
|
|
|
def _dir_latest_mtime(root: str) -> float:
|
|
latest = -1.0
|
|
if not os.path.isdir(root):
|
|
return latest
|
|
for dirpath, _dirnames, filenames in os.walk(root):
|
|
for name in filenames:
|
|
if name != "SKILL.md":
|
|
continue
|
|
try:
|
|
latest = max(latest, os.path.getmtime(os.path.join(dirpath, name)))
|
|
except OSError:
|
|
continue
|
|
return latest
|
|
|
|
|
|
def load_skills(*, force: bool = False) -> dict[str, dict]:
|
|
"""Return skill_id → skill record (metadata + body)."""
|
|
global _skill_cache, _skill_dir_mtime
|
|
root = skills_dir()
|
|
mtime = _dir_latest_mtime(root)
|
|
|
|
if not force and _skill_cache and mtime == _skill_dir_mtime:
|
|
return _skill_cache
|
|
|
|
skills: dict[str, dict] = {}
|
|
if os.path.isdir(root):
|
|
for entry in sorted(os.listdir(root)):
|
|
skill_path = os.path.join(root, entry, "SKILL.md")
|
|
if not os.path.isfile(skill_path):
|
|
continue
|
|
record = _parse_skill_file(skill_path)
|
|
if record:
|
|
skills[record["id"]] = record
|
|
|
|
_skill_cache = skills
|
|
_skill_dir_mtime = mtime
|
|
return skills
|
|
|
|
|
|
def list_skills_summary() -> list[dict]:
|
|
return [
|
|
{
|
|
"id": s["id"],
|
|
"name": s["name"],
|
|
"description": s["description"],
|
|
"priority": s["priority"],
|
|
"rule_ids": s.get("rule_ids") or [],
|
|
"match": s.get("match") or {},
|
|
"path_hint": f"skills/{s['id']}/SKILL.md",
|
|
}
|
|
for s in sorted(load_skills().values(), key=lambda x: (-x["priority"], x["id"]))
|
|
]
|
|
|
|
|
|
def get_skill(skill_id: str) -> dict | None:
|
|
return load_skills().get(skill_id)
|
|
|
|
|
|
def read_skill_raw(skill_id: str) -> str:
|
|
skill = get_skill(skill_id)
|
|
if not skill:
|
|
return ""
|
|
try:
|
|
with open(skill["path"], encoding="utf-8") as fh:
|
|
return fh.read()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def save_skill_raw(skill_id: str, content: str) -> dict:
|
|
skill = get_skill(skill_id)
|
|
if not skill:
|
|
raise ValueError(f"Unknown skill: {skill_id}")
|
|
# Validate frontmatter parses before writing.
|
|
if content.startswith("---"):
|
|
parts = content.split("---", 2)
|
|
if len(parts) >= 3:
|
|
parsed = yaml.safe_load(parts[1])
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("Skill frontmatter must be a YAML mapping")
|
|
path = skill["path"]
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write(content if content.endswith("\n") else content + "\n")
|
|
invalidate_skills_cache()
|
|
updated = get_skill(skill_id)
|
|
if not updated:
|
|
raise ValueError(f"Skill {skill_id} missing after save")
|
|
return updated
|
|
|
|
|
|
def _skill_match_score(text: str, skill: dict, *, matched_rule_id: str | None) -> int:
|
|
match = skill.get("match") or {}
|
|
keywords = match.get("keywords") or []
|
|
all_of = match.get("all_of") or []
|
|
|
|
if all_of and not all(_text_has_term(text, t) for t in all_of):
|
|
keyword_score = 0
|
|
else:
|
|
keyword_score = 0
|
|
for kw in keywords:
|
|
if _text_has_term(text, kw):
|
|
keyword_score += 2 if " " in str(kw).strip() else 1
|
|
if all_of:
|
|
keyword_score += len(all_of) * 2
|
|
|
|
rule_boost = 0
|
|
rule_ids = skill.get("rule_ids") or []
|
|
if matched_rule_id and matched_rule_id in rule_ids:
|
|
rule_boost = 10
|
|
|
|
return keyword_score + rule_boost
|
|
|
|
|
|
def _truncate_body(body: str, max_chars: int) -> str:
|
|
if max_chars <= 0 or len(body) <= max_chars:
|
|
return body
|
|
return body[: max_chars - 20].rstrip() + "\n\n… (truncated)"
|
|
|
|
|
|
def match_skills(
|
|
title: str,
|
|
issue: str,
|
|
*,
|
|
matched_rule_id: str | None = None,
|
|
max_skills: int | None = None,
|
|
) -> list[MatchedSkill]:
|
|
"""Return up to *max_skills* best-matching procedural guides."""
|
|
limit = max_skills if max_skills is not None else settings.skills_max_matched
|
|
max_body = settings.skills_max_body_chars
|
|
text = f"{title} {issue}".lower()
|
|
skills = load_skills()
|
|
|
|
scored: list[tuple[int, int, dict]] = []
|
|
for skill in skills.values():
|
|
score = _skill_match_score(text, skill, matched_rule_id=matched_rule_id)
|
|
if score <= 0:
|
|
continue
|
|
scored.append((score, skill.get("priority") or 0, skill))
|
|
|
|
scored.sort(key=lambda t: (-t[0], -t[1], t[2]["id"]))
|
|
out: list[MatchedSkill] = []
|
|
for score, _prio, skill in scored[:limit]:
|
|
out.append(
|
|
MatchedSkill(
|
|
id=skill["id"],
|
|
name=skill["name"],
|
|
description=skill["description"],
|
|
body=_truncate_body(skill["body"], max_body),
|
|
priority=skill.get("priority") or 0,
|
|
score=score,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def skill_guidance_block(matched: list[MatchedSkill | dict] | None) -> str:
|
|
if not matched:
|
|
return ""
|
|
lines = ["\n\n--- Agent skills (procedural guides) ---"]
|
|
for item in matched:
|
|
if isinstance(item, dict):
|
|
name = item.get("name", "")
|
|
sid = item.get("id", "")
|
|
description = item.get("description", "")
|
|
body = item.get("body", "")
|
|
else:
|
|
name, sid, description, body = item.name, item.id, item.description, item.body
|
|
lines.append(f"\n### Skill: {name} ({sid})")
|
|
if description:
|
|
lines.append(description)
|
|
if body:
|
|
lines.append(body)
|
|
lines.append(
|
|
"\nFollow these guides when investigating. Rules define device scope; "
|
|
"skills provide domain-specific procedure and proven root causes."
|
|
)
|
|
return "\n".join(lines)
|