runapprentice 0.4.2

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 an OpenAI or LangChain app). 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 --local --data commands work end to end from a CSV. Hosted setup requires the console or Python SDK to create a task and upload its dataset; the CLI has no create or upload command. 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")
resp = OpenAI().chat.completions.create(   # from openai import OpenAI
    model="gpt-5-mini",
    messages=best.messages(input=ticket_text),
)

datasets.upload() replaces prior rows for this task where the source is an upload and the tier is silver or raw. Gold rows and SDK-captured traces stay intact. Omitting prompt or passing None clears the task's stored baseline prompt. Check DatasetStatus.replaced_rows to see how many rows were removed.

best.text is instruction text, not a template. It holds no input placeholder and often contains literal JSON braces, so best.text.format(input=...) raises KeyError and best.text.replace("{input}", ...) drops the ticket with no error at all. messages() renders the message shape the backend recorded when it scored this version. Match the rest of the request too if you want the reported score to describe your call: JSON metrics were scored with response_format={"type": "json_object"} (on the Responses API, that is text={"format": {"type": "json_object"}}), on the model in report.detail["student_model"]. Versions optimized before 0.4.0 carry no recorded shape and fall back to the task's current baseline prompt. The keyword names are the artifact's input_variables: input for a plain task, question and context for RAG, your own names if you registered a template.

Live capture: OpenAI (one wrapper)

from openai import OpenAI
from runapprentice.openai import wrap

openai = wrap(OpenAI(), client, task="ticket-triage")

response = openai.responses.create(
    model="gpt-5-mini",
    input="Triage this ticket.",
)
# create() and parse() are captured, on chat.completions and responses alike
# (parse posts on its own, so patching create alone would miss it).

Both sync OpenAI and async AsyncOpenAI clients work. Capture is fail-open and returns every OpenAI result unchanged. Calls with stream=True and calls through responses.stream(...) pass through uncaptured because reading them would consume the caller's stream.

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

openai = wrap(OpenAI(), client, task="ticket-triage", redact=lambda s: my_scrubber(s))

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 including replaced_rows	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, baseline_prompt=None, budget=None, metric=None)	Job	400 if fewer than the backend's configured verified-row minimum (20 by default)
client.preview_optimization(task, baseline_prompt=None, budget=None, metric=None)	OptimizationPreview	ApprenticeError on API failure
client.train(task)	Job	400 if fewer gold rows than the backend's training minimum
client.job(job_id) / job.refresh() / job.wait(poll_seconds=, timeout_seconds=)	Job	ApprenticeError on timeout
client.cancel_job(job_id)	dict (cancel_requested, status)	409 if job is not cancellable
job.cancel()	dict (cancel_requested, status)	409 if job is not cancellable
client.jobs(task)	list[JobSummary]	ApprenticeError on API failure
client.local_runs(task)	list[LocalRun]	ApprenticeError on API failure
client.datasets.rows(task, tier, limit=200)	list[DatasetRow]	ApprenticeError on API failure
client.optimize_local(task, data=None, **options)	dict (local GEPA result)	ApprenticeError if no data and no API key
client.train_local(task, data=None, effort=None, **options)	dict (local MLX result)	ApprenticeError if no data and no API key
client.register_local_run(task, base_model, dataset_row_count, seed, config=None)	LocalRun	ApprenticeError on API failure
client.complete_local_run(run_id, payload)	dict	ApprenticeError on API failure
client.capture(task, output, input= or inputs=, model=, latency_ms=, ...)	trace_id or None	never raises
client.flush_captures(timeout=2.0)	None (blocks until buffered traces deliver or timeout; no-op in sync mode)	nothing by design
client.close()	None (flushes and closes transport)	nothing by design
job.report()	typed optimize, train, or generate report; raw dict for future shapes	if job has no report
client.prompts.get(task, version=None)	PromptVersion(version, text, score, baseline_prompt)	404 if never optimized
PromptVersion.messages(user_prompt=None, **inputs)	list[dict] reproducing the scored call	ValueError if a keyword is not one of the artifact's input_variables
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
wrap(openai_client, client, task, redact=None)	the same OpenAI client, with chat completions and responses captured	never raises from capture; OpenAI call errors pass through unchanged
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.