Get 2,500 events tracked for freeSign up now

Capsera
Start free
All tutorials
Syed Raza14 min

Per-role cost in a LangChain coding agent

A triager, parallel file readers, an editor and a reviewer in LangChain 1.x on GPT-4.1, then per-agent cost and a blocking budget on the role that fans out.

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

This builds a coding agent in LangChain 1.x against OpenAI: a triager picks which files to read from a bug report, screeners read those files in parallel, an editor rewrites the one file at fault, and a reviewer approves the patch or sends it back once. One fix is four model calls at the floor and nine at the cap, and where in that range a given report lands is decided by two models rather than by your code. 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 fans out.

What you will build

fixer.py, a single-file agent with four roles and a supervisor:

  • triager — one call. Reads an index of the repository and the report, names at most MAX_CANDIDATES files worth opening.
  • screener — one call per named file, run in parallel. Reads one file and answers SUSPECT or CLEAR. The multiplier.
  • editor — one call, on the stronger model. Rewrites one file in full.
  • reviewer — one call. Approves the patch or names the single change needed, which buys at most MAX_REVISIONS more editor passes.

The agent proposes; it never writes to your working tree. The editor returns file contents and the diff is computed in Python with difflib, so what you read at the end is a real unified diff of a change that was never applied.

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 supervisor 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 the triager, the screeners and the reviewer, gpt-4.1 for the editor.

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 — Fix the repository the agent may read

Start with the repository, not the agent. A coding agent pointed at a real checkout has an unbounded number of calls in it, because every tool call it makes to look around is a call you pay for. An agent handed a fixed snapshot has a call count you can multiply out before you run anything, which is what makes the rest of this page measurable.

The snapshot below is a small service with a real bug in it: retries that sleep after the final attempt, wrapped in retries, so one transient failure spends longer than the worker's deadline allows. Two files are implicated, one file looks implicated and is not, and a test states the expected behaviour.

"""fixer.py — the repository snapshot a coding agent is allowed to read."""

SNAPSHOT = {
    "app/config.py": (
        "RETRY_ATTEMPTS = 3\n"
        "RETRY_BASE_DELAY = 0.5\n"
        "JOB_DEADLINE_SECONDS = 5.0\n"
    ),
    "app/retry.py": (
        "import time\n"
        "\n"
        "from app.config import RETRY_ATTEMPTS, RETRY_BASE_DELAY\n"
        "\n"
        "\n"
        "class TransientError(Exception):\n"
        "    pass\n"
        "\n"
        "\n"
        "def with_retries(fn, attempts=RETRY_ATTEMPTS):\n"
        "    last = None\n"
        "    for i in range(attempts):\n"
        "        try:\n"
        "            return fn()\n"
        "        except TransientError as exc:\n"
        "            last = exc\n"
        "        time.sleep(RETRY_BASE_DELAY * 2**i)\n"
        "    raise last\n"
    ),
    "app/client.py": (
        "from app.retry import with_retries\n"
        "\n"
        "\n"
        "def post_json(session, url, body):\n"
        "    return with_retries(lambda: session.post(url, json=body, timeout=2.0))\n"
        "\n"
        "\n"
        "def submit(session, job):\n"
        "    # belt and braces: retry the whole submission as well\n"
        "    return with_retries(lambda: post_json(session, '/v1/jobs', job))\n"
    ),
    "app/queue_worker.py": (
        "import time\n"
        "\n"
        "from app.client import submit\n"
        "from app.config import JOB_DEADLINE_SECONDS\n"
        "\n"
        "\n"
        "def run_job(session, job):\n"
        "    started = time.monotonic()\n"
        "    try:\n"
        "        return submit(session, job)\n"
        "    finally:\n"
        "        if time.monotonic() - started > JOB_DEADLINE_SECONDS:\n"
        "            raise TimeoutError('job %s exceeded the deadline' % job['id'])\n"
    ),
    "app/metrics.py": (
        "RETRIES = {}\n"
        "\n"
        "\n"
        "def record_retry(name):\n"
        "    RETRIES[name] = RETRIES.get(name, 0) + 1\n"
        "\n"
        "\n"
        "def snapshot():\n"
        "    return dict(RETRIES)\n"
    ),
    "tests/test_retry.py": (
        "import time\n"
        "\n"
        "from app.retry import TransientError, with_retries\n"
        "\n"
        "\n"
        "def test_gives_up_without_a_trailing_sleep():\n"
        "    calls = []\n"
        "\n"
        "    def always_fails():\n"
        "        calls.append(1)\n"
        "        raise TransientError('boom')\n"
        "\n"
        "    started = time.monotonic()\n"
        "    try:\n"
        "        with_retries(always_fails, attempts=2)\n"
        "    except TransientError:\n"
        "        pass\n"
        "    assert len(calls) == 2\n"
        "    assert time.monotonic() - started < 1.0\n"
    ),
}

ISSUE = (
    "Jobs that hit one transient failure never come back. The worker raises "
    "the deadline error and the job is marked failed instead of retried, and "
    "tests/test_retry.py fails on timing. It started after we added retries "
    "in the client. Nothing in the logs says which layer gave up."
)


def file_index() -> str:
    return "\n".join(
        f"{path} ({len(source.splitlines())} lines)"
        for path, source in SNAPSHOT.items()
    )


def file_text(path: str) -> str:
    return SNAPSHOT[path]


if __name__ == "__main__":
    print(file_index())

file_index() is the cheap artefact that makes the expensive one avoidable. The triager sees paths and line counts, not source, so the call that decides where to look costs a fraction of the calls that do the looking. On a real repository you would build the same index from git ls-files and keep it to one line per file for the same reason.

Step 2 — Triage, with the fan-out bound in code

The triager decides how wide the agent goes. So the cap lives in Python, not in the prompt — a model asked for "at most four" will occasionally name six, and the slice is what makes that harmless.

# fragment
import re

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

FAST_MODEL = "gpt-4.1-mini"
MAX_CANDIDATES = 4

_BULLET = re.compile(r"^\s*(?:[-*•]|\d+[.)])\s*")


def _lines(text: str, limit: int) -> list[str]:
    cleaned = (_BULLET.sub("", line).strip() for line in text.splitlines())
    return [line for line in cleaned if line][:limit]


TRIAGE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You triage bug reports against a repository index. Name the files "
            "that must be read to explain the report. One path per line, copied "
            "exactly from the index, no numbering and no commentary. At most "
            "{max_candidates} paths.",
        ),
        ("human", "Report:\n{report}\n\nRepository index:\n{index}"),
    ]
)

triager_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
triage_chain = TRIAGE_PROMPT | triager_llm | StrOutputParser()


def triage(report: str) -> list[str]:
    raw = triage_chain.invoke(
        {
            "report": report,
            "index": file_index(),
            "max_candidates": MAX_CANDIDATES,
        }
    )
    return [p for p in _lines(raw, MAX_CANDIDATES) if p in SNAPSHOT]

Lines rather than JSON. A one-path-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.

The if p in SNAPSHOT filter is not defensive tidiness. A model naming a file that does not exist is the normal failure here — app/retries.py for app/retry.py — and without the filter that path either raises a KeyError or, worse, reaches a screening call about a file nobody can read.

Step 3 — Screen the candidates in parallel

Four sequential reads that do not depend on each other is four round trips of latency for nothing. batch() runs them concurrently and max_concurrency bounds how many are in flight.

# fragment
SCREEN_CONCURRENCY = 4

SCREEN_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You decide whether one source file is implicated in a bug report. "
            "Reply with exactly one line. If it is implicated, write SUSPECT, a "
            "colon, and at most twenty words naming the construct at fault. If "
            "it is not, write CLEAR and nothing else. Judge only the file you "
            "are given and never guess at code you cannot see.",
        ),
        ("human", "Report:\n{report}\n\nFile {path}:\n{source}"),
    ]
)

screener_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=96, temperature=0)
screen_chain = SCREEN_PROMPT | screener_llm | StrOutputParser()


def screen(report: str, paths: list[str]) -> list[dict[str, str]]:
    payloads = [
        {"report": report, "path": p, "source": file_text(p)} for p in paths
    ]
    verdicts = screen_chain.batch(
        payloads, config={"max_concurrency": SCREEN_CONCURRENCY}
    )
    return [
        {"path": p, "verdict": v.strip()}
        for p, v in zip(paths, verdicts, strict=True)
        if v.strip().upper().startswith("SUSPECT")
    ]

One file per screening call, not the whole candidate set per call. That is what keeps widening the search affordable: the prompt is a fixed instruction plus one file, so raising MAX_CANDIDATES multiplies a small number instead of growing a large one. It also means a CLEAR verdict is a judgement about a file the model actually had in front of it.

max_tokens=96 on the screener is load-bearing rather than tidiness. Output tokens are the expensive half of a call, this role runs once per candidate, and the single line it is asked for fits easily. A screener allowed to write a paragraph is the same search at several times the price, and it hands the editor a longer prompt too.

zip(..., strict=True) because pairing paths to verdicts by position is only safe if the lengths agree, and a silent truncation here would blame the wrong file.

Step 4 — Draft the patch, and diff it in Python

Do not ask a model for a diff. Unified diff format needs correct hunk headers and line counts, getting them wrong is invisible until git apply refuses the patch, and a rejected patch is a re-run of the most expensive call on the page. Ask for the file, and compute the diff yourself.

# fragment
import difflib

EDIT_MODEL = "gpt-4.1"

BEGIN = "<<<BEGIN FILE>>>"
END = "<<<END FILE>>>"

EDIT_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You fix one Python file. Return the complete new contents of that "
            "file between a line reading <<<BEGIN FILE>>> and a line reading "
            "<<<END FILE>>>, with nothing outside those markers. Change as "
            "little as possible: keep the existing style, imports and public "
            "names. If the report cannot be fixed in this file, return the file "
            "unchanged.",
        ),
        (
            "human",
            "Report:\n{report}\n\nScreening notes:\n{notes}\n\n"
            "File to change: {path}\n\nCurrent contents:\n{source}\n\n"
            "Required change from review: {feedback}",
        ),
    ]
)

editor_llm = ChatOpenAI(model=EDIT_MODEL, max_tokens=2048, temperature=0)
edit_chain = EDIT_PROMPT | editor_llm | StrOutputParser()


def format_notes(findings: list[dict[str, str]]) -> str:
    return "\n".join(f"{f['path']}: {f['verdict']}" for f in findings)


def _between(text: str) -> str:
    if BEGIN in text and END in text:
        return text.split(BEGIN, 1)[1].split(END, 1)[0].strip("\n")
    return text.strip()


def draft(report: str, notes: str, path: str, feedback: str = "") -> str:
    raw = edit_chain.invoke(
        {
            "report": report,
            "notes": notes,
            "path": path,
            "source": file_text(path),
            "feedback": feedback or "none",
        }
    )
    return _between(raw)


def diff_for(path: str, new_source: str) -> str:
    return "".join(
        difflib.unified_diff(
            file_text(path).splitlines(keepends=True),
            new_source.splitlines(keepends=True),
            fromfile=f"a/{path}",
            tofile=f"b/{path}",
        )
    )

Explicit markers rather than a bare response, because a model told to return only code will sometimes wrap it in a fenced block anyway, and _between is two lines while a fence parser that survives fences inside the file is not. When the markers are missing the whole response is used, so a reply that ignored the instruction still produces a patch instead of an exception.

difflib.unified_diff also gives you a free check the model cannot fake: an empty diff means the editor changed nothing. That is a cheap way to catch the common non-answer — a polite explanation of the bug, correctly wrapped in markers, with the file returned as it was.

The editor is the only role on the stronger model, because it is the only one whose output ends up in your repository. 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 5 — Review, with the loop bounded

The reviewer is where a coding agent turns into a runaway loop if you let it. Edit, review, reject, edit again is a cycle with no natural end: the reviewer always has one more objection available, and every lap is one editor call on the expensive model plus one reviewer call. The bound belongs in Python, as a constant you can read, not in a prompt asking the reviewer to be reasonable.

# fragment
MAX_REVISIONS = 1

REVIEW_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You review one patch against the report it claims to fix. Reply "
            "with exactly one line. Write APPROVE if the patch fixes the "
            "reported behaviour and changes nothing else. Otherwise write "
            "REVISE, a colon, and one sentence naming the single change "
            "required. Do not restate the patch and do not suggest "
            "improvements that the report did not ask for.",
        ),
        ("human", "Report:\n{report}\n\nPatch:\n{diff}"),
    ]
)

reviewer_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=128, temperature=0)
review_chain = REVIEW_PROMPT | reviewer_llm | StrOutputParser()


def review(report: str, diff: str) -> str:
    return review_chain.invoke({"report": report, "diff": diff}).strip()

"Do not suggest improvements that the report did not ask for" is in the prompt for the same reason MAX_REVISIONS is in the code. A reviewer that asks for type hints on an unrelated function gets them, and you pay for a second editor pass on the expensive model to buy a change nobody reported.

The reviewer reads the diff, not the whole file. The diff is the smallest thing that answers the question it was asked, and on a large file it is a fraction of the input tokens the editor just spent.

Step 6 — The supervisor

One function, and every branch the agent can take is visible in it.

# fragment
def fix_issue(report: str) -> dict[str, str]:
    paths = triage(report)
    if not paths:
        raise RuntimeError("the triager named no file that exists in the index")

    findings = screen(report, paths)
    if not findings:
        raise RuntimeError("screening implicated no file")

    target = findings[0]["path"]
    notes = format_notes(findings)
    feedback = ""

    # Runs at least once, so patch and verdict are always bound below.
    for _ in range(MAX_REVISIONS + 1):
        patch = diff_for(target, draft(report, notes, target, feedback))
        if not patch:
            raise RuntimeError(f"the editor returned {target} unchanged")
        verdict = review(report, patch)
        if verdict.upper().startswith("APPROVE"):
            return {"path": target, "diff": patch, "review": verdict}
        feedback = verdict

    return {"path": target, "diff": patch, "review": verdict}

findings[0] picks the first implicated file in the triager's order, and the editor changes exactly one file. A real fixer would rank the suspects, and ranking is another model call; a multi-file patch is a different and much more expensive program, because the editor's cost then scales with the number of files it rewrites rather than being one call.

Count the calls: one triager, up to four screeners, then one or two editors and one or two reviewers. Four at the floor, nine at the cap. Both of those numbers come from your constants; nothing in your file decides where between them a given report lands.

The whole file

"""fixer.py — a coding agent: triage, screen, patch, review."""
import difflib
import re

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

FAST_MODEL = "gpt-4.1-mini"
EDIT_MODEL = "gpt-4.1"

MAX_CANDIDATES = 4
SCREEN_CONCURRENCY = 4
MAX_REVISIONS = 1

BEGIN = "<<<BEGIN FILE>>>"
END = "<<<END FILE>>>"

SNAPSHOT = {
    "app/config.py": (
        "RETRY_ATTEMPTS = 3\n"
        "RETRY_BASE_DELAY = 0.5\n"
        "JOB_DEADLINE_SECONDS = 5.0\n"
    ),
    "app/retry.py": (
        "import time\n"
        "\n"
        "from app.config import RETRY_ATTEMPTS, RETRY_BASE_DELAY\n"
        "\n"
        "\n"
        "class TransientError(Exception):\n"
        "    pass\n"
        "\n"
        "\n"
        "def with_retries(fn, attempts=RETRY_ATTEMPTS):\n"
        "    last = None\n"
        "    for i in range(attempts):\n"
        "        try:\n"
        "            return fn()\n"
        "        except TransientError as exc:\n"
        "            last = exc\n"
        "        time.sleep(RETRY_BASE_DELAY * 2**i)\n"
        "    raise last\n"
    ),
    "app/client.py": (
        "from app.retry import with_retries\n"
        "\n"
        "\n"
        "def post_json(session, url, body):\n"
        "    return with_retries(lambda: session.post(url, json=body, timeout=2.0))\n"
        "\n"
        "\n"
        "def submit(session, job):\n"
        "    # belt and braces: retry the whole submission as well\n"
        "    return with_retries(lambda: post_json(session, '/v1/jobs', job))\n"
    ),
    "app/queue_worker.py": (
        "import time\n"
        "\n"
        "from app.client import submit\n"
        "from app.config import JOB_DEADLINE_SECONDS\n"
        "\n"
        "\n"
        "def run_job(session, job):\n"
        "    started = time.monotonic()\n"
        "    try:\n"
        "        return submit(session, job)\n"
        "    finally:\n"
        "        if time.monotonic() - started > JOB_DEADLINE_SECONDS:\n"
        "            raise TimeoutError('job %s exceeded the deadline' % job['id'])\n"
    ),
    "app/metrics.py": (
        "RETRIES = {}\n"
        "\n"
        "\n"
        "def record_retry(name):\n"
        "    RETRIES[name] = RETRIES.get(name, 0) + 1\n"
        "\n"
        "\n"
        "def snapshot():\n"
        "    return dict(RETRIES)\n"
    ),
    "tests/test_retry.py": (
        "import time\n"
        "\n"
        "from app.retry import TransientError, with_retries\n"
        "\n"
        "\n"
        "def test_gives_up_without_a_trailing_sleep():\n"
        "    calls = []\n"
        "\n"
        "    def always_fails():\n"
        "        calls.append(1)\n"
        "        raise TransientError('boom')\n"
        "\n"
        "    started = time.monotonic()\n"
        "    try:\n"
        "        with_retries(always_fails, attempts=2)\n"
        "    except TransientError:\n"
        "        pass\n"
        "    assert len(calls) == 2\n"
        "    assert time.monotonic() - started < 1.0\n"
    ),
}

ISSUE = (
    "Jobs that hit one transient failure never come back. The worker raises "
    "the deadline error and the job is marked failed instead of retried, and "
    "tests/test_retry.py fails on timing. It started after we added retries "
    "in the client. Nothing in the logs says which layer gave up."
)

_BULLET = re.compile(r"^\s*(?:[-*•]|\d+[.)])\s*")

TRIAGE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You triage bug reports against a repository index. Name the files "
            "that must be read to explain the report. One path per line, copied "
            "exactly from the index, no numbering and no commentary. At most "
            "{max_candidates} paths.",
        ),
        ("human", "Report:\n{report}\n\nRepository index:\n{index}"),
    ]
)

SCREEN_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You decide whether one source file is implicated in a bug report. "
            "Reply with exactly one line. If it is implicated, write SUSPECT, a "
            "colon, and at most twenty words naming the construct at fault. If "
            "it is not, write CLEAR and nothing else. Judge only the file you "
            "are given and never guess at code you cannot see.",
        ),
        ("human", "Report:\n{report}\n\nFile {path}:\n{source}"),
    ]
)

EDIT_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You fix one Python file. Return the complete new contents of that "
            "file between a line reading <<<BEGIN FILE>>> and a line reading "
            "<<<END FILE>>>, with nothing outside those markers. Change as "
            "little as possible: keep the existing style, imports and public "
            "names. If the report cannot be fixed in this file, return the file "
            "unchanged.",
        ),
        (
            "human",
            "Report:\n{report}\n\nScreening notes:\n{notes}\n\n"
            "File to change: {path}\n\nCurrent contents:\n{source}\n\n"
            "Required change from review: {feedback}",
        ),
    ]
)

REVIEW_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You review one patch against the report it claims to fix. Reply "
            "with exactly one line. Write APPROVE if the patch fixes the "
            "reported behaviour and changes nothing else. Otherwise write "
            "REVISE, a colon, and one sentence naming the single change "
            "required. Do not restate the patch and do not suggest "
            "improvements that the report did not ask for.",
        ),
        ("human", "Report:\n{report}\n\nPatch:\n{diff}"),
    ]
)

triager_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=256, temperature=0)
screener_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=96, temperature=0)
editor_llm = ChatOpenAI(model=EDIT_MODEL, max_tokens=2048, temperature=0)
reviewer_llm = ChatOpenAI(model=FAST_MODEL, max_tokens=128, temperature=0)

triage_chain = TRIAGE_PROMPT | triager_llm | StrOutputParser()
screen_chain = SCREEN_PROMPT | screener_llm | StrOutputParser()
edit_chain = EDIT_PROMPT | editor_llm | StrOutputParser()
review_chain = REVIEW_PROMPT | reviewer_llm | StrOutputParser()


def file_index() -> str:
    return "\n".join(
        f"{path} ({len(source.splitlines())} lines)"
        for path, source in SNAPSHOT.items()
    )


def file_text(path: str) -> str:
    return SNAPSHOT[path]


def _lines(text: str, limit: int) -> list[str]:
    cleaned = (_BULLET.sub("", line).strip() for line in text.splitlines())
    return [line for line in cleaned if line][:limit]


def _between(text: str) -> str:
    if BEGIN in text and END in text:
        return text.split(BEGIN, 1)[1].split(END, 1)[0].strip("\n")
    return text.strip()


def format_notes(findings: list[dict[str, str]]) -> str:
    return "\n".join(f"{f['path']}: {f['verdict']}" for f in findings)


def triage(report: str) -> list[str]:
    raw = triage_chain.invoke(
        {
            "report": report,
            "index": file_index(),
            "max_candidates": MAX_CANDIDATES,
        }
    )
    return [p for p in _lines(raw, MAX_CANDIDATES) if p in SNAPSHOT]


def screen(report: str, paths: list[str]) -> list[dict[str, str]]:
    payloads = [
        {"report": report, "path": p, "source": file_text(p)} for p in paths
    ]
    verdicts = screen_chain.batch(
        payloads, config={"max_concurrency": SCREEN_CONCURRENCY}
    )
    return [
        {"path": p, "verdict": v.strip()}
        for p, v in zip(paths, verdicts, strict=True)
        if v.strip().upper().startswith("SUSPECT")
    ]


def draft(report: str, notes: str, path: str, feedback: str = "") -> str:
    raw = edit_chain.invoke(
        {
            "report": report,
            "notes": notes,
            "path": path,
            "source": file_text(path),
            "feedback": feedback or "none",
        }
    )
    return _between(raw)


def diff_for(path: str, new_source: str) -> str:
    return "".join(
        difflib.unified_diff(
            file_text(path).splitlines(keepends=True),
            new_source.splitlines(keepends=True),
            fromfile=f"a/{path}",
            tofile=f"b/{path}",
        )
    )


def review(report: str, diff: str) -> str:
    return review_chain.invoke({"report": report, "diff": diff}).strip()


def fix_issue(report: str) -> dict[str, str]:
    paths = triage(report)
    if not paths:
        raise RuntimeError("the triager named no file that exists in the index")

    findings = screen(report, paths)
    if not findings:
        raise RuntimeError("screening implicated no file")

    target = findings[0]["path"]
    notes = format_notes(findings)
    feedback = ""

    # Runs at least once, so patch and verdict are always bound below.
    for _ in range(MAX_REVISIONS + 1):
        patch = diff_for(target, draft(report, notes, target, feedback))
        if not patch:
            raise RuntimeError(f"the editor returned {target} unchanged")
        verdict = review(report, patch)
        if verdict.upper().startswith("APPROVE"):
            return {"path": target, "diff": patch, "review": verdict}
        feedback = verdict

    return {"path": target, "diff": patch, "review": verdict}


if __name__ == "__main__":
    result = fix_issue(ISSUE)
    print(result["diff"])
    print(result["review"])

python fixer.py

You get a unified diff against one file and one line of review. Which file it picks is the interesting part: app/retry.py if the screening notes lead with the trailing sleep, app/client.py if they lead with the nested retry call. Both are real causes of the reported behaviour and the one-file editor can only fix one of them, which is the honest limit of an agent shaped like this.

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 four to nine 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 fixer.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 agent makes is recorded: the triager's, each screener's inside batch(), every editor pass, every reviewer pass. 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 = fix_issue(ISSUE)
    print(result["diff"])
    print(result["review"])
    capsera.shutdown()

capsera.shutdown() flushes pending events and stops the worker. A long-running service does not need it; a script, a pre-commit hook 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 decorator only says who:

# fragment
@capsera.agent("triager", team="repo-fixer", task_type="file-selection")
def triage(report: str) -> list[str]:
    ...  # body unchanged


@capsera.agent("screener", team="repo-fixer", task_type="file-screening")
def screen(report: str, paths: list[str]) -> list[dict[str, str]]:
    ...  # body unchanged


@capsera.agent("editor", team="repo-fixer", task_type="patch")
def draft(report: str, notes: str, path: str, feedback: str = "") -> str:
    ...  # body unchanged


@capsera.agent("reviewer", team="repo-fixer", task_type="patch-review")
def review(report: str, diff: 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. screen does its work in screen_chain.batch(...), which dispatches to worker threads rather than calling on the thread you were on, and the screener 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 four candidate files are four screener 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. Screen the files yourself with a bare ThreadPoolExecutor and the same decorator records unknown instead of screener, 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=SCREEN_CONCURRENCY) as pool:
    verdicts = list(pool.map(screen_one, paths))

# Keeps it: hand each worker a copy of the caller's context.
ctx = contextvars.copy_context()
with ThreadPoolExecutor(max_workers=SCREEN_CONCURRENCY) as pool:
    verdicts = list(pool.map(lambda p: ctx.run(screen_one, p), paths))

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. fix_issue 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-fix total, use the team field — which is what team="repo-fixer" on all four decorators is for.

Cost the second editor pass separately

The review loop is the part of this agent whose cost you cannot see from the role rows alone. A run with two editor passes and a run with one both report as editor, so the row grows and nothing in it says the growth came from patches being sent back. capsera.tag() opens a sub-scope inside a function you have already decorated, which is exactly this case:

# fragment
import capsera


@capsera.agent("editor", team="repo-fixer", task_type="patch")
def draft(report: str, notes: str, path: str, feedback: str = "") -> str:
    if not feedback:
        return _draft(report, notes, path, "")
    # A revision is the same agent doing a different job. Cost it as one.
    with capsera.tag("editor", team="repo-fixer", task_type="patch-revision"):
        return _draft(report, notes, path, feedback)

_draft is the original body, moved down one level and otherwise unchanged. The revision keeps the agent name, so a budget scoped to editor still counts every pass, and it changes the task type, so the share of editor spend that came from rework is a number you can read instead of a suspicion. Use tag() for a genuine sub-scope like this one. It is the wrong tool for naming an iteration: tagging each screening call with its file path would give you one row per file, none of which exists on the next run, and an agent budget cannot accumulate against a name that lived for one fix. 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 repository or which team a fix was for without fragmenting the agent dimension that budgets and cost attribution depend on.

Put a budget on the role that multiplies

The triager makes one call per run. The editor and the reviewer make one each per attempt, and MAX_REVISIONS decides how many attempts there can be. All three of those counts are in your file. The screener's is not: it comes from the triager's output, which comes from how many files in your repository look relevant to a bug report — and that number is nowhere in fixer.py. Point this agent at a repository with two thousand files and raise MAX_CANDIDATES to the width that finds anything, and the screener is the only row that moves. That is multi-agent amplification in one sentence, and it is why the cap belongs on the screener rather than on the run.

Create the budget in the dashboard with scope agent and the agent set to screener — 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 screening 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 screener 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 screening call means for the fix. 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


@capsera.agent("screener", team="repo-fixer", task_type="file-screening")
def screen(report: str, paths: list[str]) -> list[dict[str, str]]:
    payloads = [
        {"report": report, "path": p, "source": file_text(p)} for p in paths
    ]
    verdicts = screen_chain.batch(
        payloads,
        config={"max_concurrency": SCREEN_CONCURRENCY},
        return_exceptions=True,
    )

    findings: list[dict[str, str]] = []
    for path, verdict in zip(paths, verdicts, strict=True):
        if isinstance(verdict, capsera.BudgetExceededError):
            # Do not draft a patch from a repository the agent only half read.
            raise verdict
        if isinstance(verdict, BaseException):
            raise RuntimeError(f"screening {path} failed: {verdict}")
        if verdict.strip().upper().startswith("SUSPECT"):
            findings.append({"path": path, "verdict": verdict.strip()})
    return findings

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 file that went unread. Without it, one blocked screening call ends the round and the remaining candidates are never even attempted.

Both cases stop the run here, and that is the judgement call worth arguing with. An unread file is a file that might hold the bug, and the next thing the agent would do is spend its most expensive call rewriting whichever file it did read. A patch produced that way is not incomplete, it is wrong, and its cost is a human review cycle rather than a token count. Degrading instead — drafting from the files that were read and telling the reviewer which were not — is defensible if the reviewer is given that fact, but it is the harder version to get right, and the safe default for a program that proposes code changes is to refuse.

The refusal then needs one place to land:

# fragment
import capsera


def run_once(report: str) -> dict[str, str] | None:
    """Returns None when a budget stopped the run."""
    try:
        return fix_issue(report)
    except capsera.BudgetExceededError as exc:
        print(f"screener budget reached, no patch produced: {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. Narrowing MAX_CANDIDATES and rerunning, queueing the report until the period rolls over, and failing are all defensible; for an agent that files patches from a queue, failing usually is.

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 screener stops screening calls. The triager, the editor and the reviewer are different agents with no budget on them, so a run that gets past screening still reaches the expensive editor call and still pays for it. That is the right shape here — the screener is the row that grows with the repository — but it is worth deciding on purpose rather than discovering: if what you want capped is the editor's model spend, that is a second budget on editor, not a side effect of this one.

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 SCREEN_CONCURRENCY — four as shipped, and whatever you raise it to when the repository gets bigger. Set the budget below a figure you cannot exceed, or lower the concurrency; it is the concurrency that decides the overshoot, not MAX_CANDIDATES.

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

Four agent rows — triager, screener, editor, reviewer — under one team, repo-fixer, with a cost per run for the fix as a whole, and the editor row split by task type into patch and patch-revision.

Read each row as call count and cost per call, not as a total. Three things can move on this agent and each has a different fix. The screener's call count moving means the triager named more files, or the repository grew. The screener's cost per call moving means the files it named were bigger, which is a property of your repository and not of your prompt. And a patch-revision share that climbs means the reviewer is rejecting more patches — the most expensive of the three, because every rejection buys another editor call on the stronger model plus another review of it.

Which row is largest is not predictable in advance, and that is the point of measuring it rather than reasoning about it. The editor makes one or two calls on the more expensive model with a whole file in the prompt and a whole file in the response; the screener makes a handful of small ones on the cheaper model. Where the crossover falls depends on how wide you set MAX_CANDIDATES and on the size of the file being rewritten, and it decides whether the next thing to change is the editor's model or the width of the search.

Output tokens are worth watching on the editor specifically. It returns the complete file, so the same change costs more in a long file than a short one for reasons that have nothing to do with the change — the one place where asking for a diff instead of a file would pay, if a model could be trusted to produce hunk headers.

The screener and the editor both send text that repeats. Every screening call in a round carries the same instruction with a different file, and a revision pass sends the file the first pass already sent. The cache read share column is what tells you whether any of that repetition 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 fixer.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 coding agent? Call capsera.init() once, which patches the provider client ChatOpenAI constructs underneath, so every call the agent makes is already recorded. Then put @capsera.agent("<role>") on each role function — the triager, the screener, the editor, the reviewer. 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 agent in a code-fixing pipeline should carry the budget? The screening role, because its call count is the one no line of your code sets: the triager names the files and the repository decides how many there are to name. The editor and the reviewer are one call each per attempt. A budget with scope agent on the screener stops the fan-out before the editor is reached, which is also the expensive call.

What should a coding agent do when a budget blocks part of its file screening? Stop before the editor. An unread file is a file that might hold the bug, and the next thing the agent would do is spend its most expensive call rewriting whichever file it did read — a patch produced that way is not incomplete, it is wrong. Pass return_exceptions=True to batch() so a blocked call comes back as an item rather than aborting the round, tell a BudgetExceededError apart from a flaky call, and refuse to draft.

Why do two runs of the same coding agent cost different amounts? Three things move between runs and none of them is a line you wrote. The triager decides how many files get screened, up to the cap you set. The screener's input tokens scale with the size of whichever files it named, which differ. And the reviewer decides whether the editor runs a second time. Watch call count and cost per call per role rather than the total, because those three causes have three different fixes.

Questions this page answers

How do I get per-agent cost for a LangChain coding agent?
Call capsera.init() once, which patches the provider client ChatOpenAI constructs underneath, so every call the agent makes is already recorded. Then put @capsera.agent("<role>") on each role function — the triager, the screener, the editor, the reviewer. 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 agent in a code-fixing pipeline should carry the budget?
The screening role, because its call count is the one no line of your code sets: the triager names the files and the repository decides how many there are to name. The editor and the reviewer are one call each per attempt. A budget with scope agent on the screener stops the fan-out before the editor is reached, which is also the expensive call.
What should a coding agent do when a budget blocks part of its file screening?
Stop before the editor. An unread file is a file that might hold the bug, and the next thing the agent would do is spend its most expensive call rewriting whichever file it did read — a patch produced that way is not incomplete, it is wrong. Pass return_exceptions=True to batch() so a blocked call comes back as an item rather than aborting the round, tell a BudgetExceededError apart from a flaky call, and refuse to draft.
Why do two runs of the same coding agent cost different amounts?
Three things move between runs and none of them is a line you wrote. The triager decides how many files get screened, up to the cap you set. The screener's input tokens scale with the size of whichever files it named, which differ. And the reviewer decides whether the editor runs a second time. Watch call count and cost per call per role rather than the total, because those three causes have three different fixes.

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

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

See pricing