The Roadmap for Mastering LLMOps in 2026

# llm_with_tracing.py

# Purpose: A production-ready LLM call wrapper with full observability.

# Every call is traced in Langfuse: input, output, tokens, cost, latency.

#

# Prerequisites:

#   pip install langfuse anthropic python-dotenv

#

# Setup:

#   1. Create a free account at https://cloud.langfuse.com

#   2. Get your keys from Settings > API Keys

#   3. Create a .env file with the variables below

#

# Run:

#   python llm_with_tracing.py

 

import os

import time

from dotenv import load_dotenv

import anthropic

from langfuse import Langfuse

 

# Load environment variables from .env file

load_dotenv()

 

# Required environment variables in your .env:

# LANGFUSE_PUBLIC_KEY=pk-lf-…

# LANGFUSE_SECRET_KEY=sk-lf-…

# LANGFUSE_HOST=https://cloud.langfuse.com   (or your self-hosted URL)

# ANTHROPIC_API_KEY=sk-ant-…

 

# Initialize clients

langfuse_client = Langfuse()          # Reads keys automatically from environment

anthropic_client = anthropic.Anthropic()  # Reads ANTHROPIC_API_KEY from environment

 

# ── Configuration ─────────────────────────────────────────────────────────────

# Store your prompt here, not inline in the API call.

# This makes it versionable and testable independently.

SYSTEM_PROMPT = “”“You are a helpful customer support assistant.

Answer questions clearly and concisely.

If you do not know something, say so directly — do not guess.”“”

 

MODEL = “claude-sonnet-4-20250514”

 

# Anthropic’s pricing as of mid-2026 (update when pricing changes)

# Used to calculate cost per call for cost tracking

COST_PER_INPUT_TOKEN  = 3.00 / 1_000_000   # $3.00 per million input tokens

COST_PER_OUTPUT_TOKEN = 15.00 / 1_000_000  # $15.00 per million output tokens

 

 

def call_llm_with_tracing(

    user_message: str,

    session_id: str = “default-session”,

    user_id: str = “anonymous”

) -> str:

    “”“

    Make a traced LLM call. Every call creates a Langfuse trace with:

    – Full input and output

    – Token usage (input, output, total)

    – Calculated cost in USD

    – Latency in milliseconds

    – Model used and session context

 

    Parameters:

        user_message : The message from the user

        session_id   : Groups related calls into one conversation in Langfuse

        user_id      : Associates the call with a specific user for analytics

 

    Returns:

        The LLM response as a string

    ““”

 

    # Create a top-level trace for this user interaction

    # The trace appears in the Langfuse dashboard as one unit of work

    trace = langfuse_client.trace(

        name=“customer-support-call”,

        session_id=session_id,

        user_id=user_id,

        input={“user_message”: user_message, “system_prompt”: SYSTEM_PROMPT}

    )

 

    # Create a generation span inside the trace

    # This captures model-specific details: model name, tokens, cost

    generation = trace.generation(

        name=“claude-completion”,

        model=MODEL,

        input={

            “system”: SYSTEM_PROMPT,

            “messages”: [{“role”: “user”, “content”: user_message}]

        }

    )

 

    start_time = time.time()

 

    try:

        # Make the API call

        response = anthropic_client.messages.create(

            model=MODEL,

            max_tokens=1024,

            system=SYSTEM_PROMPT,

            messages=[{“role”: “user”, “content”: user_message}]

        )

 

        latency_ms = int((time.time() – start_time) * 1000)

 

        # Extract the response text

        response_text = response.content[0].text

 

        # Extract token usage from the response

        input_tokens  = response.usage.input_tokens

        output_tokens = response.usage.output_tokens

        total_tokens  = input_tokens + output_tokens

 

        # Calculate cost for this call

        cost_usd = (

            input_tokens  * COST_PER_INPUT_TOKEN +

            output_tokens * COST_PER_OUTPUT_TOKEN

        )

 

        # Update the generation span with results

        # This data populates the Langfuse cost and token dashboards

        generation.end(

            output=response_text,

            usage={

                “input”:  input_tokens,

                “output”: output_tokens,

                “total”:  total_tokens,

                “unit”:   “TOKENS”

            },

            metadata={

                “latency_ms”: latency_ms,

                “cost_usd”:   round(cost_usd, 6),

                “model”:      MODEL

            }

        )

 

        # Update the trace with the final output

        trace.update(

            output={“response”: response_text},

            metadata={“total_cost_usd”: round(cost_usd, 6)}

        )

 

        # Print a summary to stdout for local visibility

        print(f“\n{‘─’ * 60}”)

        print(f“User:    {user_message}”)

        print(f“Claude:  {response_text}”)

        print(f“Tokens:  {input_tokens} in / {output_tokens} out / {total_tokens} total”)

        print(f“Cost:    ${cost_usd:.6f}”)

        print(f“Latency: {latency_ms}ms”)

        print(f“Trace:   {langfuse_client.base_url}/trace/{trace.id}”)

        print(f“{‘─’ * 60}\n”)

 

        return response_text

 

    except Exception as e:

        # Record the error in the trace so it shows up in Langfuse

        generation.end(

            output=None,

            metadata={“error”: str(e), “latency_ms”: int((time.time() – start_time) * 1000)}

        )

        trace.update(output={“error”: str(e)})

 

        # Always flush before raising — ensures the error trace is sent

        langfuse_client.flush()

        raise

 

    finally:

        # Flush sends all buffered events to Langfuse

        # In a long-running service, Langfuse flushes automatically.

        # In a script, you must flush manually before the process exits.

        langfuse_client.flush()

 

 

# ── Run a demonstration ────────────────────────────────────────────────────────

if __name__ == “__main__”:

    # Simulate two turns of a customer support conversation

    test_messages = [

        “What is your return policy for electronics?”,

        “Can I return an item I bought 45 days ago?”

    ]

 

    session = “demo-session-001”

 

    for i, message in enumerate(test_messages):

        print(f“\nCall {i + 1}/{len(test_messages)}”)

        try:

            call_llm_with_tracing(

                user_message=message,

                session_id=session,

                user_id=“test-user-42”

            )

        except Exception as e:

            print(f“Error on call {i + 1}: {e}”)