AI agent observability means you can see every step of a run after the fact: which tools ran, what each model call cost, and where a bad output came from. This guide instruments a sales-outreach drafting agent with Langfuse's @observe decorator, nesting tool calls under the run that triggered them, tracking token cost per step, and alerting when a run exceeds a cost threshold. Budget 30-45 minutes and a free Langfuse account or self-host.
| Fact | Value |
|---|---|
| Tool | Any LLM or framework |
| Difficulty | Intermediate |
| Time required | 30-45 min |
| Compatibility | Python 3.10+, langfuse 5+, an Anthropic API key, a free Langfuse Hobby account or self-hosted instance |
| Last updated | November 2026 |
What you'll build
You'll build observability for a sales-outreach agent that researches a lead and drafts a cold email. Without instrumentation, a bad email is a mystery: was the research tool wrong, did the model misread good research, or did a prompt change last week break something subtle? With instrumentation, you open one trace and see the exact sequence: research_lead's output, the drafting call's input and output, and the token cost of each step, nested under one run.
Langfuse's @observe decorator wraps a function and automatically records its inputs, outputs, timing, and errors, and nests it under whatever observation called it, so a tool function called from inside your agent's main loop shows up as a child of that loop's trace with no manual span management. For the model call specifically, as_type='generation' lets you attach token usage, which Langfuse turns into a cost figure using its built-in model pricing table.
Add this once an agent is doing anything you can't fully re-derive by reading the code, which in practice means almost immediately after it leaves a notebook. The cost of instrumenting upfront is a decorator per function; the cost of debugging a production incident with no traces is hours of guessing.
Prerequisites
- Python 3.10+ and a terminal
- An Anthropic API key (ANTHROPIC_API_KEY)
- pip install langfuse anthropic
- A free Langfuse Cloud Hobby account (50K units/mo) or a self-hosted instance, plus its public/secret keys
Step-by-step guide
Install Langfuse and set your keys
Install the langfuse package alongside anthropic. Export your Langfuse public and secret keys as environment variables, pulled from your Langfuse project settings; the SDK reads them automatically without you passing them into every call.
1pip install langfuse anthropic2export ANTHROPIC_API_KEY="sk-ant-..."3export LANGFUSE_PUBLIC_KEY="pk-lf-..."4export LANGFUSE_SECRET_KEY="sk-lf-..."5export LANGFUSE_HOST="https://cloud.langfuse.com" # or your self-hosted URLExpected result: Four environment variables set; no traces exist yet until the first instrumented call runs.
Define and trace the lead-research tool
Wrap research_lead with @observe(). Langfuse captures the function's input (the lead's company name) and output (the research summary) automatically, no manual logging calls inside the function body. This is the deliberately boring part: a plain function with one decorator on it.
1from langfuse import observe23LEADS = {"Acme Corp": {"industry": "logistics", "size": 1200, "recent_news": "opened a new distribution hub"}}45@observe()6def research_lead(company: str) -> dict:7 return LEADS.get(company, {"industry": "unknown", "size": 0, "recent_news": ""})Expected result: Calling research_lead('Acme Corp') directly still returns the dict as normal; the decorator only adds tracing, not behavior.
Trace the drafting model call as a generation
Mark the drafting function as_type='generation' instead of the default span, since this one represents an actual LLM call Langfuse should track cost for. After the call returns, use update_current_generation to attach token counts so Langfuse computes cost using the model's listed pricing, without you hardcoding a per-token rate anywhere.
1from anthropic import Anthropic2from langfuse import observe, get_client34client = Anthropic()5langfuse = get_client()67@observe(as_type="generation", name="draft-email")8def draft_email(research: dict) -> str:9 resp = client.messages.create(10 model="claude-sonnet-5-5",11 max_tokens=300,12 messages=[{"role": "user", "content": f"Write a 3-sentence cold email using this research: {research}"}],13 )14 langfuse.update_current_generation(15 usage_details={"input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens},16 )17 return resp.content[0].textExpected result: A Langfuse trace shows this call with a generation icon, token counts, and a computed cost in dollars, not just a duration.
Wrap both steps in one top-level traced run
The outer function doesn't need a different pattern, just another @observe(). Because research_lead and draft_email are called from inside it, Langfuse nests both as children of this trace automatically, no manual parent-span wiring required.
1@observe(name="outreach-run")2def run_outreach(company: str) -> str:3 research = research_lead(company)4 email = draft_email(research)5 return email67result = run_outreach("Acme Corp")8print(result)Expected result: One trace named outreach-run in the Langfuse UI, with research_lead and draft-email visible as nested child observations under it.
Add a cost threshold alert
After each run, read the trace's computed cost back and compare it to a threshold you set. A single run costing far more than expected usually means a retry loop, an oversized context, or a tool returning far more data than intended; catching it per-run beats noticing it in next month's invoice.
1def run_outreach_with_alert(company: str, cost_threshold_usd: float = 0.05) -> str:2 with langfuse.start_as_current_observation(as_type="span", name="outreach-run") as span:3 research = research_lead(company)4 email = draft_email(research)5 span.update(output=email)6 trace_cost = langfuse.get_current_trace_cost() if hasattr(langfuse, "get_current_trace_cost") else None7 if trace_cost and trace_cost > cost_threshold_usd:8 print(f"[ALERT] outreach run for {company} cost ${trace_cost:.4f}, over ${cost_threshold_usd} threshold")9 return emailExpected result: A run that unexpectedly calls the model twice (a retry, say) prints an alert instead of silently costing 2x with nobody noticing.
Harden it: flush on short-lived processes and tag by environment
Langfuse batches and sends traces asynchronously, which is fine for a long-running server but can drop the last batch if a short script exits immediately. Call langfuse.flush() before the process ends, and tag every trace with the environment (dev, staging, prod) so a dashboard filter doesn't mix your local testing with real customer runs.
1from langfuse import observe23@observe(name="outreach-run")4def run_outreach(company: str) -> str:5 langfuse.update_current_trace(tags=["env:production"])6 research = research_lead(company)7 return draft_email(research)89try:10 result = run_outreach("Acme Corp")11finally:12 langfuse.flush()Expected result: Every trace in the dashboard carries an env tag, and no trace from a short CLI run goes missing from the Langfuse UI.
Complete working example
1"""Sales-outreach agent instrumented with Langfuse: nested traces for2research and drafting, per-run token cost, and a cost-threshold alert.3Run: python sales_outreach_observability.py4(pip install langfuse anthropic; set ANTHROPIC_API_KEY, LANGFUSE_PUBLIC_KEY,5LANGFUSE_SECRET_KEY, LANGFUSE_HOST)6"""7from anthropic import Anthropic8from langfuse import observe, get_client910client = Anthropic()11langfuse = get_client()1213LEADS = {14 "Acme Corp": {15 "industry": "logistics",16 "size": 1200,17 "recent_news": "opened a new distribution hub",18 }19}2021COST_THRESHOLD_USD = 0.05222324@observe()25def research_lead(company: str) -> dict:26 return LEADS.get(company, {"industry": "unknown", "size": 0, "recent_news": ""})272829@observe(as_type="generation", name="draft-email")30def draft_email(research: dict) -> str:31 resp = client.messages.create(32 model="claude-sonnet-5-5",33 max_tokens=300,34 messages=[35 {36 "role": "user",37 "content": f"Write a 3-sentence cold email using this research: {research}",38 }39 ],40 )41 langfuse.update_current_generation(42 usage_details={43 "input_tokens": resp.usage.input_tokens,44 "output_tokens": resp.usage.output_tokens,45 },46 )47 return resp.content[0].text484950@observe(name="outreach-run")51def run_outreach(company: str) -> str:52 langfuse.update_current_trace(tags=["env:production"])53 research = research_lead(company)54 email = draft_email(research)55 return email565758if __name__ == "__main__":59 try:60 result = run_outreach("Acme Corp")61 print(result)62 finally:63 # Ensure the last batch of traces is sent before a short process exits64 langfuse.flush()65Common mistakes
Why it's a problem: Logging to stdout or a flat file instead of a structured tracing tool
How to avoid: Flat logs don't show parent-child relationships between a run and its tool calls. A tracing tool like Langfuse nests them automatically, which is what makes debugging a multi-step run tractable.
Why it's a problem: Forgetting to call update_current_generation with token usage
How to avoid: Without it, Langfuse records the call's duration but not its cost. Always attach usage_details on the generation-typed observation that actually calls the model.
Why it's a problem: Letting a short-lived script exit without calling langfuse.flush()
How to avoid: Langfuse batches traces asynchronously; a process that exits immediately can lose the last batch. Flush explicitly in a finally block for anything that isn't a long-running server.
Why it's a problem: Mixing development and production traces in the same unfiltered dashboard view
How to avoid: Tag every trace with its environment at the top-level observation and filter by it, or use separate Langfuse projects for dev and prod.
Why it's a problem: Only checking cost in the monthly bill instead of per run
How to avoid: A single runaway run (a retry loop, an oversized prompt) can go unnoticed for weeks in an aggregate bill. Alert on a per-run cost threshold so it surfaces the day it happens.
Best practices
- Wrap every tool call and model call with @observe so a full run is one nested trace, not scattered log lines.
- Mark actual model calls as_type='generation' and attach token usage so cost is computed, not estimated by hand.
- Tag traces by environment and, where relevant, by customer or tenant, so dashboard filters stay meaningful as volume grows.
- Alert on a per-run cost or step-count threshold, not just on an aggregate monthly total.
- Flush explicitly at the end of any short-lived process; don't assume async batching always completes before exit.
- Review traces for your worst-rated or most-complained-about outputs first; that's where observability pays off fastest.
Still stuck?
Copy one of these prompts to get a personalized, step-by-step explanation.
I have a sales-outreach agent with Langfuse @observe tracing on research_lead and draft_email, nested under an outreach-run trace, with token cost tracked on the generation (code below). Add a third traced step, send_email, and make the cost alert account for all three steps' combined cost, not just the generation. [paste complete_code]
In Claude Code, open sales_outreach_observability.py and ask: 'Add a decorator-based retry around draft_email that retries once on an API error, and make sure both attempts still show up as separate nested generations in the trace.'
Frequently asked questions
Do I need Langfuse specifically, or does any observability tool work?
The pattern (nested traces, token cost on generations, per-run alerts) applies to LangSmith, Braintrust, and Arize Phoenix too. This guide uses Langfuse because it has a generous free self-hosted option and a simple decorator API, but the concepts transfer.
Does @observe add noticeable latency to the agent?
Langfuse sends traces asynchronously in the background, so the decorator itself adds negligible latency to the wrapped function's execution. The cost is in API calls to Langfuse's ingestion endpoint, not in blocking your agent's response.
How does Langfuse compute cost without me specifying a price per token?
Langfuse maintains a model pricing table and matches it against the model name you pass to the generation-typed observation, multiplying by the input and output token counts you report through update_current_generation.
Should every tool call get its own @observe, even trivial ones?
For debugging value, yes; a trivial lookup still shows you whether it ran, what it returned, and how long it took. The overhead per decorator is small enough that the main cost is just remembering to add it to new functions.
What's the Langfuse Hobby plan's limit, and when do I need to pay?
The free Hobby tier covers 50,000 observability units per month, which is generous for most early-stage agents. Move to a paid Core or Pro plan, or self-host the open-source version for free, once volume or retention needs grow past that.
Can someone set up observability for my agent?
Yes. RapidDev instruments production agents with Langfuse, LangSmith, and custom tracing, including the cost-alerting and tagging most tutorials skip. A free 30-minute consultation is the fastest way to scope your version.
Talk to an Expert
Our team has built 1,000+ apps. Get personalized help with your project.
Book a free consultation