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

Merge pull request #316 from ggozad/feature/create-skill

Browse source 
Add create-skill command for generating standalone skill packages
This commit is contained in:
Yiorgis Gozadinos 2026-03-24 17:46:03 +02:00 committed by GitHub
commit 757b58d5db
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1251 additions and 314 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -1,6 +1,10 @@
# Changelog
## [Unreleased]
### Added
- **`create-skill` CLI command**: Generate standalone skill packages with embedded LanceDB databases. Generated packages register as `haiku.skills` entry points.
## [0.35.0] - 2026-03-24
### Added

58
docs/cli.md
View file

@ -247,6 +247,64 @@ Flags:
See [RLM Agent](agents/rlm.md) for details on capabilities and configuration.
## Create Skill
Generate a standalone skill package with an embedded database:
```bash
haiku-rag create-skill --name myskill --db /path/to/database.lancedb
```
The generated package is a pip-installable Python package that registers as a `haiku.skills` entry point.
### Options
| Flag | Description | Default |
|------|-------------|---------|
| `--name` | Skill name (lowercase alphanumeric and hyphens, required) | — |
| `--db` | Path to LanceDB database to embed (required) | — |
| `--description` | Skill description | Standard RAG description |
| `--tools` | Comma-separated tool names, or `all` | `all` |
| `--preamble` | Custom preamble for skill instructions | Standard RAG preamble |
| `--config-file` | Path to `haiku.rag.yaml` to embed | None |
| `--output` / `-o` | Output directory | Current directory |
### Available Tools
`analyze`, `ask`, `get_document`, `list_documents`, `research`, `search`
### Example
```bash
# Generate a skill with specific tools and custom preamble
haiku-rag create-skill \
--name medic \
--db /path/to/medic.lancedb \
--tools search,ask \
--config-file /path/to/haiku.rag.yaml \
--description "Military medic knowledge base" \
--preamble "You are a military medic expert."
# Install the generated package
uv pip install -e ./medic-skill
# Use with haiku-skills
haiku-skills chat --use-entrypoints --skill medic
```
### Generated Package Structure
```
{name}-skill/
├── pyproject.toml
└── {name}_skill/
├── __init__.py # create_skill() entry point
├── SKILL.md # Skill metadata and instructions
└── assets/
├── {name}.lancedb/ # Embedded database
└── haiku.rag.yaml # Optional config
```
## Server
Start services (requires at least one flag):

26
docs/skills/index.md
View file

@ -39,6 +39,32 @@ agent = Agent(
result = await agent.run("What documents do we have?")
```
## Generating Custom Skills
Use `create-skill` to generate a standalone skill package with an embedded database:
```bash
haiku-rag create-skill \
--name recipes \
--db /path/to/recipes.lancedb \
--tools search,ask \
--description "Recipe knowledge base" \
--preamble "You are a recipe expert."
```
This generates a pip-installable package (`recipes-skill/`) that bundles the database and registers as a `haiku.skills` entry point. After installing (`uv pip install -e ./recipes-skill`), the skill is automatically discovered:
```bash
haiku-skills list --use-entrypoints
# recipes — Recipe knowledge base
haiku-skills chat --use-entrypoints --skill recipes
```
Since each generated skill is self-contained with its own database and instructions, you can generate multiple skills for different domains and run them together. The agent sees each skill's description and routes questions to the appropriate knowledge base automatically.
See [CLI: Create Skill](../cli.md#create-skill) for all options.
## Database Path Resolution
Both skills resolve the database path in the same order:

73
haiku_rag_slim/haiku/rag/cli.py
View file

@ -716,5 +716,78 @@ def serve(
)
@_cli.command(
"create-skill",
help="Generate a standalone skill package with an embedded database",
)
def create_skill_cmd( # pragma: no cover
name: str = typer.Option(
...,
"--name",
help="Skill name (lowercase alphanumeric and hyphens)",
),
db: Path = typer.Option(
...,
"--db",
help="Path to the LanceDB database to embed",
),
description: str | None = typer.Option(
None,
"--description",
help="Skill description (default: standard RAG description)",
),
tools: str = typer.Option(
"all",
"--tools",
help="Comma-separated tool names, or 'all'",
),
preamble: str | None = typer.Option(
None,
"--preamble",
help="Custom preamble for the skill instructions",
),
config_file: Path | None = typer.Option(
None,
"--config-file",
help="Path to haiku.rag.yaml to embed in the skill",
),
output: Path = typer.Option(
Path("."),
"--output",
"-o",
help="Output directory for the generated package",
),
):
"""Generate a standalone haiku.skills package with an embedded database."""
from haiku.rag.skill_generator import (
AVAILABLE_TOOLS,
DEFAULT_DESCRIPTION,
generate_skill,
)
if description is None:
description = DEFAULT_DESCRIPTION
if tools.strip().lower() == "all":
tool_names = sorted(AVAILABLE_TOOLS)
else:
tool_names = [t.strip() for t in tools.split(",") if t.strip()]
try:
result = generate_skill(
db_path=db,
output_dir=output,
name=name,
description=description,
tool_names=tool_names,
config_path=config_file,
preamble=preamble,
)
typer.echo(f"Skill generated: {result}")
except ValueError as e:
typer.echo(f"Error: {e}", err=True)
raise typer.Exit(1)
if __name__ == "__main__": # pragma: no cover
cli()

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

@ -0,0 +1,142 @@
import pathlib
import shutil
from importlib.metadata import version
from jinja2 import Environment, PackageLoader
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=False,
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)
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
pkg_name = name.replace("-", "_")
rag_version = version("haiku.rag-slim")
env = _get_env()
context = {
"name": name,
"pkg_name": pkg_name,
"description": description,
"tool_names": tool_names,
"preamble": preamble,
"rag_version": rag_version,
}
result_dir = output_dir / f"{name}-skill"
pkg_dir = result_dir / f"{pkg_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,
)
pkg_name = name.replace("-", "_")
assets_dir = result / f"{pkg_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 >= {{ rag_version }}",
"haiku-skills >= 0.10.0",
]
[project.entry-points."haiku.skills"]
{{ name }} = "{{ pkg_name }}_skill:create_skill"

437
haiku_rag_slim/haiku/rag/skills/_tools.py Normal file
View file

@ -0,0 +1,437 @@
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.config.models import AppConfig
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.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(
qa_history: list[QAHistoryEntry],
query: str,
config: AppConfig,
) -> list[QAHistoryEntry]:
from haiku.rag.embeddings import get_embedder
from haiku.rag.tools.qa import PRIOR_ANSWER_RELEVANCE_THRESHOLD
from haiku.rag.utils import cosine_similarity
if not qa_history:
return []
embedder = get_embedder(config)
query_embedding = await embedder.embed_query(query)
to_embed = []
to_embed_indices = []
for i, qa in enumerate(qa_history):
if qa.question_embedding is None:
to_embed.append(qa.question)
to_embed_indices.append(i)
if to_embed:
new_embeddings = await embedder.embed_documents(to_embed)
for i, idx in enumerate(to_embed_indices):
qa_history[idx].question_embedding = new_embeddings[i]
matches = []
for qa in qa_history:
if qa.question_embedding is not None:
similarity = cosine_similarity(query_embedding, qa.question_embedding)
if similarity >= PRIOR_ANSWER_RELEVANCE_THRESHOLD:
matches.append(qa)
return matches
async def skill_search(
db_path: Path,
config: AppConfig,
query: str,
limit: int | None = None,
document_filter: str | None = None,
) -> tuple[str, list[SearchResult]]:
from haiku.rag.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
results = await rag.search(
query,
limit=limit,
filter=document_filter,
)
results = await rag.expand_context(results)
formatted = "\n\n---\n\n".join(
r.format_for_agent(rank=i + 1, total=len(results))
for i, r in enumerate(results)
)
return formatted, list(results)
async def skill_list_documents(
db_path: Path,
config: AppConfig,
limit: int | None = None,
offset: int | None = None,
) -> list[dict[str, Any]]:
from haiku.rag.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
documents = await rag.list_documents(limit, offset)
return [
{
"id": doc.id,
"title": doc.title,
"uri": doc.uri,
"metadata": doc.metadata,
"created_at": str(doc.created_at),
"updated_at": str(doc.updated_at),
}
for doc in documents
]
async def skill_get_document(
db_path: Path,
config: AppConfig,
query: str,
) -> dict[str, Any] | None:
from haiku.rag.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
document = await rag.resolve_document(query)
if document is None:
return None
return {
"id": document.id,
"content": document.content,
"title": document.title,
"uri": document.uri,
"metadata": document.metadata,
"created_at": str(document.created_at),
"updated_at": str(document.updated_at),
}
async def skill_ask(
db_path: Path,
config: AppConfig,
question: str,
qa_history: list[QAHistoryEntry] | None = None,
document_filter: str | None = None,
) -> tuple[str, list[Citation]]:
from haiku.rag.client import HaikuRAG
from haiku.rag.utils import format_citations
ask_question = question
if qa_history:
matches = await find_relevant_prior_qa(qa_history, question, config)
if matches:
prior_parts = []
for qa in matches:
part = f"Q: {qa.question}\nA: {qa.answer}"
if qa.citations:
part += "\n" + format_citations(qa.citations)
prior_parts.append(part)
ask_question = (
"Context from prior questions in this session:\n\n"
+ "\n\n---\n\n".join(prior_parts)
+ "\n\n---\n\nCurrent question: "
+ question
)
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
answer, citations = await rag.ask(
ask_question,
filter=document_filter,
)
return answer, citations
async def skill_research(
db_path: Path,
config: AppConfig,
question: str,
document_filter: str | None = None,
) -> tuple[str, str, str]:
from haiku.rag.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
report = await rag.research(question, filter=document_filter)
parts = [
f"# {report.title}",
f"\n## Executive Summary\n{report.executive_summary}",
]
if report.main_findings:
parts.append("\n## Main Findings")
for finding in report.main_findings:
parts.append(f"- {finding}")
if report.conclusions:
parts.append("\n## Conclusions")
for conclusion in report.conclusions:
parts.append(f"- {conclusion}")
if report.limitations:
parts.append("\n## Limitations")
for limitation in report.limitations:
parts.append(f"- {limitation}")
if report.recommendations:
parts.append("\n## Recommendations")
for rec in report.recommendations:
parts.append(f"- {rec}")
parts.append(f"\n## Sources\n{report.sources_summary}")
formatted = "\n".join(parts)
return formatted, report.title, report.executive_summary
async def skill_analyze(
db_path: Path,
config: AppConfig,
question: str,
document: str | None = None,
filter: str | None = None,
) -> tuple[str, str, str | None]:
from haiku.rag.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
documents = [document] if document else None
result = await rag.rlm(question, documents=documents, filter=filter)
output = result.answer
if result.program:
output += f"\n\nProgram:\n{result.program}"
return output, result.answer, result.program
def update_documents_state(
documents_state: list[DocumentInfo],
doc_dicts: list[dict[str, Any]],
) -> None:
for doc_dict in doc_dicts:
doc_info = DocumentInfo(
id=str(doc_dict["id"]),
title=doc_dict["title"] or "Untitled",
uri=doc_dict.get("uri") or "",
created=doc_dict.get("created_at", ""),
)
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[BaseModel]) -> 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: AppConfig,
state_type: type[BaseModel],
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

275
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):
@ -78,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()
@ -89,275 +85,14 @@ def create_skill(
else:
db_path = config.storage.data_dir / "haiku.rag.lancedb"
async def _find_relevant_prior_qa(
state: RAGState, query: str
) -> list[QAHistoryEntry]:
from haiku.rag.embeddings import get_embedder
from haiku.rag.tools.qa import PRIOR_ANSWER_RELEVANCE_THRESHOLD
from haiku.rag.utils import cosine_similarity
if not state.qa_history:
return []
embedder = get_embedder(config)
query_embedding = await embedder.embed_query(query)
to_embed = []
to_embed_indices = []
for i, qa in enumerate(state.qa_history):
if qa.question_embedding is None:
to_embed.append(qa.question)
to_embed_indices.append(i)
if to_embed:
new_embeddings = await embedder.embed_documents(to_embed)
for i, idx in enumerate(to_embed_indices):
state.qa_history[idx].question_embedding = new_embeddings[i]
matches = []
for qa in state.qa_history:
if qa.question_embedding is not None:
similarity = cosine_similarity(query_embedding, qa.question_embedding)
if similarity >= PRIOR_ANSWER_RELEVANCE_THRESHOLD:
matches.append(qa)
return matches
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.client import HaikuRAG
state = (
ctx.deps.state
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState)
else None
)
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
results = await rag.search(
query,
limit=limit,
filter=state.document_filter if state else None,
)
results = await rag.expand_context(results)
if state:
state.searches[query] = list(results)
return "\n\n---\n\n".join(
r.format_for_agent(rank=i + 1, total=len(results))
for i, r in enumerate(results)
)
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.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
documents = await rag.list_documents(limit, offset)
result = [
{
"id": doc.id,
"title": doc.title,
"uri": doc.uri,
"metadata": doc.metadata,
"created_at": str(doc.created_at),
"updated_at": str(doc.updated_at),
}
for doc in documents
]
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState):
for doc_dict in result:
doc_info = DocumentInfo(
id=str(doc_dict["id"]),
title=doc_dict["title"] or "Untitled",
uri=doc_dict.get("uri") or "",
created=doc_dict.get("created_at", ""),
)
if not any(d.id == doc_info.id for d in ctx.deps.state.documents):
ctx.deps.state.documents.append(doc_info)
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.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
document = await rag.resolve_document(query)
if document is None:
return None
result = {
"id": document.id,
"content": document.content,
"title": document.title,
"uri": document.uri,
"metadata": document.metadata,
"created_at": str(document.created_at),
"updated_at": str(document.updated_at),
}
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState):
doc_info = DocumentInfo(
id=str(result["id"]),
title=result["title"] or "Untitled",
uri=result.get("uri") or "",
created=result.get("created_at", ""),
)
if not any(d.id == doc_info.id for d in ctx.deps.state.documents):
ctx.deps.state.documents.append(doc_info)
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.client import HaikuRAG
from haiku.rag.utils import format_citations
state = (
ctx.deps.state
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState)
else None
)
ask_question = question
if state:
matches = await _find_relevant_prior_qa(state, question)
if matches:
prior_parts = []
for qa in matches:
part = f"Q: {qa.question}\nA: {qa.answer}"
if qa.citations:
part += "\n" + format_citations(qa.citations)
prior_parts.append(part)
ask_question = (
"Context from prior questions in this session:\n\n"
+ "\n\n---\n\n".join(prior_parts)
+ "\n\n---\n\nCurrent question: "
+ question
)
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
answer, citations = await rag.ask(
ask_question,
filter=state.document_filter if state else None,
)
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState):
next_index = len(ctx.deps.state.citations) + 1
for citation in citations:
citation.index = next_index
next_index += 1
ctx.deps.state.citations.extend(citations)
ctx.deps.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.client import HaikuRAG
state = (
ctx.deps.state
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RAGState)
else None
)
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
report = await rag.research(
question, filter=state.document_filter if state else None
)
if state:
state.reports.append(
ResearchEntry(
question=question,
title=report.title,
executive_summary=report.executive_summary,
)
)
state.qa_history.append(
QAHistoryEntry(question=question, answer=report.executive_summary)
)
parts = [
f"# {report.title}",
f"\n## Executive Summary\n{report.executive_summary}",
]
if report.main_findings:
parts.append("\n## Main Findings")
for finding in report.main_findings:
parts.append(f"- {finding}")
if report.conclusions:
parts.append("\n## Conclusions")
for conclusion in report.conclusions:
parts.append(f"- {conclusion}")
if report.limitations:
parts.append("\n## Limitations")
for limitation in report.limitations:
parts.append(f"- {limitation}")
if report.recommendations:
parts.append("\n## Recommendations")
for rec in report.recommendations:
parts.append(f"- {rec}")
parts.append(f"\n## Sources\n{report.sources_summary}")
return "\n".join(parts)
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,
)

48
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,48 +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.client import HaikuRAG
async with HaikuRAG(db_path, config=config, read_only=True) as rag:
documents = [document] if document else None
result = await rag.rlm(question, documents=documents, filter=filter)
output = result.answer
if result.program:
output += f"\n\nProgram:\n{result.program}"
if ctx.deps and ctx.deps.state and isinstance(ctx.deps.state, RLMState):
ctx.deps.state.analyses.append(
AnalysisEntry(
question=question,
answer=result.answer,
program=result.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",

352
tests/test_skill_generator.py Normal file
View file

@ -0,0 +1,352 @@
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("my-recipes", "A skill.")
def test_rejects_invalid_name(self):
with pytest.raises(ValueError):
validate_metadata("Bad_Name!", "A skill.")
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_dashed_name_uses_underscores_for_python(self, tmp_path):
result = render_templates(
output_dir=tmp_path,
name="my-recipes",
description="A recipe skill.",
tool_names=["search", "ask"],
)
assert result == tmp_path / "my-recipes-skill"
pkg = result / "my_recipes_skill"
assert (pkg / "__init__.py").is_file()
init = (pkg / "__init__.py").read_text()
assert '"my-recipes.lancedb"' in init
assert 'state_namespace="my-recipes"' in init
toml = (result / "pyproject.toml").read_text()
assert 'name = "my-recipes-skill"' in toml
assert 'my-recipes = "my_recipes_skill:create_skill"' in toml
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 >= " 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):
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" },