Log distributed traces

When working with complex LLM applications, it is common to need to track a traces across multiple services. Opik supports distributed tracing out of the box when integrating using function decorators using a mechanism that is similar to how OpenTelemetry implements distributed tracing.

For the purposes of this guide, we will assume that you have a simple LLM application that is made up of two services: a client and a server. We will assume that the client will create the trace and span, while the server will add a nested span. In order to do this, the trace_id and span_id will be passed in the headers of the request from the client to the server.

The Python SDK includes some helper functions to make it easier to fetch headers in the client and ingest them in the server:

1
from opik import track, opik_context
2
3
@track()
4
def my_client_function(prompt: str) -> str:
5
    headers = {}
6
7
    # Update the headers to include Opik Trace ID and Span ID
8
    headers.update(opik_context.get_distributed_trace_headers())
9
10
    # Make call to backend service
11
    response = requests.post("http://.../generate_response", headers=headers, json={"prompt": prompt})
12
    return response.json()

On the server side, you can pass the headers to your decorated function:

1
from opik import track
2
from fastapi import FastAPI, Request
3
4
@track()
5
def my_llm_application():
6
    pass
7
8
app = FastAPI()  # Or Flask, Django, or any other framework
9
10
11
@app.post("/generate_response")
12
def generate_llm_response(request: Request) -> str:
13
    return my_llm_application(opik_distributed_trace_headers=request.headers)

Using the distributed_headers Context Manager

As an alternative to passing opik_distributed_trace_headers as a parameter, you can use the distributed_headers() context manager for more explicit control over distributed header handling. This approach provides automatic cleanup, error handling, and optional data flushing.

1
from opik import track
2
from opik.decorator.context_manager import distributed_headers
3
from fastapi import FastAPI, Request
4
5
@track()
6
def my_llm_application():
7
    pass
8
9
app = FastAPI()  # Or Flask, Django, or any other framework
10
11
12
@app.post("/generate_response")
13
def generate_llm_response(request: Request) -> str:
14
    # Extract distributed headers from the request
15
    headers = {
16
        "opik_trace_id": request.headers.get("opik_trace_id"),
17
        "opik_parent_span_id": request.headers.get("opik_parent_span_id"),
18
    }
19
20
    # Use the context manager to handle distributed headers
21
    with distributed_headers(headers, flush=False):
22
        result = my_llm_application()
23
24
    return result

The distributed_headers() context manager accepts two parameters:

• headers: A dictionary containing the distributed trace headers (opik_trace_id and opik_parent_span_id)
• flush (optional): Whether to flush the Opik client data after the root span is processed. Defaults to False. Set to True if you want to ensure immediate data transmission.

For more details and additional examples, see the distributed_headers context manager API reference.