Production AgentOps Pipeline: OpenTelemetry, OpenInference & Langfuse Tracing for Autonomous Multi-Agent Systems

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

The observability ceiling in multi-agent systems

Single-agent applications fail predictably. A multi-agent system fails combinatorially. When an orchestrator fan-outs to five specialists, each calling three tools, each invoking a model, the number of possible failure paths grows to thousands per run. Log files and ad-hoc print statements cannot answer the only question that matters in production: which agent, on which hop, for which task, made the decision that broke the business outcome?

The industry converged on a three-layer answer, and it is now a de-facto standard in the 2026 agent observability ecosystem:

1. OpenTelemetry (OTel) as the transport and signal model, emitting traces, metrics, and logs over OTLP.
2. OpenInference as the semantic convention layer that teaches OTel what "a model call," "a tool call," "an agent turn," or "an embedding" means.
3. Langfuse as the storage, analysis, and evaluation plane where traces become dashboards, scores, and regression gates.

This article builds a production AgentOps pipeline that wires all three together. The code is real, runnable, and framework-agnostic: it works whether your fleet is built on CrewAI, LangGraph, AutoGen, Google ADK, or raw SDK calls, because the conventions sit at the instrumentation boundary, not inside any single framework.

Why vanilla OTel is not enough for agents

OpenTelemetry gives you spans, attributes, and propagation. But a span is only useful if the semantic vocabulary is agreed upon. An arbitrary span called worker.run tells you nothing about prompt tokens, tool latency, retrieval source, or agent hops. OpenInference fixes this by defining a semantic-convention catalog:

• openinference.span.kind — values such as GEN_AI, AGENT, TOOL, CHAIN, RAG, and EVALUATOR.
• openinference.span.cost — token and currency cost attributes for a model call.
• openinference.tool.name and openinference.tool.description — metadata captured on every tool invocation.
• openinference.agent.message — serialized message payloads that let you replay the exact conversation an agent saw.

The OpenInference semantic conventions are now adopted by Langfuse, Arize Phoenix, and dozens of tracing backends, which means one instrumentation pass produces traces that render identically across vendors. That portability is the entire point: your AgentOps pipeline should never lock you into a single dashboard.

The three pillars and how they interact

OpenInference ships as Python auto-instrumentation packages (openinference-instrumentation-openai, -langchain, -crewai, -llama-index, -pydantic-ai) that patch SDK internals with a single TracerProvider() call. Everything below is written against the OTel-native Langfuse exporter, so spans flow through a real OTLP pipeline rather than a proprietary agent SDK hook.

Reference architecture

                          +-------------------------------------------+
                          |               WORKLOAD FLEET               |
                          |  Orchestrator -> Researcher -> Writer     |
                          |  (CrewAI / LangGraph / ADK / AutoGen)     |
                          +---------------------+---------------------+
                                                |
                          OpenInference auto-   |  OTLP over gRPC/HTTP
                          instrumentation      v
                          +-------------------------------------------+
                          |  OTel SDK + TracerProvider + Processor    |
                          |  - BatchSpanProcessor (128 MB queue)      |
                          |  - Sampler: parent-based, 10% tail        |
                          +---------------------+---------------------+
                                                |
                          OTLP/HTTP +           |  batching, retry,
                          bearer token          v  backoff (2s..60s)
                          +-------------------------------------------+
                          |                 LANGFUSE                   |
                          |  ingestion -> clickhouse -> api           |
                          |  projects, traces, evals, prompts, scores |
                          +---------------------+---------------------+
                                                |
                          langfuse SDK          v
                          +-------------------------------------------+
                          |  Evaluators (LLM-judge, heuristic)        |
                          |  Regression gates on every release        |
                          +-------------------------------------------+

The critical design decision is the batch processor with an unbounded-capacity queue and a non-blocking exporter thread. Agent workloads burst; if the exporter blocks the agent thread, you add latency to a system that is already slow. A dedicated span processor keeps observability off the critical path.

Project layout

agentops/
├── .env
├── tracing/
│   ├── __init__.py
│   ├── setup.py
│   └── semantic.py
├── agents/
│   ├── orchestrator.py
│   └── researcher.py
├── evaluators/
│   └── judge.py
└── main.py

Environment configuration

# .env
LANGFUSE_PUBLIC_KEY=pf-xxxxx
LANGFUSE_SECRET_KEY=sk-xxxxx
LANGFUSE_HOST=https://cloud.langfuse.com
OTEL_SERVICE_NAME=production-agentops
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
OTEL_EXPORTER_OTLP_ENDPOINT=https://your-langfuse-otlp.example.com
OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer%20lf-otlp-token
OTEL_TRACES_SAMPLER=parentbased_traceidratio
OTEL_TRACES_SAMPLER_ARG=1.0
AGENTOPS_SPAN_QUEUE_SIZE=131072

Run the agent under opentelemetry-instrument for zero-code instrumentation, or call setup_tracing() at module import time. Both approaches are shown below; production systems typically use the CLI launcher so the SDK config stays declarative.

Bootstrapping the tracer

# tracing/setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from openinference.instrumentation.langchain import LangChainInstrumentor
from openinference.instrumentation.openai import OpenAIInstrumentor
from openinference.instrumentation.crewai import CrewAIInstrumentor
from opentelemetry.sdk.resources import Resource
import os

def setup_tracing() -> None:
    resource = Resource.create({"service.name": os.getenv("OTEL_SERVICE_NAME", "agentops")})
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"),
        headers=_parse_headers(os.getenv("OTEL_EXPORTER_OTLP_HEADERS", "")),
    )
    provider.add_span_processor(BatchSpanProcessor(exporter, max_queue_size=131_072))
    trace.set_tracer_provider(provider)
    LangChainInstrumentor().instrument()
    OpenAIInstrumentor().instrument()
    CrewAIInstrumentor().instrument()

def _parse_headers(raw: str) -> dict:
    out = {}
    for pair in raw.split(","):
        key, _, value = pair.partition("=")
        if value:
            out[key.strip()] = value.strip().replace("%20", " ")
    return out

The instrumentors read environment variables by default, which is why .env values map cleanly onto OTel's standard names. Langfuse's OTLP endpoint is a drop-in OTLP/HTTP receiver, so no vendor-specific exporter code is needed in your service.

Instrumenting a multi-agent run with semantic spans

# agents/orchestrator.py
from opentelemetry import trace
from openinference.semconv.trace import SpanAttributes as SemConv

tracer = trace.get_tracer("production-agentops.orchestrator")

def run_pipeline(request: dict) -> dict:
    with tracer.start_as_current_span("pipeline.run") as span:
        span.set_attribute(SemConv.OPENINFERENCE_SPAN_KIND, "AGENT")
        span.set_attribute("session.id", request["session_id"])
        span.set_attribute("input.task", request["task"])
        try:
            researcher = run_researcher(request["task"])
            writer = run_writer(researcher)
            span.set_attribute("output.status", "ok")
            return writer
        except ToolError as exc:
            span.set_attribute("output.status", "error")
            span.set_attribute("error.message", str(exc))
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(exc)))
            raise

Every span carries session.id and a correlation ID from upstream so that Langfuse can stitch a single multi-hop request into one timeline, even when it spans service boundaries. Use propagation headers (W3C traceparent) at the RPC layer so the orchestrator and each specialist share one trace ID.

Retry rules and error handling

Agents fail for different reasons than services: rate limits, tool timeouts, malformed JSON from models, and duplicate tool calls. Treat each class separately.

# evaluators/retry.py
import random, time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RateLimited(RuntimeError):
    pass

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, max=8),
    retry=retry_if_exception_type((RateLimited, TimeoutError)),
    reraise=True,
)
def call_with_retries(fn, *args, **kwargs):
    return fn(*args, **kwargs)

def retry_on_429(response):
    if response.status_code == 429:
        jitter = random.uniform(0.5, 1.5)
        time.sleep(2 * jitter)
        raise RateLimited("rate limited, backing off")

Add a circuit breaker at the tool layer: after 10 consecutive failures in a 60-second window, fail fast for 30 seconds and emit a metric. The same metric feeds an alert that pages the on-call engineer before the queue backs up.

Evaluating traces and gating releases

Langfuse stores every trace; the evaluation plane turns them into scores. A lightweight LLM-judge evaluator scores a sample of traces on correctness, tool use efficiency, and hallucination risk.

# evaluators/judge.py
from langfuse import Langfuse
from langfuse.evaluations import evaluate

lf = Langfuse(public_key=..., secret_key=..., host=...)

def factual_accuracy_judge(output, expected):
    return lf.evaluate("factual-accuracy", input={"output": output, "expected": expected})

results = evaluate(
    dataset_name="production-weekly-sample",
    evaluators=[factual_accuracy_judge],
    max_retries=3,
    retry_delay=1.5,
)
print("Mean score:", results.get_mean_scores())

Key insight: A trace you cannot score is a log line you cannot act on. The pipeline is not complete until every run feeds a regression gate, and every release blocks on a minimum accuracy threshold.

The full feedback loop: run -> trace -> score -> dataset -> prompt tweak -> regression test -> promote. This closes the loop that makes AgentOps a discipline rather than a dashboard.

Choosing a backend: Langfuse vs the field

Langfuse wins on portability: because it ingests standard OTel spans annotated with OpenInference semantics, you can migrate to another backend without re-instrumenting a single agent.

Hardening for production

• Sampling: keep parentbased_traceidratio=1.0 for interactive traces; sample high-volume internal RAG calls at 10% tail sampling.
• PII control: configure Langfuse masking rules so spans never persist raw PII such as customer email addresses.
• Alerting: connect Langfuse scores and OTel error metrics to PagerDuty so a degradation in factual accuracy pages the team before users complain.
• Cost attribution: rely on OpenInference cost attributes per span to allocate spend per agent per project in the monthly review.

The same pipeline pattern also applies to your latest AI workflow experiments, and you can track the newest observability vendors and model context protocol updates as the ecosystem evolves. For day-to-day release coverage, follow the latest AI news on agent tooling releases.

Summary

Standardized OTel emission, OpenInference semantics, and Langfuse analytics give you a pipeline where every agent decision is a queryable, scoreable, replayable artifact. The observability tax is small — a batch processor, three instrumentor calls, and environment variables — and the payoff is the difference between debugging by guesswork and debugging by timeline.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.