Get 2,500 events tracked for freeSign up now

Capsera
Start free
All tutorials
Syed Raza15 min

Per-agent cost in a LangGraph supervisor loop

A supervisor that delegates requirements to specialists in LangGraph 1.x on GPT-4.1, then four cost rows and a blocking budget on the role that multiplies.

Framework
LangGraph 1.x
Provider
openai
Python
3.11+
Capsera
pip install capsera

This builds a bid desk in LangGraph 1.x against OpenAI: a coordinator reads a board of customer requirements and delegates the open ones to specialists, the specialists answer in parallel from a fixed evidence pack, an auditor flags answers the evidence does not support, and the loop runs again before a writer assembles the response. With the constants this page ships with, one response is between 2 and 14 model calls, and where in that range a run lands is decided by a model rather than by your code. The last third adds Capsera: one init() call to capture every request, four decorator lines to say which role made it, and a blocking budget on the one role whose call count a model decides.

What you will build

desk.py, a single-file LangGraph app with four roles and one loop:

  • coordinator — one call per round. Reads the board and writes assignments: a requirement id and the specialist to send it to. Decides when the desk is done.
  • responder — one call per assignment, run in parallel. Answers one requirement from one slice of the evidence pack, or says it cannot. The multiplier.
  • auditor — one call per round. Checks the answers against the evidence and flags anything unsupported.
  • writer — one call, on the stronger model. Assembles the response and names what is still open.

The graph is a cycle, not a pipeline. The coordinator's successor is a conditional edge that returns one Send per assignment, the branches merge through a reducer, the auditor's edge points back at the coordinator, and the coordinator is what breaks the cycle by assigning nothing. That shape is the whole reason this is a cost problem: a pipeline's call count is the length of the pipeline, and a loop's is whatever the model in the middle of it decides.

Each role is its own function, and the function that only routes is not one of them. That matters later — it is the seam the instrumentation attaches to — and it is the better shape regardless, because a supervisor that inlines four model calls has nothing you can name, budget, or move to a different model.

Written against LangGraph 1.x with langchain-openai, calling the OpenAI API directly rather than through a gateway. Models: gpt-4.1-mini for the coordinator, the responders and the auditor, gpt-4.1 for the writer.

Prerequisites

Python 3.11 or newer.

pip install "langgraph>=1.0,<2" langchain-openai

The major is pinned so this page stays true as LangGraph moves. langchain-openai is left unpinned on purpose: install the one that matches your langgraph 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 — Fix the evidence the desk may quote

Start with the evidence pack, not the agents. A desk with a search tool has an unbounded number of calls in it, because every lookup is a call. A desk handed a fixed pack has a call count you can multiply out before you run anything, and more importantly it has a defensible answer to "where did this claim come from", which is the actual job.

Everything below is invented for this page. It describes a fictional vendor, not Capsera, and it is written the way a real pack is: every entry states a limit next to a capability, because the limits are what an answer gets wrong.

"""desk.py — the evidence pack and the requirement list a bid desk works from."""

EVIDENCE = {
    "sso": (
        "SAML 2.0 single sign-on and SCIM 2.0 user provisioning, on the "
        "Business plan and above. OIDC is not supported."
    ),
    "audit-log": (
        "Immutable audit log of administrative actions, retained 400 days, "
        "exportable as JSON over the API. Data-plane read events are not "
        "logged."
    ),
    "certs": (
        "SOC 2 Type II, report dated 2026-03-31, no exceptions. ISO 27001 is "
        "not held; certification work is scheduled for 2027."
    ),
    "residency": (
        "Tenant data is stored in eu-west-1 or us-east-1, chosen at signup. "
        "The region cannot be changed afterwards without a migration."
    ),
    "uptime": (
        "Contractual uptime commitment is 99.9% per calendar month, measured "
        "on the API endpoint, with service credits. Scheduled maintenance is "
        "excluded."
    ),
    "support": (
        "Support hours are 08:00-20:00 UTC on business days. A 24/7 pager "
        "rotation is available as a paid add-on."
    ),
    "pricing": (
        "Per-seat annual pricing with volume tiers at 50, 250 and 1000 seats. "
        "Multi-year discounts are approved case by case."
    ),
    "onboarding": (
        "Standard onboarding runs six weeks: two integration, two pilot, two "
        "rollout. Historical data import is scoped separately."
    ),
    "integrations": (
        "Prebuilt connectors for Okta, Workday and Snowflake, plus a REST API "
        "and outbound webhooks. There is no native SAP connector."
    ),
}

AREA_BRIEF = {
    "security": "You answer security and compliance requirements.",
    "commercial": "You answer pricing, contract and service-level requirements.",
    "delivery": "You answer rollout, integration and support requirements.",
}

AREA_EVIDENCE = {
    "security": ("sso", "audit-log", "certs", "residency"),
    "commercial": ("pricing", "uptime", "support"),
    "delivery": ("onboarding", "integrations", "support"),
}

REQUIREMENTS = [
    {
        "id": "R1",
        "text": "Describe your single sign-on and user provisioning support, "
                "including which identity protocols you accept.",
    },
    {
        "id": "R2",
        "text": "State your third-party security certifications and the date "
                "of the most recent report.",
    },
    {
        "id": "R3",
        "text": "Confirm where customer data is held and what happens if we "
                "need a different region later.",
    },
    {
        "id": "R4",
        "text": "State your uptime commitment, how it is measured, and the "
                "remedy when it is missed.",
    },
    {
        "id": "R5",
        "text": "Describe your implementation timeline and what you need from "
                "us at each phase.",
    },
    {
        "id": "R6",
        "text": "State your penetration-testing cadence and who performs the "
                "tests.",
    },
]

REQUIREMENT_IDS = {r["id"] for r in REQUIREMENTS}


def requirement_text(rid: str) -> str:
    for requirement in REQUIREMENTS:
        if requirement["id"] == rid:
            return requirement["text"]
    return ""


def evidence_for(area: str) -> str:
    return "\n".join(f"[{key}] {EVIDENCE[key]}" for key in AREA_EVIDENCE[area])


def evidence_all() -> str:
    return "\n".join(f"[{key}] {value}" for key, value in EVIDENCE.items())


if __name__ == "__main__":
    for area in AREA_BRIEF:
        print(area, len(AREA_EVIDENCE[area]), "entries")

AREA_EVIDENCE is the decision that makes the fan-out affordable. A responder gets three or four entries, not the whole pack, so the prompt is a fixed instruction plus a few lines — widening the desk to eighty requirements multiplies a small number instead of a large one. The auditor is the only role that sees everything, and it runs once per round rather than once per requirement.

R6 is in the list on purpose and nothing in the pack settles it. A desk that cannot tell "we do not have this written down" from "the answer is no" is a desk that invents a penetration-testing schedule, and that is a worse failure than an unanswered question on a form.

Step 2 — The state, and where the loop is bounded

Two things in the state are not obvious. answers needs a reducer because branches write to it concurrently — operator.add concatenates the lists instead of the last branch overwriting the rest. And rounds exists because a supervisor loop has no natural end: the coordinator can always find one more thing to improve, and every lap it takes is a coordinator call, an auditor call and up to MAX_ASSIGNMENTS responder calls.

# fragment
import operator
from typing import Annotated, TypedDict

MAX_ROUNDS = 2          # laps of the loop. A real desk raises this to ~3.
MAX_ASSIGNMENTS = 4     # delegations per lap. A real desk raises this to ~12.
ROUND_CONCURRENCY = 4   # responders in flight at once


class Answer(TypedDict):
    id: str
    area: str
    status: str  # "ok" or "insufficient"
    text: str


class Assignment(TypedDict):
    """The payload one branch receives. Not the graph's state."""

    id: str
    area: str


class DeskState(TypedDict):
    answers: Annotated[list[Answer], operator.add]
    flags: list[str]
    assignments: list[Assignment]
    rounds: int
    response: str


def flagged_ids(flags: list[str]) -> set[str]:
    return {flag.split(":", 1)[0].strip() for flag in flags}


def board_text(answers: list[Answer], flags: list[str]) -> str:
    latest = {answer["id"]: answer for answer in answers}
    flagged = flagged_ids(flags)
    lines = []
    for requirement in REQUIREMENTS:
        answer = latest.get(requirement["id"])
        if answer is None:
            mark = "OPEN"
        elif requirement["id"] in flagged:
            mark = "FLAGGED"
        else:
            mark = answer["status"].upper()
        lines.append(f"{requirement['id']} [{mark}] {requirement['text']}")
    return "\n".join(lines)


def answers_text(answers: list[Answer]) -> str:
    latest = {answer["id"]: answer for answer in answers}
    return "\n\n".join(
        f"{a['id']} ({a['area']}, {a['status']}): {a['text']}"
        for a in latest.values()
    )

status on Answer is there before any of this is instrumented. A responder that cannot settle a requirement from its slice of the pack says so, and the difference between "answered" and "could not answer" has to survive into the final document — otherwise the writer treats a refusal as prose and buries it in a paragraph. Counting what did not get answered is part of the job, not part of the monitoring.

Both formatters keep the latest answer per requirement. Round two can re-answer something round one got flagged for, both writes land in answers through the reducer, and last-write-wins is what stops the auditor reading a draft that has already been replaced.

Multiply the first two constants and you have the cost model of this program: at most MAX_ROUNDS + 1 coordinator calls, MAX_ROUNDS × MAX_ASSIGNMENTS responder calls, MAX_ROUNDS auditor calls and one writer call. Fourteen, as shipped. Two, if the coordinator decides on its first look that there is nothing to delegate.

Step 3 — The coordinator, and the delegation it is allowed to make

The coordinator is the supervisor: it is the only role that decides what runs next, and everything it decides costs money. So the constraints live in Python rather than in the prompt. A model asked for "at most four" will occasionally write six, and a model asked not to repeat itself will occasionally assign the same requirement to two specialists — which is two branches, two calls and two answers to the same question.

# fragment
import re

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.types import Send

FAST_MODEL = "gpt-4.1-mini"

COORDINATOR_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You run a bid desk. You are given a board of requirements and the "
            "specialists available. Assign work as one line per requirement: "
            "the requirement id, a space, and one specialist name, copied "
            "exactly. Assign a requirement marked OPEN. Assign one marked "
            "FLAGGED only if the audit note says a better answer is possible "
            "from the evidence. Assign nothing in any other state. Write "
            "NOTHING TO ASSIGN on a single line when the board is complete. At "
            "most {max_assignments} lines, no commentary.",
        ),
        (
            "human",
            "Specialists:\n{specialists}\n\nBoard:\n{board}\n\n"
            "Audit notes:\n{notes}",
        ),
    ]
)

coordinator_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
coordinate_chain = COORDINATOR_PROMPT | coordinator_llm | StrOutputParser()

_ASSIGNMENT = re.compile(r"^\s*(?:[-*•]\s*)?([A-Za-z]+\d+)\s*[:\-]?\s*([A-Za-z]+)\s*$")


def parse_assignments(raw: str) -> list[Assignment]:
    out: list[Assignment] = []
    seen: set[str] = set()
    for line in raw.splitlines():
        match = _ASSIGNMENT.match(line)
        if not match:
            continue
        rid, area = match.group(1).upper(), match.group(2).lower()
        if rid in REQUIREMENT_IDS and area in AREA_BRIEF and rid not in seen:
            seen.add(rid)
            out.append({"id": rid, "area": area})
        if len(out) == MAX_ASSIGNMENTS:
            break
    return out


def coordinate(state: DeskState) -> dict:
    raw = coordinate_chain.invoke(
        {
            "specialists": "\n".join(
                f"{name}: {brief}" for name, brief in AREA_BRIEF.items()
            ),
            "board": board_text(state["answers"], state["flags"]),
            "notes": "\n".join(state["flags"]) or "none",
            "max_assignments": MAX_ASSIGNMENTS,
        }
    )
    return {
        "assignments": parse_assignments(raw),
        "rounds": state["rounds"] + 1,
    }


def dispatch(state: DeskState) -> list[Send] | str:
    if state["rounds"] > MAX_ROUNDS or not state["assignments"]:
        return "write"
    return [Send("respond", assignment) for assignment in state["assignments"]]

Lines rather than JSON. A one-per-line contract degrades into a shorter list when the model ignores you; a malformed JSON object degrades into an exception, and a retry on a parse failure is a call you pay for twice.

Three filters in parse_assignments, and each one is a call that does not happen. rid in REQUIREMENT_IDS drops a requirement the coordinator invented. area in AREA_BRIEF drops a specialist that does not exist — the usual version is a plausible-sounding legal. rid not in seen drops the duplicate. Without them the graph either raises inside a branch or spends a real call answering a question nobody asked.

dispatch is not a node. It is the function a conditional edge calls to decide where to go next, and returning a list of Send objects is how LangGraph is told to run the same node once per item with a different payload. It makes no model call, so it will not appear in the cost report later — which is the correct outcome for the function that does the routing but none of the work.

The round bound is checked in dispatch rather than in the coordinator's prompt, because a bound a model can talk itself out of is not a bound. When rounds passes MAX_ROUNDS the desk writes with whatever it has, and anything the last audit flagged travels to the writer as an open item instead of buying another lap.

Step 4 — The responder

One requirement per call, one slice of the pack per call. Every responder in a round is independent, so Send runs them as separate branches and max_concurrency in the run config bounds how many are in flight.

# fragment
RESPONDER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "{brief} Answer the requirement in at most three sentences using "
            "only the evidence given, and cite the evidence key you relied on "
            "in square brackets. State the limits the evidence states. If the "
            "evidence does not settle the requirement, reply INSUFFICIENT, a "
            "colon, and the one fact you would need. Never state a capability "
            "the evidence does not state.",
        ),
        ("human", "Requirement {id}: {requirement}\n\nEvidence:\n{evidence}"),
    ]
)

responder_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
respond_chain = RESPONDER_PROMPT | responder_llm | StrOutputParser()


def respond(assignment: Assignment) -> dict:
    text = respond_chain.invoke(
        {
            "brief": AREA_BRIEF[assignment["area"]],
            "id": assignment["id"],
            "requirement": requirement_text(assignment["id"]),
            "evidence": evidence_for(assignment["area"]),
        }
    ).strip()
    status = "insufficient" if text.upper().startswith("INSUFFICIENT") else "ok"
    return {
        "answers": [
            {
                "id": assignment["id"],
                "area": assignment["area"],
                "status": status,
                "text": text,
            }
        ]
    }

One function, three specialists. The difference between a security responder and a commercial one is a line of brief and a slice of evidence, both of them data, and writing three near-identical functions to hold two strings buys nothing. It is worth being deliberate about, though, because it decides what your cost report looks like: one function is one agent row for all the answering the desk does, and three functions would be three rows. Split them when you would set different budgets or different models for them, not because the org chart has three specialists in it. The next section shows how to get the per-specialism breakdown without the split.

max_tokens=256 is load-bearing rather than tidiness. Output tokens are the expensive half of a call, this is the role that runs once per assignment, and three sentences fit easily. A responder allowed to write a page is the same desk at several times the price, and it hands the auditor and the writer longer prompts too.

Step 5 — The auditor and the writer

The auditor is the only role that sees the whole pack and every answer at once, which makes it the one prompt that grows with the size of the desk. One line per flag is what keeps that survivable.

# fragment
WRITE_MODEL = "gpt-4.1"

AUDITOR_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You check draft answers against the evidence pack. Write one line "
            "for each answer that is wrong: the requirement id, a colon, and "
            "at most twenty words naming what is unsupported. Flag an answer "
            "only if it states something the evidence does not, or drops a "
            "limit the evidence states. Do not flag an answer for style, "
            "length or tone. Write NONE on a single line if every answer is "
            "supported.",
        ),
        ("human", "Answers:\n{answers}\n\nEvidence:\n{evidence}"),
    ]
)

WRITER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You assemble a response to a customer requirement list. One "
            "section per requirement, in the order given, headed by the "
            "requirement id. Use the drafted answers as written in substance "
            "and add no capability that is not in them. Where a requirement "
            "has no answer, where its answer is not marked ok, or where the "
            "audit notes name it, write one line saying what the desk must "
            "still confirm. End with a list of the open items.",
        ),
        (
            "human",
            "Requirements:\n{board}\n\nDrafted answers:\n{answers}\n\n"
            "Audit notes:\n{notes}",
        ),
    ]
)

auditor_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=384, temperature=0)
writer_llm = ChatOpenAI(model=WRITE_MODEL, max_tokens=1536, temperature=0)

audit_chain = AUDITOR_PROMPT | auditor_llm | StrOutputParser()
write_chain = WRITER_PROMPT | writer_llm | StrOutputParser()


def parse_flags(raw: str) -> list[str]:
    flags: list[str] = []
    for line in raw.splitlines():
        head, _, rest = line.partition(":")
        rid = head.strip().upper()
        if rid in REQUIREMENT_IDS:
            flags.append(f"{rid}: {rest.strip()}")
    return flags


def audit(state: DeskState) -> dict:
    raw = audit_chain.invoke(
        {
            "answers": answers_text(state["answers"]),
            "evidence": evidence_all(),
        }
    )
    return {"flags": parse_flags(raw)}


def write(state: DeskState) -> dict:
    return {
        "response": write_chain.invoke(
            {
                "board": board_text(state["answers"], state["flags"]),
                "answers": answers_text(state["answers"]) or "(none)",
                "notes": "\n".join(state["flags"]) or "none",
            }
        )
    }

parse_flags normalises the id before storing it, because the coordinator reads these notes back and a flag recorded as r3 matches no requirement in board_text. A flag that silently fails to attach is worse than no auditor: the board says the answer is fine, and the round you paid for changed nothing.

The auditor is one call for the whole round, not one per answer. That is a choice. A per-answer auditor would be a second fan-out, a second thing to bound and a second thing to budget, in exchange for checks that never see each other — and half of what an auditor catches on a requirement list is two answers contradicting one another.

The writer is the only role on the stronger model, because it is the only one whose output a person sends to a customer. The other three produce input for another model. Whether that split is right is a judgement to revisit with real numbers rather than settle by guessing now.

Step 6 — Wire the loop

# fragment
from langgraph.graph import END, START, StateGraph

builder = StateGraph(DeskState)
builder.add_node("coordinate", coordinate)
builder.add_node("respond", respond)
builder.add_node("audit", audit)
builder.add_node("write", write)

builder.add_edge(START, "coordinate")
builder.add_conditional_edges("coordinate", dispatch, ["respond", "write"])
builder.add_edge("respond", "audit")
builder.add_edge("audit", "coordinate")
builder.add_edge("write", END)

desk = builder.compile()


def run_desk() -> dict:
    state = desk.invoke(
        {"answers": [], "flags": [], "rounds": 0},
        config={"max_concurrency": ROUND_CONCURRENCY},
    )
    answered = {answer["id"] for answer in state["answers"]}
    unresolved = {a["id"] for a in state["answers"] if a["status"] != "ok"}
    unresolved |= flagged_ids(state["flags"])
    unresolved |= REQUIREMENT_IDS - answered
    return {"response": state["response"], "open": sorted(unresolved)}

add_conditional_edges("coordinate", dispatch, ["respond", "write"]) is both the fan-out and the exit. The third argument tells the graph which nodes the edge can reach, so both targets are registered even though no plain edge points at either. respond then has one ordinary edge to audit, and the reducer on answers is what makes that a join: the branches merge before the auditor runs. audit points back at coordinate, which closes the cycle.

The graph has its own recursion limit as a backstop against a cycle that never terminates, but a backstop stops the run by raising rather than by producing a response — a desk that hits it has spent every call and has no document. The bound you actually want is rounds in the state, because it is checked before the fan-out and it exits through the writer.

run_desk computes the open items in Python rather than reading them out of the writer's prose. Three different things make a requirement open — never delegated, answered insufficient, or flagged by the auditor — and a caller that needs to route this to a human wants a list of ids, not a paragraph it has to parse.

The whole file

"""desk.py — a bid desk: coordinate, delegate, audit, write."""
import operator
import re
from typing import Annotated, TypedDict

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

FAST_MODEL = "gpt-4.1-mini"
WRITE_MODEL = "gpt-4.1"

MAX_ROUNDS = 2          # laps of the loop. A real desk raises this to ~3.
MAX_ASSIGNMENTS = 4     # delegations per lap. A real desk raises this to ~12.
ROUND_CONCURRENCY = 4   # responders in flight at once

EVIDENCE = {
    "sso": (
        "SAML 2.0 single sign-on and SCIM 2.0 user provisioning, on the "
        "Business plan and above. OIDC is not supported."
    ),
    "audit-log": (
        "Immutable audit log of administrative actions, retained 400 days, "
        "exportable as JSON over the API. Data-plane read events are not "
        "logged."
    ),
    "certs": (
        "SOC 2 Type II, report dated 2026-03-31, no exceptions. ISO 27001 is "
        "not held; certification work is scheduled for 2027."
    ),
    "residency": (
        "Tenant data is stored in eu-west-1 or us-east-1, chosen at signup. "
        "The region cannot be changed afterwards without a migration."
    ),
    "uptime": (
        "Contractual uptime commitment is 99.9% per calendar month, measured "
        "on the API endpoint, with service credits. Scheduled maintenance is "
        "excluded."
    ),
    "support": (
        "Support hours are 08:00-20:00 UTC on business days. A 24/7 pager "
        "rotation is available as a paid add-on."
    ),
    "pricing": (
        "Per-seat annual pricing with volume tiers at 50, 250 and 1000 seats. "
        "Multi-year discounts are approved case by case."
    ),
    "onboarding": (
        "Standard onboarding runs six weeks: two integration, two pilot, two "
        "rollout. Historical data import is scoped separately."
    ),
    "integrations": (
        "Prebuilt connectors for Okta, Workday and Snowflake, plus a REST API "
        "and outbound webhooks. There is no native SAP connector."
    ),
}

AREA_BRIEF = {
    "security": "You answer security and compliance requirements.",
    "commercial": "You answer pricing, contract and service-level requirements.",
    "delivery": "You answer rollout, integration and support requirements.",
}

AREA_EVIDENCE = {
    "security": ("sso", "audit-log", "certs", "residency"),
    "commercial": ("pricing", "uptime", "support"),
    "delivery": ("onboarding", "integrations", "support"),
}

REQUIREMENTS = [
    {
        "id": "R1",
        "text": "Describe your single sign-on and user provisioning support, "
                "including which identity protocols you accept.",
    },
    {
        "id": "R2",
        "text": "State your third-party security certifications and the date "
                "of the most recent report.",
    },
    {
        "id": "R3",
        "text": "Confirm where customer data is held and what happens if we "
                "need a different region later.",
    },
    {
        "id": "R4",
        "text": "State your uptime commitment, how it is measured, and the "
                "remedy when it is missed.",
    },
    {
        "id": "R5",
        "text": "Describe your implementation timeline and what you need from "
                "us at each phase.",
    },
    {
        "id": "R6",
        "text": "State your penetration-testing cadence and who performs the "
                "tests.",
    },
]

REQUIREMENT_IDS = {r["id"] for r in REQUIREMENTS}

_ASSIGNMENT = re.compile(r"^\s*(?:[-*•]\s*)?([A-Za-z]+\d+)\s*[:\-]?\s*([A-Za-z]+)\s*$")


class Answer(TypedDict):
    id: str
    area: str
    status: str
    text: str


class Assignment(TypedDict):
    id: str
    area: str


class DeskState(TypedDict):
    answers: Annotated[list[Answer], operator.add]
    flags: list[str]
    assignments: list[Assignment]
    rounds: int
    response: str


COORDINATOR_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You run a bid desk. You are given a board of requirements and the "
            "specialists available. Assign work as one line per requirement: "
            "the requirement id, a space, and one specialist name, copied "
            "exactly. Assign a requirement marked OPEN. Assign one marked "
            "FLAGGED only if the audit note says a better answer is possible "
            "from the evidence. Assign nothing in any other state. Write "
            "NOTHING TO ASSIGN on a single line when the board is complete. At "
            "most {max_assignments} lines, no commentary.",
        ),
        (
            "human",
            "Specialists:\n{specialists}\n\nBoard:\n{board}\n\n"
            "Audit notes:\n{notes}",
        ),
    ]
)

RESPONDER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "{brief} Answer the requirement in at most three sentences using "
            "only the evidence given, and cite the evidence key you relied on "
            "in square brackets. State the limits the evidence states. If the "
            "evidence does not settle the requirement, reply INSUFFICIENT, a "
            "colon, and the one fact you would need. Never state a capability "
            "the evidence does not state.",
        ),
        ("human", "Requirement {id}: {requirement}\n\nEvidence:\n{evidence}"),
    ]
)

AUDITOR_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You check draft answers against the evidence pack. Write one line "
            "for each answer that is wrong: the requirement id, a colon, and "
            "at most twenty words naming what is unsupported. Flag an answer "
            "only if it states something the evidence does not, or drops a "
            "limit the evidence states. Do not flag an answer for style, "
            "length or tone. Write NONE on a single line if every answer is "
            "supported.",
        ),
        ("human", "Answers:\n{answers}\n\nEvidence:\n{evidence}"),
    ]
)

WRITER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You assemble a response to a customer requirement list. One "
            "section per requirement, in the order given, headed by the "
            "requirement id. Use the drafted answers as written in substance "
            "and add no capability that is not in them. Where a requirement "
            "has no answer, where its answer is not marked ok, or where the "
            "audit notes name it, write one line saying what the desk must "
            "still confirm. End with a list of the open items.",
        ),
        (
            "human",
            "Requirements:\n{board}\n\nDrafted answers:\n{answers}\n\n"
            "Audit notes:\n{notes}",
        ),
    ]
)

coordinator_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
responder_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
auditor_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=384, temperature=0)
writer_llm = ChatOpenAI(model=WRITE_MODEL, max_tokens=1536, temperature=0)

coordinate_chain = COORDINATOR_PROMPT | coordinator_llm | StrOutputParser()
respond_chain = RESPONDER_PROMPT | responder_llm | StrOutputParser()
audit_chain = AUDITOR_PROMPT | auditor_llm | StrOutputParser()
write_chain = WRITER_PROMPT | writer_llm | StrOutputParser()


def requirement_text(rid: str) -> str:
    for requirement in REQUIREMENTS:
        if requirement["id"] == rid:
            return requirement["text"]
    return ""


def evidence_for(area: str) -> str:
    return "\n".join(f"[{key}] {EVIDENCE[key]}" for key in AREA_EVIDENCE[area])


def evidence_all() -> str:
    return "\n".join(f"[{key}] {value}" for key, value in EVIDENCE.items())


def flagged_ids(flags: list[str]) -> set[str]:
    return {flag.split(":", 1)[0].strip() for flag in flags}


def board_text(answers: list[Answer], flags: list[str]) -> str:
    latest = {answer["id"]: answer for answer in answers}
    flagged = flagged_ids(flags)
    lines = []
    for requirement in REQUIREMENTS:
        answer = latest.get(requirement["id"])
        if answer is None:
            mark = "OPEN"
        elif requirement["id"] in flagged:
            mark = "FLAGGED"
        else:
            mark = answer["status"].upper()
        lines.append(f"{requirement['id']} [{mark}] {requirement['text']}")
    return "\n".join(lines)


def answers_text(answers: list[Answer]) -> str:
    latest = {answer["id"]: answer for answer in answers}
    return "\n\n".join(
        f"{a['id']} ({a['area']}, {a['status']}): {a['text']}"
        for a in latest.values()
    )


def parse_assignments(raw: str) -> list[Assignment]:
    out: list[Assignment] = []
    seen: set[str] = set()
    for line in raw.splitlines():
        match = _ASSIGNMENT.match(line)
        if not match:
            continue
        rid, area = match.group(1).upper(), match.group(2).lower()
        if rid in REQUIREMENT_IDS and area in AREA_BRIEF and rid not in seen:
            seen.add(rid)
            out.append({"id": rid, "area": area})
        if len(out) == MAX_ASSIGNMENTS:
            break
    return out


def parse_flags(raw: str) -> list[str]:
    flags: list[str] = []
    for line in raw.splitlines():
        head, _, rest = line.partition(":")
        rid = head.strip().upper()
        if rid in REQUIREMENT_IDS:
            flags.append(f"{rid}: {rest.strip()}")
    return flags


def coordinate(state: DeskState) -> dict:
    raw = coordinate_chain.invoke(
        {
            "specialists": "\n".join(
                f"{name}: {brief}" for name, brief in AREA_BRIEF.items()
            ),
            "board": board_text(state["answers"], state["flags"]),
            "notes": "\n".join(state["flags"]) or "none",
            "max_assignments": MAX_ASSIGNMENTS,
        }
    )
    return {
        "assignments": parse_assignments(raw),
        "rounds": state["rounds"] + 1,
    }


def dispatch(state: DeskState) -> list[Send] | str:
    if state["rounds"] > MAX_ROUNDS or not state["assignments"]:
        return "write"
    return [Send("respond", assignment) for assignment in state["assignments"]]


def respond(assignment: Assignment) -> dict:
    text = respond_chain.invoke(
        {
            "brief": AREA_BRIEF[assignment["area"]],
            "id": assignment["id"],
            "requirement": requirement_text(assignment["id"]),
            "evidence": evidence_for(assignment["area"]),
        }
    ).strip()
    status = "insufficient" if text.upper().startswith("INSUFFICIENT") else "ok"
    return {
        "answers": [
            {
                "id": assignment["id"],
                "area": assignment["area"],
                "status": status,
                "text": text,
            }
        ]
    }


def audit(state: DeskState) -> dict:
    raw = audit_chain.invoke(
        {
            "answers": answers_text(state["answers"]),
            "evidence": evidence_all(),
        }
    )
    return {"flags": parse_flags(raw)}


def write(state: DeskState) -> dict:
    return {
        "response": write_chain.invoke(
            {
                "board": board_text(state["answers"], state["flags"]),
                "answers": answers_text(state["answers"]) or "(none)",
                "notes": "\n".join(state["flags"]) or "none",
            }
        )
    }


builder = StateGraph(DeskState)
builder.add_node("coordinate", coordinate)
builder.add_node("respond", respond)
builder.add_node("audit", audit)
builder.add_node("write", write)

builder.add_edge(START, "coordinate")
builder.add_conditional_edges("coordinate", dispatch, ["respond", "write"])
builder.add_edge("respond", "audit")
builder.add_edge("audit", "coordinate")
builder.add_edge("write", END)

desk = builder.compile()


def run_desk() -> dict:
    state = desk.invoke(
        {"answers": [], "flags": [], "rounds": 0},
        config={"max_concurrency": ROUND_CONCURRENCY},
    )
    answered = {answer["id"] for answer in state["answers"]}
    unresolved = {a["id"] for a in state["answers"] if a["status"] != "ok"}
    unresolved |= flagged_ids(state["flags"])
    unresolved |= REQUIREMENT_IDS - answered
    return {"response": state["response"], "open": sorted(unresolved)}


if __name__ == "__main__":
    result = run_desk()
    print(result["response"])
    if result["open"]:
        print("\nopen items:", ", ".join(result["open"]))

python desk.py

You get a response document with one section per requirement and a list of open items at the end. R6 should be in that list every time — nothing in the pack gives a penetration-testing cadence — and whether R2 joins it is the interesting part: the pack states a SOC 2 report and states that ISO 27001 is not held, so an answer that mentions only the first is exactly the kind of omission the auditor exists to flag.

The application 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 calls behind it are indistinguishable — same key, same account, three of the four roles on the same model. Nor can you tell a two-round run from a one-round run, which is the difference that matters, because the second round is a coordinator call, an auditor call and up to four responder calls that the first round decided to buy.

pip install capsera

At the top of desk.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 program makes is recorded: the coordinator's, each responder branch's, the auditor's, the writer's. The prompts are untouched, the chains are untouched, the graph 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_desk()
    print(result["response"])
    if result["open"]:
        print("\nopen items:", ", ".join(result["open"]))
    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.

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 decorators only say who:

# fragment
@capsera.langgraph_node("coordinator", team="bid-desk", task_type="delegation")
def coordinate(state: DeskState) -> dict:
    ...  # body unchanged


@capsera.langgraph_node("responder", team="bid-desk", task_type="requirement-answer")
def respond(assignment: Assignment) -> dict:
    ...  # body unchanged


@capsera.langgraph_node("auditor", team="bid-desk", task_type="answer-check")
def audit(state: DeskState) -> dict:
    ...  # body unchanged


@capsera.langgraph_node("writer", team="bid-desk", task_type="assembly")
def write(state: DeskState) -> dict:
    ...  # body unchanged

Four added lines, and that is the entire diff. No node body changed, no prompt changed, no edge changed, no model construction changed, no state field added, no argument threaded through the graph. Every LLM call made while a decorated function is on the stack is attributed to that name, including calls made by the functions it calls and by the framework it invokes.

langgraph_node is @capsera.agent plus a record of the framework, so use it on graph nodes and agent on plain functions. Every role here is a node, so every role here takes langgraph_node (coverage table).

dispatch is deliberately not decorated. It is the busiest decision in the program and it makes no model call, so a decorator on it would add a fifth row that can never have spend on it. Decorate the function that makes the call. For a desk-level total, use the team field — which is what team="bid-desk" on all four decorators is for.

The fan-out keeps its identity. The decorator is on respond, and LangGraph opens each branch and runs the node inside it, so the scope is entered wherever the branch runs and the model call happens on the same stack a moment later. Nothing has to cross a thread boundary for that to hold. This program starts no threads of its own; if you add a pool inside a node, the rule changes and the LangGraph coverage note has the case. Async needs nothing extra — Capsera keeps the attribution stack in a ContextVar, which is snapshotted per task, so branches awaited concurrently under asyncio.gather keep their own attribution.

Split the responder by specialism, not by invocation

One responder function is one row, and one row hides the thing you most want to know about a desk: whether the security answers cost three times what the commercial ones do. capsera.tag() opens a sub-scope inside a function you have already decorated, which is exactly this case:

# fragment
import capsera


@capsera.langgraph_node("responder", team="bid-desk", task_type="requirement-answer")
def respond(assignment: Assignment) -> dict:
    # assignment["area"] is a key of AREA_BRIEF. parse_assignments dropped
    # anything else before this branch existed.
    with capsera.tag("responder", team="bid-desk", task_type=assignment["area"]):
        return _respond(assignment)

_respond is the original body, moved down one level and otherwise unchanged. The agent name stays responder, so a budget scoped to it still counts every answer whatever the specialism, and the task type splits the row three ways so the per-specialism cost is a number you can read instead of a suspicion.

The comment above the tag is the load-bearing part. Tag with a value from a set you control in code. assignment["area"] came out of a model, but parse_assignments checked it against AREA_BRIEF before it ever reached a branch, so the tag can only ever be one of three strings. Tag with assignment["id"] instead and you get one row per requirement, none of which exists on the next bid, and an agent budget cannot accumulate against a name that lived for one document. The per-invocation axis has its own fields on the same event — customer_id and cost_center on @capsera.langgraph_node, session_id on capsera.tag — so you can slice by which bid or which customer a run was for without fragmenting the agent dimension that budgets and cost attribution depend on.

Put a budget on the role that multiplies

The coordinator makes one call per round and the auditor makes one call per round; MAX_ROUNDS is in your file. The writer makes one call per run. The responder's count is the one no line of your code sets: it is however many assignments a model chose to write, up to the cap, on however many rounds the same model chose to take. Point this desk at a real requirement list, raise MAX_ASSIGNMENTS to the width that gets through eighty requirements, and the responder is the only row that moves. That is multi-agent amplification in one sentence, and it is why the cap belongs on the responder rather than on the run.

Create the budget in the dashboard with scope agent and the agent set to responder — 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 responder spend 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 responder 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, and the cost of it is one round trip to Capsera per call.

Then decide what a blocked answer means for the document. An exception that leaves a node surfaces from invoke(), so this is a decision you make rather than one you inherit — let it out and the run ends, and the other answers in that round do not come back to you either:

# fragment
import capsera


@capsera.langgraph_node("responder", team="bid-desk", task_type="requirement-answer")
def respond(assignment: Assignment) -> dict:
    try:
        with capsera.tag("responder", team="bid-desk", task_type=assignment["area"]):
            return _respond(assignment)
    except capsera.BudgetExceededError as exc:
        # Keep the branch alive so the round's other answers survive, and mark
        # the requirement unanswered so the writer and run_desk both see it.
        return {
            "answers": [
                {
                    "id": assignment["id"],
                    "area": assignment["area"],
                    "status": "insufficient",
                    "text": f"INSUFFICIENT: not answered, responder budget: {exc}",
                }
            ]
        }

Catching it in the node is the whole difference between a desk that returns a document with three open items and a desk that returns a traceback. The marker uses the insufficient status the app already has, so nothing downstream needed editing: board_text shows the requirement as unresolved, the coordinator is told not to reassign it, the writer's existing instruction writes the "still to confirm" line for it, and run_desk puts the id in open.

That is the opposite call from the one a code-fixing agent should make, and the reason is what the output is for. A patch built from half the evidence is wrong and costs a human review cycle to discover; a response document with its open items named is exactly what a bid desk hands to a person anyway. Degrading is right here only because the degradation is visible in the output. If your writer would smooth a missing answer into a confident paragraph, catching the error is worse than raising it.

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 responder stops responder calls. The coordinator, the auditor and the writer are different agents with no budget on them, so a run that is fully blocked at the responder still pays for three coordinator calls, two auditor calls and one writer call on the stronger model — and produces a document with nothing in it. If what you want is for the desk to give up rather than assemble an empty response, that is a check on open in run_desk, not a side effect of the budget.

A parallel round 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 ROUND_CONCURRENCY — four as shipped, and whatever you raise it to when the requirement list gets longer. Set the budget below a figure you cannot exceed, or lower the concurrency; it is the concurrency that decides the overshoot, not MAX_ASSIGNMENTS.

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 desk has not been run against a real key, and a number invented here would not be yours anyway — it depends on your requirement list, how many rounds your coordinator takes, and today's prices. What is worth predicting is the shape of the result.

Four agent rows — coordinator, responder, auditor, writer — under one team, bid-desk, with a cost per run for the response as a whole, and the responder row split by task type into security, commercial and delivery.

Read each row as call count and cost per call, not as a total. In a supervisor loop the call count is the more informative half, because it is the part a model chose. A coordinator row with three calls on it instead of two is a desk that took an extra lap, and that lap dragged an auditor call and up to four responder calls along with it — which means the coordinator's own row moves least when the run gets expensive, and is still the row that tells you why.

Which row is largest is not predictable in advance, and that is the point of measuring it rather than reasoning about it. The writer makes one call on the more expensive model with every answer in the prompt and a whole document in the response; the responders make several small ones on the cheaper model. Where the crossover falls depends on how many requirements you put on the board, and it decides whether the next thing to change is the writer's model or the width of the delegation.

The three responder task types are worth comparing directly, because their input sizes are fixed by AREA_EVIDENCE and their output sizes are not. A specialism whose cost per call runs well above the others is telling you either that its slice of the pack is too big or that its answers are running long, and those have different fixes — move an entry out of the slice, or tighten the brief.

Repetition is the other thing to watch. Every responder call in a round carries the same brief with a different requirement, the auditor sends the whole pack once per round, and a second round re-sends answers the first round already sent. The cache read share column is what tells you whether any of that was billed twice, and it is a number to read rather than one to predict: a shared prefix has to be long and identical to be worth anything, and reordering a prompt so the repeated part comes first is the change that 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 and line that made the call, which points at desk.py rather than into langgraph, 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 out of a LangGraph supervisor architecture? Call capsera.init() once, which patches the provider client ChatOpenAI constructs underneath, so every call the graph makes is already recorded. Then put @capsera.langgraph_node("<name>") on each node that makes a call — the coordinator, the responder, the auditor, the writer. The decorator says who made the call; it does not change whether the call is captured, and no node body, prompt, edge or model construction changes.

Which agent should carry the budget in a supervisor-and-specialists system? The delegate, not the supervisor. The supervisor makes one call per round and the round count is a constant in your file; the delegate makes one call per assignment, and the assignment count is written by a model. Put the budget on the role whose call count you did not choose, and remember that capping it caps that role only — the roles downstream of it still run and still spend.

How do I split one agent's cost by specialism without creating a row per invocation? Use capsera.tag() inside the decorated function with a value from a set you control in code, never with free text from a model. Three specialisms give one responder row split three ways, which stays readable and stays budgetable because the agent name is unchanged. A tag built from a requirement id or a customer name gives you one row per invocation, none of which exists on the next run, and an agent budget cannot accumulate against a name that lived for one bid.

What happens to a LangGraph fan-out when a budget blocks one branch? That is your decision, and you make it inside the node. An exception that leaves a node surfaces from invoke(), so letting BudgetExceededError out ends the run and the round's other answers do not come back to you. Catching it in the node and returning a marked-unanswered result keeps the rest of the round and hands the writer a response with its open items named, which for a document assembled from evidence is the better failure.

Questions this page answers

How do I get per-agent cost out of a LangGraph supervisor architecture?
Call capsera.init() once, which patches the provider client ChatOpenAI constructs underneath, so every call the graph makes is already recorded. Then put @capsera.langgraph_node("<name>") on each node that makes a call — the coordinator, the responder, the auditor, the writer. The decorator says who made the call; it does not change whether the call is captured, and no node body, prompt, edge or model construction changes.
Which agent should carry the budget in a supervisor-and-specialists system?
The delegate, not the supervisor. The supervisor makes one call per round and the round count is a constant in your file; the delegate makes one call per assignment, and the assignment count is written by a model. Put the budget on the role whose call count you did not choose, and remember that capping it caps that role only — the roles downstream of it still run and still spend.
How do I split one agent's cost by specialism without creating a row per invocation?
Use capsera.tag() inside the decorated function with a value from a set you control in code, never with free text from a model. Three specialisms give one responder row split three ways, which stays readable and stays budgetable because the agent name is unchanged. A tag built from a requirement id or a customer name gives you one row per invocation, none of which exists on the next run, and an agent budget cannot accumulate against a name that lived for one bid.
What happens to a LangGraph fan-out when a budget blocks one branch?
That is your decision, and you make it inside the node. An exception that leaves a node surfaces from invoke(), so letting BudgetExceededError out ends the run and the round's other answers do not come back to you. Catching it in the node and returning a marked-unanswered result keeps the rest of the round and hands the writer a response with its open items named, which for a document assembled from evidence is the better failure.

Give every agent an identity, a budget, and hard limits.

One line of code. Anthropic, OpenAI, and Google Gemini.

See pricing