Cost per document in a LangChain invoice workflow
Triage, extraction, exception notes and a run report in LangChain 1.x on GPT-4.1, then a blocking budget on the role whose call count the triage chose.
- Framework
- LangChain 1.x
- Provider
- openai
- Python
- 3.11+
- Capsera
- pip install capsera
This builds an accounts-payable intake workflow in LangChain 1.x against OpenAI:
triage sorts an inbound mail queue, an extractor reads the fields off the
documents that are actually invoices, Python does the arithmetic and the
duplicate check, and an adjudicator writes the exception note for anything that
fails. Seven documents is eight model calls at the floor and twenty-four at the
cap, and the role that decides where in that range a run lands is not the role
that runs most often. The last third of this page adds Capsera: one init()
call to capture every request, four decorator lines to say which role made it,
and a blocking budget on the role that multiplies.
What you will build
intake.py, a single-file workflow with four roles and a supervisor:
- triage — one call per document in the queue, on the cheap model. Returns
one word:
invoice,credit-note,statementorother. Its job is to stop the expensive role from ever seeing a delivery note. - extractor — one call per document triage called an invoice, on the stronger model. Copies out the supplier, the invoice number, the purchase order, every charge line and the stated total. The multiplier.
- adjudicator — one call per invoice that failed a check, on the cheap model. Writes the note a person will read and picks a disposition from a fixed vocabulary.
- reporter — one call. Writes the end-of-run note for the mailbox from facts the supervisor assembled, not from the documents.
Everything that decides whether money moves is Python, not a model call. Summing the charge lines, comparing that sum to the stated total, matching the supplier against the approved list, catching an invoice number that is already posted, and testing the total against the auto-approval limit are arithmetic and set membership. A workflow that asks a model to add up an invoice has a correctness problem and a cost curve nothing in the prompts explains.
Each role is its own function. That matters later — it is the seam the instrumentation attaches to — and it is the better shape regardless, because a pipeline that inlines four model calls has nothing you can name, budget, or move to a different model.
Written against LangChain 1.x with langchain-openai, calling the OpenAI API
directly rather than through a gateway. Models: gpt-4.1-mini for triage, the
adjudicator and the reporter; gpt-4.1 for the extractor.
Prerequisites
Python 3.11 or newer.
pip install "langchain>=1.0,<2" langchain-openai
The major is pinned so this page stays true as LangChain moves.
langchain-openai is left unpinned on purpose: install the one that matches your
langchain major.
You need an OpenAI API key. ChatOpenAI reads OPENAI_API_KEY from the
environment, so export it and no code has to touch it:
export OPENAI_API_KEY=...
Step 1 — The queue, and what an invoice has to satisfy before it posts
Start with the rules, not the model. An intake workflow is a gate on money leaving the company, and every check below is a thing that has gone wrong in a real ledger: a document that looks like an invoice but is a statement listing invoices you have already paid, a second copy of something posted last month, and a total that disagrees with its own charge lines.
"""intake.py — the inbound queue and the checks an invoice must pass to post."""
import re
from dataclasses import dataclass
from decimal import Decimal
# Minor units throughout. 400_000 is 4000.00 in the ledger currency.
AUTO_APPROVE_LIMIT = 400_000
KNOWN_SUPPLIERS = {
"Northgate Freight Ltd": "NG-",
"Pellier Tooling SARL": "PT-",
"Halversen Labs": "HL-",
"Tamsin Print Co": "TP-",
}
# Invoice numbers this ledger has already paid out on.
POSTED_LEDGER = frozenset({"INV-4471"})
INBOX = {
"msg-0191": (
"From: billing@northgate-freight.example\n"
"Subject: Invoice INV-4482\n\n"
"Northgate Freight Ltd\n"
"Invoice INV-4482, dated 2026-10-28, PO NG-7741.\n"
"Line haulage, Rotterdam to Leeds, 3 loads at 420.00 = 1260.00\n"
"Waiting time, 2 hours at 55.00 = 110.00\n"
"Total due 1370.00, payable in 30 days."
),
"msg-0192": (
"From: accounts@pellier-tooling.example\n"
"Subject: Invoice INV-4483\n\n"
"Pellier Tooling SARL\n"
"Invoice INV-4483, dated 2026-10-29, PO PT-2210.\n"
"Carbide inserts, 40 boxes at 86.50 = 3460.00\n"
"Tool holders, 6 at 240.00 = 1440.00\n"
"Total due 4900.00, net 45 days."
),
"msg-0193": (
"From: billing@northgate-freight.example\n"
"Subject: Statement of account, October\n\n"
"Northgate Freight Ltd, statement of account.\n"
"Open items: INV-4471 1180.00, INV-4482 1370.00.\n"
"Balance 2550.00. No payment is due on this document."
),
"msg-0194": (
"From: accounts@pellier-tooling.example\n"
"Subject: Credit note CN-0117\n\n"
"Pellier Tooling SARL\n"
"Credit note CN-0117 against INV-4455, dated 2026-10-30.\n"
"Return of 4 tool holders at 240.00 = 960.00 credited to your account."
),
"msg-0195": (
"From: billing@halversen-labs.example\n"
"Subject: Invoice INV-4471, second copy\n\n"
"Halversen Labs\n"
"Invoice INV-4471, dated 2026-10-02, PO HL-0098.\n"
"Calibration service, 1 at 1180.00 = 1180.00\n"
"Total due 1180.00. Our records show this as unpaid."
),
"msg-0196": (
"From: no-reply@courier.example\n"
"Subject: Delivery confirmation\n\n"
"Consignment NG-7741-02 was delivered on 2026-10-27 and signed for.\n"
"This notification is for information only."
),
"msg-0197": (
"From: billing@tamsin-print.example\n"
"Subject: Invoice INV-4490\n\n"
"Tamsin Print Co\n"
"Invoice INV-4490, dated 2026-11-01, PO TP-0042.\n"
"Perfect-bound catalogues, 2000 at 1.15 = 2300.00\n"
"Delivery = 95.00\n"
"Total due 2295.00, 30 days."
),
}
_MONEY = re.compile(r"-?\d{1,3}(?:,\d{3})*(?:\.\d{1,2})?|-?\d+(?:\.\d{1,2})?")
_NORM = re.compile(r"[^a-z0-9]+")
@dataclass
class Invoice:
doc_id: str
supplier: str
number: str
date: str
po: str
lines: list[tuple[str, int]]
total_cents: int
def money(raw: str) -> int | None:
"""Minor units, or None if the text is not an amount this ledger accepts."""
text = raw.strip().replace(" ", "")
if not _MONEY.fullmatch(text):
return None
return int(Decimal(text.replace(",", "")).scaleb(2).to_integral_value())
def fmt(cents: int) -> str:
return f"{Decimal(cents).scaleb(-2):.2f}"
def line_sum(inv: Invoice) -> int:
return sum(amount for _, amount in inv.lines)
def match_supplier(name: str) -> str | None:
key = _NORM.sub("", name.lower())
for known in KNOWN_SUPPLIERS:
if _NORM.sub("", known.lower()) == key:
return known
return None
def checks(inv: Invoice, seen: set[str]) -> list[str]:
"""Everything that has to be true before this invoice may post."""
problems: list[str] = []
supplier = match_supplier(inv.supplier)
if supplier is None:
problems.append(f"supplier {inv.supplier!r} is not on the approved list")
if not inv.number:
problems.append("no invoice number on the document")
elif inv.number in seen:
problems.append(f"invoice number {inv.number} has already been posted")
if not inv.lines:
problems.append("no charge lines were read from the document")
elif line_sum(inv) != inv.total_cents:
problems.append(
f"charge lines sum to {fmt(line_sum(inv))}, "
f"the document states {fmt(inv.total_cents)}"
)
if supplier is not None:
prefix = KNOWN_SUPPLIERS[supplier]
if inv.po.strip().lower() in {"", "none"}:
problems.append("no purchase order reference on the document")
elif not inv.po.upper().startswith(prefix):
problems.append(
f"purchase order {inv.po} does not match {supplier}'s prefix {prefix}"
)
if inv.total_cents > AUTO_APPROVE_LIMIT:
problems.append(
f"total {fmt(inv.total_cents)} is over the auto-approval limit "
f"of {fmt(AUTO_APPROVE_LIMIT)}"
)
return problems
if __name__ == "__main__":
print(f"{len(INBOX)} documents in the queue")
for doc_id, text in INBOX.items():
print(f" {doc_id}: {text.splitlines()[1]}")
Money is Decimal scaled to minor units and compared as integers. Floats are the
classic way to make an intake workflow disagree with a ledger by a cent that
nobody can find, and the comparison line_sum(inv) != inv.total_cents is the
check this whole page exists to protect — it has to be exact.
_MONEY is stricter than it looks, and deliberately so. It accepts 2300.00
and 1,370.00, and it rejects 1370,00. A decimal comma stripped as a thousands
separator turns a reading into one a hundred times too large, silently, in the
field that decides what gets paid. Rejecting it sends the document to a person
instead. This fixture set is single-format; a real queue is not, and the regex is
where you decide what the ledger accepts — not the prompt.
match_supplier is the other half of that idea. The model reads the name off the
document; resolving it against master data happens here, against a list you
control, with punctuation and case normalised away. Asking a model to tell you
whether "Pellier Tooling SARL." is on your approved supplier list is asking it to
have your database in its head.
seen is passed in rather than read from POSTED_LEDGER directly, because two
copies of the same invoice can arrive in the same batch. The supervisor adds each
accepted number to the set as it goes, so the second copy fails against the first
even though neither is in the ledger yet.
Step 2 — Sort the queue before you read any of it carefully
The cheapest call in this workflow is the one that stops a more expensive one from happening. Triage reads one document and returns one word.
# fragment
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
FAST_MODEL = "gpt-4.1-mini"
TRIAGE_CONCURRENCY = 4
KINDS = ("invoice", "credit-note", "statement", "other")
TRIAGE_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You sort inbound accounts-payable mail into exactly one category. "
"Reply with one word and nothing else.\n"
"invoice — asks this company to pay a specific amount it has not "
"paid yet\n"
"credit-note — reduces what this company owes\n"
"statement — lists open or past items and asks for no new payment\n"
"other — anything else, including delivery notes and notifications\n"
"A document that lists invoice numbers and a balance is a "
"statement, never an invoice, however much it looks like one.",
),
("human", "{document}"),
]
)
triage_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=8, temperature=0)
triage_chain = TRIAGE_PROMPT | triage_llm | StrOutputParser()
def _kind(raw: str) -> str:
text = raw.strip().lower().replace(" ", "-")
for kind in KINDS:
if text.startswith(kind):
return kind
return "other"
def triage(docs: list[tuple[str, str]]) -> dict[str, str]:
replies = triage_chain.batch(
[{"document": text} for _, text in docs],
config={"max_concurrency": TRIAGE_CONCURRENCY},
)
return {
doc_id: _kind(reply) for (doc_id, _), reply in zip(docs, replies, strict=True)
}
max_tokens=8 is a cost control and a format control at once. A model given room
to explain its answer will explain its answer, and every token of that is paid
for on a call that needed one word.
An unrecognised reply becomes other, which means the document is not read and
not paid. That is the safe direction for this workflow and the wrong direction
for some others: a dropped invoice is a late payment, a mis-sorted statement is a
double payment. Late is recoverable. The supervisor carries the other list out
with the result so the choice is visible rather than silent.
The statement is the document this step exists for. msg-0193 names two invoice
numbers and a balance, and an extractor handed it will dutifully produce an
invoice for 2550.00 covering two invoices, one of which is already paid. One
cheap call keeps it out of the expensive role entirely — which is the cost shape
of every intake workflow worth building: spend a little on everything to spend a
lot on less.
Step 3 — Copy the fields out, and add them up in Python
The extractor transcribes. It does not compute, correct or decide.
# fragment
READ_MODEL = "gpt-4.1"
EXTRACT_CONCURRENCY = 3
EXTRACT_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You read one supplier invoice and copy out what it says. One "
"field per line, in this exact form, and nothing else:\n"
"supplier :: name exactly as printed\n"
"number :: invoice number\n"
"date :: YYYY-MM-DD\n"
"po :: purchase order reference, or none\n"
"line :: description :: amount\n"
"total :: amount\n"
"One line row per charge, including discounts, surcharges and "
"delivery, in the order they are printed. Amounts are digits and a "
"decimal point, with no currency symbol. Copy what the document "
"states and never compute, correct or round anything — if the "
"charges do not add up to the stated total, copy both as printed. "
"If a field is not printed, write none.\n"
"Recheck note: {note}",
),
("human", "{document}"),
]
)
extractor_llm = ChatOpenAI(model=READ_MODEL, max_tokens=700, temperature=0)
extract_chain = EXTRACT_PROMPT | extractor_llm | StrOutputParser()
_FIELDS = {"supplier", "number", "date", "po", "total"}
def parse_invoice(doc_id: str, raw: str) -> Invoice | None:
fields: dict[str, str] = {}
lines: list[tuple[str, int]] = []
for row in raw.splitlines():
parts = [part.strip() for part in row.split("::")]
if len(parts) == 3 and parts[0].lower() == "line":
amount = money(parts[2])
if amount is not None:
lines.append((parts[1], amount))
elif len(parts) == 2 and parts[0].lower() in _FIELDS:
fields[parts[0].lower()] = parts[1]
total = money(fields.get("total", ""))
if total is None:
return None
return Invoice(
doc_id=doc_id,
supplier=fields.get("supplier", ""),
number=fields.get("number", ""),
date=fields.get("date", ""),
po=fields.get("po", "none"),
lines=lines,
total_cents=total,
)
def extract(docs: list[tuple[str, str]], notes: dict[str, str]) -> dict[str, Invoice]:
replies = extract_chain.batch(
[
{"document": text, "note": notes.get(doc_id, "none")}
for doc_id, text in docs
],
config={"max_concurrency": EXTRACT_CONCURRENCY},
)
out: dict[str, Invoice] = {}
for (doc_id, _), reply in zip(docs, replies, strict=True):
invoice = parse_invoice(doc_id, reply)
if invoice is not None:
out[doc_id] = invoice
return out
"Never compute, correct or round anything" is the single most important sentence
in this file. A model asked to read an invoice whose lines sum to 2395.00 under a
stated total of 2295.00 will usually hand back a tidy document — it will drop the
delivery line, or quietly report 2395.00 as the total — because a consistent
invoice is what invoices look like. Both of those destroy the only evidence that
something is wrong. The prompt asks for a faithful copy precisely so the
disagreement survives into Python, where checks can see it.
An unparseable reply returns None and the document simply does not appear in
the result. The supervisor diffs the documents it sent against the invoices it
got back, so an extraction that failed shows up as a named document waiting on a
person rather than as an absence.
The recheck note is a field in the same prompt rather than a second prompt. The
first pass sends none; a re-read sends the mismatch. Two prompts that differ by
one field are two prompts to keep in step, and they drift.
max_concurrency bounds how many documents are in flight on the expensive model.
It is also, as the budget section gets to, the number that decides how far a cap
can overshoot.
Step 4 — Write the exception note, and pick from a vocabulary
An invoice that fails a check is not a failure; it is work for a person. The adjudicator's job is to make that work short, and to put the item in a state the rest of the system understands.
# fragment
ADJUDICATE_CONCURRENCY = 3
DISPOSITIONS = ("hold", "query-supplier", "route-to-approver")
ADJUDICATE_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"An invoice failed one or more checks before posting. Write the "
"note the accounts-payable team will read, and choose what happens "
"to it. Exactly two lines, in this form:\n"
"disposition :: one of hold, query-supplier, route-to-approver\n"
"note :: one sentence under thirty words naming the check that "
"failed and what has to happen next\n"
"Choose route-to-approver only when the sole problem is that the "
"total is over the auto-approval limit. Choose query-supplier when "
"the document disagrees with itself or with our records and the "
"supplier has to answer. Choose hold for anything else. Never "
"approve an invoice and never recommend paying one.",
),
(
"human",
"Invoice {number} from {supplier}, total {total}.\n"
"Failed checks:\n{problems}",
),
]
)
adjudicator_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=160, temperature=0)
adjudicate_chain = ADJUDICATE_PROMPT | adjudicator_llm | StrOutputParser()
def _ruling(raw: str) -> tuple[str, str]:
disposition, note = "hold", ""
for row in raw.splitlines():
key, _, value = row.partition("::")
key, value = key.strip().lower(), value.strip()
if key == "disposition" and value.lower() in DISPOSITIONS:
disposition = value.lower()
elif key == "note" and value:
note = value
return disposition, note
def adjudicate(cases: list[tuple[Invoice, list[str]]]) -> dict[str, tuple[str, str]]:
replies = adjudicate_chain.batch(
[
{
"number": inv.number or "unknown",
"supplier": inv.supplier or "unknown",
"total": fmt(inv.total_cents),
"problems": "\n".join(f"- {problem}" for problem in problems),
}
for inv, problems in cases
],
config={"max_concurrency": ADJUDICATE_CONCURRENCY},
)
return {
inv.doc_id: _ruling(reply)
for (inv, _), reply in zip(cases, replies, strict=True)
}
The disposition is checked against DISPOSITIONS and defaults to hold. The
model picks from a vocabulary; it does not get to invent a workflow state. A
queue that acquires a status nobody wrote a handler for is a queue items
disappear into, and hold is the one default that costs a delay rather than a
payment.
The adjudicator sees the failed checks, not the document. Its input is four short fields and a list of sentences Python already wrote, which is why it belongs on the cheap model even though it is the role whose output a human reads. The expensive reading already happened; this call only has to say what it means.
Step 5 — Report from the facts, not from the documents
# fragment
REPORT_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You write the end-of-run note for an accounts-payable mailbox. "
"Say how many documents were read, how many invoices are ready to "
"post and for what total, and what is waiting on a person and why. "
"Use only the figures you are given and never calculate a new one. "
"If the run status is not complete, say so in the first sentence "
"and name the documents that were not processed. Under 120 words, "
"plain sentences, no bullet list and no sign-off.",
),
("human", "{facts}"),
]
)
reporter_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=320, temperature=0)
report_chain = REPORT_PROMPT | reporter_llm | StrOutputParser()
def report(facts: str) -> str:
return report_chain.invoke({"facts": facts})
The reporter never sees an invoice. It is given lines the supervisor assembled from values Python already computed, which is the difference between a summary that is true by construction and one that is a second, unchecked reading of the same documents. "Never calculate a new one" is in the prompt for the same reason the extractor is told not to compute: the totals in this note will be read by someone deciding whether the run went well.
Step 6 — The supervisor
One function, and every branch the workflow can take is visible in it.
# fragment
from typing import Any
MAX_RECHECKS = 2
def ruling_for(rulings: dict[str, tuple[str, str]], doc_id: str) -> tuple[str, str]:
return rulings.get(doc_id, ("hold", "no note was returned"))
def run_intake(inbox: dict[str, str], posted: frozenset[str]) -> dict[str, Any]:
docs = list(inbox.items())
kinds = triage(docs)
payable = [(doc_id, text) for doc_id, text in docs if kinds[doc_id] == "invoice"]
invoices = extract(payable, {})
# Python found the disagreement. A mismatch is usually a charge line the
# extractor skipped, so re-read those documents once with the sums named.
mismatched = [
doc_id
for doc_id, inv in invoices.items()
if inv.lines and line_sum(inv) != inv.total_cents
]
rechecked = mismatched[:MAX_RECHECKS]
if rechecked:
notes = {
doc_id: (
f"Your previous reading listed charge lines summing to "
f"{fmt(line_sum(invoices[doc_id]))} under a stated total of "
f"{fmt(invoices[doc_id].total_cents)}. Read every charge line "
f"again, including any discount, surcharge or delivery line, "
f"and copy the stated total exactly as printed."
)
for doc_id in rechecked
}
invoices.update(extract([(d, inbox[d]) for d in rechecked], notes))
seen = set(posted)
ready: list[Invoice] = []
cases: list[tuple[Invoice, list[str]]] = []
for doc_id, inv in invoices.items():
problems = checks(inv, seen)
if problems:
cases.append((inv, problems))
else:
ready.append(inv)
seen.add(inv.number)
rulings = adjudicate(cases) if cases else {}
sorted_as = ", ".join(
f"{kind}={sum(1 for value in kinds.values() if value == kind)}"
for kind in KINDS
)
facts = "\n".join(
[
f"documents read: {len(docs)}",
f"sorted as: {sorted_as}",
f"ready to post: {len(ready)} totalling "
f"{fmt(sum(inv.total_cents for inv in ready))}",
*(
f"ready: {inv.number} {inv.supplier} {fmt(inv.total_cents)}"
for inv in ready
),
*(
f"exception: {inv.number or inv.doc_id} "
f"({ruling_for(rulings, inv.doc_id)[0]}) "
f"{'; '.join(problems)}"
for inv, problems in cases
),
"run status: complete",
]
)
return {
"summary": report(facts),
"ready": [
{"number": inv.number, "supplier": inv.supplier, "total": fmt(inv.total_cents)}
for inv in ready
],
"exceptions": [
{
"doc_id": inv.doc_id,
"number": inv.number,
"disposition": ruling_for(rulings, inv.doc_id)[0],
"note": ruling_for(rulings, inv.doc_id)[1],
"problems": problems,
}
for inv, problems in cases
],
"rechecked": rechecked,
"not_invoices": [doc_id for doc_id, kind in kinds.items() if kind != "invoice"],
"unread": [doc_id for doc_id, _ in payable if doc_id not in invoices],
}
MAX_RECHECKS bounds the obvious loop. A document whose lines never reconcile
can be re-read forever, and each lap is a full call on the expensive model over
the whole document. Two is defensible because the second reading either finds the
line it skipped or it does not; a third reading of the same text at temperature
zero is unlikely to be different, and the failure then belongs to the
adjudicator, which costs one cheap call instead.
not_invoices and unread travel in the result rather than being swallowed. The
first is triage's decisions, which is where a mis-sorted invoice is recoverable.
The second is documents triage sent to the extractor that came back unparseable.
Neither has an error attached to it, and neither will surface anywhere else.
Count the calls: seven triage, up to seven extractions, up to two re-reads, up to seven adjudications, one report. Eight at the floor and twenty-four at the cap. The seven and the one come from the queue and from your file. Everything between is a decision another model made.
The whole file
"""intake.py — an accounts-payable intake workflow: triage, extract, adjudicate, report."""
import re
from dataclasses import dataclass
from decimal import Decimal
from typing import Any
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
FAST_MODEL = "gpt-4.1-mini"
READ_MODEL = "gpt-4.1"
TRIAGE_CONCURRENCY = 4
EXTRACT_CONCURRENCY = 3
ADJUDICATE_CONCURRENCY = 3
MAX_RECHECKS = 2
AUTO_APPROVE_LIMIT = 400_000
KINDS = ("invoice", "credit-note", "statement", "other")
DISPOSITIONS = ("hold", "query-supplier", "route-to-approver")
KNOWN_SUPPLIERS = {
"Northgate Freight Ltd": "NG-",
"Pellier Tooling SARL": "PT-",
"Halversen Labs": "HL-",
"Tamsin Print Co": "TP-",
}
POSTED_LEDGER = frozenset({"INV-4471"})
INBOX = {
"msg-0191": (
"From: billing@northgate-freight.example\n"
"Subject: Invoice INV-4482\n\n"
"Northgate Freight Ltd\n"
"Invoice INV-4482, dated 2026-10-28, PO NG-7741.\n"
"Line haulage, Rotterdam to Leeds, 3 loads at 420.00 = 1260.00\n"
"Waiting time, 2 hours at 55.00 = 110.00\n"
"Total due 1370.00, payable in 30 days."
),
"msg-0192": (
"From: accounts@pellier-tooling.example\n"
"Subject: Invoice INV-4483\n\n"
"Pellier Tooling SARL\n"
"Invoice INV-4483, dated 2026-10-29, PO PT-2210.\n"
"Carbide inserts, 40 boxes at 86.50 = 3460.00\n"
"Tool holders, 6 at 240.00 = 1440.00\n"
"Total due 4900.00, net 45 days."
),
"msg-0193": (
"From: billing@northgate-freight.example\n"
"Subject: Statement of account, October\n\n"
"Northgate Freight Ltd, statement of account.\n"
"Open items: INV-4471 1180.00, INV-4482 1370.00.\n"
"Balance 2550.00. No payment is due on this document."
),
"msg-0194": (
"From: accounts@pellier-tooling.example\n"
"Subject: Credit note CN-0117\n\n"
"Pellier Tooling SARL\n"
"Credit note CN-0117 against INV-4455, dated 2026-10-30.\n"
"Return of 4 tool holders at 240.00 = 960.00 credited to your account."
),
"msg-0195": (
"From: billing@halversen-labs.example\n"
"Subject: Invoice INV-4471, second copy\n\n"
"Halversen Labs\n"
"Invoice INV-4471, dated 2026-10-02, PO HL-0098.\n"
"Calibration service, 1 at 1180.00 = 1180.00\n"
"Total due 1180.00. Our records show this as unpaid."
),
"msg-0196": (
"From: no-reply@courier.example\n"
"Subject: Delivery confirmation\n\n"
"Consignment NG-7741-02 was delivered on 2026-10-27 and signed for.\n"
"This notification is for information only."
),
"msg-0197": (
"From: billing@tamsin-print.example\n"
"Subject: Invoice INV-4490\n\n"
"Tamsin Print Co\n"
"Invoice INV-4490, dated 2026-11-01, PO TP-0042.\n"
"Perfect-bound catalogues, 2000 at 1.15 = 2300.00\n"
"Delivery = 95.00\n"
"Total due 2295.00, 30 days."
),
}
_MONEY = re.compile(r"-?\d{1,3}(?:,\d{3})*(?:\.\d{1,2})?|-?\d+(?:\.\d{1,2})?")
_NORM = re.compile(r"[^a-z0-9]+")
_FIELDS = {"supplier", "number", "date", "po", "total"}
TRIAGE_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You sort inbound accounts-payable mail into exactly one category. "
"Reply with one word and nothing else.\n"
"invoice — asks this company to pay a specific amount it has not "
"paid yet\n"
"credit-note — reduces what this company owes\n"
"statement — lists open or past items and asks for no new payment\n"
"other — anything else, including delivery notes and notifications\n"
"A document that lists invoice numbers and a balance is a "
"statement, never an invoice, however much it looks like one.",
),
("human", "{document}"),
]
)
EXTRACT_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You read one supplier invoice and copy out what it says. One "
"field per line, in this exact form, and nothing else:\n"
"supplier :: name exactly as printed\n"
"number :: invoice number\n"
"date :: YYYY-MM-DD\n"
"po :: purchase order reference, or none\n"
"line :: description :: amount\n"
"total :: amount\n"
"One line row per charge, including discounts, surcharges and "
"delivery, in the order they are printed. Amounts are digits and a "
"decimal point, with no currency symbol. Copy what the document "
"states and never compute, correct or round anything — if the "
"charges do not add up to the stated total, copy both as printed. "
"If a field is not printed, write none.\n"
"Recheck note: {note}",
),
("human", "{document}"),
]
)
ADJUDICATE_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"An invoice failed one or more checks before posting. Write the "
"note the accounts-payable team will read, and choose what happens "
"to it. Exactly two lines, in this form:\n"
"disposition :: one of hold, query-supplier, route-to-approver\n"
"note :: one sentence under thirty words naming the check that "
"failed and what has to happen next\n"
"Choose route-to-approver only when the sole problem is that the "
"total is over the auto-approval limit. Choose query-supplier when "
"the document disagrees with itself or with our records and the "
"supplier has to answer. Choose hold for anything else. Never "
"approve an invoice and never recommend paying one.",
),
(
"human",
"Invoice {number} from {supplier}, total {total}.\n"
"Failed checks:\n{problems}",
),
]
)
REPORT_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You write the end-of-run note for an accounts-payable mailbox. "
"Say how many documents were read, how many invoices are ready to "
"post and for what total, and what is waiting on a person and why. "
"Use only the figures you are given and never calculate a new one. "
"If the run status is not complete, say so in the first sentence "
"and name the documents that were not processed. Under 120 words, "
"plain sentences, no bullet list and no sign-off.",
),
("human", "{facts}"),
]
)
triage_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=8, temperature=0)
extractor_llm = ChatOpenAI(model=READ_MODEL, max_tokens=700, temperature=0)
adjudicator_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=160, temperature=0)
reporter_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=320, temperature=0)
triage_chain = TRIAGE_PROMPT | triage_llm | StrOutputParser()
extract_chain = EXTRACT_PROMPT | extractor_llm | StrOutputParser()
adjudicate_chain = ADJUDICATE_PROMPT | adjudicator_llm | StrOutputParser()
report_chain = REPORT_PROMPT | reporter_llm | StrOutputParser()
@dataclass
class Invoice:
doc_id: str
supplier: str
number: str
date: str
po: str
lines: list[tuple[str, int]]
total_cents: int
def money(raw: str) -> int | None:
text = raw.strip().replace(" ", "")
if not _MONEY.fullmatch(text):
return None
return int(Decimal(text.replace(",", "")).scaleb(2).to_integral_value())
def fmt(cents: int) -> str:
return f"{Decimal(cents).scaleb(-2):.2f}"
def line_sum(inv: Invoice) -> int:
return sum(amount for _, amount in inv.lines)
def match_supplier(name: str) -> str | None:
key = _NORM.sub("", name.lower())
for known in KNOWN_SUPPLIERS:
if _NORM.sub("", known.lower()) == key:
return known
return None
def checks(inv: Invoice, seen: set[str]) -> list[str]:
problems: list[str] = []
supplier = match_supplier(inv.supplier)
if supplier is None:
problems.append(f"supplier {inv.supplier!r} is not on the approved list")
if not inv.number:
problems.append("no invoice number on the document")
elif inv.number in seen:
problems.append(f"invoice number {inv.number} has already been posted")
if not inv.lines:
problems.append("no charge lines were read from the document")
elif line_sum(inv) != inv.total_cents:
problems.append(
f"charge lines sum to {fmt(line_sum(inv))}, "
f"the document states {fmt(inv.total_cents)}"
)
if supplier is not None:
prefix = KNOWN_SUPPLIERS[supplier]
if inv.po.strip().lower() in {"", "none"}:
problems.append("no purchase order reference on the document")
elif not inv.po.upper().startswith(prefix):
problems.append(
f"purchase order {inv.po} does not match {supplier}'s prefix {prefix}"
)
if inv.total_cents > AUTO_APPROVE_LIMIT:
problems.append(
f"total {fmt(inv.total_cents)} is over the auto-approval limit "
f"of {fmt(AUTO_APPROVE_LIMIT)}"
)
return problems
def parse_invoice(doc_id: str, raw: str) -> Invoice | None:
fields: dict[str, str] = {}
lines: list[tuple[str, int]] = []
for row in raw.splitlines():
parts = [part.strip() for part in row.split("::")]
if len(parts) == 3 and parts[0].lower() == "line":
amount = money(parts[2])
if amount is not None:
lines.append((parts[1], amount))
elif len(parts) == 2 and parts[0].lower() in _FIELDS:
fields[parts[0].lower()] = parts[1]
total = money(fields.get("total", ""))
if total is None:
return None
return Invoice(
doc_id=doc_id,
supplier=fields.get("supplier", ""),
number=fields.get("number", ""),
date=fields.get("date", ""),
po=fields.get("po", "none"),
lines=lines,
total_cents=total,
)
def _kind(raw: str) -> str:
text = raw.strip().lower().replace(" ", "-")
for kind in KINDS:
if text.startswith(kind):
return kind
return "other"
def _ruling(raw: str) -> tuple[str, str]:
disposition, note = "hold", ""
for row in raw.splitlines():
key, _, value = row.partition("::")
key, value = key.strip().lower(), value.strip()
if key == "disposition" and value.lower() in DISPOSITIONS:
disposition = value.lower()
elif key == "note" and value:
note = value
return disposition, note
def ruling_for(rulings: dict[str, tuple[str, str]], doc_id: str) -> tuple[str, str]:
return rulings.get(doc_id, ("hold", "no note was returned"))
def triage(docs: list[tuple[str, str]]) -> dict[str, str]:
replies = triage_chain.batch(
[{"document": text} for _, text in docs],
config={"max_concurrency": TRIAGE_CONCURRENCY},
)
return {
doc_id: _kind(reply) for (doc_id, _), reply in zip(docs, replies, strict=True)
}
def extract(docs: list[tuple[str, str]], notes: dict[str, str]) -> dict[str, Invoice]:
replies = extract_chain.batch(
[
{"document": text, "note": notes.get(doc_id, "none")}
for doc_id, text in docs
],
config={"max_concurrency": EXTRACT_CONCURRENCY},
)
out: dict[str, Invoice] = {}
for (doc_id, _), reply in zip(docs, replies, strict=True):
invoice = parse_invoice(doc_id, reply)
if invoice is not None:
out[doc_id] = invoice
return out
def adjudicate(cases: list[tuple[Invoice, list[str]]]) -> dict[str, tuple[str, str]]:
replies = adjudicate_chain.batch(
[
{
"number": inv.number or "unknown",
"supplier": inv.supplier or "unknown",
"total": fmt(inv.total_cents),
"problems": "\n".join(f"- {problem}" for problem in problems),
}
for inv, problems in cases
],
config={"max_concurrency": ADJUDICATE_CONCURRENCY},
)
return {
inv.doc_id: _ruling(reply)
for (inv, _), reply in zip(cases, replies, strict=True)
}
def report(facts: str) -> str:
return report_chain.invoke({"facts": facts})
def run_intake(inbox: dict[str, str], posted: frozenset[str]) -> dict[str, Any]:
docs = list(inbox.items())
kinds = triage(docs)
payable = [(doc_id, text) for doc_id, text in docs if kinds[doc_id] == "invoice"]
invoices = extract(payable, {})
mismatched = [
doc_id
for doc_id, inv in invoices.items()
if inv.lines and line_sum(inv) != inv.total_cents
]
rechecked = mismatched[:MAX_RECHECKS]
if rechecked:
notes = {
doc_id: (
f"Your previous reading listed charge lines summing to "
f"{fmt(line_sum(invoices[doc_id]))} under a stated total of "
f"{fmt(invoices[doc_id].total_cents)}. Read every charge line "
f"again, including any discount, surcharge or delivery line, "
f"and copy the stated total exactly as printed."
)
for doc_id in rechecked
}
invoices.update(extract([(d, inbox[d]) for d in rechecked], notes))
seen = set(posted)
ready: list[Invoice] = []
cases: list[tuple[Invoice, list[str]]] = []
for doc_id, inv in invoices.items():
problems = checks(inv, seen)
if problems:
cases.append((inv, problems))
else:
ready.append(inv)
seen.add(inv.number)
rulings = adjudicate(cases) if cases else {}
sorted_as = ", ".join(
f"{kind}={sum(1 for value in kinds.values() if value == kind)}"
for kind in KINDS
)
facts = "\n".join(
[
f"documents read: {len(docs)}",
f"sorted as: {sorted_as}",
f"ready to post: {len(ready)} totalling "
f"{fmt(sum(inv.total_cents for inv in ready))}",
*(
f"ready: {inv.number} {inv.supplier} {fmt(inv.total_cents)}"
for inv in ready
),
*(
f"exception: {inv.number or inv.doc_id} "
f"({ruling_for(rulings, inv.doc_id)[0]}) "
f"{'; '.join(problems)}"
for inv, problems in cases
),
"run status: complete",
]
)
return {
"summary": report(facts),
"ready": [
{
"number": inv.number,
"supplier": inv.supplier,
"total": fmt(inv.total_cents),
}
for inv in ready
],
"exceptions": [
{
"doc_id": inv.doc_id,
"number": inv.number,
"disposition": ruling_for(rulings, inv.doc_id)[0],
"note": ruling_for(rulings, inv.doc_id)[1],
"problems": problems,
}
for inv, problems in cases
],
"rechecked": rechecked,
"not_invoices": [doc_id for doc_id, kind in kinds.items() if kind != "invoice"],
"unread": [doc_id for doc_id, _ in payable if doc_id not in invoices],
}
if __name__ == "__main__":
result = run_intake(INBOX, POSTED_LEDGER)
print(result["summary"])
print()
for item in result["ready"]:
print(f"post {item['number']} {item['supplier']} {item['total']}")
for item in result["exceptions"]:
label = item["number"] or item["doc_id"]
print(f"{item['disposition']} {label} — {item['note']}")
if result["rechecked"]:
print(f"re-read: {', '.join(result['rechecked'])}")
if result["not_invoices"]:
print(f"not invoices: {', '.join(result['not_invoices'])}")
if result["unread"]:
print(f"could not be read: {', '.join(result['unread'])}")
python intake.py
You get a mailbox note, a list of invoices ready to post, and one line per
exception with the disposition and the reason. The interesting output is the
disposition list: INV-4471 held as a duplicate against the ledger, INV-4483
routed to an approver for being over the limit, and INV-4490 queried because
its charge lines and its stated total disagree by 100.00 — a disagreement that
only exists in the result because the extractor was told to copy rather than
tidy.
The workflow is finished. Nothing below changes what it does.
Seeing what it cost
What you do not have is any idea which of the four roles spent the money. The provider's invoice has one line for the API key, and the eight to twenty-four calls behind it are indistinguishable — same key, same account, three of the four roles on the same model.
pip install capsera
At the top of intake.py, above everything else:
# fragment
import os
import capsera
capsera.init(api_key=os.environ["CAPSERA_API_KEY"], endpoint="https://api.capsera.ai")
That is the capture step, and it is already complete. capsera.init() patches
the provider client that ChatOpenAI constructs underneath, so from that call on
every request the workflow makes is recorded: each triage call inside batch(),
each extraction, each adjudication, the report. The prompts are untouched, the
chains are untouched, batch() is untouched, and no traffic changed route — the
SDK runs inside your process rather than in front of it, so there is no proxy, no
base_url to swap and nothing new in the request path.
One addition for a script: a background thread ships events on an interval, so a process that exits immediately can exit before the queue drains.
# fragment
if __name__ == "__main__":
result = run_intake(INBOX, POSTED_LEDGER)
print(result["summary"])
capsera.shutdown()
capsera.shutdown() flushes pending events and stops the worker. A long-running
service does not need it; a script, a cron job or a notebook cell does, and
omitting it is the usual reason a first run reports nothing. An intake workflow
is almost always the cron-job case.
Run it now and you have a total, a per-model split and token counts per call. You
also have every one of those events attributed to unknown, because nothing has
said which role made it. unknown is a real row, not a dropped one — an
unattributed cost is still a cost — but it is one row where you wanted four.
Name the roles
The calls are already captured. The decorator only says who:
# fragment
@capsera.agent("triage", team="ap-intake", task_type="document-sort")
def triage(docs: list[tuple[str, str]]) -> dict[str, str]:
... # body unchanged
@capsera.agent("extractor", team="ap-intake", task_type="field-extraction")
def extract(docs: list[tuple[str, str]], notes: dict[str, str]) -> dict[str, Invoice]:
... # body unchanged
@capsera.agent("adjudicator", team="ap-intake", task_type="exception-note")
def adjudicate(cases: list[tuple[Invoice, list[str]]]) -> dict[str, tuple[str, str]]:
... # body unchanged
@capsera.agent("reporter", team="ap-intake", task_type="run-summary")
def report(facts: str) -> str:
... # body unchanged
Four added lines, and that is the entire diff. No function body changed, no call site changed, no model construction changed, no argument threaded through the supervisor. Every LLM call made while a decorated function is on the stack is attributed to that agent, including calls made by the functions it calls and by the framework it invokes.
The fan-out keeps its identity. triage, extract and adjudicate all do
their work in .batch(...), which dispatches to worker threads rather than
calling on the thread you were on, and the decorated scope is still what gets
recorded. Capsera holds the attribution stack in a ContextVar and LangChain's
batch() dispatches through a context-copying executor, so the scope the caller
was in is the scope the worker sees — verified empirically on LangChain 1.3 with
capsera 0.3.0. batch() also produces one event per underlying call rather than
one per batch, so seven documents are seven triage events
(coverage table).
The credit there belongs to LangChain, and it is worth knowing where the guarantee
stops. A ContextVar does not cross a thread boundary on its own — it crosses
when the code starting the thread copies the context, which batch() does and a
hand-rolled pool does not. Sort the queue yourself with a bare
ThreadPoolExecutor and the same decorator records unknown instead of
triage, with no error to tell you. Measured on the same versions:
# fragment
import contextvars
from concurrent.futures import ThreadPoolExecutor
# Loses the scope: workers start with a fresh context.
with ThreadPoolExecutor(max_workers=TRIAGE_CONCURRENCY) as pool:
kinds = list(pool.map(sort_one, documents))
# Keeps it: hand each worker a copy of the caller's context.
ctx = contextvars.copy_context()
with ThreadPoolExecutor(max_workers=TRIAGE_CONCURRENCY) as pool:
kinds = list(pool.map(lambda doc: ctx.run(sort_one, doc), documents))
Async needs nothing extra — a ContextVar is snapshotted per task, so roles
awaited concurrently under asyncio.gather keep their own attribution.
Leave the supervisor undecorated. run_intake makes no LLM call of its own;
every call inside it happens while one of the four roles is on the stack, and the
innermost scope is the one recorded. Decorating it would add a row that can never
have spend on it. For a per-run total, use the team field — which is what
team="ap-intake" on all four decorators is for.
Cost the re-reads separately
The recheck pass is the part of this workflow whose cost you cannot see from the
role rows alone. A queue where every document reconciled first time and a queue
where two documents had to be read twice both report as extractor, on the
expensive model, over the whole document text, and the row says nothing about
which. capsera.tag() opens a sub-scope inside a function you have already
decorated, which is exactly this case:
# fragment
import capsera
@capsera.agent("extractor", team="ap-intake", task_type="field-extraction")
def extract(docs: list[tuple[str, str]], notes: dict[str, str]) -> dict[str, Invoice]:
if not notes:
return _extract(docs, notes)
# A recheck is the same role reading the same document again because the
# arithmetic did not agree. Cost the second pass separately from the first.
with capsera.tag("extractor", team="ap-intake", task_type="extract-recheck"):
return _extract(docs, notes)
_extract is the original body, moved down one level and otherwise unchanged.
The recheck keeps the agent name, so a budget scoped to extractor still counts
both passes, and it changes the task type, so the share of extraction spend that
came from documents that did not reconcile is a number you can read instead of a
suspicion. That number is also a document-quality signal: a supplier whose
invoices always need a second reading is a supplier whose template your prompt
does not handle.
Use tag() for a genuine sub-scope like this one. It is the wrong tool for naming
an iteration: tagging each document with its message id would give you one row per
message, none of which appear in tomorrow's queue, and an agent budget cannot
accumulate against a name that lived for one run. The per-invocation axis has its
own fields on the same event — customer_id and cost_center on
@capsera.agent, session_id on capsera.tag — so you can slice by which entity
or which cost centre a batch was processed for without fragmenting the agent
dimension that budgets and cost attribution depend
on.
Put a budget on the role that multiplies
Both of the batched roles scale with the queue, and only one of them is a risk.
Triage runs once per document, so its call count is the queue depth: a number you can read off the mailbox before the run starts, on the cheap model, with one word coming back. Double the queue and you have doubled a small, predictable row.
The extractor's count is not in your file. It is the number of documents triage
decided were invoices, plus the number of those that failed arithmetic, capped at
MAX_RECHECKS. A queue of seven where triage calls one document an invoice is one
expensive call; the same seven where triage calls all of them invoices is seven
plus two. Nothing you wrote decides which, and the extractor is also the role on
the stronger model carrying the full document text in its prompt — so both halves
of the multiplication grow together. That is
multi-agent amplification in one sentence,
and it is why the cap belongs on the extractor rather than on the run.
Create the budget in the dashboard with scope agent and the agent set to
extractor — the same string as the decorator. That string is the whole join
between the two halves of this page: a budget can only act on a boundary the
events carry, and spend attributed to unknown cannot be governed by a
per-agent budget. Pick the amount from a week of
your own queue rather than a figure from a tutorial, and set the action to
block.
Enforcement is on by default (enable_budget_enforcement=True in init()) and
the check runs before the provider call. When the extractor is over its
budget, BudgetExceededError is raised inside your process and the request never
leaves it, so no tokens are spent on it. This is
pre-call enforcement: the check happens where
your code is, not in front of the provider, and a blocked call costs you the check
rather than the completion.
Then decide what a blocked document means. batch() raises on the first failure
and abandons the remaining inputs unless you tell it otherwise, so this is a
decision you make rather than one you inherit:
# fragment
import capsera
def _extract(
docs: list[tuple[str, str]], notes: dict[str, str]
) -> tuple[dict[str, Invoice], list[str]]:
"""Returns the invoices read, and the documents a budget stopped."""
replies = extract_chain.batch(
[
{"document": text, "note": notes.get(doc_id, "none")}
for doc_id, text in docs
],
config={"max_concurrency": EXTRACT_CONCURRENCY},
return_exceptions=True,
)
out: dict[str, Invoice] = {}
deferred: list[str] = []
for (doc_id, _), reply in zip(docs, replies, strict=True):
if isinstance(reply, capsera.BudgetExceededError):
deferred.append(doc_id)
continue
if isinstance(reply, BaseException):
raise RuntimeError(f"reading {doc_id} failed: {reply}")
invoice = parse_invoice(doc_id, reply)
if invoice is not None:
out[doc_id] = invoice
return out, deferred
return_exceptions=True asks batch() to hand failures back as items in the
result list instead of raising on the first one, which is what lets you tell a
budget block from a flaky call and name the documents that went unread. Without
it, one blocked document ends the extraction pass and the rest are never even
attempted.
_extract now returns two values, so the decorated extract wrapper above
returns them too — its annotation becomes
tuple[dict[str, Invoice], list[str]] and its body is otherwise the same two
lines. The decorator and the tag() scope do not change.
Here is the trap, and it is worse in a workflow automation than anywhere else. A run that drops a blocked document looks like a clean run over a smaller queue. A partial answer looks partial; a batch of five processed cleanly looks exactly like a batch of five. Nobody is reading the documents — that is the point of the workflow — so a supplier whose invoice was never opened has no row, no exception, no note, and discovers the problem by not being paid. The deferred ids have to reach the result, and the roles downstream have to be told:
# fragment
from typing import Any
def run_intake(inbox: dict[str, str], posted: frozenset[str]) -> dict[str, Any]:
docs = list(inbox.items())
kinds = triage(docs)
payable = [(doc_id, text) for doc_id, text in docs if kinds[doc_id] == "invoice"]
invoices, deferred = extract(payable, {})
# Deferred documents are not exceptions. Nothing is wrong with them and the
# adjudicator has nothing to say about them; they were never read.
seen = set(posted)
ready, cases = [], []
for doc_id, inv in invoices.items():
problems = checks(inv, seen)
if problems:
cases.append((inv, problems))
else:
ready.append(inv)
seen.add(inv.number)
rulings = adjudicate(cases) if cases else {}
status = "complete" if not deferred else (
f"INCOMPLETE — {len(deferred)} document(s) were not read because the "
f"extraction budget was reached: {', '.join(deferred)}"
)
facts = "\n".join(
[
f"documents read: {len(docs) - len(deferred)} of {len(docs)}",
f"ready to post: {len(ready)}",
f"run status: {status}",
]
)
return {
"summary": report(facts),
# Not a finished run. Do not close the queue without reading `deferred`.
"partial": bool(deferred),
"deferred": deferred,
"ready": [inv.number for inv in ready],
"exceptions": [
{"number": inv.number, "disposition": ruling_for(rulings, inv.doc_id)[0]}
for inv, _ in cases
],
}
The reporter still runs. That is the opposite of what you would do in a pipeline producing one document, where a budget that stopped the work usually means skipping the summary too — and it is right here because the report is one cheap call and it is the only thing that tells a person the run was short. The run that cannot afford to read can still afford to say so.
Deferred documents are kept out of cases on purpose. An adjudicator handed an
invoice that was never read has no failed check to describe and will write a note
about a document it cannot see, at one cheap call each, converting a budget
problem into something that reads like a document problem in the queue a person
works from.
The single-call roles still need somewhere for a refusal to land. report calls
invoke() directly, so there is no return_exceptions to catch a block for it —
if you also put a budget on reporter or on the ap-intake team, it surfaces as
a raise:
# fragment
from typing import Any
import capsera
def run_once(inbox: dict[str, str], posted: frozenset[str]) -> dict[str, Any] | None:
"""Returns None when a budget stopped a role that makes a single call."""
try:
return run_intake(inbox, posted)
except capsera.BudgetExceededError as exc:
print(f"intake stopped before the run could be reported: {exc}")
return None
BudgetExceededError is the only exception the SDK raises deliberately;
everything else fails open. Do not catch it and retry immediately — the budget
will still be exceeded. Leaving the queue for the next scheduled run is usually
the right answer for an intake workflow, because the documents are still in the
mailbox and nothing has been posted; just make sure the thing that reports the
run knows it processed nothing.
Three limits to know before you rely on this as a hard ceiling.
The blast radius is the role, not the run. A budget on extractor stops
extraction calls. Triage is a different agent with no budget on it, and it runs
first, over every document in the queue — so a run that is blocked before it reads
anything has still paid for the whole queue to be sorted. That is cheap and it is
not zero, and it is worth knowing which way round your two batched roles sit. If
what you want capped is the whole workflow, that is a team-scoped budget on
ap-intake, not a side effect of this one.
A parallel pass can overshoot. The check reads budget state computed from
delivered events, and events are delivered in batches on an interval
(flush_interval_ms, 500 ms by default), so a burst of concurrent calls can each
pass a check taken before any of them was recorded. Here the burst is
EXTRACT_CONCURRENCY — three as shipped, and whatever you raise it to when the
queue gets deeper. Set the budget below a figure you cannot exceed, or lower the
concurrency; it is the concurrency that decides the overshoot, not the queue
length.
It fails open, on purpose. If the pre-call check does not complete within
budget_check_timeout (1 second by default), the call is allowed. A network
problem between your service and Capsera should not halt production traffic. Of
the three actions, block and margin downgrade act at runtime; throttle records
the decision but does not delay the call. The full behaviour is in
budgets and enforcement.
What this run actually cost
No dollar figures on this page: this workflow has not been run against a real key, and a number invented here would not be yours anyway — it depends on your queue, how much of it triage calls an invoice, and today's prices. What is worth predicting is the shape of the result.
Four agent rows — triage, extractor, adjudicator, reporter — under one
team, ap-intake, with a cost per run for the batch as
a whole, and the extractor row split by task type into field-extraction and
extract-recheck.
Read each row as call count and cost per call, not as a total. Three things move
on this workflow and each has a different fix. The triage call count moving means
the queue got deeper, which is a business fact rather than a defect. The
extractor's call count moving while the queue stayed the same means triage is
passing more documents through, which is a sorting change that costs the stronger
model every time and is fixed in the triage prompt, not the extraction one. And
the extract-recheck share moving means more documents are failing arithmetic on
the first reading, which is the signal that a supplier changed their invoice
layout.
Expect the extractor row to dominate the cost and the triage row to dominate the call count. That inversion is the whole design: the cheapest call runs most often so the dearest one runs least. If the two rows ever converge, triage has stopped filtering and is just a tax on every document.
The triage row is also the one with the most repetition in it. Every triage call
in a run carries an identical instruction and differs only in the document after
it, so a long, identical prefix sits in front of the whole queue. The cache read
share column is what tells you whether any of it was billed seven times instead of
once; it is a number to read rather than one to predict, because a shared prefix
has to be long enough and identical enough to be worth anything, and putting the
instruction before the document is what earns it. Turning on
enable_prompt_analysis=True in init() puts a figure on how often one
system-prompt hash repeats, which is the evidence for whether that restructuring
is worth doing: prompt caching that never
engaged.
Each event also carries the file that made the call, which points at intake.py
rather than into langchain_core, so a row you did not expect traces back to code
rather than to a model name.
Questions this page answers
How do I get per-agent cost for a LangChain document intake workflow? Call
capsera.init() once, which patches the provider client ChatOpenAI constructs
underneath, so every call the workflow makes is already recorded. Then put
@capsera.agent("<role>") on each role function — triage, the extractor, the
adjudicator, the reporter. The decorator says who made the call; it does not
change whether the call is captured, and no function body, call site or model
construction changes.
Which role in an invoice intake workflow should carry the budget? The extractor. Triage runs once per document, so its call count is the queue depth — a number you can read before the run starts. The extractor runs once per document triage called an invoice, plus once per re-read, so its count is a decision another model made, and it is the role on the stronger model with the whole document in its prompt.
What should a workflow automation do when a budget blocks part of a batch? Carry the blocked document ids out with the result and mark the run incomplete. A workflow that drops a blocked item reports a clean run over a shorter batch, and the supplier whose invoice was never read has no row anywhere. Keep the run report running even when extraction was blocked — the cheap call at the end is the thing that makes the block visible.
Why does triage cost less than extraction when it runs more often? Because the two halves move in opposite directions. Triage sends one document to the cheap model and takes one word back, so its output is near zero and its input is a single piece of mail. The extractor sends the same document to the stronger model and takes a full field list back, and it only runs on the documents triage passed. More calls at the cheap end is what buys fewer calls at the expensive end.
Questions this page answers
- How do I get per-agent cost for a LangChain document intake workflow?
- Call capsera.init() once, which patches the provider client ChatOpenAI constructs underneath, so every call the workflow makes is already recorded. Then put @capsera.agent("<role>") on each role function — triage, the extractor, the adjudicator, the reporter. The decorator says who made the call; it does not change whether the call is captured, and no function body, call site or model construction changes.
- Which role in an invoice intake workflow should carry the budget?
- The extractor. Triage runs once per document, so its call count is the queue depth — a number you can read before the run starts. The extractor runs once per document triage called an invoice, plus once per re-read, so its count is a decision another model made, and it is the role on the stronger model with the whole document in its prompt.
- What should a workflow automation do when a budget blocks part of a batch?
- Carry the blocked document ids out with the result and mark the run incomplete. A workflow that drops a blocked item reports a clean run over a shorter batch, and the supplier whose invoice was never read has no row anywhere. Keep the run report running even when extraction was blocked — the cheap call at the end is the thing that makes the block visible.
- Why does triage cost less than extraction when it runs more often?
- Because the two halves move in opposite directions. Triage sends one document to the cheap model and takes one word back, so its output is near zero and its input is a single piece of mail. The extractor sends the same document to the stronger model and takes a full field list back, and it only runs on the documents triage passed. More calls at the cheap end is what buys fewer calls at the expensive end.
Give every agent an identity, a budget, and hard limits.
One line of code. Anthropic, OpenAI, and Google Gemini.
See pricing