runapprentice 0.2.0

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

runapprentice

The Python SDK for Apprentice, a tool that cuts frontier-LLM cost on repeatable tasks. You give it a dataset of inputs and correct outputs for one task (uploaded as a CSV, or captured live from a LangChain app with one callback). Apprentice runs prompt optimization (DSPy GEPA) against the verified rows (gold plus silver), or fine-tunes a small open model on your Mac, and reports the score change on a held-out slice, so the gain is measured, never claimed. Gated, eval-verified takeover of production traffic is still in development.

Docs: docs.runapprentice.com · Reproducible benchmark: apprentice-benchmark

This README documents the implemented surface. Anything not shown here does not exist yet.

CLI (no code, no account required)

uv add 'runapprentice[optimize]'   # add [local] too for MLX fine-tuning on Apple silicon

apprentice optimize <task> --local --data golden.csv              # free, your OPENAI_API_KEY
apprentice train <task> --local --data golden.csv --effort high   # free, fine-tunes on your Mac

Both commands work on a CSV alone, no Apprentice account. No golden dataset yet? apprentice login also lets Apprentice generate a starter set and route rows to a subject-matter expert for review, and tracks every run in the console. apprentice --help (or any subcommand --help) documents every flag; --json gives clean machine-readable output.

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)

Capture records the call; feedback records whether it worked. That score is what the console's Drift tab charts, and what decides when a retrain is worth doing.

trace_id = client.capture(task="support-triage", input=question, output=answer)
if trace_id:  # None means the capture was rejected locally; capture never raises
    client.feedback(trace_id, good=True)                  # thumbs
    client.feedback(trace_id, score=0.5, note="partial")  # graded

In async mode (the default), feedback() first waits up to two seconds for pending captures to deliver, so it can never race its own trace; if the queue cannot drain in time it raises ApprenticeError instead of sending a request that would 404.

Wire it to a signal you already have: a thumbs up or down, the user accepting or discarding the output, a downstream check that passed or failed. Do not add a second model to grade the first one, this score decides when to retrain. With no real signal, send nothing: captured rows still become gold when a human verifies them.

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; ApprenticeError if pending captures cannot drain in time, or on API failure
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.