diff options
| -rw-r--r-- | llamachat/skills.py | 202 | ||||
| -rwxr-xr-x | test_llamachat.py | 136 |
2 files changed, 338 insertions, 0 deletions
diff --git a/llamachat/skills.py b/llamachat/skills.py new file mode 100644 index 0000000..e36b66e --- /dev/null +++ b/llamachat/skills.py @@ -0,0 +1,202 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# llamachat - a small native chat client for a local llama.cpp router +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +"""Skills as instruction files, loaded into the model's context on demand. + +One SKILL.md file per skill inside the skills directory. Each file starts +with a YAML frontmatter block carrying name and description, then the +instructions. Only those two keys are read; parsing is a line scan because +adding a YAML dependency for two keys would be absurd. +""" + +import json +import re +from dataclasses import dataclass +from pathlib import Path + +LOAD_SKILL_NAME = "load_skill" + +# The model is told which skill it just received. +SKILL_HEADER = "[skill: {name}]" + +# Appended to the final round's tool result; see tool_message(). A note +# must ride on the result because a trailing system message is rejected by +# the chat template ("System message must be at the beginning"). +NO_MORE_TOOLS = ( + "This was your last available tool use for this turn and tools are now " + "unavailable. Write your answer now using the information above. Do not " + "request another tool." +) + +# A description is shortened to this for the model's directory, so listing +# every skill never costs a real chunk of context per request. +DIRECTORY_CHARS = 140 + + +@dataclass +class Skill: + name: str + description: str + text: str + + +def _parse(path: Path) -> Skill: + """Read one SKILL.md, pulling name/description out of its frontmatter.""" + text = path.read_text(encoding="utf-8", errors="replace") + name = path.parent.name + description = "" + body = text + if text.startswith("---"): + # split("---", 2) takes only the opening and closing fences, so a + # body that itself contains "---" is left intact. + parts = text.split("---", 2) + if len(parts) == 3: + body = parts[2] + for line in parts[1].splitlines(): + key, sep, value = line.partition(":") + if not sep: + continue + key, value = key.strip(), value.strip() + if key == "name" and value: + name = value + elif key == "description": + description = value + return Skill(name=name, description=description, text=body.strip()) + + +def _short(text: str) -> str: + """First line of a description, capped, so the directory stays compact.""" + stripped = text.strip() + first = stripped.splitlines()[0] if stripped else "" + return first[:DIRECTORY_CHARS] + + +class SkillStore: + """The skills directory: one SKILL.md per skill, read only.""" + + def __init__(self, directory: Path): + self.dir = Path(directory) + + def _skills(self) -> list[Skill]: + if not self.dir.is_dir(): + return [] + found = [] + for path in sorted(self.dir.glob("*/SKILL.md")): + try: + found.append(_parse(path)) + except OSError: + continue # a broken skill must not take down the chat + return found + + def names(self) -> list[str]: + return [skill.name for skill in self._skills()] + + def load(self, name: str) -> Skill | None: + for skill in self._skills(): + if skill.name == name: + return skill + return None + + def directory(self) -> str: + """Compact name-description lines for the model to choose from.""" + return "\n".join( + f"- {s.name}: {_short(s.description)}" if s.description else f"- {s.name}" + for s in self._skills() + ) + + def tool_schema(self) -> dict: + """The OpenAI `load_skill` function schema for the model to call.""" + return { + "type": "function", + "function": { + "name": LOAD_SKILL_NAME, + "description": ( + "Load a skill's full instructions into context. Use this " + "when the task matches one of these skills.\n\n" + f"Available skills:\n{self.directory()}" + ), + "parameters": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "The skill to load.", + "enum": self.names(), + }, + }, + "required": ["name"], + }, + }, + } + + def match_mentions(self, text: str) -> list[str]: + """Skills whose names appear in `text` as whole words, in store order.""" + found = [] + for skill in self._skills(): + if re.search(r"\b" + re.escape(skill.name) + r"\b", text, re.IGNORECASE): + found.append(skill.name) + return found + + +def parse_commands(text: str, names: set[str]) -> tuple[str, list[str]]: + """Split /name tokens off `text`; return (remaining, loaded names). + + Only tokens whose name is a known skill are treated as commands and + stripped; any other /word survives as plain text. + """ + loaded: list[str] = [] + remaining: list[str] = [] + for token in text.split(): + if token.startswith("/") and len(token) > 1: + candidate = token[1:] + if candidate in names: + loaded.append(candidate) + continue + remaining.append(token) + return " ".join(remaining), loaded + + +def parse_name(arguments: str) -> str: + """The skill name out of a tool call's JSON arguments, or ''.""" + try: + parsed = json.loads(arguments or "{}") + except ValueError: + return "" + if not isinstance(parsed, dict): + return "" + name = parsed.get("name") + return name.strip() if isinstance(name, str) else "" + + +def tool_message( + call_id: str, skill: Skill | None, error: str = "", last: bool = False +) -> dict: + """The `role: tool` message carrying a loaded skill back to the model. + + A failure is reported rather than swallowed, for the same reason as + search.tool_message: the model asked for the tool and is waiting on it. + With `last`, the message also says no further tools are available. + """ + if error: + content = f"error: {error}" + elif skill is not None: + content = f"{SKILL_HEADER.format(name=skill.name)}\n{skill.text}" + else: + content = "error: no skill" + if last: + content = f"{content}\n\n{NO_MORE_TOOLS}" + return { + "role": "tool", + "tool_call_id": call_id, + "content": content, + } diff --git a/test_llamachat.py b/test_llamachat.py index 407c5dc..ef73fe3 100755 --- a/test_llamachat.py +++ b/test_llamachat.py @@ -55,6 +55,37 @@ ctx-size = 16384 ngl = all """ +SKILLS_SAMPLE = { + "firefly-cli/SKILL.md": ( + "---\n" + "name: firefly-cli\n" + "description: Operate a Firefly III instance from the command line.\n" + "---\n" + "\n" + "# firefly-cli\n" + "Run `firefly auth test` first.\n" + ), + "test-build-slackbuild/SKILL.md": ( + "---\n" + "name: test-build-slackbuild\n" + "description: Drive the sbo-dockerbuild test-build tool.\n" + "---\n" + "\n" + "Load the skill, then test-build.\n" + ), + "no-frontmatter/SKILL.md": ( + "# no-frontmatter\n" + "Skill with no frontmatter; the name falls back to the directory.\n" + ), +} + + +def _write_skills(root: Path, sample: dict) -> None: + for rel, content in sample.items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + def test_presets(): with tempfile.TemporaryDirectory() as tmp: @@ -3325,6 +3356,107 @@ def test_title_prompt(): print("ok title prompt") +def test_skills_store(): + from llamachat import skills + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_skills(root, SKILLS_SAMPLE) + store = skills.SkillStore(root) + + assert store.names() == [ + "firefly-cli", "no-frontmatter", "test-build-slackbuild" + ] + + skill = store.load("firefly-cli") + assert skill is not None + assert skill.name == "firefly-cli" + assert skill.description == "Operate a Firefly III instance from the command line." + assert skill.text.startswith("# firefly-cli") + + # A name that does not exist, and an absent directory, both yield None. + assert store.load("nope") is None + assert skills.SkillStore(Path(tmp) / "missing").names() == [] + + schema = store.tool_schema() + function = schema["function"] + assert function["name"] == "load_skill" + assert function["parameters"]["required"] == ["name"] + assert function["parameters"]["properties"]["name"]["enum"] == store.names() + assert "firefly-cli" in function["description"] + + directory = store.directory() + assert "firefly-cli" in directory + assert "test-build-slackbuild" in directory + print("ok skills store") + + +def test_skills_mentions(): + from llamachat import skills + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_skills(root, {"firefly-cli/SKILL.md": SKILLS_SAMPLE["firefly-cli/SKILL.md"]}) + store = skills.SkillStore(root) + + assert store.match_mentions("use firefly-cli for this") == ["firefly-cli"] + assert store.match_mentions("USE FIREFLY-CLI now") == ["firefly-cli"] + # No word boundary, no mention: a substring must not load a skill. + assert store.match_mentions("firefly-cli2 rocks") == [] + assert store.match_mentions("the fireflies are out") == [] + assert store.match_mentions("nothing here") == [] + print("ok skills mentions") + + +def test_skills_parse_commands(): + from llamachat import skills + + known = {"firefly-cli", "handoff"} + text, loaded = skills.parse_commands( + "/firefly-cli fix my budget /unknown stays", known + ) + assert loaded == ["firefly-cli"] + assert text == "fix my budget /unknown stays" + + text, loaded = skills.parse_commands("no commands here", known) + assert loaded == [] + assert text == "no commands here" + + # A lone command leaves nothing behind. + text, loaded = skills.parse_commands("/handoff", known) + assert loaded == ["handoff"] + assert text == "" + + # A name that is not a known skill is kept as plain text. + text, loaded = skills.parse_commands("/nope", known) + assert loaded == [] + assert text == "/nope" + print("ok skills parse commands") + + +def test_skill_tool_message(): + from llamachat import skills + + skill = skills.Skill(name="firefly-cli", description="d", text="body text") + msg = skills.tool_message("call_1", skill) + assert msg["role"] == "tool" + assert msg["tool_call_id"] == "call_1" + assert "[skill: firefly-cli]" in msg["content"] + assert "body text" in msg["content"] + + err = skills.tool_message("call_1", None, error="No such skill") + assert "error: No such skill" in err["content"] + + last = skills.tool_message("call_1", skill, last=True) + assert skills.NO_MORE_TOOLS in last["content"] + + assert skills.parse_name('{"name": "firefly-cli"}') == "firefly-cli" + assert skills.parse_name("not json") == "" + assert skills.parse_name('{"query": "x"}') == "" + assert skills.parse_name("") == "" + print("ok skill tool message") + + if __name__ == "__main__": test_presets() test_real_presets() @@ -3387,4 +3519,8 @@ if __name__ == "__main__": test_code_block_wrapping() test_code_block_copy_links() test_code_block_copy_states() + test_skills_store() + test_skills_mentions() + test_skills_parse_commands() + test_skill_tool_message() print("\nall checks passed") |
