runapprentice 0.1.0

Cut LLM cost without losing quality: capture traffic, build golden datasets, and optimize prompts.

runapprentice

Capture your LLM traffic, build a golden dataset, and optimize your prompt against it. Training a small model to take over from the frontier model is planned for a later release.

This README documents the implemented surface (v0.1). Anything not shown here does not exist yet.

Install

uv add runapprentice[langchain]    # [langchain] extra enables zero-code capture

Quickstart (the whole trial journey)

from runapprentice import Apprentice

client = Apprentice(
    api_key="ap_live_...",   # create once in the console, or env APPRENTICE_API_KEY
)                            # talks to the hosted API; set base_url only to self-host

# 1. Create a task (a "task" = one repeatable LLM job you want to make cheaper)
client.tasks.create("ticket-triage")

# 2. Point us at the data you already have (CSV with input/output columns)
client.datasets.upload(
    "ticket-triage",
    path="golden.csv",
    input_col="input",
    output_col="output",
    prompt="...your current system prompt...",   # the baseline to beat
)

# 3. Optimize the prompt against your data (GEPA, held-out eval)
job = client.optimize("ticket-triage")
report = job.wait().report()
print(report.baseline_score, "->", report.optimized_score)

# 4. Pull the optimized prompt back into your code (versioned)
best = client.prompts.get("ticket-triage")
print(best.version, best.text)

Live capture: LangChain (one line, zero code changes)

# init_chat_model comes from your LangChain model package (for example
# langchain[openai]); runapprentice[langchain] only adds langchain-core for
# the callback.
from langchain.chat_models import init_chat_model
from runapprentice.langchain import ApprenticeCallback

model = init_chat_model(
    "gpt-5.5",   # whatever YOU already use. We observe, we don't choose
    callbacks=[ApprenticeCallback("ticket-triage", client)],
)
# Every call now logs input/output/model/latency/tokens to the task's dataset (raw tier).

Fail-open guarantee: the capture path can NEVER break your app. If the Apprentice backend is down, unreachable, or slow, your LLM calls proceed untouched. The only loss is the trace. First drop logs a WARNING, repeats log at DEBUG.

PII redaction runs in your process, before anything is transmitted:

ApprenticeCallback("ticket-triage", client, redact=lambda s: my_scrubber(s))

RAG quickstart (grounding + refusal-aware)

For retrieval-augmented tasks, each row carries the exact context the model saw (the retrieved passages), not just the question. Pick the rag_composite metric so optimization rewards grounding and correct refusals, not just answer overlap.

from runapprentice import Apprentice
client = Apprentice(api_key="ap_live_...")

# rag_composite scores answer correctness + faithfulness to context + refusal
# correctness (it should say "not enough information" when the context lacks it).
client.tasks.create("support-rag", metric="rag_composite")

client.datasets.upload("support-rag", rows=[
    {"inputs": {"question": q, "context": retrieved_passages}, "output": gold_answer}
    for q, retrieved_passages, gold_answer in your_golden_set
])

client.prompts.register("support-rag", {
    "format": "f-string",
    "messages": [
        {"role": "system", "template":
            "Answer using only the context. If the context does not contain the "
            "answer, say you do not have enough information.\n{context}"},
        {"role": "human", "template": "{question}"},
    ],
    "input_variables": ["context", "question"],
})

report = client.optimize("support-rag").wait().report()
print(report.baseline_score, "->", report.optimized_score)

RAG capture for simple chains. For a standard LangChain RAG chain, the ApprenticeCallback records the retrieved context from on_retriever_end, capturing the {question, context} shape. For custom formatting, multiple retrievers, or non-standard chains, call client.capture(..., inputs={"question": ..., "context": ...}) explicitly.

Register a LangChain prompt directly

prompts.register also accepts a LangChain ChatPromptTemplate, no raw dict needed:

from langchain_core.prompts import ChatPromptTemplate
client.prompts.register("support-rag", ChatPromptTemplate.from_messages([
    ("system", "Answer using only the context. If it lacks the answer, refuse.\n{context}"),
    ("human", "{question}"),
]))
# round-trip the optimized prompt back into LangChain:
optimized = client.prompts.to_langchain("support-rag")

Metric menu

metric=	Use for	Scored by
auto (default)	let the backend infer from your rows	inferred
json_f1	JSON / structured extraction	deterministic
text_f1	short free-text answers	deterministic
semantic_f1	free-text / RAG answers (default for RAG)	LLM judge
rag_faithfulness	is every claim supported by the context	LLM judge
rag_composite	RAG grounding and refusal correctness	LLM judge

RAG rows auto-route to semantic_f1; pass metric="rag_composite" explicitly when you want grounding + refusal optimized together.

Feedback (your end-users' signal)

client.feedback(trace_id, good=True)                  # thumbs
client.feedback(trace_id, score=0.5, note="partial")  # graded

Debugging

import runapprentice
runapprentice.enable_debug_logging()   # or env APPRENTICE_DEBUG=1

Shows every API call with status + latency, every capture attempt, every job poll. Control-plane errors are designed to tell you what to do (e.g. unreachable backend includes the URL it tried; optimize with too few rows tells you the count and the minimum).

API reference (implemented surface)

Call	Returns	Raises
Apprentice(api_key=, base_url=, timeout=)	client	ApprenticeError on bad config
client.tasks.create(name, metric="auto")	dict (created bool)	ApprenticeError on HTTP failure
client.datasets.upload(task, path= or rows=, input_col=, output_col=, prompt=)	DatasetStatus	same; also if both/neither of path/rows
client.datasets.status(task)	DatasetStatus(gold, silver, raw, ready_for_optimization)	same
client.prompts.register(task, template)	dict	ImportError if a LangChain template is passed without the [langchain] extra
client.prompts.to_langchain(task= or artifact=)	LangChain prompt	ImportError without the [langchain] extra
client.optimize(task)	Job	400 if fewer than the backend's configured verified-row minimum (20 by default)
client.job(job_id) / job.refresh() / job.wait(poll_seconds=, timeout_seconds=)	Job	ApprenticeError on timeout
job.report()	OptimizationReport(baseline_score, optimized_score, optimized_prompt, ...)	if job has no report
client.prompts.get(task, version=None)	PromptVersion(version, text, score)	404 if never optimized
client.prompts.history(task)	list[PromptVersion]	ApprenticeError on API failure
client.feedback(trace_id, good=, score=, note=)	None	404 unknown trace
client.post_trace_failopen(record)	trace_id or None	never raises
ApprenticeCallback(task, client, redact=None)	LangChain callback	never raises into your chain

Data tiers (how your rows are treated)

• raw: captured from live traffic, unverified
• silver: uploaded by you (curated) or passed deterministic checks
• gold: human-verified
• Optimization uses gold + silver; eval gates will use gold only.

For AI coding tools

Module docstrings and this README are the source of truth. Key invariants an agent must preserve when editing this package:

1. post_trace_failopen and everything in runapprentice/langchain.py must never raise into the caller. Capture is fail-open by contract (see tests/test_failopen.py).
2. Control-plane methods must raise ApprenticeError with actionable messages.
3. Dependencies stay minimal: httpx + pydantic; LangChain only via the [langchain] extra; never import from apprentice-api.
4. Required tests for any new method: add a row to ../needed_test.md.