:::tip 🎮 Interactive Playground Visualize this concept: Try the MLflow Experiment Tracking demo on the EngineersOfAI Playground - no code required. :::

LLMOps Platforms

The Three Months That Built Nothing

In early 2024, a twelve-person AI product team was building an enterprise coding assistant. They had moved fast - six months from idea to five hundred paying users. What they had not built was any LLMOps infrastructure: no tracing, no eval platform, no prompt registry. Every prompt was a hardcoded string. Every quality assessment was "does it look right in the demo?"

They recognized the problem and initiated a platform evaluation. They shortlisted five tools. Each engineer had opinions rooted in familiarity, not evidence. The LangChain users wanted LangSmith. The ML team wanted W&B Weave because they were already using W&B for experiment tracking. The compliance lead wanted Langfuse because it was self-hostable. Someone had read a blog post about Helicone and thought zero-code-change tracing sounded compelling. They built sample integrations, wrote comparison documents, held async debates that generated two hundred Slack messages.

Three months passed. They shipped no LLMOps capabilities. The platform evaluation became a blocking artifact - analysis paralysis in which the search for the optimal solution prevented them from having any solution. Meanwhile, their production system was running completely blind: they had no visibility into which prompts were causing failures, no cost attribution by feature, and no quality trend data. Users were churning, and they could not diagnose why.

The resolution was undramatic. The CTO set a two-week deadline. They picked Langfuse because self-hosting satisfied the compliance team. They integrated it in three days. They had production tracing running by day five. The insight was not about which platform to choose - any of the five would have served them adequately at their scale. The insight was that the cost of choosing nothing vastly exceeds the cost difference between the options. The best platform is the one that is running in production.

Why This Ecosystem Exists

Generic observability tools - Datadog, Grafana, Sentry, New Relic - were built for deterministic software. They record HTTP request latency, error rates, database query times, and exception stack traces. They have no model of what an LLM application does:

• What exact prompt was sent to which model at which version
• How many input and output tokens were consumed, and at what per-token cost
• What the quality of the output was, according to any rubric
• How to search production calls by prompt version or A/B experiment arm
• How to run automated evals against a golden dataset and attach scores to traces

The LLMOps platform ecosystem emerged in 2023 specifically to fill this gap. By 2024, a half-dozen serious platforms had emerged with overlapping but distinct feature sets. They provide varying combinations of four capabilities: tracing (capture and store every LLM call with full context), evaluation (score output quality on traces or datasets), prompt management (version and serve prompts from a registry), and cost tracking (token usage and spend attributed by model, feature, user, and experiment).

The LLMOps Platform Landscape

Platform Comparison Matrix

Feature	LangSmith	Langfuse	W&B Weave	Arize Phoenix	Helicone	PromptLayer
Tracing	Yes	Yes	Yes	Yes	Yes	Yes
Eval framework	Yes (rich)	Yes	Yes	Yes	No	Basic
Prompt registry	Yes (Hub)	Yes	No	No	No	Yes
Cost tracking	Yes	Yes	Yes	Yes	Yes	Yes
Self-host	No	Yes (Docker/K8s)	No	Yes (local)	No	No
Open source	No	Yes (MIT)	No	Yes (Apache)	No	No
Free tier	Yes (limited)	Yes (cloud)	Yes	Yes	Yes	Yes
LangChain native	Yes (zero config)	Partial	No	No	No	No
OTEL support	Partial	Yes	No	Yes (native)	No	No
Best for	LangChain teams	Self-host/compliance	ML exp tracking	Local debugging	Zero-code tracing	Prompt versioning
Pricing model	Per trace	Per event	Per seat/trace	Free (cloud pricing)	Per request	Per request

LangSmith

LangSmith is the production observability and evaluation platform from LangChain, Inc. If your team uses LangChain or LangGraph, LangSmith is the path of least resistance - LANGCHAIN_TRACING_V2=true in your environment and every LangChain call is traced automatically. If you do not use LangChain, it works with any Python code through its SDK.

What LangSmith Does Best

• Trace search: filter production traces by model, latency bucket, cost, tag, feedback score, and user in a powerful web UI with full-text search over prompt content
• Dataset management: sample traces from production into a golden eval dataset with two clicks in the UI - the fastest path from "production failure" to "eval case"
• Online evaluation: configure automated LLM judges that run on every new production trace matching a filter, then track score distributions over time
• Prompt Hub: publish prompt versions to a registry, pull them by name from any codebase, and link specific prompt versions to specific eval runs
• LangChain/LangGraph integration: LANGCHAIN_TRACING_V2=true is all you need - the full chain, every retrieval, every LLM call, every tool call, structured as a tree

LangSmith Integration

# langsmith_integration.py
"""
LangSmith integration with the Anthropic SDK.
Three patterns: wrapped client, @traceable decorator, manual SDK.
"""
import anthropic
import os
from typing import Optional
from langsmith import Client, traceable
from langsmith.wrappers import wrap_anthropic

# --- Setup ---
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY", "")
os.environ["LANGCHAIN_PROJECT"] = "coding-assistant-prod"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

# -----------------------------------------------------------------------
# Pattern 1: Wrapped Anthropic Client - zero code change per call site
# -----------------------------------------------------------------------
# Every call to wrapped_client.messages.create() is auto-traced in LangSmith
# with the full prompt, response, token counts, latency, and model version.
wrapped_client = wrap_anthropic(anthropic.Anthropic())


def generate_code_suggestion(code_context: str, instruction: str) -> str:
    """Every call here is automatically traced in LangSmith."""
    response = wrapped_client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system=(
            "You are an expert software engineer. "
            "Generate clean, well-commented code based on the instruction."
        ),
        messages=[
            {
                "role": "user",
                "content": f"Code context:\n```\n{code_context}\n```\n\nInstruction: {instruction}",
            }
        ],
    )
    return response.content[0].text


# -----------------------------------------------------------------------
# Pattern 2: @traceable Decorator - parent trace with named child spans
# -----------------------------------------------------------------------
@traceable(
    name="coding-assistant-pipeline",
    tags=["production", "code-gen"],
    metadata={"prompt_version": "v2.3"},
)
def coding_assistant_pipeline(
    user_message: str,
    user_id: str,
    file_path: Optional[str] = None,
) -> dict:
    """
    Traced as a root span. Nested @traceable calls become child spans.
    LangSmith displays the full tree: root → retrieval → LLM call.
    """
    # Retrieve code context (child span)
    context = retrieve_code_context(file_path or "")

    # Call LLM (child span)
    response = call_llm_for_code(user_message, context)

    # Post-process
    result = {
        "suggestion": response,
        "context_file": file_path,
        "context_length": len(context),
    }
    return result


@traceable(name="code-context-retrieval", run_type="retrieval")
def retrieve_code_context(file_path: str) -> str:
    """Simulated retrieval - traced as a retrieval span in LangSmith."""
    # In production: read file content, query vector DB, etc.
    return f"# Existing code from {file_path}\ndef placeholder(): pass"


@traceable(name="llm-code-generation", run_type="llm")
def call_llm_for_code(instruction: str, context: str) -> str:
    """LLM call - traced as an LLM span."""
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="You are a senior software engineer. Write clean, production-ready code.",
        messages=[
            {
                "role": "user",
                "content": f"Context:\n```python\n{context}\n```\n\nTask: {instruction}",
            }
        ],
    )
    return response.content[0].text


# -----------------------------------------------------------------------
# Pattern 3: Manual SDK - full control over trace structure
# -----------------------------------------------------------------------
def run_with_manual_trace(user_message: str, user_id: str) -> str:
    """
    Manual tracing gives precise control over span attributes,
    error handling, and feedback attachment.
    """
    ls_client = Client()
    run_id = None

    try:
        run = ls_client.create_run(
            name="code-suggestion-manual",
            run_type="chain",
            inputs={"message": user_message, "user_id": user_id},
            tags=["production", "manual-trace"],
            metadata={"prompt_version": "v2.3", "feature": "inline-completion"},
        )
        run_id = run.id

        client = anthropic.Anthropic()
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system="You are a senior software engineer.",
            messages=[{"role": "user", "content": user_message}],
        )
        output = response.content[0].text

        ls_client.update_run(
            run_id=run_id,
            outputs={"response": output},
            extra={
                "usage": {
                    "input_tokens": response.usage.input_tokens,
                    "output_tokens": response.usage.output_tokens,
                }
            },
        )
        return output

    except Exception as e:
        if run_id:
            ls_client.update_run(run_id=run_id, error=str(e))
        raise


# -----------------------------------------------------------------------
# LangSmith Eval: Run against a golden dataset
# -----------------------------------------------------------------------
def run_langsmith_eval():
    """
    Build a dataset from examples, run the pipeline against it,
    evaluate results with an LLM judge, compare experiment runs.
    """
    from langsmith.evaluation import evaluate, LangChainStringEvaluator

    ls_client = Client()

    # Create or retrieve a named dataset
    try:
        dataset = ls_client.create_dataset(
            "code-gen-eval-v1",
            description="Golden eval set for inline code generation",
        )
        ls_client.create_examples(
            inputs=[
                {"message": "Write a function to compute Fibonacci numbers"},
                {"message": "Write a binary search implementation"},
                {"message": "Parse a JSON config file with error handling"},
            ],
            outputs=[
                {"expected_contains": "def fibonacci"},
                {"expected_contains": "def binary_search"},
                {"expected_contains": "json.load"},
            ],
            dataset_id=dataset.id,
        )
    except Exception:
        # Dataset already exists - load it
        dataset = ls_client.read_dataset(dataset_name="code-gen-eval-v1")

    # Target function: takes dataset input, returns output
    def app_under_test(inputs: dict) -> dict:
        code = generate_code_suggestion("", inputs["message"])
        return {"response": code}

    # Run eval with LangSmith's built-in CoT QA judge
    results = evaluate(
        app_under_test,
        data="code-gen-eval-v1",
        evaluators=[LangChainStringEvaluator("cot_qa")],
        experiment_prefix="code-gen-v2.3",
        num_repetitions=1,
    )

    # Print summary
    summary = results.get_aggregate_feedback()
    print(f"Eval complete. Summary: {summary}")
    return results

:::tip LangSmith's Killer Feature The ability to click any production trace and add it to an eval dataset in two clicks is uniquely powerful. When a user reports a bad output, you find the trace, click "Add to dataset," and your eval set automatically improves. Over time, your eval set becomes a real representation of what users actually ask. :::

Langfuse

Langfuse is the leading open-source LLMOps platform. It is self-hostable via Docker or Kubernetes, deeply instrumented, and provides a Python SDK with decorator-based tracing that feels nearly invisible in your code. For teams with data residency or compliance requirements - HIPAA, GDPR, SOC2 - Langfuse is often the only viable option without building a custom solution.

What Langfuse Does Best

• Self-hosting: full deployment on your own infrastructure with a single docker-compose up - your LLM prompts and responses never leave your network
• Decorator-based tracing: @observe() wraps any function as a Langfuse trace, with nested @observe() calls automatically becoming child spans, mirroring your function call graph
• Trace hierarchy: parent traces, child spans, generation spans - the structure maps directly to your code's call hierarchy without configuration
• Prompt management: full CRUD API for prompt versions with labels (development, staging, production), variable templating, and pull-by-name at runtime with a local 60-second cache
• Score API: attach quality scores from any external evaluator to any trace - LLM judge, human review, task-specific metrics - all queryable in the UI

Langfuse Integration

# langfuse_integration.py
"""
Langfuse integration with the Anthropic SDK.
Four patterns: decorator, low-level SDK, prompt management, batch scoring.
"""
import anthropic
import os
from dataclasses import dataclass
from typing import Optional
from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe

# Initialize Langfuse (reads LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY from env)
# For self-hosted: set LANGFUSE_HOST=https://your-langfuse.internal.example.com
langfuse = Langfuse()


# -----------------------------------------------------------------------
# Pattern 1: Decorator-based tracing - recommended for most teams
# -----------------------------------------------------------------------
@observe()
def customer_support_pipeline(user_message: str, user_id: str) -> str:
    """
    @observe() creates a root Langfuse trace for this function.
    Nested @observe() calls create child spans automatically - no IDs to pass.
    The trace name defaults to the function name: "customer_support_pipeline".
    """
    # Attach user context to the root trace
    langfuse_context.update_current_trace(
        user_id=user_id,
        session_id=f"session-{user_id}",
        tags=["production", "support-v3"],
        metadata={"channel": "web-chat"},
    )

    context = retrieve_knowledge_base(user_message)
    response = generate_support_response(user_message, context)

    return response


@observe(name="knowledge-base-retrieval")
def retrieve_knowledge_base(query: str) -> str:
    """Traced as a retrieval span. Attach retrieval-specific metadata."""
    langfuse_context.update_current_observation(
        input=query,
        metadata={
            "retriever": "faiss-v2",
            "top_k": 5,
            "index_version": "2024-11-01",
        },
    )
    # Simulated vector DB query
    return "Policy: Refunds are processed within 5-7 business days..."


@observe(name="llm-response-generation", as_type="generation")
def generate_support_response(user_message: str, context: str) -> str:
    """
    as_type="generation" marks this as an LLM generation span.
    Langfuse automatically tracks token usage and inferred cost for known models.
    """
    client = anthropic.Anthropic()

    system = (
        "You are a helpful customer support agent. "
        "Answer based only on the provided context. "
        "Be concise and empathetic."
    )
    messages = [
        {
            "role": "user",
            "content": f"Context:\n{context}\n\nCustomer question: {user_message}",
        }
    ]

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system=system,
        messages=messages,
    )
    output = response.content[0].text

    # Attach the complete LLM call context to the generation span
    langfuse_context.update_current_observation(
        model="claude-opus-4-6",
        input={"system": system, "messages": messages},
        output=output,
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens,
            "unit": "TOKENS",
        },
        metadata={"prompt_version": "support-v3.2"},
    )
    return output


# -----------------------------------------------------------------------
# Pattern 2: Low-level SDK - maximum control over span structure
# -----------------------------------------------------------------------
def run_with_manual_langfuse_trace(user_message: str, user_id: str) -> str:
    """
    Manual SDK gives precise control over span hierarchy, attributes,
    and allows attaching scores immediately after generation.
    """
    client = anthropic.Anthropic()

    trace = langfuse.trace(
        name="support-request-manual",
        user_id=user_id,
        session_id=f"session-{user_id}-manual",
        tags=["production", "manual"],
        metadata={"feature": "email-support", "prompt_version": "support-v3.2"},
    )

    retrieval_span = trace.span(
        name="context-retrieval",
        input={"query": user_message},
        metadata={"retriever": "faiss-v2"},
    )
    retrieved_context = "Relevant policy content..."
    retrieval_span.end(
        output={"context": retrieved_context, "num_docs": 3}
    )

    generation = trace.generation(
        name="claude-response",
        model="claude-opus-4-6",
        model_parameters={"max_tokens": 512},
        input={"messages": [{"role": "user", "content": user_message}]},
    )

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system="You are a helpful support agent.",
        messages=[{"role": "user", "content": user_message}],
    )
    output = response.content[0].text

    generation.end(
        output=output,
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens,
        },
    )

    # Attach automated quality score immediately
    langfuse.score(
        trace_id=trace.id,
        name="auto-quality-score",
        value=_score_response_quality(output),
        data_type="NUMERIC",
        comment="Scored by LLM judge - claude-haiku-4-5",
    )

    return output


def _score_response_quality(response: str) -> float:
    """Score a response using a cheap, fast LLM judge."""
    judge = anthropic.Anthropic()
    result = judge.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=16,
        system=(
            "You are a quality evaluator. "
            "Score the response from 0.0 to 1.0. "
            "Output ONLY the numeric score, e.g. 0.87"
        ),
        messages=[
            {
                "role": "user",
                "content": f"Response to evaluate:\n{response}\n\nScore (0.0-1.0):",
            }
        ],
    )
    try:
        return float(result.content[0].text.strip())
    except ValueError:
        return 0.5


# -----------------------------------------------------------------------
# Pattern 3: Prompt Management - pull versioned prompts at runtime
# -----------------------------------------------------------------------
def using_langfuse_prompt_registry(user_message: str) -> str:
    """
    Pull the current production prompt from Langfuse at runtime.
    Changes made in the Langfuse UI take effect on the next request
    without a code deploy - the SDK caches for 60 seconds by default.
    """
    client = anthropic.Anthropic()

    # Pull the prompt labeled "production" - cached for 60 seconds
    prompt = langfuse.get_prompt("customer-support-v3", label="production")

    # Compile the template with runtime variables
    compiled = prompt.compile(
        customer_message=user_message,
        current_date="2026-03-15",
        company_name="Acme Corp",
    )

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": compiled}],
    )

    return response.content[0].text


# -----------------------------------------------------------------------
# Pattern 4: Batch scoring of production traces
# -----------------------------------------------------------------------
def score_production_traces_batch(limit: int = 100) -> dict:
    """
    Pull recent production traces, run the LLM judge on each,
    and attach scores. Run this as a nightly job.
    """
    traces = langfuse.fetch_traces(
        limit=limit,
        tags=["production"],
        order_by="timestamp",
    )

    scored = 0
    low_quality = 0
    total_score = 0.0

    for trace in traces.data:
        # Skip already-scored traces
        if any(s.name == "auto-quality-score" for s in (trace.scores or [])):
            continue

        full_trace = langfuse.fetch_trace(trace.id)
        # Extract the LLM output from the generation span
        output = _extract_output_from_trace(full_trace)
        if not output:
            continue

        score = _score_response_quality(output)
        langfuse.score(
            trace_id=trace.id,
            name="auto-quality-score",
            value=score,
            data_type="NUMERIC",
        )

        total_score += score
        scored += 1
        if score < 0.7:
            low_quality += 1

    avg_score = total_score / scored if scored > 0 else 0.0
    return {
        "scored": scored,
        "low_quality": low_quality,
        "avg_score": round(avg_score, 3),
        "alert": avg_score < 0.75,
    }


def _extract_output_from_trace(trace) -> Optional[str]:
    """Extract the LLM response text from a Langfuse trace object."""
    for obs in getattr(trace, "observations", []):
        if getattr(obs, "type", "") == "GENERATION" and obs.output:
            if isinstance(obs.output, str):
                return obs.output
            if isinstance(obs.output, dict):
                return obs.output.get("text", "")
    return None

:::warning Langfuse Self-Host Gotcha When self-hosting Langfuse with Docker, set NEXTAUTH_SECRET and SALT to stable random strings before the first run. If you reset these values, all existing sessions are invalidated and all stored prompt labels become inaccessible until you re-apply the old values. Store these secrets in your secrets manager before first launch. :::

W&B Weave

Weights & Biases Weave is W&B's LLM-focused observability layer, integrated into the broader W&B ML experiment tracking platform. It is the natural choice for teams already using W&B for model training and hyperparameter sweeps, because it collapses LLM tracing, dataset versioning, and eval results into the same UI where the team already lives.

What W&B Weave Does Best

• Experiment tracking integration: link LLM eval runs to the training runs that produced the model - unified view across the full ML pipeline
• Dataset versioning: version your eval datasets with W&B Artifacts - exact reproducibility across experiment runs
• @weave.op() simplicity: the simplest decorator API of any platform - two lines of setup, one decorator, and everything is traced
• Eval framework: weave.Evaluation runs your scoring functions across a dataset, stores results, and generates comparison tables across experiment runs

W&B Weave Integration

# wandb_weave_integration.py
"""
W&B Weave integration with the Anthropic SDK.
Best for teams already using W&B for ML experiment tracking.
"""
import anthropic
import asyncio
import weave
from dataclasses import dataclass
from typing import Optional

# One-line initialization - links this session to a W&B project
weave.init("engineersofai/coding-assistant")


# -----------------------------------------------------------------------
# Pattern 1: @weave.op() - the simplest possible tracing
# -----------------------------------------------------------------------
@weave.op()
def generate_code_review(code: str, language: str) -> str:
    """
    @weave.op() traces inputs, outputs, latency, and exceptions.
    No other instrumentation needed for basic tracing.
    """
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system=(
            f"You are an expert {language} code reviewer. "
            "Identify bugs, security issues, and style improvements. "
            "Be specific and actionable."
        ),
        messages=[
            {
                "role": "user",
                "content": f"Review this {language} code:\n```{language}\n{code}\n```",
            }
        ],
    )
    return response.content[0].text


@weave.op()
def code_review_pipeline(code: str, language: str = "python") -> dict:
    """Composable pipeline - each @weave.op() call appears as a child span."""
    review = generate_code_review(code, language)
    return {
        "review": review,
        "language": language,
        "code_length": len(code.split("\n")),
    }


# -----------------------------------------------------------------------
# Pattern 2: Weave Eval Framework - dataset + scoring functions
# -----------------------------------------------------------------------
@dataclass
class CodeReviewExample:
    code: str
    language: str
    expected_issues: list[str]  # Issues the review must mention


EVAL_DATASET = [
    CodeReviewExample(
        code="def divide(a, b):\n    return a / b",
        language="python",
        expected_issues=["ZeroDivisionError", "division by zero", "zero"],
    ),
    CodeReviewExample(
        code="import subprocess\ncmd = input('Enter command: ')\nsubprocess.run(cmd, shell=True)",
        language="python",
        expected_issues=["injection", "security", "shell=True", "unsafe"],
    ),
    CodeReviewExample(
        code="password = 'hardcoded_secret_123'\ndb.connect(password=password)",
        language="python",
        expected_issues=["hardcoded", "secret", "credential", "environment"],
    ),
]


@weave.op()
def score_review_completeness(
    code: str,
    language: str,
    expected_issues: list[str],
    model_output: dict,
) -> dict:
    """
    Weave scorer - returns a dict of metric name → value.
    Weave automatically aggregates these across the eval dataset.
    """
    review_text = model_output.get("review", "").lower()
    issues_mentioned = [
        issue for issue in expected_issues
        if any(kw in review_text for kw in issue.lower().split())
    ]
    completeness = len(issues_mentioned) / len(expected_issues)
    return {
        "completeness": completeness,
        "issues_found": len(issues_mentioned),
        "issues_expected": len(expected_issues),
        "all_issues_caught": completeness == 1.0,
    }


@weave.op()
def score_review_length(
    code: str,
    language: str,
    expected_issues: list[str],
    model_output: dict,
) -> dict:
    """Second scorer - ensures reviews are substantive but not bloated."""
    review = model_output.get("review", "")
    word_count = len(review.split())
    return {
        "word_count": word_count,
        "length_appropriate": 50 < word_count < 500,
    }


def run_weave_eval():
    """Run the full eval suite - results appear in the W&B UI."""
    # Convert dataclasses to dicts for Weave
    dataset = [
        {
            "code": ex.code,
            "language": ex.language,
            "expected_issues": ex.expected_issues,
        }
        for ex in EVAL_DATASET
    ]

    evaluation = weave.Evaluation(
        name="code-review-eval-v1",
        dataset=dataset,
        scorers=[score_review_completeness, score_review_length],
    )

    # Run evaluation (async)
    results = asyncio.run(evaluation.evaluate(code_review_pipeline))
    print(f"Eval results: {results}")
    return results


# -----------------------------------------------------------------------
# Pattern 3: Custom attributes on ops
# -----------------------------------------------------------------------
@weave.op()
def generate_with_metadata(prompt: str, prompt_version: str) -> dict:
    """Weave stores all input dict fields - include metadata as inputs."""
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    return {
        "response": response.content[0].text,
        "prompt_version": prompt_version,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
    }

Arize Phoenix

Arize Phoenix is an open-source, locally-runnable LLM observability tool built on OpenTelemetry. Unlike cloud-first platforms, Phoenix runs in your browser as a local server and is ideal for debugging during development before you have a production observability stack. It also runs as a deployed server for production use.

What Phoenix Does Best

• Zero latency overhead in production: OTEL spans are written asynchronously to a collector - no synchronous SDK call on the critical path
• Local debugging: px.launch_app() opens a full trace explorer in your browser during development
• OTEL native: Phoenix uses OpenTelemetry as its trace protocol, which means it integrates with any OTEL-compatible collector or backend
• Anthropic auto-instrumentation: AnthropicInstrumentor() automatically traces every Anthropic SDK call without decorator changes

Arize Phoenix Integration

# arize_phoenix_integration.py
"""
Arize Phoenix integration with the Anthropic SDK via OpenTelemetry.
Two modes: local development (px.launch_app()) and production server.
"""
import anthropic
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# -----------------------------------------------------------------------
# Setup: Development mode (local Phoenix server)
# -----------------------------------------------------------------------
def setup_phoenix_dev():
    """Start a local Phoenix server and configure OTEL to send to it."""
    import phoenix as px
    from phoenix.otel import register
    from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

    # Start Phoenix in your browser at http://localhost:6006
    px.launch_app()

    tracer_provider = register(
        project_name="coding-assistant-dev",
        endpoint="http://localhost:4317",
    )

    # Auto-instrument the Anthropic SDK - every client.messages.create()
    # call becomes an OTEL span automatically. No decorator needed.
    AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)

    return trace.get_tracer("coding-assistant")


# -----------------------------------------------------------------------
# Setup: Production mode (Phoenix server or OTEL collector)
# -----------------------------------------------------------------------
def setup_phoenix_prod():
    """Configure OTEL for production Phoenix deployment."""
    from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

    phoenix_endpoint = os.getenv(
        "PHOENIX_COLLECTOR_ENDPOINT",
        "http://your-phoenix-server:4317",
    )

    tracer_provider = TracerProvider()
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(endpoint=phoenix_endpoint)
        )
    )
    trace.set_tracer_provider(tracer_provider)

    AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)
    return trace.get_tracer("coding-assistant")


# -----------------------------------------------------------------------
# Application code - unchanged from non-instrumented version
# -----------------------------------------------------------------------
def run_coding_assistant(user_request: str, file_context: str) -> str:
    """
    After AnthropicInstrumentor().instrument() is called in setup,
    this function needs ZERO additional instrumentation.
    Every client.messages.create() call is automatically traced
    with full prompt, response, token counts, and latency.
    """
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="You are an expert software engineer.",
        messages=[
            {
                "role": "user",
                "content": f"File context:\n```python\n{file_context}\n```\n\nRequest: {user_request}",
            }
        ],
    )
    return response.content[0].text


# -----------------------------------------------------------------------
# Custom spans for pipeline steps
# -----------------------------------------------------------------------
def run_rag_pipeline_with_phoenix(query: str, tracer) -> str:
    """
    Use the OTEL tracer directly for custom spans around non-LLM steps.
    The Anthropic call inside is auto-instrumented as a child span.
    """
    client = anthropic.Anthropic()

    with tracer.start_as_current_span("rag-pipeline") as root_span:
        root_span.set_attribute("query", query)
        root_span.set_attribute("feature", "code-completion")

        # Retrieval span
        with tracer.start_as_current_span("vector-retrieval") as retrieval_span:
            retrieved_context = "Relevant code from codebase..."
            retrieval_span.set_attribute("num_results", 5)
            retrieval_span.set_attribute("retriever_version", "faiss-v2")

        # LLM call - auto-instrumented as a child span by AnthropicInstrumentor
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=512,
            system="You are a coding assistant. Use the provided context.",
            messages=[
                {
                    "role": "user",
                    "content": f"Context:\n{retrieved_context}\n\nQuery: {query}",
                }
            ],
        )
        output = response.content[0].text

        root_span.set_attribute("response_length", len(output))
        return output

:::info Phoenix vs Langfuse for Self-Hosting Phoenix is easier to run locally for development (one Python call to px.launch_app()). Langfuse is more complete for production self-hosting (full prompt management, team access, score API, richer UI). A common pattern: use Phoenix locally during development for fast debugging, and Langfuse self-hosted in production for team-wide visibility and compliance. :::

Helicone

Helicone is a proxy-based LLM observability tool. You route your Anthropic API calls through Helicone's proxy server, and it records them automatically with zero changes to your application code beyond the base_url and an API key header. This is its primary advantage and its primary limitation simultaneously.

What Helicone Does Best

• Zero code change per call site: point the client at a different URL and add one header - your existing codebase is fully traced
• Response caching: cache responses for identical prompts with Helicone-Cache-Enabled: true - can dramatically reduce costs for repetitive prompts
• Rate limiting: enforce per-user rate limits via headers without application-layer code
• Cost and usage dashboards: immediate visibility into spend, request count, and latency breakdown in the Helicone UI

Helicone Integration

# helicone_integration.py
"""
Helicone integration with the Anthropic SDK.
The only change: base_url points to Helicone's proxy.
All other code is identical to normal Anthropic usage.
"""
import anthropic
import os
from typing import Optional


def build_helicone_client(
    user_id: Optional[str] = None,
    feature: Optional[str] = None,
    session_id: Optional[str] = None,
    enable_cache: bool = False,
) -> anthropic.Anthropic:
    """
    Build an Anthropic client routed through Helicone.
    All properties are attached as headers and appear in the Helicone UI
    as filterable dimensions.
    """
    headers = {
        "Helicone-Auth": f"Bearer {os.getenv('HELICONE_API_KEY')}",
    }

    if user_id:
        headers["Helicone-User-Id"] = user_id
    if feature:
        headers["Helicone-Property-Feature"] = feature
    if session_id:
        headers["Helicone-Session-Id"] = session_id
    if enable_cache:
        headers["Helicone-Cache-Enabled"] = "true"
        headers["Helicone-Cache-Max-Age"] = "3600"  # 1 hour

    return anthropic.Anthropic(
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        base_url="https://anthropic.helicone.ai",
        default_headers=headers,
    )


def generate_support_response_helicone(
    user_message: str,
    user_id: str,
) -> str:
    """
    Identical to normal Anthropic code - just uses the Helicone client.
    Every call is recorded in Helicone with user_id attribution.
    """
    client = build_helicone_client(
        user_id=user_id,
        feature="customer-support",
        enable_cache=False,
    )

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system="You are a helpful customer support agent.",
        messages=[{"role": "user", "content": user_message}],
    )
    return response.content[0].text


def generate_with_caching(prompt: str) -> str:
    """
    Enable Helicone's response cache for repetitive prompts.
    Identical prompts return cached responses instantly at no API cost.
    Useful for: FAQ responses, static lookups, template generation.
    """
    client = build_helicone_client(
        feature="cached-faq",
        enable_cache=True,
    )

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

:::warning Helicone's Network Overhead and Compliance Trade-off Helicone adds a network round-trip (typically 15-50ms) to every API call. For latency-sensitive applications (chat interfaces, real-time completions), this can be material. More critically: your complete prompts and responses travel through Helicone's servers. For HIPAA-regulated applications, PHI in prompts, or financial data with strict data handling requirements, this is a compliance blocker. Evaluate whether the zero-code-change convenience is worth the latency and data residency trade-offs before adopting it for production. :::

Building the Abstraction Layer

The most consequential decision in LLMOps platform adoption is not which platform to choose - it is whether to build a thin abstraction layer from day one. If you instrument your code directly against Langfuse's @observe() and later need to switch to LangSmith, every traced function requires changes. Build the abstraction once; switch platforms in a configuration file.

# observability/tracer.py
"""
Thin observability abstraction layer.
Switch backends by changing OBSERVABILITY_BACKEND env var - no app code changes.
Supported backends: langfuse, langsmith, none (local dev / testing)
"""
import os
import functools
from typing import Any, Callable, Optional, TypeVar

F = TypeVar("F", bound=Callable[..., Any])

BACKEND = os.getenv("OBSERVABILITY_BACKEND", "langfuse").lower()

# -----------------------------------------------------------------------
# Backend: Langfuse
# -----------------------------------------------------------------------
if BACKEND == "langfuse":
    from langfuse.decorators import observe as _lf_observe, langfuse_context as _lf_ctx
    from langfuse import Langfuse

    _langfuse_client = Langfuse()

    def observe(name: Optional[str] = None, as_type: Optional[str] = None):
        """Wrap a function as a Langfuse trace or span."""
        kwargs: dict = {}
        if name:
            kwargs["name"] = name
        if as_type:
            kwargs["as_type"] = as_type
        return _lf_observe(**kwargs)

    def set_trace_attributes(**kwargs):
        """Attach attributes to the current trace root."""
        _lf_ctx.update_current_trace(**kwargs)

    def set_span_attributes(**kwargs):
        """Attach attributes to the current span."""
        _lf_ctx.update_current_observation(**kwargs)

    def attach_score(trace_id: str, name: str, value: float, comment: str = ""):
        """Attach a quality score to a trace by ID."""
        _langfuse_client.score(
            trace_id=trace_id,
            name=name,
            value=value,
            data_type="NUMERIC",
            comment=comment,
        )

# -----------------------------------------------------------------------
# Backend: LangSmith
# -----------------------------------------------------------------------
elif BACKEND == "langsmith":
    from langsmith import traceable as _ls_traceable, Client as _LSClient

    _ls_client = _LSClient()

    def observe(name: Optional[str] = None, as_type: Optional[str] = None):
        """Wrap a function as a LangSmith traceable."""
        return _ls_traceable(name=name or "")

    def set_trace_attributes(**kwargs):
        """LangSmith handles metadata via decorator parameters - no-op at call site."""
        pass

    def set_span_attributes(**kwargs):
        """No-op for LangSmith (metadata set at decorator level)."""
        pass

    def attach_score(trace_id: str, name: str, value: float, comment: str = ""):
        """Attach feedback to a LangSmith run."""
        _ls_client.create_feedback(
            run_id=trace_id,
            key=name,
            score=value,
            comment=comment,
        )

# -----------------------------------------------------------------------
# Backend: none (development / testing - no-op all observability)
# -----------------------------------------------------------------------
else:

    def observe(name: Optional[str] = None, as_type: Optional[str] = None):
        """No-op decorator for local dev and tests."""
        def decorator(f: F) -> F:
            @functools.wraps(f)
            def wrapper(*args, **kwargs):
                return f(*args, **kwargs)
            return wrapper  # type: ignore
        return decorator

    def set_trace_attributes(**kwargs):
        pass

    def set_span_attributes(**kwargs):
        pass

    def attach_score(trace_id: str, name: str, value: float, comment: str = ""):
        pass

# application code - uses the abstraction, not the platform directly
# observability/tracer.py is the ONLY file that imports from langfuse or langsmith

import anthropic
from observability.tracer import observe, set_trace_attributes, set_span_attributes

@observe(name="customer-support-pipeline")
def handle_support_request(user_message: str, user_id: str) -> str:
    """
    Application code is identical regardless of which platform is active.
    Set OBSERVABILITY_BACKEND=langfuse, langsmith, or none.
    """
    set_trace_attributes(user_id=user_id, tags=["production"])

    context = fetch_context(user_message)
    response = call_model(user_message, context)
    return response


@observe(name="context-fetch")
def fetch_context(query: str) -> str:
    set_span_attributes(metadata={"retriever": "faiss-v2"})
    return "Relevant knowledge base content..."


@observe(name="model-call", as_type="generation")
def call_model(message: str, context: str) -> str:
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system="You are a helpful support agent.",
        messages=[{"role": "user", "content": f"{context}\n\n{message}"}],
    )
    set_span_attributes(
        model="claude-opus-4-6",
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens,
        },
    )
    return response.content[0].text

When to Build vs Buy

Build Custom When

• You have data residency requirements that self-hosted Langfuse still cannot satisfy (edge deployment, air-gapped environments, specific regional jurisdiction constraints)
• Your tracing needs are structurally incompatible with the available platforms (e.g., you need to trace across dozens of services with a custom span schema that no platform models correctly)
• You have a dedicated platform engineering team and the investment is justified by scale: typically 50+ engineers building LLM features full-time
• You need to integrate LLM tracing deeply into an existing internal observability stack (e.g., every trace must appear in an on-premise Jaeger instance alongside your existing service traces)

Do Not Build Custom When

• You are under fifty engineers building LLM features
• You are still finding product-market fit - observability platform is not your differentiator
• Any of the five platforms does 80% or more of what you need
• You lack a dedicated platform team to maintain the custom system long-term

The platforms have invested millions of dollars in query performance, UI design, and evaluation tooling. A custom solution built in a sprint will not match them for two to three years. You will spend engineering capacity on maintenance rather than product.

Integration Sequence: The Right Order

Do not try to set up all six capabilities simultaneously. Teams that attempt this spend three months on infrastructure instead of product. Do them in order. Each layer builds on the previous: you cannot set up online eval before you have tracing; you cannot build a golden dataset before you have traces to sample from; you cannot run CI eval gates before you have an eval dataset.

Production Engineering Notes

Start with Traces, Not Evals

The most common adoption mistake: teams set up the platform but only use the eval features, leaving production tracing unimplemented. Traces are more valuable day-to-day because they enable debugging specific production failures. "A user reported that the AI gave wrong information" is answerable in thirty seconds if you have the trace (pull the trace for that user and timestamp, read the exact prompt and response). Without tracing, the same question requires hours of log scraping and speculation. Set up tracing in week one. Set up evals in week two.

Set Cost Budget Alerts Immediately

Every platform provides cost tracking. Set a budget alert (for example, $500/day on inference) in the first week. Unexpected cost spikes from prompt changes or traffic growth are common and caught within hours if you have an alert. Without an alert, you discover them on the monthly invoice. This is not a hypothetical - nearly every team running LLMs in production has a story about a runaway cost event they found too late.

Score in Production, Not Just in CI

The platforms all support attaching quality scores to production traces. Use this capability. Run an LLM judge on a sample - 5-10% of production traffic - daily. Plot the score distribution trend over a rolling seven-day window. This is your primary signal for silent quality degradation: the failure mode where prompt changes or model provider updates cause gradual quality decline that no individual user notices enough to report, but that accumulates into measurable churn.

Never Store Prompts Only in the Platform UI

Platform-specific prompt UIs are convenient for quick iteration. They are dangerous as the source of truth. If your production system pulls prompts from the platform and the platform has an outage (or you need to roll back to a previous version quickly), you need a reliable fallback. Store canonical prompt versions in version-controlled YAML files in your repository. Publish them to the platform for serving and UI visibility, but the YAML file is the source of truth. The platform is a cache and a viewer.

# Correct: prompts stored in YAML, published to platform, pulled at runtime
# scripts/publish_prompts.py

import yaml
import anthropic
from langfuse import Langfuse
from pathlib import Path

langfuse = Langfuse()

def publish_all_prompts(prompts_dir: str = "prompts/") -> None:
    """
    Read all prompt YAML files and publish them to Langfuse.
    Run this as part of your CI/CD pipeline on every merge to main.
    """
    prompts_path = Path(prompts_dir)
    for yaml_file in prompts_path.glob("*.yaml"):
        with open(yaml_file) as f:
            spec = yaml.safe_load(f)

        # Create or update the prompt in Langfuse
        langfuse.create_prompt(
            name=spec["name"],
            prompt=spec["template"],
            labels=spec.get("labels", ["staging"]),
            config={
                "model": spec.get("model", "claude-opus-4-6"),
                "max_tokens": spec.get("max_tokens", 512),
                "version": spec.get("version", "unknown"),
            },
        )
        print(f"Published: {spec['name']} v{spec.get('version', '?')}")


# Example prompts/customer-support.yaml:
# name: customer-support
# version: "3.2"
# model: claude-opus-4-6
# max_tokens: 512
# labels: [production]
# template: |
#   You are a helpful customer support agent for {{company_name}}.
#   Answer based only on the provided context.
#   Context: {{context}}
#   Customer message: {{customer_message}}

:::danger Platform Lock-In on Core Business Logic Avoid writing business logic inside platform-specific constructs. Platform tracing SDKs (decorators, wrapped clients) are acceptable - switching platforms is a day of work with the abstraction layer shown above. But never put conditional logic, prompt assembly, or output parsing inside platform-specific callback handlers or eval functions. Keep the platform layer thin: it observes and records what your application does. It does not drive what your application does. :::

Interview Q&A

Q1: Compare LangSmith and Langfuse. When would you choose each?

LangSmith and Langfuse are the two most broadly adopted LLMOps platforms and cover similar ground, but they differ on the decisions that matter most at the team level. Choose LangSmith when your team uses LangChain or LangGraph - LANGCHAIN_TRACING_V2=true gives zero-configuration tracing of the entire chain including every retrieval, tool call, and LLM generation as a structured tree. LangSmith is also stronger on dataset management for evals: sampling production traces into a golden dataset is a two-click operation in the UI, and the evaluator framework is tightly integrated. Choose Langfuse when you need self-hosting for data residency, HIPAA compliance, GDPR, or per-trace cost control at scale. Langfuse's @observe() decorator is cleaner than LangSmith's @traceable for non-LangChain codebases - it automatically creates parent-child span hierarchies from your function call graph without configuration. Langfuse is MIT-licensed and community-governed, which matters to teams that want to avoid SaaS vendor dependency for a critical observability path. At a practical level, both platforms do 90% of what most teams need - the deciding factors are LangChain dependency and self-hosting requirement.

Q2: What is the most important information to capture in an LLM trace?

The non-negotiable minimum is: the exact prompt (system prompt and user messages verbatim), the exact response, the model name and version, input and output token counts, latency in milliseconds, and a timestamp. Everything else is secondary. The reason the exact prompt must be captured verbatim is diagnostic: when a user reports a bad output, the first question is always "what exactly did we send?" Without the complete prompt in the trace, you cannot distinguish between a prompt engineering bug, a model regression, or a user input edge case. Model version is critical because providers silently update models on the same endpoint identifier - a trace without the exact version reported in the API response cannot be attributed to a specific model behavior. Token counts are required for cost attribution and for detecting prompt inflation (prompts growing gradually over time due to context accumulation). Latency is essential for SLA debugging. Quality scores can be attached asynchronously after the fact; the trace itself must be captured synchronously on the critical path.

Q3: How does proxy-based observability (Helicone) differ from SDK-based observability (Langfuse, LangSmith)?

Proxy-based observability intercepts your API calls at the network layer. You point your SDK's base_url at the observability vendor's proxy server. The proxy receives the request, records it, and forwards it to the model provider. Zero code changes per call site. Latency cost: 15-50ms per call for the additional network hop. Compliance concern: your complete prompts and responses transit through the observability vendor's infrastructure - for PHI, financial data, or any regulated content, this is typically a blocker. SDK-based observability instruments your code through decorators or wrapped clients. The SDK records the trace asynchronously (negligible latency impact on the critical path). Your prompts go directly from your server to the model provider - the observability data is sent separately to the observability backend. SDK-based is preferred for production applications for three reasons: latency (no added hop), data path (prompts never leave your network on the way to the model), and control (you can choose what to record and what to redact). Proxy-based is useful for quick PoCs, for organizations that cannot modify the application code, or for teams that need response caching at the proxy layer.

Q4: How would you instrument a multi-step RAG pipeline so the full request is traceable end-to-end?

Using Langfuse as an example: apply @observe() to the top-level pipeline function (creates the root trace), @observe(name="retrieval") to the vector DB query function (creates a retrieval child span), and @observe(name="llm-generation", as_type="generation") to the model call function (creates a generation child span). Langfuse links them automatically from the Python call stack - no trace ID threading required. Within each span, use langfuse_context.update_current_observation() to attach relevant metadata: for the retrieval span, attach the query text, number of documents retrieved, retriever version, and similarity scores; for the generation span, attach the model name, token counts, input and output in full, and the prompt version. After the pipeline completes, attach a quality score to the root trace via langfuse.score(). The resulting trace gives you: end-to-end latency broken down by step, cost by model call, retrieval quality signals, and an LLM judge score - all linked to the same root trace ID, queryable in the UI, and filterable by any attribute.

Q5: When should a team build custom LLMOps tooling instead of using an existing platform?

Almost never, and later than most teams think. The conditions that justify a custom build are: first, data residency requirements that self-hosted Langfuse cannot satisfy - specifically air-gapped environments, edge deployments, or jurisdictions where the Langfuse self-hosted architecture itself is non-compliant; second, span schema requirements that are structurally incompatible with the platforms' data models and cannot be resolved with metadata fields; third, a dedicated platform engineering team whose output is justified by scale (typically 50+ engineers building LLM-powered features full time) and where the observability platform is a differentiating capability rather than commodity infrastructure. The argument against building is strong: the major platforms represent years of engineering investment in query performance, UI design, and evaluation framework ergonomics. A custom solution built in a sprint will take two to three years to approach their quality. The practical answer for 95% of teams: pick a platform (the decision matters less than the fact of deciding), build the abstraction layer shown in this lesson so you can switch in a day, and invest your engineering capacity in the AI product instead of the observability infrastructure.

Q6: What is the integration sequence you recommend for a team adopting LLMOps for the first time?

Week one: basic tracing. Every production LLM call captured with the full prompt, response, token counts, and latency. No quality scoring yet - just visibility. Week two: cost alerts. Set budget thresholds by model and by feature. You will almost certainly discover a call site consuming more tokens than expected. Week three: prompt registry. Move all production prompt strings from code into the platform's prompt registry. Prompts are now versioned, named, and pulled at runtime - changes require no code deploy. Week four: online eval. Configure an LLM judge to score a 10% sample of production traffic daily. Establish the baseline quality distribution. Month two: golden dataset. Curate 50-100 real production examples that represent your hardest cases, edge cases, and known failure modes. This dataset is the foundation of your eval system. Month three: CI eval gate. Add the golden dataset to your CI pipeline. Every pull request must pass the eval gate before merge. The sequence matters because each layer depends on the previous: you cannot build a golden dataset before you have traces to sample from; you cannot run CI gates before you have a dataset.