Build OpenTelemetry distributed trace pipelines

Loading diagram...

• Line 1: Declares a Mermaid flowchart with left-to-right (LR) layout direction.
• Line 2: Defines node A ("FastAPI Request") connecting to node B ("OTel Middleware Span"), showing that incoming API requests are first intercepted by OpenTelemetry middleware to create a root tracing span.
• Line 3: Connects the middleware span B to node C ("RAG Retrieval Span"), representing a child span that traces the retrieval-augmented generation document lookup.
• Line 4: Connects the middleware span B to node D ("Prompt Build Span"), representing a parallel child span that traces the construction of the LLM prompt.
• Line 5: Connects the prompt build span D to node E ("LiteLLM Call Span"), showing that after the prompt is built, a child span traces the actual LLM API call made through LiteLLM.
• Line 6: Connects the LiteLLM call span E to node F ("Guardrail Span"), representing a subsequent span that traces safety/guardrail checks applied to the LLM output.
• Line 7: Connects the guardrail span F to node G ("Response Span"), representing the final processing span that traces the construction and return of the API response.
• Line 8: Connects the middleware span B to node H ("BatchSpanProcessor"), showing that all collected spans are forwarded to OpenTelemetry's batch processor for efficient, buffered export.
• Line 9: Connects the batch processor H to node I ("OTLPSpanExporter"), indicating spans are serialized and sent out via the OpenTelemetry Protocol exporter.
• Line 10: Connects the OTLP exporter I to node J ("Langfuse OTLP Endpoint"), showing the primary telemetry destination is Langfuse's OTLP-compatible ingestion endpoint.
• Line 11: Connects the OTLP exporter I to node K ("Jaeger / Datadog Optional"), showing that additional observability backends can optionally receive the same span data.
• Line 12: Connects the Langfuse endpoint J to node L ("Langfuse Trace Viewer"), representing the final UI where developers visualize and inspect the full distributed trace.

1import base64
2from opentelemetry import trace
3from opentelemetry.sdk.trace import TracerProvider
4from opentelemetry.sdk.trace.export import BatchSpanProcessor
5from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
6    OTLPSpanExporter,
7)
8from opentelemetry.sdk.resources import Resource
9
10class OTelTraceProvider:
11    """Configures OpenTelemetry to export spans to Langfuse."""
12
13    def __init__(
14        self,
15        service_name: str,
16        langfuse_host: str,
17        langfuse_public_key: str,
18        langfuse_secret_key: str,
19        service_version: str = "0.1.0",
20        environment: str = "development",
21    ):
22        self._resource = Resource.create({
23            "service.name": service_name,
24            "service.version": service_version,
25            "deployment.environment": environment,
26        })
27        credentials = f"{langfuse_public_key}:{langfuse_secret_key}"
28        encoded = base64.b64encode(credentials.encode()).decode()
29        self._exporter = OTLPSpanExporter(
30            endpoint=f"{langfuse_host}/api/public/otel/v1/traces",
31            headers={"Authorization": f"Basic {encoded}"},
32        )
33        self._provider: TracerProvider | None = None
34
35    def setup_provider(self) -> TracerProvider:
36        self._provider = TracerProvider(resource=self._resource)
37        processor = BatchSpanProcessor(
38            self._exporter,
39            max_queue_size=2048,
40            max_export_batch_size=512,
41            schedule_delay_millis=5000,
42        )
43        self._provider.add_span_processor(processor)
44        trace.set_tracer_provider(self._provider)
45        return self._provider
46
47    def get_tracer(self, name: str = __name__) -> trace.Tracer:
48        if self._provider is None:
49            raise RuntimeError("Call setup_provider() before get_tracer()")
50        return self._provider.get_tracer(name)
51
52    def shutdown(self) -> None:
53        if self._provider is not None:
54            self._provider.shutdown()

• Lines 1-7: Import the core OpenTelemetry SDK components. The base64 module is needed to encode Langfuse credentials for HTTP Basic Auth. The OTLPSpanExporter from the proto.http package sends spans over HTTP/protobuf, which Langfuse's OTLP endpoint accepts.
• Lines 10-11: The class docstring makes the intent explicit—this provider targets Langfuse specifically, not a generic OTLP backend.
• Lines 13-21: The constructor accepts Langfuse connection parameters alongside standard service metadata. The service_version and environment default to safe development values, preventing accidental production data pollution when developers forget to override them.
• Lines 22-25: The Resource encodes three semantic conventions that Langfuse uses to group and filter traces: service.name appears as the project identifier, service.version enables version-to-version latency comparison, and deployment.environment separates staging from production traces.
• Lines 26-28: The Langfuse public and secret keys are concatenated with a colon separator and Base64-encoded, producing the value expected by HTTP Basic Auth. This is the same encoding pattern used by curl's -u flag.
• Lines 29-32: The OTLPSpanExporter is instantiated with the full Langfuse OTLP path and the encoded authorization header. Langfuse validates these credentials on every batch submission.
• Lines 35-43: setup_provider() creates the TracerProvider with the resource, then attaches a BatchSpanProcessor configured with explicit tuning parameters. The max_queue_size of 2048 prevents memory growth under burst traffic, max_export_batch_size of 512 balances network efficiency against latency, and schedule_delay_millis of 5000 means spans flush every 5 seconds—acceptable for observability but not real-time alerting.
• Lines 44-45: trace.set_tracer_provider() registers this provider globally, meaning any library that calls trace.get_tracer() will route spans through this pipeline. This is critical for auto-instrumentation libraries like opentelemetry-instrumentation-fastapi to work without explicit wiring.
• Lines 47-49: get_tracer() returns a named Tracer instance. The RuntimeError guard ensures developers cannot accidentally create spans before the provider is initialized, catching misconfigured startup sequences early.
• Lines 51-53: shutdown() flushes all pending spans and releases resources. Call this in FastAPI's lifespan shutdown hook to avoid losing the final batch of spans when the process terminates.

1from fastapi import FastAPI, Request
2from opentelemetry import trace, context
3from opentelemetry.trace import SpanKind, StatusCode
4
5app = FastAPI()
6tracer = trace.get_tracer("genai.service")
7
8@app.post("/api/chat")
9async def chat_endpoint(request: Request):
10    body = await request.json()
11    user_query = body["query"]
12
13    with tracer.start_as_current_span(
14        "chat_request", kind=SpanKind.SERVER
15    ) as root_span:
16        root_span.set_attribute("user.query_length", len(user_query))
17
18        # Phase 1: RAG retrieval
19        with tracer.start_as_current_span("rag_retrieval") as rag_span:
20            docs = await retrieve_documents(user_query)
21            rag_span.set_attribute("rag.document_count", len(docs))
22            rag_span.set_attribute("rag.strategy", "hybrid_search")
23
24        # Phase 2: Prompt construction
25        with tracer.start_as_current_span("prompt_build") as prompt_span:
26            prompt = build_prompt(user_query, docs)
27            prompt_span.set_attribute("prompt.template_version", "v2.3")
28            prompt_span.set_attribute("prompt.token_estimate", len(prompt) // 4)
29
30        # Phase 3: LLM invocation
31        with tracer.start_as_current_span(
32            "llm_call", kind=SpanKind.CLIENT
33        ) as llm_span:
34            llm_span.set_attribute("gen_ai.system", "openai")
35            llm_span.set_attribute("gen_ai.request.model", "gpt-4o")
36            response = await call_litellm(prompt)
37            llm_span.set_attribute(
38                "gen_ai.usage.prompt_tokens", response.usage.prompt_tokens
39            )
40            llm_span.set_attribute(
41                "gen_ai.usage.completion_tokens",
42                response.usage.completion_tokens,
43            )
44            llm_span.set_attribute("gen_ai.response.model", response.model)
45
46        # Phase 4: Guardrail post-processing
47        with tracer.start_as_current_span("guardrail_check") as guard_span:
48            result = await apply_guardrails(response.choices[0].message.content)
49            guard_span.set_attribute("guardrail.passed", result.is_safe)
50            if not result.is_safe:
51                guard_span.set_status(StatusCode.ERROR, "Guardrail blocked")
52
53    return {"response": result.content, "trace_id": format(root_span.context.trace_id, "032x")}

• Lines 1-3: Import FastAPI alongside OTel's context module (needed for propagation in async scenarios) and SpanKind/StatusCode enums for span classification and error signaling.
• Lines 5-6: The tracer is obtained at module level after the global provider has been set. The name "genai.service" appears in Langfuse as the instrumentation scope, letting you filter spans by origin.
• Lines 9-12: The endpoint extracts the user query from the request body. In production, you would validate with Pydantic models, but the focus here is on trace instrumentation.
• Lines 14-17: start_as_current_span creates the root span with SpanKind.SERVER, signaling this is an inbound request handler. Langfuse maps this root span to a top-level trace. The user.query_length attribute enables correlation between query complexity and latency.
• Lines 20-23: The RAG retrieval span captures how many documents were returned and which search strategy was used. These attributes let you build Langfuse dashboards comparing latency across retrieval strategies.
• Lines 26-29: The prompt construction span records the template version (critical for correlating with prompt A/B tests from another goal) and an estimated token count. The // 4 heuristic approximates tokens for English text—use tiktoken in production for accuracy.
• Lines 32-45: The LLM invocation span uses SpanKind.CLIENT because this is an outbound call to an external service. The gen_ai.* attributes follow the emerging OpenTelemetry GenAI semantic conventions. Langfuse specifically recognizes gen_ai.system, gen_ai.request.model, gen_ai.usage.prompt_tokens, and gen_ai.usage.completion_tokens to populate its cost and token usage panels.
• Lines 48-53: The guardrail span records whether the response passed safety checks. When result.is_safe is False, the span status is set to StatusCode.ERROR with a descriptive message. Langfuse surfaces errored spans with red highlights in the trace waterfall, making it trivial to spot blocked responses during quality reviews.
• Line 55: The response includes the trace_id formatted as a 32-character hex string. This ID can be passed to frontend clients for user feedback correlation (covered in another goal), allowing users to report issues that link directly to the full distributed trace.