diff --git a/integrations/otel/distributed_tracing/option_a_explicit_ids.py b/integrations/otel/distributed_tracing/option_a_explicit_ids.py index e7db751..9b5ed72 100644 --- a/integrations/otel/distributed_tracing/option_a_explicit_ids.py +++ b/integrations/otel/distributed_tracing/option_a_explicit_ids.py @@ -144,10 +144,10 @@ def llm_attrs(model: str, prompt, completion, usage: dict) -> dict: `usage` keys map under gen_ai.usage.* (input_tokens, output_tokens, etc.). """ attrs = { - "gen_ai.system": "openai", # -> provider - "gen_ai.request.model": model, # -> model, forces span type = llm - "input": prompt, # -> input - "output": completion, # -> output + "gen_ai.system": "openai", # -> provider + "gen_ai.request.model": model, # -> model, forces span type = llm + "input": prompt, # -> input + "output": completion, # -> output } for token_type, count in usage.items(): attrs[f"gen_ai.usage.{token_type}"] = count @@ -167,32 +167,50 @@ def backend_request_1(session_id: str, user_prompt: str) -> dict: then dispatches it to the UI/external service (out of process). """ trace_id = new_id() - s_orchestrator = new_id() # root span (closes last, in finalize()) + s_orchestrator = new_id() # root span (closes last, in finalize()) s_routing = new_id() s_agent = new_id() s_llm1 = new_id() - s_dispatch = new_id() # tool_dispatch: tool result must nest UNDER this - - emit_span("routing.invoke", trace_id=trace_id, span_id=s_routing, - parent_span_id=s_orchestrator, duration_s=0.1) - emit_span("routing.generic_agent", trace_id=trace_id, span_id=s_agent, - parent_span_id=s_routing, duration_s=0.1) - emit_span("llm.request", trace_id=trace_id, span_id=s_llm1, parent_span_id=s_agent, - duration_s=0.3, - attributes=llm_attrs( - model="gpt-4o", - prompt=[{"role": "user", "content": user_prompt}], - completion=[{"role": "assistant", "tool_calls": [ - {"id": "call_1", "name": "web_search", "arguments": {"query": user_prompt}}]}], - usage={"input_tokens": 412, "output_tokens": 37}, - )) - emit_span("tool_dispatch", trace_id=trace_id, span_id=s_dispatch, - parent_span_id=s_llm1, duration_s=0.02, - attributes={ - "gen_ai.tool.name": "web_search", - "gen_ai.tool.call.id": "call_1", - "input": {"query": user_prompt}, - }) + s_dispatch = new_id() # tool_dispatch: tool result must nest UNDER this + + emit_span( + "routing.invoke", trace_id=trace_id, span_id=s_routing, parent_span_id=s_orchestrator, duration_s=0.1 + ) + emit_span( + "routing.generic_agent", trace_id=trace_id, span_id=s_agent, parent_span_id=s_routing, duration_s=0.1 + ) + emit_span( + "llm.request", + trace_id=trace_id, + span_id=s_llm1, + parent_span_id=s_agent, + duration_s=0.3, + attributes=llm_attrs( + model="gpt-4o", + prompt=[{"role": "user", "content": user_prompt}], + completion=[ + { + "role": "assistant", + "tool_calls": [ + {"id": "call_1", "name": "web_search", "arguments": {"query": user_prompt}} + ], + } + ], + usage={"input_tokens": 412, "output_tokens": 37}, + ), + ) + emit_span( + "tool_dispatch", + trace_id=trace_id, + span_id=s_dispatch, + parent_span_id=s_llm1, + duration_s=0.02, + attributes={ + "gen_ai.tool.name": "web_search", + "gen_ai.tool.call.id": "call_1", + "input": {"query": user_prompt}, + }, + ) # Persist the two ids the second request needs to stitch the trace correctly. SESSION_STORE[session_id] = { @@ -221,28 +239,42 @@ def backend_request_2(session_id: str, tool_result: dict) -> str: # THE KEY LINE: parent is the dispatch span id from request 1 — no longer a # disconnected root. - emit_span("tool_execution", trace_id=trace_id, span_id=s_tool, - parent_span_id=ctx["dispatch_span_id"], duration_s=0.2, - attributes={ - "gen_ai.tool.name": "web_search", - "gen_ai.tool.call.id": "call_1", - "input": ctx["user_prompt"], - "output": tool_result, - }) - emit_span("routing.invoke", trace_id=trace_id, span_id=s_routing2, - parent_span_id=s_tool, duration_s=0.1) - emit_span("routing.generic_agent", trace_id=trace_id, span_id=s_agent2, - parent_span_id=s_routing2, duration_s=0.1) + emit_span( + "tool_execution", + trace_id=trace_id, + span_id=s_tool, + parent_span_id=ctx["dispatch_span_id"], + duration_s=0.2, + attributes={ + "gen_ai.tool.name": "web_search", + "gen_ai.tool.call.id": "call_1", + "input": ctx["user_prompt"], + "output": tool_result, + }, + ) + emit_span("routing.invoke", trace_id=trace_id, span_id=s_routing2, parent_span_id=s_tool, duration_s=0.1) + emit_span( + "routing.generic_agent", + trace_id=trace_id, + span_id=s_agent2, + parent_span_id=s_routing2, + duration_s=0.1, + ) final_answer = "The capital of France is Paris." - emit_span("llm.request", trace_id=trace_id, span_id=s_llm2, parent_span_id=s_agent2, - duration_s=0.3, - attributes=llm_attrs( - model="gpt-4o", - prompt=[{"role": "tool", "content": tool_result}], - completion=[{"role": "assistant", "content": final_answer}], - usage={"input_tokens": 690, "output_tokens": 122}, - )) + emit_span( + "llm.request", + trace_id=trace_id, + span_id=s_llm2, + parent_span_id=s_agent2, + duration_s=0.3, + attributes=llm_attrs( + model="gpt-4o", + prompt=[{"role": "tool", "content": tool_result}], + completion=[{"role": "assistant", "content": final_answer}], + usage={"input_tokens": 690, "output_tokens": 122}, + ), + ) ctx["final_answer"] = final_answer return final_answer @@ -254,14 +286,19 @@ def finalize(session_id: str) -> None: thread_id groups the trace into a conversation thread in Opik. """ ctx = SESSION_STORE[session_id] - emit_span("orchestrator_request", trace_id=ctx["trace_id"], - span_id=ctx["orchestrator_span_id"], parent_span_id=None, duration_s=0.05, - attributes={ - "thread_id": session_id, # -> Opik thread grouping - "input": ctx["user_prompt"], # -> trace input - "output": ctx["final_answer"], # -> trace output - "opik.tags": ["tool-call", "distributed", "demo"], - }) + emit_span( + "orchestrator_request", + trace_id=ctx["trace_id"], + span_id=ctx["orchestrator_span_id"], + parent_span_id=None, + duration_s=0.05, + attributes={ + "thread_id": session_id, # -> Opik thread grouping + "input": ctx["user_prompt"], # -> trace input + "output": ctx["final_answer"], # -> trace output + "opik.tags": ["tool-call", "distributed", "demo"], + }, + ) def print_tree() -> None: diff --git a/integrations/otel/distributed_tracing/option_b_w3c_propagation.py b/integrations/otel/distributed_tracing/option_b_w3c_propagation.py index 3ce70d5..0861a48 100644 --- a/integrations/otel/distributed_tracing/option_b_w3c_propagation.py +++ b/integrations/otel/distributed_tracing/option_b_w3c_propagation.py @@ -90,8 +90,7 @@ def set_attrs(span, attributes: dict) -> None: """OTel attribute values must be primitives; JSON-encode structured payloads.""" span.set_attributes( - {k: (v if isinstance(v, (str, bool, int, float)) else json.dumps(v)) - for k, v in attributes.items()} + {k: (v if isinstance(v, (str, bool, int, float)) else json.dumps(v)) for k, v in attributes.items()} ) @@ -130,28 +129,47 @@ def backend_request_1(session_id: str, user_prompt: str) -> dict: with tracer.start_as_current_span("routing.generic_agent"): time.sleep(0.05) with tracer.start_as_current_span("llm.request") as llm1: - set_attrs(llm1, llm_attrs( - model="gpt-4o", - prompt=[{"role": "user", "content": user_prompt}], - completion=[{"role": "assistant", "tool_calls": [ - {"id": "call_1", "name": "web_search", "arguments": {"query": user_prompt}}]}], - usage={"input_tokens": 412, "output_tokens": 37}, - )) + set_attrs( + llm1, + llm_attrs( + model="gpt-4o", + prompt=[{"role": "user", "content": user_prompt}], + completion=[ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call_1", + "name": "web_search", + "arguments": {"query": user_prompt}, + } + ], + } + ], + usage={"input_tokens": 412, "output_tokens": 37}, + ), + ) time.sleep(0.2) # Inject WHILE tool_dispatch is the current span so the carrier # encodes it as the parent for the tool execution in request 2. with tracer.start_as_current_span("tool_dispatch") as dispatch: - set_attrs(dispatch, { - "gen_ai.tool.name": "web_search", - "gen_ai.tool.call.id": "call_1", - "input": {"query": user_prompt}, - }) + set_attrs( + dispatch, + { + "gen_ai.tool.name": "web_search", + "gen_ai.tool.call.id": "call_1", + "input": {"query": user_prompt}, + }, + ) inject(carrier) # -> carrier["traceparent"] = 00---01 - SESSION_STORE[session_id] = {"orchestrator": orchestrator, "carrier": carrier, - "user_prompt": user_prompt} - return {"tool": "web_search", "args": {"query": user_prompt}, - "session_id": session_id, "traceparent": carrier.get("traceparent")} + SESSION_STORE[session_id] = {"orchestrator": orchestrator, "carrier": carrier, "user_prompt": user_prompt} + return { + "tool": "web_search", + "args": {"query": user_prompt}, + "session_id": session_id, + "traceparent": carrier.get("traceparent"), + } def backend_request_2(session_id: str, tool_result: dict) -> str: @@ -163,12 +181,15 @@ def backend_request_2(session_id: str, tool_result: dict) -> str: # THE KEY LINE: context=parent_ctx -> tool_execution becomes a child of # tool_dispatch, sharing the same trace id. with tracer.start_as_current_span("tool_execution", context=parent_ctx) as tool: - set_attrs(tool, { - "gen_ai.tool.name": "web_search", - "gen_ai.tool.call.id": "call_1", - "input": ctx_data["user_prompt"], - "output": tool_result, - }) + set_attrs( + tool, + { + "gen_ai.tool.name": "web_search", + "gen_ai.tool.call.id": "call_1", + "input": ctx_data["user_prompt"], + "output": tool_result, + }, + ) time.sleep(0.1) with tracer.start_as_current_span("routing.invoke"): time.sleep(0.05) @@ -176,12 +197,15 @@ def backend_request_2(session_id: str, tool_result: dict) -> str: time.sleep(0.05) final_answer = "The capital of France is Paris." with tracer.start_as_current_span("llm.request") as llm2: - set_attrs(llm2, llm_attrs( - model="gpt-4o", - prompt=[{"role": "tool", "content": tool_result}], - completion=[{"role": "assistant", "content": final_answer}], - usage={"input_tokens": 690, "output_tokens": 122}, - )) + set_attrs( + llm2, + llm_attrs( + model="gpt-4o", + prompt=[{"role": "tool", "content": tool_result}], + completion=[{"role": "assistant", "content": final_answer}], + usage={"input_tokens": 690, "output_tokens": 122}, + ), + ) time.sleep(0.2) ctx_data["final_answer"] = final_answer diff --git a/integrations/otel/distributed_tracing/run.sh b/integrations/otel/distributed_tracing/run.sh new file mode 100644 index 0000000..18e8a35 --- /dev/null +++ b/integrations/otel/distributed_tracing/run.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +set -e + +export OPIK_PROJECT_NAME="otel-distributed-tracing" + +uv sync + +# Both scripts support DRY_RUN mode when credentials are absent, +# and send real traces to Opik when OPIK_API_KEY + OPIK_WORKSPACE are set. +uv run python option_a_explicit_ids.py +uv run python option_b_w3c_propagation.py