danvics/haiku.rag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Skill generator core

Browse source
This commit is contained in:
Yiorgis Gozadinos 2026-03-24 15:11:54 +02:00
parent a778ea68a8
commit 26da4a0624
No known key found for this signature in database
10 changed files with 885 additions and 194 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

140
haiku_rag_slim/haiku/rag/skill_generator/__init__.py Normal file
View file

@ -0,0 +1,140 @@
import pathlib
import shutil
from jinja2 import Environment, PackageLoader, select_autoescape
AVAILABLE_TOOLS: set[str] = {
"list_documents",
"get_document",
"search",
"ask",
"research",
"analyze",
}
DEFAULT_PREAMBLE = (
"You are a RAG (Retrieval Augmented Generation) assistant "
"with access to a document knowledge base.\n"
"Use your tools to search and answer questions. "
"Never make up information — always use tools to get facts "
"from the knowledge base."
)
DEFAULT_DESCRIPTION = (
"Search, retrieve and analyze documents using RAG (Retrieval Augmented Generation)."
)
def _get_env() -> Environment:
return Environment(
loader=PackageLoader("haiku.rag.skill_generator", "templates"),
autoescape=select_autoescape(),
keep_trailing_newline=True,
lstrip_blocks=True,
trim_blocks=True,
)
def validate_metadata(name: str, description: str) -> None:
from haiku.skills import SkillMetadata
SkillMetadata(name=name, description=description)
if not name.isidentifier():
raise ValueError(f"{name!r} is not a valid Python identifier")
if not name.islower():
raise ValueError(f"{name!r} must be lowercase")
def validate_tools(tools: list[str]) -> None:
if not tools:
raise ValueError("tools must contain at least one tool")
unknown = set(tools) - AVAILABLE_TOOLS
if unknown:
raise ValueError(
f"Unknown tools: {', '.join(sorted(unknown))}."
f" Available: {', '.join(sorted(AVAILABLE_TOOLS))}"
)
def validate_db_path(db_path: pathlib.Path) -> None:
if not db_path.exists():
raise ValueError(f"db_path does not exist: {db_path}")
if not db_path.is_dir():
raise ValueError(f"db_path is not a directory: {db_path}")
def validate_output_dir(output_dir: pathlib.Path, name: str) -> None:
if not output_dir.exists():
raise ValueError(f"output_dir does not exist: {output_dir}")
target = output_dir / f"{name}-skill"
if target.exists():
raise ValueError(f"Target directory already exists: {target}")
def render_templates(
output_dir: pathlib.Path,
name: str,
description: str,
tool_names: list[str],
preamble: str | None = None,
) -> pathlib.Path:
if preamble is None:
preamble = DEFAULT_PREAMBLE
env = _get_env()
context = {
"name": name,
"description": description,
"tool_names": tool_names,
"preamble": preamble,
}
result_dir = output_dir / f"{name}-skill"
pkg_dir = result_dir / f"{name}_skill"
assets_dir = pkg_dir / "assets"
assets_dir.mkdir(parents=True)
# Render pyproject.toml
template = env.get_template("pyproject.toml.j2")
(result_dir / "pyproject.toml").write_text(template.render(context))
# Render __init__.py
template = env.get_template("__init__.py.j2")
(pkg_dir / "__init__.py").write_text(template.render(context))
# Render SKILL.md
template = env.get_template("SKILL.md.j2")
(pkg_dir / "SKILL.md").write_text(template.render(context))
return result_dir
def generate_skill(
db_path: pathlib.Path,
output_dir: pathlib.Path,
name: str,
description: str,
tool_names: list[str],
config_path: pathlib.Path | None = None,
preamble: str | None = None,
) -> pathlib.Path:
validate_metadata(name, description)
validate_tools(tool_names)
validate_db_path(db_path)
validate_output_dir(output_dir, name)
result = render_templates(
output_dir=output_dir,
name=name,
description=description,
tool_names=tool_names,
preamble=preamble,
)
assets_dir = result / f"{name}_skill" / "assets"
shutil.copytree(db_path, assets_dir / f"{name}.lancedb")
if config_path is not None:
shutil.copy2(config_path, assets_dir / "haiku.rag.yaml")
return result

54
haiku_rag_slim/haiku/rag/skill_generator/templates/SKILL.md.j2 Normal file
View file

@ -0,0 +1,54 @@
---
name: {{ name }}
description: {{ description }}
---
# {{ name }}
{{ preamble }}
## How to decide which tool to use
{% if "ask" in tool_names %}
**Default rule:** If the user is asking a question, use **ask**. Only use **search** when the user explicitly wants to browse or find passages.
{% endif %}
{% if "list_documents" in tool_names %}
- **list_documents** — Use when the user wants to browse or see what documents are available (e.g., "what documents do you have?", "show me the documents", "list available docs").
{% endif %}
{% if "get_document" in tool_names %}
- **get_document** — Use when the user wants the full content of a specific document (e.g., "get the paper about X", "show me document Y"). Accepts a document ID, title, or URI — partial matches work.
{% endif %}
{% if "search" in tool_names %}
- **search** — Use when the user wants to browse, explore, or find specific passages across documents (e.g., "search for embeddings", "find mentions of transformers"). Returns all matching results as sources.
{% endif %}
{% if "ask" in tool_names %}
- **ask** — Use for factual questions that need a synthesized answer (e.g., "what is DocLayNet?", "explain the methodology"). Searches, synthesizes, and returns only the chunks actually used as citations. Always include the citations in your response.
{% endif %}
{% if "research" in tool_names %}
- **research** — Deep multi-agent research that produces comprehensive reports. **Only use when the user explicitly requests deep research** (e.g., "do a deep research on X", "research this topic thoroughly"). Never call this tool on your own — it is slow and expensive.
{% endif %}
{% if "analyze" in tool_names %}
- **analyze** — Use for complex analytical questions that require computation, aggregation, or data traversal across documents (e.g., "how many pages?", "compare table 3 across documents", "calculate average word count"). Executes Python code in a sandboxed interpreter.
{% endif %}
{% if "search" in tool_names %}
## When search returns irrelevant results
If your first search returns results that clearly don't match the question, **do not keep searching with variations**. Instead:
{% if "ask" in tool_names %}
- Use **ask** if the question is factual
{% endif %}
- Report that the knowledge base doesn't contain relevant information
{% endif %}
{% if "get_document" in tool_names %}
## When the user mentions a specific document
If the user says "search in [doc]", "find in [doc]", or "answer from [doc]":
- Extract the **topic** as the `query`/`question` parameter
- Use **get_document** or **list_documents** first to identify the document, then search/ask with a filter
Examples:
- "search for embeddings in the ML paper" -> first identify "ML paper", then search for "embeddings"
- "what does the DocLayNet paper say about annotations?" -> ask with question="what are the annotation methods?"
{% endif %}

77
haiku_rag_slim/haiku/rag/skill_generator/templates/__init__.py.j2 Normal file
View file

@ -0,0 +1,77 @@
from pathlib import Path
from pydantic import BaseModel, Field
from haiku.skills.models import Skill
from haiku.skills.parser import parse_skill_md
{% if "ask" in tool_names or "research" in tool_names %}
from haiku.rag.agents.research.models import Citation
{% endif %}
{% if "search" in tool_names or "list_documents" in tool_names or "get_document" in tool_names %}
from haiku.rag.tools.document import DocumentInfo
{% endif %}
{% if "ask" in tool_names %}
from haiku.rag.tools.qa import QAHistoryEntry
{% endif %}
{% if "search" in tool_names %}
from haiku.rag.store.models.chunk import SearchResult
{% endif %}
{% if "research" in tool_names %}
from haiku.rag.skills._tools import ResearchEntry
{% endif %}
{% if "analyze" in tool_names %}
from haiku.rag.skills._tools import AnalysisEntry
{% endif %}
_TOOL_NAMES = {{ tool_names | tojson }}
_ASSETS_DIR = Path(__file__).resolve().parent / "assets"
_DB_PATH = _ASSETS_DIR / "{{ name }}.lancedb"
_CONFIG_PATH = _ASSETS_DIR / "haiku.rag.yaml"
class SkillState(BaseModel):
{% if "ask" in tool_names or "research" in tool_names %}
citations: list[Citation] = Field(default_factory=list)
{% endif %}
{% if "ask" in tool_names %}
qa_history: list[QAHistoryEntry] = Field(default_factory=list)
{% endif %}
document_filter: str | None = None
{% if "search" in tool_names %}
searches: dict[str, list[SearchResult]] = Field(default_factory=dict)
{% endif %}
{% if "search" in tool_names or "list_documents" in tool_names or "get_document" in tool_names %}
documents: list[DocumentInfo] = Field(default_factory=list)
{% endif %}
{% if "research" in tool_names %}
reports: list[ResearchEntry] = Field(default_factory=list)
{% endif %}
{% if "analyze" in tool_names %}
analyses: list[AnalysisEntry] = Field(default_factory=list)
{% endif %}
def _get_config():
if _CONFIG_PATH.exists():
from haiku.rag.config import AppConfig, load_yaml_config
return AppConfig.model_validate(load_yaml_config(_CONFIG_PATH))
from haiku.rag.config import get_config
return get_config()
def create_skill() -> Skill:
from haiku.rag.skills._tools import create_skill_tools
metadata, instructions = parse_skill_md(Path(__file__).parent / "SKILL.md")
config = _get_config()
tools = create_skill_tools(_DB_PATH, config, SkillState, _TOOL_NAMES)
return Skill(
metadata=metadata,
instructions=instructions,
tools=list(tools.values()),
state_type=SkillState,
state_namespace="{{ name }}",
)

16
haiku_rag_slim/haiku/rag/skill_generator/templates/pyproject.toml.j2 Normal file
View file

@ -0,0 +1,16 @@
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "{{ name }}-skill"
version = "0.1.0"
description = "{{ description }}"
requires-python = ">=3.12"
dependencies = [
"haiku.rag-slim >= 0.35",
"haiku-skills >= 0.10.0",
]
[project.entry-points."haiku.skills"]
{{ name }} = "{{ name }}_skill:create_skill"

216
haiku_rag_slim/haiku/rag/skills/_tools.py
View file

@ -1,9 +1,25 @@
from pathlib import Path
from typing import Any
from pydantic import BaseModel
from pydantic_ai import RunContext
from haiku.rag.agents.research.models import Citation
from haiku.rag.tools.document import DocumentInfo
from haiku.rag.tools.qa import QAHistoryEntry
from haiku.skills.state import SkillRunDeps
class ResearchEntry(BaseModel):
question: str
title: str
executive_summary: str
class AnalysisEntry(BaseModel):
question: str
answer: str
program: str | None = None
async def find_relevant_prior_qa(
@ -217,3 +233,203 @@ def update_documents_state(
)
if not any(d.id == doc_info.id for d in documents_state):
documents_state.append(doc_info)
def _get_state(ctx: RunContext[SkillRunDeps], state_type: type) -> Any:
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, state_type):
return ctx.deps.state
return None
def create_skill_tools(
db_path: Path,
config: Any,
state_type: type,
tool_names: list[str],
) -> dict[str, Any]:
"""Create tool closures for a skill.
Returns a dict mapping tool name to async callable.
Each tool extracts state from RunContext, calls the shared implementation,
and updates state.
"""
tools: dict[str, Any] = {}
if "search" in tool_names:
async def search(
ctx: RunContext[SkillRunDeps], query: str, limit: int | None = None
) -> str:
"""Search the knowledge base using hybrid search (vector + full-text).
Returns ranked results with content and metadata.
Args:
query: The search query.
limit: Maximum number of results.
"""
state = _get_state(ctx, state_type)
formatted, results = await skill_search(
db_path,
config,
query,
limit=limit,
document_filter=state.document_filter if state else None,
)
if state:
state.searches[query] = results
return formatted
tools["search"] = search
if "list_documents" in tool_names:
async def list_documents(
ctx: RunContext[SkillRunDeps],
limit: int | None = None,
offset: int | None = None,
) -> list[dict[str, Any]]:
"""List documents in the knowledge base with optional pagination.
Args:
limit: Maximum number of documents to return.
offset: Number of documents to skip.
"""
result = await skill_list_documents(db_path, config, limit, offset)
state = _get_state(ctx, state_type)
if state:
update_documents_state(state.documents, result)
return result
tools["list_documents"] = list_documents
if "get_document" in tool_names:
async def get_document(
ctx: RunContext[SkillRunDeps], query: str
) -> dict[str, Any] | None:
"""Retrieve a document by ID, title, or URI.
Args:
query: Document ID, title, or URI to look up.
"""
result = await skill_get_document(db_path, config, query)
if result is not None:
state = _get_state(ctx, state_type)
if state:
update_documents_state(state.documents, [result])
return result
tools["get_document"] = get_document
if "ask" in tool_names:
async def ask(ctx: RunContext[SkillRunDeps], question: str) -> str:
"""Ask a question and get an answer with citations from the knowledge base.
Args:
question: The question to ask.
"""
from haiku.rag.utils import format_citations
state = _get_state(ctx, state_type)
answer, citations = await skill_ask(
db_path,
config,
question,
qa_history=state.qa_history if state else None,
document_filter=state.document_filter if state else None,
)
if state:
next_index = len(state.citations) + 1
for citation in citations:
citation.index = next_index
next_index += 1
state.citations.extend(citations)
state.qa_history.append(
QAHistoryEntry(
question=question, answer=answer, citations=citations
)
)
if citations:
answer += "\n\n" + format_citations(citations)
return answer
tools["ask"] = ask
if "research" in tool_names:
async def research(ctx: RunContext[SkillRunDeps], question: str) -> str:
"""Conduct deep multi-agent research on a question.
Iteratively searches, analyzes, and synthesizes information from the
knowledge base to produce a comprehensive research report.
Only use when the user explicitly requests deep research.
Args:
question: The research question to investigate.
"""
state = _get_state(ctx, state_type)
formatted, title, executive_summary = await skill_research(
db_path,
config,
question,
document_filter=state.document_filter if state else None,
)
if state:
state.reports.append(
ResearchEntry(
question=question,
title=title,
executive_summary=executive_summary,
)
)
state.qa_history.append(
QAHistoryEntry(question=question, answer=executive_summary)
)
return formatted
tools["research"] = research
if "analyze" in tool_names:
async def analyze(
ctx: RunContext[SkillRunDeps],
question: str,
document: str | None = None,
filter: str | None = None,
) -> str:
"""Answer complex analytical questions using code execution.
Use this for questions requiring computation, aggregation, or
data traversal across documents.
Args:
question: The question to answer.
document: Optional document ID or title to pre-load for analysis.
filter: Optional SQL WHERE clause to filter documents.
"""
output, answer, program = await skill_analyze(
db_path, config, question, document=document, filter=filter
)
state = _get_state(ctx, state_type)
if state:
state.analyses.append(
AnalysisEntry(
question=question,
answer=answer,
program=program,
)
)
return output
tools["analyze"] = analyze
return tools

158
haiku_rag_slim/haiku/rag/skills/rag.py
View file

@ -4,15 +4,14 @@ from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
from pydantic_ai import RunContext
from haiku.rag.agents.research.models import Citation
from haiku.rag.skills._tools import ResearchEntry
from haiku.rag.store.models.chunk import SearchResult
from haiku.rag.tools.document import DocumentInfo
from haiku.rag.tools.qa import QAHistoryEntry
from haiku.skills.models import Skill, SkillMetadata, SkillSource, StateMetadata
from haiku.skills.parser import parse_skill_md
from haiku.skills.state import SkillRunDeps
AGENT_PREAMBLE = """You are a helpful research assistant powered by haiku.rag, a knowledge base system.
@ -22,11 +21,7 @@ CRITICAL RULES:
3. When a skill returns citations, always include them in your response
"""
class ResearchEntry(BaseModel):
question: str
title: str
executive_summary: str
_RAG_TOOLS = ["search", "list_documents", "get_document", "ask", "research"]
class RAGState(BaseModel):
@ -64,12 +59,6 @@ def state_metadata() -> StateMetadata:
)
def _get_state(ctx: RunContext[SkillRunDeps]) -> RAGState | None:
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState):
return ctx.deps.state
return None
def create_skill(
db_path: Path | None = None,
config: Any = None,
@ -84,6 +73,7 @@ def create_skill(
config: haiku.rag AppConfig instance. If None, uses get_config().
"""
from haiku.rag.config import get_config
from haiku.rag.skills._tools import create_skill_tools
if config is None:
config = get_config()
@ -95,152 +85,14 @@ def create_skill(
else:
db_path = config.storage.data_dir / "haiku.rag.lancedb"
async def search(
ctx: RunContext[SkillRunDeps], query: str, limit: int | None = None
) -> str:
"""Search the knowledge base using hybrid search (vector + full-text).
Returns ranked results with content and metadata.
Args:
query: The search query.
limit: Maximum number of results.
"""
from haiku.rag.skills._tools import skill_search
state = _get_state(ctx)
formatted, results = await skill_search(
db_path,
config,
query,
limit=limit,
document_filter=state.document_filter if state else None,
)
if state:
state.searches[query] = results
return formatted
async def list_documents(
ctx: RunContext[SkillRunDeps],
limit: int | None = None,
offset: int | None = None,
) -> list[dict[str, Any]]:
"""List documents in the knowledge base with optional pagination.
Args:
limit: Maximum number of documents to return.
offset: Number of documents to skip.
"""
from haiku.rag.skills._tools import (
skill_list_documents,
update_documents_state,
)
result = await skill_list_documents(db_path, config, limit, offset)
state = _get_state(ctx)
if state:
update_documents_state(state.documents, result)
return result
async def get_document(
ctx: RunContext[SkillRunDeps], query: str
) -> dict[str, Any] | None:
"""Retrieve a document by ID, title, or URI.
Args:
query: Document ID, title, or URI to look up.
"""
from haiku.rag.skills._tools import (
skill_get_document,
update_documents_state,
)
result = await skill_get_document(db_path, config, query)
if result is not None:
state = _get_state(ctx)
if state:
update_documents_state(state.documents, [result])
return result
async def ask(ctx: RunContext[SkillRunDeps], question: str) -> str:
"""Ask a question and get an answer with citations from the knowledge base.
Args:
question: The question to ask.
"""
from haiku.rag.skills._tools import skill_ask
from haiku.rag.utils import format_citations
state = _get_state(ctx)
answer, citations = await skill_ask(
db_path,
config,
question,
qa_history=state.qa_history if state else None,
document_filter=state.document_filter if state else None,
)
if state:
next_index = len(state.citations) + 1
for citation in citations:
citation.index = next_index
next_index += 1
state.citations.extend(citations)
state.qa_history.append(
QAHistoryEntry(question=question, answer=answer, citations=citations)
)
if citations:
answer += "\n\n" + format_citations(citations)
return answer
async def research(ctx: RunContext[SkillRunDeps], question: str) -> str:
"""Conduct deep multi-agent research on a question.
Iteratively searches, analyzes, and synthesizes information from the
knowledge base to produce a comprehensive research report.
Only use when the user explicitly requests deep research.
Args:
question: The research question to investigate.
"""
from haiku.rag.skills._tools import skill_research
state = _get_state(ctx)
formatted, title, executive_summary = await skill_research(
db_path,
config,
question,
document_filter=state.document_filter if state else None,
)
if state:
state.reports.append(
ResearchEntry(
question=question,
title=title,
executive_summary=executive_summary,
)
)
state.qa_history.append(
QAHistoryEntry(question=question, answer=executive_summary)
)
return formatted
tools = create_skill_tools(db_path, config, RAGState, _RAG_TOOLS)
return Skill(
metadata=skill_metadata(),
source=SkillSource.ENTRYPOINT,
path=_skill_path,
instructions=instructions(),
tools=[
search,
list_documents,
get_document,
ask,
research,
],
tools=list(tools.values()),
state_type=STATE_TYPE,
state_namespace=STATE_NAMESPACE,
)

45
haiku_rag_slim/haiku/rag/skills/rlm.py
View file

@ -4,17 +4,10 @@ from pathlib import Path
from typing import Any
from pydantic import BaseModel
from pydantic_ai import RunContext
from haiku.rag.skills._tools import AnalysisEntry
from haiku.skills.models import Skill, SkillMetadata, SkillSource, StateMetadata
from haiku.skills.parser import parse_skill_md
from haiku.skills.state import SkillRunDeps
class AnalysisEntry(BaseModel):
question: str
answer: str
program: str | None = None
class RLMState(BaseModel):
@ -61,6 +54,7 @@ def create_skill(
config: haiku.rag AppConfig instance. If None, uses get_config().
"""
from haiku.rag.config import get_config
from haiku.rag.skills._tools import create_skill_tools
if config is None:
config = get_config()
@ -72,45 +66,14 @@ def create_skill(
else:
db_path = config.storage.data_dir / "haiku.rag.lancedb"
async def analyze(
ctx: RunContext[SkillRunDeps],
question: str,
document: str | None = None,
filter: str | None = None,
) -> str:
"""Answer complex analytical questions using code execution.
Use this for questions requiring computation, aggregation, or
data traversal across documents.
Args:
question: The question to answer.
document: Optional document ID or title to pre-load for analysis.
filter: Optional SQL WHERE clause to filter documents.
"""
from haiku.rag.skills._tools import skill_analyze
output, answer, program = await skill_analyze(
db_path, config, question, document=document, filter=filter
)
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RLMState):
ctx.deps.state.analyses.append(
AnalysisEntry(
question=question,
answer=answer,
program=program,
)
)
return output
tools = create_skill_tools(db_path, config, RLMState, ["analyze"])
return Skill(
metadata=skill_metadata(),
source=SkillSource.ENTRYPOINT,
path=_skill_path,
instructions=instructions(),
tools=[analyze],
tools=list(tools.values()),
state_type=STATE_TYPE,
state_namespace=STATE_NAMESPACE,
)

1
haiku_rag_slim/pyproject.toml
View file

@ -26,6 +26,7 @@ dependencies = [
"docling-core>=2.70.2",
"haiku.skills>=0.10.0",
"httpx>=0.28.1",
"jinja2>=3.1.0",
"jsonpatch>=1.33",
"lancedb==0.29.2",
"pathspec>=1.0.4",

370
tests/test_skill_generator.py Normal file
View file

@ -0,0 +1,370 @@
import pytest
from haiku.rag.skill_generator import (
AVAILABLE_TOOLS,
generate_skill,
render_templates,
validate_db_path,
validate_metadata,
validate_output_dir,
validate_tools,
)
class TestAvailableTools:
def test_all_tools_present(self):
assert AVAILABLE_TOOLS == {
"list_documents",
"get_document",
"search",
"ask",
"research",
"analyze",
}
class TestValidateMetadata:
def test_valid(self):
validate_metadata("recipes", "A skill.")
def test_valid_with_numbers(self):
validate_metadata("recipes123", "A skill.")
def test_rejects_underscores(self):
with pytest.raises(ValueError, match="name"):
validate_metadata("my_recipes", "A skill.")
def test_rejects_hyphens(self):
with pytest.raises(ValueError, match="identifier"):
validate_metadata("my-recipes", "A skill.")
def test_rejects_uppercase(self):
with pytest.raises(ValueError, match="lowercase"):
validate_metadata("Recipes", "A skill.")
def test_rejects_empty_name(self):
with pytest.raises(ValueError, match="name"):
validate_metadata("", "A skill.")
def test_rejects_not_identifier(self):
with pytest.raises(ValueError, match="identifier"):
validate_metadata("123abc", "A skill.")
def test_rejects_spaces_in_name(self):
with pytest.raises(ValueError, match="name"):
validate_metadata("my recipes", "A skill.")
def test_rejects_special_chars(self):
with pytest.raises(ValueError, match="name"):
validate_metadata("my@recipes", "A skill.")
def test_rejects_empty_description(self):
with pytest.raises(ValueError, match="description"):
validate_metadata("recipes", "")
def test_rejects_too_long_description(self):
with pytest.raises(ValueError, match="description"):
validate_metadata("recipes", "x" * 1025)
class TestValidateTools:
def test_valid_single_tool(self):
validate_tools(["search"])
def test_valid_multiple_tools(self):
validate_tools(["list_documents", "get_document", "search", "ask"])
def test_valid_all_tools(self):
validate_tools(list(AVAILABLE_TOOLS))
def test_rejects_empty(self):
with pytest.raises(ValueError, match="at least one"):
validate_tools([])
def test_rejects_unknown_tool(self):
with pytest.raises(ValueError, match="Unknown"):
validate_tools(["search", "bogus"])
class TestValidateDbPath:
def test_valid_path(self, tmp_path):
db_path = tmp_path / "test.lancedb"
db_path.mkdir()
validate_db_path(db_path)
def test_rejects_nonexistent(self, tmp_path):
db_path = tmp_path / "nonexistent.lancedb"
with pytest.raises(ValueError, match="does not exist"):
validate_db_path(db_path)
def test_rejects_file(self, tmp_path):
db_path = tmp_path / "test.lancedb"
db_path.touch()
with pytest.raises(ValueError, match="not a directory"):
validate_db_path(db_path)
class TestValidateOutputDir:
def test_valid_output_dir(self, tmp_path):
validate_output_dir(tmp_path, "recipes")
def test_rejects_nonexistent(self, tmp_path):
output_dir = tmp_path / "nonexistent"
with pytest.raises(ValueError, match="does not exist"):
validate_output_dir(output_dir, "recipes")
def test_rejects_existing_target(self, tmp_path):
target = tmp_path / "recipes-skill"
target.mkdir()
with pytest.raises(ValueError, match="already exists"):
validate_output_dir(tmp_path, "recipes")
class TestRenderTemplates:
def test_output_structure(self, tmp_path):
result = render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["list_documents", "get_document", "search", "ask"],
)
assert result == tmp_path / "recipes-skill"
assert result.is_dir()
pkg = result / "recipes_skill"
assert (pkg / "__init__.py").is_file()
assert (pkg / "SKILL.md").is_file()
assert (pkg / "assets").is_dir()
def test_tool_names_list_matches_selection(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=["search", "ask"],
)
init = tmp_path / "docs-skill" / "docs_skill" / "__init__.py"
content = init.read_text()
assert '["search", "ask"]' in content
def test_create_skill_tools_called(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=list(AVAILABLE_TOOLS),
)
init = tmp_path / "docs-skill" / "docs_skill" / "__init__.py"
content = init.read_text()
assert (
"create_skill_tools(_DB_PATH, config, SkillState, _TOOL_NAMES)" in content
)
def test_tool_names_in_init(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search", "ask"],
)
init = tmp_path / "recipes-skill" / "recipes_skill" / "__init__.py"
content = init.read_text()
assert '"search"' in content
assert '"ask"' in content
def test_pyproject_toml(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search"],
)
toml = tmp_path / "recipes-skill" / "pyproject.toml"
content = toml.read_text()
assert 'name = "recipes-skill"' in content
assert 'description = "A recipe skill."' in content
assert 'recipes = "recipes_skill:create_skill"' in content
assert "haiku.rag-slim >= 0.35" in content
def test_skill_md_conditionals(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=["search"],
)
skill_md = tmp_path / "docs-skill" / "docs_skill" / "SKILL.md"
content = skill_md.read_text()
assert "**search**" in content
assert "**ask**" not in content
assert "**list_documents**" not in content
assert "**research**" not in content
assert "**analyze**" not in content
def test_skill_md_includes_all_selected(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=["search", "ask", "analyze"],
)
skill_md = tmp_path / "docs-skill" / "docs_skill" / "SKILL.md"
content = skill_md.read_text()
assert "**search**" in content
assert "**ask**" in content
assert "**analyze**" in content
def test_custom_preamble(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=["search"],
preamble="You are a docs expert.",
)
skill_md = tmp_path / "docs-skill" / "docs_skill" / "SKILL.md"
content = skill_md.read_text()
assert "You are a docs expert." in content
def test_state_namespace_is_skill_name(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search"],
)
init = tmp_path / "recipes-skill" / "recipes_skill" / "__init__.py"
content = init.read_text()
assert 'state_namespace="recipes"' in content
def test_analyze_state_fields(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="docs",
description="A docs skill.",
tool_names=["search", "analyze"],
)
init = tmp_path / "docs-skill" / "docs_skill" / "__init__.py"
content = init.read_text()
assert "analyses" in content
def test_imports_from_shared_tools(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search", "ask", "analyze"],
)
init = tmp_path / "recipes-skill" / "recipes_skill" / "__init__.py"
content = init.read_text()
assert "from haiku.rag.skills._tools import create_skill_tools" in content
def test_generated_python_is_valid(self, tmp_path):
render_templates(
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=list(AVAILABLE_TOOLS),
)
pkg = tmp_path / "recipes-skill" / "recipes_skill"
for py_file in pkg.glob("*.py"):
source = py_file.read_text()
compile(source, str(py_file), "exec")
def _make_fake_lancedb(path):
path.mkdir()
(path / "data.lance").touch()
return path
class TestGenerateSkill:
def test_end_to_end(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
result = generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search", "ask"],
)
assert result == tmp_path / "recipes-skill"
assets = result / "recipes_skill" / "assets"
assert (assets / "recipes.lancedb").is_dir()
assert (assets / "recipes.lancedb" / "data.lance").is_file()
assert not (assets / "haiku.rag.yaml").exists()
def test_with_config(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
config_file = tmp_path / "haiku.rag.yaml"
config_file.write_text("storage:\n data_dir: /tmp\n")
result = generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search"],
config_path=config_file,
)
assets = result / "recipes_skill" / "assets"
assert (assets / "haiku.rag.yaml").is_file()
assert (assets / "haiku.rag.yaml").read_text() == (
"storage:\n data_dir: /tmp\n"
)
def test_rejects_invalid_name(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
with pytest.raises(ValueError, match="name"):
generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="Bad-Name",
description="A skill.",
tool_names=["search"],
)
def test_rejects_invalid_tools(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
with pytest.raises(ValueError, match="Unknown"):
generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="recipes",
description="A skill.",
tool_names=["bogus"],
)
def test_rejects_nonexistent_db(self, tmp_path):
with pytest.raises(ValueError, match="does not exist"):
generate_skill(
db_path=tmp_path / "nope.lancedb",
output_dir=tmp_path,
name="recipes",
description="A skill.",
tool_names=["search"],
)
def test_rejects_existing_target(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
(tmp_path / "recipes-skill").mkdir()
with pytest.raises(ValueError, match="already exists"):
generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="recipes",
description="A skill.",
tool_names=["search"],
)
def test_with_preamble(self, tmp_path):
db_path = _make_fake_lancedb(tmp_path / "test.lancedb")
result = generate_skill(
db_path=db_path,
output_dir=tmp_path,
name="recipes",
description="A recipe skill.",
tool_names=["search"],
preamble="You are a recipe expert.",
)
skill_md = result / "recipes_skill" / "SKILL.md"
content = skill_md.read_text()
assert "You are a recipe expert." in content

2
uv.lock
View file

@ -1506,6 +1506,7 @@ dependencies = [
{ name = "docling-core" },
{ name = "haiku-skills" },
{ name = "httpx" },
{ name = "jinja2" },
{ name = "jsonpatch" },
{ name = "lancedb" },
{ name = "pathspec" },
@ -1571,6 +1572,7 @@ requires-dist = [
{ name = "docling-core", specifier = ">=2.70.2" },
{ name = "haiku-skills", specifier = ">=0.10.0" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "jinja2", specifier = ">=3.1.0" },
{ name = "jsonpatch", specifier = ">=1.33" },
{ name = "lancedb", specifier = "==0.29.2" },
{ name = "mxbai-rerank", marker = "extra == 'mxbai'", specifier = ">=0.1.6" },