Langfuse Custom Tracer

What is this library?

When you build applications with AI models, you quickly need answers to questions like: how many tokens did that cost? which step is the slowest? which user is spending the most? This library answers all of those — automatically — by integrating with Langfuse, an open-source observability platform for LLM apps.

Installation

Install with pip. Pick the extras that match the AI providers you use. If you are unsure, install everything with [all].

pip install langfuse-custom-tracer

pip install langfuse-custom-tracer[env]

# Google Gemini only
pip install langfuse-custom-tracer[gemini]

# Anthropic Claude only
pip install langfuse-custom-tracer[anthropic]

# Everything — recommended for most projects
pip install langfuse-custom-tracer[all]

Quick start

Get up and running in three steps. You will need API keys from Langfuse and at least one AI provider.

Step 1 — Get your API keys

Step 2 — Create a .env file

In your project root, create a file named .env and add your keys:

# .env — never commit this file to version control
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_PUBLIC_KEY=pk-lf-...

GEMINI_API_KEY=...
ANTHROPIC_API_KEY=...

Step 3 — Run your first traced call

from langfuse_custom_tracer import load_env, observe
import os

load_env()   # reads your .env file
observe()    # turns on automatic tracing — do this before importing AI SDKs

import google.generativeai as genai
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.0-flash")

# This call is now automatically traced and sent to Langfuse
response = model.generate_content("Summarise the history of Python in 3 sentences.")
print(response.text)

Auto tracing

The recommended approach for most applications. Call observe() once at startup and every subsequent AI call is recorded automatically — no wrapping, no boilerplate.

What gets recorded on every call

Tag calls with a user ID

Call set_user() at the start of each request to tag all subsequent AI calls with that user's ID. Langfuse aggregates cost and usage per user automatically.

from langfuse_custom_tracer import observe, set_user

observe()

# In a web app, set the user at the start of each request
def handle_chat_request(user_id: str, message: str):
    set_user(user_id)

    # All AI calls below are now tagged to user_id in Langfuse
    response = model.generate_content(message)
    return response.text

Group calls into sessions

A session groups multiple traces together — useful for multi-turn conversations. All messages in one conversation appear together on the Langfuse Sessions tab.

from langfuse_custom_tracer import observe, set_user, set_session, end_session

observe()

set_user("user-123")
session_id = set_session()   # auto-generates a UUID; returns it so you can store it

response1 = model.generate_content("Hello")
response2 = model.generate_content("Tell me more")
response3 = model.generate_content("Thanks, that is helpful")

end_session()   # clears the session so the next conversation starts fresh

Attach quality scores

After a call completes, you can attach a score — useful for recording user feedback (thumbs up/down), LLM-as-a-judge evaluations, or custom metrics.

from langfuse_custom_tracer import observe, get_trace_id, score

observe()

response = model.generate_content("Explain recursion")
trace_id = get_trace_id()   # capture the ID immediately after the call

# Later — for example when the user clicks thumbs up
score("thumbs_up",    1.0,  trace_id=trace_id, comment="Very clear explanation")
score("relevance",    0.95, trace_id=trace_id, data_type="NUMERIC")
score("hallucination", False, trace_id=trace_id, data_type="BOOLEAN")

Async support

Auto tracing works with asyncio out of the box. Each asyncio task has its own isolated context via Python's contextvars, so concurrent requests never mix up each other's user IDs or session IDs.

import asyncio
from langfuse_custom_tracer import observe, set_user

observe()

async def process_user(user_id: str, messages: list):
    set_user(user_id)   # ContextVar — isolated per asyncio task
    tasks = [model.generate_content_async(m) for m in messages]
    return await asyncio.gather(*tasks)

# Both users are processed concurrently without interfering with each other
asyncio.run(asyncio.gather(
    process_user("user-1", ["msg1", "msg2"]),
    process_user("user-2", ["msg3", "msg4"]),
))

Manual tracing

When you want full control — custom names, explicit inputs and outputs, metadata — use the context manager API directly. This is particularly useful for labelling multi-step pipelines so each step is clearly visible in Langfuse.

With Google Gemini

from langfuse_custom_tracer import load_env, create_langfuse_client, GeminiTracer
import google.generativeai as genai
import os

load_env()

lf     = create_langfuse_client(os.getenv("LANGFUSE_SECRET_KEY"), os.getenv("LANGFUSE_PUBLIC_KEY"))
tracer = GeminiTracer(lf)

genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.0-flash")

with tracer.trace("invoice-processing", input={"file": "invoice.pdf"}) as span:
    with tracer.generation("extract-data", model="gemini-2.0-flash",
                          input="Extract name, amount, date") as gen:
        response = model.generate_content("Extract name, amount, date from this invoice: ...")
        usage    = tracer.extract_usage(response, model="gemini-2.0-flash")
        gen.update(output=response.text, usage_details=usage)
    span.update(output="Extraction complete")

tracer.flush()   # required at end of script

With Anthropic Claude

from langfuse_custom_tracer import load_env, create_langfuse_client, AnthropicTracer
from anthropic import Anthropic
import os

load_env()

lf     = create_langfuse_client(os.getenv("LANGFUSE_SECRET_KEY"), os.getenv("LANGFUSE_PUBLIC_KEY"))
tracer = AnthropicTracer(lf)
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

with tracer.trace("invoice-processing", input={"file": "invoice.pdf"}) as span:
    with tracer.generation("extract-data", model="claude-3-5-sonnet-20241022") as gen:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            messages=[{"role": "user", "content": "Extract name, amount, date from this invoice: ..."}]
        )
        usage = tracer.extract_usage(response, model="claude-3-5-sonnet-20241022")
        gen.update(output=response.content[0].text, usage_details=usage)
    span.update(output="Extraction complete")

tracer.flush()

Multi-step pipelines

Traces can be nested to any depth. Each tracer.trace() block inside another becomes a child span. Langfuse uses OpenTelemetry context propagation to detect the nesting automatically — you do not need to pass parent IDs manually.

with tracer.trace("document-processing", user_id="user-123",
                 metadata={"doc_type": "invoice"}) as root:

    with tracer.trace("step-1-ocr"):
        with tracer.generation("extract-text", model="gemini-2.0-flash-lite") as gen:
            text = model.generate_content("Extract all text from this image.")
            gen.update(output=text.text, usage_details=tracer.extract_usage(text, model="gemini-2.0-flash-lite"))

    with tracer.trace("step-2-classify"):
        with tracer.generation("classify-doc", model="gemini-2.0-flash") as gen:
            classification = model.generate_content(f"Classify this document: {text.text}")
            gen.update(output=classification.text, usage_details=tracer.extract_usage(classification, model="gemini-2.0-flash"))

    with tracer.trace("step-3-extract-fields"):
        with tracer.generation("get-fields", model="gemini-2.0-flash") as gen:
            fields = model.generate_content(f"Extract name, amount, date from: {text.text}")
            gen.update(output=fields.text, usage_details=tracer.extract_usage(fields, model="gemini-2.0-flash"))

    root.update(output="All steps complete")

tracer.flush()

In Langfuse you will see a clean tree: the root trace contains three child spans, each with its own latency, token count, and cost. The root trace also shows the combined totals.

Error handling

Wrap AI calls in a try/except inside the generation block. Even if the call fails, the trace is sent to Langfuse with the error attached — so you can track failure rates over time.

with tracer.trace("risky-task"):
    with tracer.generation("ai-call", model="gemini-2.0-flash") as gen:
        try:
            response = model.generate_content("...")
            usage    = tracer.extract_usage(response, model="gemini-2.0-flash")
            gen.update(output=response.text, usage_details=usage)
        except Exception as e:
            gen.update(status_code=500, error=str(e))
            raise   # re-raise so your application can handle it

Langfuse dashboard

After running your code, open cloud.langfuse.com. Each traced call appears in the Traces list. Here is what a typical trace looks like:

API reference

lf = create_langfuse_client(
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    host="https://us.cloud.langfuse.com"  # optional — US region
)

Auto tracing API

These functions are only available when using observe() mode.

Dynamic pricing

Pricing is decoupled from source code and managed via a remote JSON file. When AI providers change their prices, your cost numbers update automatically — no new library version required.

How it works

Use a custom pricing URL

Point the library at your own pricing file by setting an environment variable or passing a URL in code:

# Option 1 — environment variable
PRICING_JSON_URL=https://your-domain.com/pricing.json

# Option 2 — in code
from langfuse_custom_tracer import get_pricing_manager

pm = get_pricing_manager(url="https://your-domain.com/pricing.json")
price, version, source = pm.get_price("gemini-2.0-flash")
print(f"Input: ${price['input']} per 1M tokens (v{version}, from {source})")

Supported models

All prices are per 1 million tokens (Q1 2026). Prices update automatically via the remote JSON.

Google Gemini

Anthropic Claude

Supported SDKs

Project structure

Testing

81 unit tests with 64% overall coverage. The full suite runs in approximately 1 second.

# run the full test suite
pytest

# run with a coverage report
pytest --cov

# run a single test
pytest tests/test_gemini_tracer.py::TestGeminiTracer::test_extract_usage_basic -v

# run by module
pytest tests/test_anthropic_tracer.py -v
pytest tests/test_pricing_manager.py -v

Security