Syed Raza5 August 202612 min

Cost per agent in a LangChain incident-brief pipeline

A planner, parallel analysts, a critic and a writer in LangChain 1.x on Claude, then per-agent cost and a blocking budget on the role that multiplies.

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

This builds an incident-brief pipeline in LangChain 1.x against Claude: a planner turns a raw report into a bounded list of questions, analysts answer them in parallel from a fixed evidence bundle, a critic finds what is still missing, and a writer produces the brief. One run is a handful of Haiku calls plus a single Sonnet call, and the analyst's share of that is decided by how many questions the planner wrote — a model's output, not a constant in your file. The last third of the page adds Capsera: one init() call to capture every request, four decorator lines to say which role made it, and a budget that stops the analyst before it calls.

What you will build

brief.py, a single-file pipeline with four roles and a supervisor:

planner — one call. Turns the report into at most six questions.
analyst — one call per question, run in parallel. Answers only from a fixed evidence bundle, or says UNKNOWN.
critic — one call. Returns the gaps worth a second pass, at most two.
writer — one call, on a stronger model. Produces the brief.

The supervisor is a plain Python function, not a graph. That is deliberate: the subject here is where multi-agent cost comes from, and a readable control flow makes the role boundaries visible. If you want the same instrumentation on a LangGraph app, the node-level equivalent is langgraph_node.

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 swap a model on.

The cost problem in this tutorial only exists because one request becomes many calls. If your app makes a single call per request, there is nothing to group and nothing to amplify.

Written against LangChain 1.x with langchain-anthropic, calling Anthropic's API directly rather than through a gateway. Models: claude-haiku-4-5 for the three cheap roles, claude-sonnet-4-6 for the writer.

Prerequisites

Python 3.11 or newer.

pip install "langchain>=1.0,<2" langchain-anthropic

The major is pinned so this page stays true as LangChain moves. langchain-anthropic is left unpinned on purpose: install the one that matches your langchain major.

You need an Anthropic API key. ChatAnthropic reads ANTHROPIC_API_KEY from the environment, so export it and no code has to touch it:

export ANTHROPIC_API_KEY=...

Step 1 — Fix the evidence the analysts may read

Start with the corpus, not the agents. A fixed evidence bundle is what makes the run's cost predictable in the first place: an analyst that can search has an unbounded number of calls in it, while an analyst handed a bundle has exactly one.

"""brief.py — the evidence an analyst is allowed to read."""

EVIDENCE = {
    "deploy-log": (
        "13:58 UTC  api-gateway 8f21c3 built from main\n"
        "14:02 UTC  8f21c3 to canary, 10% of traffic\n"
        "14:09 UTC  8f21c3 autopromoted to 100%\n"
        "14:31 UTC  rollback to 7d90aa started\n"
        "14:36 UTC  rollback complete"
    ),
    "error-rates": (
        "checkout POST /orders 5xx: 0.3% at 14:00, 4% at 14:11, 11% at 14:18, "
        "0.4% at 14:38\n"
        "payments POST /charge 5xx: flat at 0.2% all afternoon\n"
        "api-gateway p99: 180ms at 14:00, 2.4s at 14:15, 190ms at 14:40"
    ),
    "runbook": (
        "api-gateway holds 40 connections per instance to payments. Pool "
        "exhaustion surfaces as 5xx on checkout, never on payments, because the "
        "gateway fails the request before forwarding it. Release 8f21c3 raised "
        "the pool acquire timeout from 250ms to 5s."
    ),
    "oncall-chat": (
        "14:14 payments oncall: dashboards green, charge volume normal\n"
        "14:22 checkout oncall: every failure I sample is a gateway 503\n"
        "14:29 platform oncall: rolling 8f21c3 back, no sev declared"
    ),
}


def evidence_text() -> str:
    return "\n\n".join(f"[{name}]\n{body}" for name, body in EVIDENCE.items())


if __name__ == "__main__":
    print(f"{len(EVIDENCE)} sources, {len(evidence_text())} characters")

Note what evidence_text() produces: one string, identical on every analyst call in a round. That single fact drives most of what the cost report will say at the end of this page.

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

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

# fragment
import re

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

FAST_MODEL = "claude-haiku-4-5"
MAX_QUESTIONS = 6

_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]


PLANNER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You triage production incidents. Given a report, write the "
            "questions an analyst must answer to explain what happened. One "
            "question per line. No numbering, no preamble. At most "
            "{max_questions} questions.",
        ),
        ("human", "{report}"),
    ]
)

planner_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
plan_chain = PLANNER_PROMPT | planner_llm | StrOutputParser()


def plan_questions(report: str) -> list[str]:
    raw = plan_chain.invoke({"report": report, "max_questions": MAX_QUESTIONS})
    return _lines(raw, MAX_QUESTIONS)

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

Each role gets its own ChatAnthropic instance even where the model string repeats. Three of these are the same model today, and the point of separating them is that changing one role's model later is a one-line edit instead of a refactor.

Step 3 — The analysts, in parallel

Six sequential calls that do not depend on each other is six round trips of latency for no reason. batch() runs them concurrently, and max_concurrency bounds how many are in flight.

# fragment
ANALYST_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You answer one question about a production incident using only the "
            "evidence below. Quote the line you relied on. If the evidence does "
            "not answer the question, reply with the single word UNKNOWN.\n\n"
            "Evidence:\n{evidence}",
        ),
        ("human", "{question}"),
    ]
)

analyst_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
analyst_chain = ANALYST_PROMPT | analyst_llm | StrOutputParser()


def investigate(questions: list[str]) -> list[dict[str, str]]:
    evidence = evidence_text()
    payloads = [{"question": q, "evidence": evidence} for q in questions]
    answers = analyst_chain.batch(payloads, config={"max_concurrency": 4})
    return [
        {"question": q, "answer": a.strip()}
        for q, a in zip(questions, answers, strict=True)
        if a.strip() != "UNKNOWN"
    ]

UNKNOWN answers are dropped rather than passed on. An analyst that pads a non-answer to look useful poisons the critic, which then asks for a second pass, which costs another round of calls — a quality failure that shows up as a cost failure two steps downstream.

zip(..., strict=True) because pairing questions to answers by position is only safe if the lengths agree, and a silent truncation here would attach the wrong answer to the wrong question.

Step 4 — The critic and the writer

The critic exists to bound the second pass. Asking "what is missing" without a limit invites a list of everything imaginable, and every item on that list is an analyst call.

# fragment
MAX_GAPS = 2
WRITER_MODEL = "claude-sonnet-4-6"

CRITIC_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You review an incident investigation for gaps. List only questions "
            "that are still unanswered and that the evidence could plausibly "
            "answer. One per line, at most {max_gaps}. If there are none, reply "
            "with the single word NONE.",
        ),
        ("human", "Report:\n{report}\n\nFindings so far:\n{findings}"),
    ]
)

WRITER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write incident briefs for engineers. Four short sections: what "
            "happened, impact, cause, what to do next. Use only the findings "
            "given. Do not speculate. Under 250 words.",
        ),
        ("human", "Report:\n{report}\n\nFindings:\n{findings}"),
    ]
)

critic_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=256, temperature=0)
writer_llm = ChatAnthropic(model=WRITER_MODEL, max_tokens=1024, temperature=0)

critic_chain = CRITIC_PROMPT | critic_llm | StrOutputParser()
writer_chain = WRITER_PROMPT | writer_llm | StrOutputParser()


def format_findings(findings: list[dict[str, str]]) -> str:
    return "\n\n".join(f"Q: {f['question']}\nA: {f['answer']}" for f in findings)


def review(report: str, findings: list[dict[str, str]]) -> list[str]:
    raw = critic_chain.invoke(
        {
            "report": report,
            "findings": format_findings(findings),
            "max_gaps": MAX_GAPS,
        }
    )
    if raw.strip().upper().startswith("NONE"):
        return []
    return _lines(raw, MAX_GAPS)


def write_brief(report: str, findings: list[dict[str, str]]) -> str:
    return writer_chain.invoke(
        {"report": report, "findings": format_findings(findings)}
    )

The writer is the only role on the stronger model, because it is the only one whose output a person reads. The other three produce input for another model, and that is the judgement worth revisiting later with real numbers rather than guessing now.

Step 5 — The supervisor

One function, and every branch in the pipeline is visible in it.

# fragment
REPORT = (
    "Checkout started failing this afternoon and recovered on its own. "
    "The payments team says they were healthy the whole time. "
    "Nobody has written anything down."
)


def brief_incident(report: str) -> str:
    questions = plan_questions(report)
    if not questions:
        raise RuntimeError("the planner returned no questions")

    findings = investigate(questions)
    gaps = review(report, findings)
    if gaps:
        findings = findings + investigate(gaps)
    return write_brief(report, findings)

Count the calls: one planner, up to six analysts, one critic, up to two more analysts, one writer. Nine or ten calls for one brief, and only three of those counts are fixed by your code.

The whole file

"""brief.py — an incident brief from a planner, analysts, a critic and a writer."""
import re

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

FAST_MODEL = "claude-haiku-4-5"
WRITER_MODEL = "claude-sonnet-4-6"
MAX_QUESTIONS = 6
MAX_GAPS = 2

EVIDENCE = {
    "deploy-log": (
        "13:58 UTC  api-gateway 8f21c3 built from main\n"
        "14:02 UTC  8f21c3 to canary, 10% of traffic\n"
        "14:09 UTC  8f21c3 autopromoted to 100%\n"
        "14:31 UTC  rollback to 7d90aa started\n"
        "14:36 UTC  rollback complete"
    ),
    "error-rates": (
        "checkout POST /orders 5xx: 0.3% at 14:00, 4% at 14:11, 11% at 14:18, "
        "0.4% at 14:38\n"
        "payments POST /charge 5xx: flat at 0.2% all afternoon\n"
        "api-gateway p99: 180ms at 14:00, 2.4s at 14:15, 190ms at 14:40"
    ),
    "runbook": (
        "api-gateway holds 40 connections per instance to payments. Pool "
        "exhaustion surfaces as 5xx on checkout, never on payments, because the "
        "gateway fails the request before forwarding it. Release 8f21c3 raised "
        "the pool acquire timeout from 250ms to 5s."
    ),
    "oncall-chat": (
        "14:14 payments oncall: dashboards green, charge volume normal\n"
        "14:22 checkout oncall: every failure I sample is a gateway 503\n"
        "14:29 platform oncall: rolling 8f21c3 back, no sev declared"
    ),
}

REPORT = (
    "Checkout started failing this afternoon and recovered on its own. "
    "The payments team says they were healthy the whole time. "
    "Nobody has written anything down."
)

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


def evidence_text() -> str:
    return "\n\n".join(f"[{name}]\n{body}" for name, body in EVIDENCE.items())


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]


PLANNER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You triage production incidents. Given a report, write the "
            "questions an analyst must answer to explain what happened. One "
            "question per line. No numbering, no preamble. At most "
            "{max_questions} questions.",
        ),
        ("human", "{report}"),
    ]
)

ANALYST_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You answer one question about a production incident using only the "
            "evidence below. Quote the line you relied on. If the evidence does "
            "not answer the question, reply with the single word UNKNOWN.\n\n"
            "Evidence:\n{evidence}",
        ),
        ("human", "{question}"),
    ]
)

CRITIC_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You review an incident investigation for gaps. List only questions "
            "that are still unanswered and that the evidence could plausibly "
            "answer. One per line, at most {max_gaps}. If there are none, reply "
            "with the single word NONE.",
        ),
        ("human", "Report:\n{report}\n\nFindings so far:\n{findings}"),
    ]
)

WRITER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write incident briefs for engineers. Four short sections: what "
            "happened, impact, cause, what to do next. Use only the findings "
            "given. Do not speculate. Under 250 words.",
        ),
        ("human", "Report:\n{report}\n\nFindings:\n{findings}"),
    ]
)

planner_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
analyst_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
critic_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=256, temperature=0)
writer_llm = ChatAnthropic(model=WRITER_MODEL, max_tokens=1024, temperature=0)

plan_chain = PLANNER_PROMPT | planner_llm | StrOutputParser()
analyst_chain = ANALYST_PROMPT | analyst_llm | StrOutputParser()
critic_chain = CRITIC_PROMPT | critic_llm | StrOutputParser()
writer_chain = WRITER_PROMPT | writer_llm | StrOutputParser()


def format_findings(findings: list[dict[str, str]]) -> str:
    return "\n\n".join(f"Q: {f['question']}\nA: {f['answer']}" for f in findings)


def plan_questions(report: str) -> list[str]:
    raw = plan_chain.invoke({"report": report, "max_questions": MAX_QUESTIONS})
    return _lines(raw, MAX_QUESTIONS)


def investigate(questions: list[str]) -> list[dict[str, str]]:
    evidence = evidence_text()
    payloads = [{"question": q, "evidence": evidence} for q in questions]
    answers = analyst_chain.batch(payloads, config={"max_concurrency": 4})
    return [
        {"question": q, "answer": a.strip()}
        for q, a in zip(questions, answers, strict=True)
        if a.strip() != "UNKNOWN"
    ]


def review(report: str, findings: list[dict[str, str]]) -> list[str]:
    raw = critic_chain.invoke(
        {
            "report": report,
            "findings": format_findings(findings),
            "max_gaps": MAX_GAPS,
        }
    )
    if raw.strip().upper().startswith("NONE"):
        return []
    return _lines(raw, MAX_GAPS)


def write_brief(report: str, findings: list[dict[str, str]]) -> str:
    return writer_chain.invoke(
        {"report": report, "findings": format_findings(findings)}
    )


def brief_incident(report: str) -> str:
    questions = plan_questions(report)
    if not questions:
        raise RuntimeError("the planner returned no questions")

    findings = investigate(questions)
    gaps = review(report, findings)
    if gaps:
        findings = findings + investigate(gaps)
    return write_brief(report, findings)


if __name__ == "__main__":
    print(brief_incident(REPORT))

python brief.py

The output is a four-section brief assembled only from the evidence bundle, which is why one runbook line about the connection pool ends up carrying more of the explanation than the whole chat log. 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 nine or ten calls behind it are indistinguishable — same key, same account, and all but one from the same model.

pip install capsera

At the top of brief.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 done. capsera.init() patches the provider clients that ChatAnthropic constructs underneath, so from that call on every request the pipeline makes is recorded: the planner's, each analyst's inside batch(), the critic's, the writer's. 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__":
    print(brief_incident(REPORT))
    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 decorator only says who:

# fragment
@capsera.agent("planner", team="incident-response", task_type="triage")
def plan_questions(report: str) -> list[str]:
    ...  # body unchanged


@capsera.agent("analyst", team="incident-response", task_type="evidence-qa")
def investigate(questions: list[str]) -> list[dict[str, str]]:
    ...  # body unchanged


@capsera.agent("critic", team="incident-response", task_type="review")
def review(report: str, findings: list[dict[str, str]]) -> list[str]:
    ...  # body unchanged


@capsera.agent("writer", team="incident-response", task_type="drafting")
def write_brief(report: str, findings: list[dict[str, 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.

Two details worth knowing before you trust the numbers:

The fan-out keeps its identity. investigate does its work in analyst_chain.batch(...), which dispatches to worker threads rather than calling on the thread you were on. The analyst 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 six questions are six analyst 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. Fan the analysts out yourself with a bare ThreadPoolExecutor and the same decorator records unknown instead of analyst, with no error to tell you. Measured on the same versions:

# fragment
# Loses the scope: workers start with a fresh context.
with ThreadPoolExecutor(max_workers=4) as pool:
    answers = list(pool.map(answer_one, questions))

# Keeps it: hand each worker the caller's context.
ctx = contextvars.copy_context()
with ThreadPoolExecutor(max_workers=4) as pool:
    answers = list(pool.map(lambda q: ctx.run(answer_one, q), questions))

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. brief_incident 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 pipeline-level total, use the team field — which is what team="incident-response" on all four decorators is for.

Put a budget on the role that multiplies

Three of the four roles make exactly one call per run, and your code decides that. The analyst's call count is decided by the planner's output: up to six in the first round, up to two more after the critic. The number that sets the run's cost is not written anywhere in brief.py, which is multi-agent amplification in one sentence, and it is why the cap belongs on the analyst rather than on the pipeline.

Create the budget in the dashboard with scope agent and the agent set to analyst — 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 an agent budget. Pick the amount from a month of your own analyst 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 analyst is over its budget, BudgetExceededError is raised inside your process and the request never leaves it, so no tokens are spent. The cost of that guarantee is one round trip to Capsera per call.

Then decide what a blocked analyst means for the brief. batch() raises on the first failure and abandons the rest of the inputs unless you tell it otherwise, so this is a decision you make rather than one you inherit:

# fragment
@capsera.agent("analyst", team="incident-response", task_type="evidence-qa")
def investigate(questions: list[str]) -> list[dict[str, str]]:
    evidence = evidence_text()
    payloads = [{"question": q, "evidence": evidence} for q in questions]
    answers = analyst_chain.batch(
        payloads,
        config={"max_concurrency": 4},
        return_exceptions=True,
    )

    findings: list[dict[str, str]] = []
    for question, answer in zip(questions, answers, strict=True):
        if isinstance(answer, capsera.BudgetExceededError):
            raise answer  # the cap is real: stop, do not ship a partial brief
        if isinstance(answer, BaseException):
            continue  # one flaky call: the other answers still stand
        if answer.strip() != "UNKNOWN":
            findings.append({"question": question, "answer": answer.strip()})
    return findings

return_exceptions=True asks batch() to hand failures back as items in the result list instead of raising, which is what separates the two cases: a single call that failed for its own reasons is survivable, and a budget block is not something to paper over with five answers out of six. An incident brief with a silent hole in it is worse than no brief.

The re-raise then has one place to land:

# fragment
import capsera


def run_once(report: str) -> str | None:
    """Returns None when a budget stopped the run."""
    try:
        return brief_incident(report)
    except capsera.BudgetExceededError as exc:
        print(f"analyst budget reached, no brief 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. Degrading (a template, a cheaper model called directly), queueing until the period rolls over, and failing are all defensible; for internal batch work, failing usually is.

Two limits to know before you rely on this as a hard ceiling:

The parallel round can overshoot. The check reads budget state computed from delivered events, and events are delivered in batches every 500 ms, so a burst of concurrent calls can each pass a check taken before any of them was recorded. With max_concurrency at 4, up to four analyst calls can clear the same recorded total. Set the budget slightly below a figure you cannot exceed, or lower max_concurrency.

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

Four agent rows — planner, analyst, critic, writer — under one team, incident-response, with a cost per run for the brief as a whole.

The analyst row is the only one that moves between two runs on the same report, because its call count came from a model rather than from your code. Watch its call count next to its cost: a jump in cost per run with flat cost per call is a planner writing more questions, and that is a different fix from a prompt getting longer.

Which row is largest is not predictable in advance, and that is the point of measuring it. The writer makes one call on the more expensive model; the analyst makes six to eight on the cheaper one. Where the crossover falls depends on the fan-out width and on the size of your evidence bundle, and it decides whether the next thing to change is the writer's model or MAX_QUESTIONS.

Input tokens per analyst call are dominated by the evidence bundle, which is byte-identical on every call in a round. Cache read share will read zero, because nothing in this code sets an Anthropic cache_control breakpoint — the prefix repeats and gets billed as fresh input every time. That is a lever this code leaves on the table, and it is invisible on an invoice: prompt caching that never engaged.

Each event also carries the file and line that made the call, which points at brief.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 multi-agent pipeline?: Call capsera.init() once, which patches the provider clients ChatAnthropic and ChatOpenAI construct underneath, so every call is already recorded. Then put @capsera.agent("<role>") on each role function. The decorator adds who made the call; it does not change whether the call is captured.
Does @capsera.agent attribution survive LangChain's batch()?: Yes. Capsera keeps 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 thread 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 a fan-out of six questions is six analyst events.
Can I put a budget on one agent in a LangChain app?: Yes, if that agent's spend is attributed. A budget with scope agent matches on the agent identifier the events carry, which is the string in the decorator. Spend attributed to unknown cannot be governed by an agent budget, so attribution comes first. With enforcement on, the check runs before the provider call and a blocking budget raises BudgetExceededError, so no tokens are spent.
What happens to a LangChain batch when a budget blocks one call?: By default batch() raises on the first failure and abandons the remaining inputs, so one blocked analyst call ends the whole round. Pass return_exceptions=True to get failures back as items in the result list, then decide per item whether to degrade on partial findings or stop the run.

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

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

See pricing