What you will build
In this tutorial you will build a document-triage agent that reads a folder of incoming files, classifies each one, extracts structured fields, and decides which ones a human must review. The whole thing fits in about ninety lines of Python with no frameworks beyond the standard library, because the interesting engineering in an agent is not the plumbing — it is the prompt layout, the tool schemas, the guardrails, and the eval set.
Because we lay out the prompt for cache reuse, the expensive part of each call is a cached prefix. That is the single biggest cost lever available in 2026, so the tutorial bakes it in rather than adding it later.
Step 0: Prerequisites
- Python 3.11 or newer. No third-party packages required.
- An API key for any OpenAI-compatible endpoint. Any provider works; only the base URL and model name change.
- A folder of test documents as plain text files. Twenty-five is plenty to start.
Step 1: Lay out the prompt so the prefix caches
The static block — instructions, tool schemas, and a worked example — goes first and never changes. The variable block — the document itself — goes last. Keeping the prefix byte-identical across calls is what turns cache reads into a rounding error.
STATIC_PREFIX = '''<role>You triage incoming documents for a finance team.</role>
<rules>
Always call triage_document exactly once.
If a required field is absent from the document, use null and add the
field name to "missing".
Never guess an invoice total. If multiple totals appear, set needs_human true.
</rules>
<example>
Document: "Receipt from Acme, $42.10, ref 9912"
triage_document(document_id="a1", doc_type="receipt", total=42.10,
currency="USD", reference="9912", missing=[], needs_human=false)
</example>'''
def build_messages(doc_id, text):
return [
{"role": "system", "content": STATIC_PREFIX},
{"role": "user", "content": f"<document id='{doc_id}'>{text}</document>"},
]
Note what is absent: no timestamp, no session id, no random ordering. Anything that changes between calls sits in the variable block.
Step 2: Define the tool schema once and reuse it
The schema is part of the cached prefix, so it must be stable and unusually explicit. Every field gets a type and a rule for when to leave it empty.
TOOLS = [{"type": "function", "function": {
"name": "triage_document",
"description": "Record the triage result for one document.",
"parameters": {"type": "object", "properties": {
"document_id": {"type": "string"},
"doc_type": {"type": "string",
"enum": ["invoice", "receipt", "contract", "other"]},
"total": {"type": ["number", "null"]},
"currency": {"type": ["string", "null"]},
"reference": {"type": ["string", "null"]},
"missing": {"type": "array", "items": {"type": "string"}},
"needs_human": {"type": "boolean"}
},
"required": ["document_id", "doc_type", "missing", "needs_human"]}}}]
Step 3: The agent loop
One call, one tool, one structured result. Resist the urge to make it a general-purpose loop on day one; a single-step agent with a strict schema is far easier to evaluate and far cheaper to run.
import json, os, urllib.request
BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.example.com/v1")
API_KEY = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "triage-small")
def post(path, payload):
req = urllib.request.Request(
BASE_URL + path, data=json.dumps(payload).encode(),
headers={"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=60) as r:
return json.load(r)
def triage(doc_id, text):
body = {"model": MODEL, "messages": build_messages(doc_id, text),
"tools": TOOLS, "tool_choice": "required",
"temperature": 0, "max_tokens": 400}
resp = post("/chat/completions", body)
call = resp["choices"][0]["message"]["tool_calls"][0]
result = json.loads(call["function"]["arguments"])
usage = resp.get("usage", {})
result["_cost"] = estimate_cost(usage)
return result
Three details that matter. Setting temperature to zero makes results reproducible enough to evaluate. Sending tool_choice: required removes an entire class of parsing failure. And capturing usage on every call is what lets you report cost per completed task later.
Step 4: Guardrails before you point it at anything real
ALLOWED_TYPES = {"invoice", "receipt", "contract", "other"}
def safe_triage(doc_id, text):
if len(text) > 60_000:
text = text[:60_000] # bounded input
if not text.strip():
return {"document_id": doc_id, "doc_type": "other", "total": None,
"missing": ["content"], "needs_human": True, "_cost": 0.0}
try:
out = triage(doc_id, text)
except Exception as e:
return {"document_id": doc_id, "doc_type": "other", "total": None,
"missing": ["api_error"], "needs_human": True,
"_error": str(e), "_cost": 0.0}
if out.get("doc_type") not in ALLOWED_TYPES:
out["needs_human"] = True # unknown enum value = review
out["document_id"] = doc_id # never trust the model's echo
return out
The pattern is worth naming: bound the input, catch every failure and route it to a human, never trust an echoed identifier, and treat any value outside your allowlist as a review case rather than an error to retry.
Step 5: Build the golden set before you tune anything
Write twenty-five cases to a JSONL file with the field values you expect. Include the ugly ones, because they are the ones that break in production.
# cases.jsonl — one line per case
{"id": "c01", "text": "Invoice 4471 from Northwind, total 1,240.00 USD",
"expect": {"doc_type": "invoice", "total": 1240.0, "needs_human": false}}
{"id": "c02", "text": "Receipt: 19.99 and 24.99",
"expect": {"needs_human": true}}
import json
def evaluate(path="cases.jsonl"):
cases = [json.loads(l) for l in open(path) if l.strip()]
ok = 0
for c in cases:
got = safe_triage(c["id"], c["text"])
if all(got.get(k) == v for k, v in c["expect"].items()):
ok += 1
print(f"{ok}/{len(cases)} passed")
return ok / len(cases)
Run this after every prompt change. A golden set is the difference between improving the agent and merely editing it.
Step 6: Measure cost per completed task
Do not track price per token; track what a finished document costs. Sum token usage per run, convert with your provider's current rates, and divide by the number of documents that passed review without a human edit.
def estimate_cost(usage, in_price=0.15, out_price=0.60):
# prices are per million tokens; update for your provider and tier
hin = usage.get("prompt_tokens_details", {}).get("cached_tokens", 0)
miss = usage.get("prompt_tokens", 0) - hin
return (miss * in_price + hin * 0.003 + usage.get("completion_tokens", 0) * out_price) / 1e6
That number is the one to put on a dashboard, because it captures both price and reliability. A cheap model with a 70 percent first-pass rate often loses to a pricier one at 95 percent.
Common failure modes and their fixes
- Wrong tool, right answer. The tool description was too vague. Add an explicit rule about when not to call it.
- Cache-hit rate near zero. Something in your static prefix changes between calls. Diff two consecutive prompts and look for a timestamp, a set, or a reordered block.
- Schema-valid but wrong. The prompt, not the schema, is the problem. Add a single negative example showing that specific mistake.
- Cost spikes after an unrelated change. A model snapshot moved under you. Pin the version and re-run the golden set.
- Silent quality drift. No output inspection in weeks. Sample ten results weekly and read them yourself; dashboards will not catch a confident wrong answer.
Ship this version before you add multi-step reasoning, retrieval, or a second tool. Ninety lines with a golden set beats a thousand lines without one, and once the eval harness exists, every later addition is a measurable improvement instead of a hopeful one.
