5. Spans for Embeddings and Tool Calls

Capsule description

In the previous capsules you configured the OpenTelemetry SDK, created your first tracer, and learned to instrument LLM calls with spans that capture prompt tokens, completion tokens, model, and latency. Those spans represent part of the work your system does — but only part. A production AI system doesn't just call the LLM. It generates embeddings to search for context. It executes tools like web searches, calculators, or database queries. Each one of those operations needs its own span if you want to understand what really happened in a request.

The problem with instrumenting only the LLM call is that it leaves you blind to the rest of the pipeline. If your RAG system takes 3 seconds, how much is embedding generation? How much is vector search? How much is prompt construction? Without spans for each step, you only know that "it was slow" — not where or why.

This capsule teaches you to create spans for the two most common operations after LLM calls: embedding API calls and tool calls in agents. You're going to learn to capture the relevant attributes of each operation, to nest spans as children of a main trace, and to understand why the span hierarchy is what makes tracing useful for debugging.

Spans for Embedding API Calls

What to capture in an embedding span

When you call an embedding API (OpenAI, Cohere, or any provider), there are specific attributes you need to record so the span is operationally useful:

Attribute                     Why?
──────────────────────────────────────────────────────────────────
gen_ai.system                 Identify the provider (openai, cohere)
gen_ai.request.model          The embedding model used
gen_ai.operation.name         "embeddings" — operation type
input_text_length             Input text size (chars)
input_token_count             Input tokens (for cost)
embedding_dimensions          Dimensions of the resulting vector
embedding_count               Number of embeddings generated
latency_ms                    Duration of the call

Instrument an embedding call

import time
from openai import OpenAI
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({
    "service.name": "ai-embedding-service",
    "service.version": "1.0.0",
})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("embedding.instrumentation", "1.0.0")

client = OpenAI()

EMBEDDING_PRICING = {
    "text-embedding-3-small": 0.00002,
    "text-embedding-3-large": 0.00013,
    "text-embedding-ada-002": 0.0001,
}


def create_embedding(
    text: str,
    model: str = "text-embedding-3-small",
) -> list[float]:
    with tracer.start_as_current_span(
        "gen_ai.embeddings",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.operation.name", "embeddings")
        span.set_attribute("input.text_length", len(text))

        start = time.perf_counter()

        try:
            response = client.embeddings.create(
                model=model,
                input=text,
            )

            duration_ms = (time.perf_counter() - start) * 1000
            embedding = response.data[0].embedding
            usage = response.usage

            span.set_attribute("gen_ai.usage.input_tokens", usage.total_tokens)
            span.set_attribute("embedding.dimensions", len(embedding))
            span.set_attribute("embedding.count", len(response.data))
            span.set_attribute("duration_ms", round(duration_ms, 2))

            price_per_1k = EMBEDDING_PRICING.get(model, 0.0001)
            cost = usage.total_tokens / 1000 * price_per_1k
            span.set_attribute("gen_ai.usage.cost_usd", round(cost, 8))

            span.set_status(StatusCode.OK)
            return embedding

        except Exception as e:
            duration_ms = (time.perf_counter() - start) * 1000
            span.set_attribute("duration_ms", round(duration_ms, 2))
            span.set_status(StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise


embedding = create_embedding("What is OpenTelemetry?")
print(f"Embedding generated: {len(embedding)} dimensions")
print(f"First 5 values: {embedding[:5]}")

The resulting span has all the information you need for debugging and cost tracking: which model was used, how many tokens it consumed, how much it cost, how long it took, and the dimensions of the resulting vector.

Instrument multiple embeddings in a batch

In a RAG system, you sometimes generate embeddings for multiple texts in a single call. The span should reflect that.

import time
from openai import OpenAI
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode

tracer = trace.get_tracer("embedding.instrumentation", "1.0.0")
client = OpenAI()

EMBEDDING_PRICING = {
    "text-embedding-3-small": 0.00002,
    "text-embedding-3-large": 0.00013,
}


def create_embeddings_batch(
    texts: list[str],
    model: str = "text-embedding-3-small",
) -> list[list[float]]:
    with tracer.start_as_current_span(
        "gen_ai.embeddings.batch",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.operation.name", "embeddings")
        span.set_attribute("input.batch_size", len(texts))
        span.set_attribute(
            "input.total_text_length",
            sum(len(t) for t in texts),
        )

        start = time.perf_counter()

        try:
            response = client.embeddings.create(
                model=model,
                input=texts,
            )

            duration_ms = (time.perf_counter() - start) * 1000
            embeddings = [item.embedding for item in response.data]
            usage = response.usage

            span.set_attribute("gen_ai.usage.input_tokens", usage.total_tokens)
            span.set_attribute("embedding.dimensions", len(embeddings[0]))
            span.set_attribute("embedding.count", len(embeddings))
            span.set_attribute("duration_ms", round(duration_ms, 2))
            span.set_attribute(
                "tokens_per_text_avg",
                round(usage.total_tokens / len(texts), 1),
            )

            price_per_1k = EMBEDDING_PRICING.get(model, 0.0001)
            cost = usage.total_tokens / 1000 * price_per_1k
            span.set_attribute("gen_ai.usage.cost_usd", round(cost, 8))

            span.set_status(StatusCode.OK)
            return embeddings

        except Exception as e:
            duration_ms = (time.perf_counter() - start) * 1000
            span.set_attribute("duration_ms", round(duration_ms, 2))
            span.set_status(StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise


documents = [
    "OpenTelemetry is the observability standard",
    "Embeddings convert text into numeric vectors",
    "FastAPI is a modern web framework for Python",
]
results = create_embeddings_batch(documents)
print(f"Embeddings generated: {len(results)}")
for i, emb in enumerate(results):
    print(f"  Doc {i+1}: {len(emb)} dimensions")

The key difference from the individual span: you capture batch_size and tokens_per_text_avg. This lets you detect batch calls with excessively long texts (which consume more tokens and cost more) and optimize the batch size.

Spans for Tool Calls in Agents

What to capture in a tool call span

When an agent executes a tool (web search, SQL query, calculator, external API call), each execution deserves its own span:

Attribute                     Why?
──────────────────────────────────────────────────────────────────
tool.name                     Which tool was executed
tool.input                    What it received as input (truncated)
tool.output                   What it returned (truncated)
tool.status                   success / error / timeout
tool.duration_ms              How long it took
tool.source                   internal / external
error.type                    If it failed, what type of error

Instrument an individual tool call

import time
import json
import math
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({
    "service.name": "ai-agent-service",
    "service.version": "1.0.0",
})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent.instrumentation", "1.0.0")


def truncate(text: str, max_len: int = 200) -> str:
    if len(text) <= max_len:
        return text
    return text[:max_len] + "..."


def tool_web_search(query: str) -> dict:
    """Simulate a web search."""
    results = {
        "query": query,
        "results": [
            {
                "title": f"Result about: {query}",
                "snippet": f"Relevant information about {query} found on the web.",
                "url": f"https://example.com/search?q={query.replace(' ', '+')}",
            }
        ],
        "total_results": 1,
    }
    time.sleep(0.3)
    return results


def tool_calculator(expression: str) -> dict:
    """Safely evaluate a mathematical expression."""
    allowed = {
        "abs": abs, "round": round, "min": min, "max": max,
        "pow": pow, "sqrt": math.sqrt, "pi": math.pi, "e": math.e,
    }
    result = eval(expression, {"__builtins__": {}}, allowed)
    return {"expression": expression, "result": result}


def execute_tool_with_span(
    tool_name: str,
    tool_fn,
    tool_input: str,
) -> dict:
    with tracer.start_as_current_span(
        f"tool.{tool_name}",
        kind=SpanKind.INTERNAL,
    ) as span:
        span.set_attribute("tool.name", tool_name)
        span.set_attribute("tool.input", truncate(tool_input))
        span.set_attribute("tool.source", "internal")

        start = time.perf_counter()

        try:
            result = tool_fn(tool_input)
            duration_ms = (time.perf_counter() - start) * 1000

            output_str = json.dumps(result, ensure_ascii=False)
            span.set_attribute("tool.output", truncate(output_str))
            span.set_attribute("tool.status", "success")
            span.set_attribute("tool.duration_ms", round(duration_ms, 2))
            span.set_status(StatusCode.OK)

            return result

        except Exception as e:
            duration_ms = (time.perf_counter() - start) * 1000
            span.set_attribute("tool.status", "error")
            span.set_attribute("tool.duration_ms", round(duration_ms, 2))
            span.set_attribute("error.type", type(e).__name__)
            span.set_status(StatusCode.ERROR, str(e))
            span.record_exception(e)

            return {"error": str(e)}


search_result = execute_tool_with_span(
    "web_search", tool_web_search, "OpenTelemetry AI monitoring"
)
print(f"Search: {json.dumps(search_result, indent=2, ensure_ascii=False)}")

calc_result = execute_tool_with_span(
    "calculator", tool_calculator, "sqrt(144) + pow(2, 10)"
)
print(f"Calculator: {calc_result}")

Each tool call produces a span named tool.<name>, the input it received, the output it produced, and how long it took. When a tool fails, the span captures the error with record_exception — this appears as an event within the span in Jaeger.

Nested Spans: The Hierarchy That Matters

The problem with flat spans

If all the operations of a request are spans at the same level (without parent-child), your trace looks like this:

[gen_ai.embeddings]        200ms
[vector_search]            50ms
[gen_ai.chat]              1200ms
[tool.web_search]          300ms
[tool.calculator]          5ms

Five loose spans. Which depends on which? Was the tool call before or after the LLM? Was the embedding for the query or for something else? Without a hierarchy, you lose the narrative.

Hierarchical spans tell a story

With parent-child relationships, the same request looks like this:

[agent.request]                                  1800ms
├── [gen_ai.embeddings]                          200ms
├── [vector_search]                              50ms
├── [prompt.construction]                        5ms
├── [gen_ai.chat]                                1200ms
│   ├── [tool.web_search]                        300ms
│   └── [tool.calculator]                        5ms
└── [response.format]                            10ms

Now the story is clear: the agent's request generated an embedding, searched vectors, built the prompt, called the LLM, the LLM decided to use two tools, and finally the response was formatted. If the request was slow, you can see exactly where: the LLM call took 1200ms, and of those, 300ms were the web search tool.

Implement a complete trace with child spans

import time
import json
import math
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({
    "service.name": "ai-agent-service",
    "service.version": "1.0.0",
})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent.instrumentation", "1.0.0")


def truncate(text: str, max_len: int = 200) -> str:
    return text[:max_len] + "..." if len(text) > max_len else text


def simulate_embedding(text: str, model: str = "text-embedding-3-small"):
    with tracer.start_as_current_span(
        "gen_ai.embeddings",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.operation.name", "embeddings")
        span.set_attribute("input.text_length", len(text))

        time.sleep(0.15)

        fake_tokens = len(text.split()) * 2
        span.set_attribute("gen_ai.usage.input_tokens", fake_tokens)
        span.set_attribute("embedding.dimensions", 1536)
        span.set_attribute("duration_ms", 150.0)
        span.set_status(StatusCode.OK)

        return [0.1] * 1536


def simulate_vector_search(query_embedding: list[float], top_k: int = 3):
    with tracer.start_as_current_span(
        "vector_search",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("vector_db.system", "pinecone")
        span.set_attribute("vector_db.top_k", top_k)
        span.set_attribute("vector_db.dimensions", len(query_embedding))

        time.sleep(0.05)

        results = [
            {"id": f"doc_{i}", "score": 0.95 - i * 0.05, "text": f"Relevant context #{i+1}"}
            for i in range(top_k)
        ]

        span.set_attribute("vector_db.results_count", len(results))
        span.set_attribute("vector_db.top_score", results[0]["score"])
        span.set_attribute("duration_ms", 50.0)
        span.set_status(StatusCode.OK)

        return results


def simulate_llm_call(prompt: str, tools_available: list[str]):
    with tracer.start_as_current_span(
        "gen_ai.chat",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", "gpt-4o-mini")
        span.set_attribute("gen_ai.operation.name", "chat")
        span.set_attribute("gen_ai.request.max_tokens", 500)
        span.set_attribute("input.prompt_length", len(prompt))
        span.set_attribute("tools.available", json.dumps(tools_available))

        time.sleep(0.5)

        for tool_name in tools_available[:2]:
            execute_tool_in_agent(tool_name)

        fake_prompt_tokens = len(prompt.split()) * 2
        fake_completion_tokens = 150

        span.set_attribute("gen_ai.usage.prompt_tokens", fake_prompt_tokens)
        span.set_attribute("gen_ai.usage.completion_tokens", fake_completion_tokens)
        span.set_attribute("gen_ai.response.finish_reason", "stop")
        span.set_attribute("tools.called_count", min(2, len(tools_available)))
        span.set_attribute("duration_ms", 800.0)
        span.set_status(StatusCode.OK)

        return "Response generated by the agent with information from tools."


def execute_tool_in_agent(tool_name: str):
    with tracer.start_as_current_span(
        f"tool.{tool_name}",
        kind=SpanKind.INTERNAL,
    ) as span:
        span.set_attribute("tool.name", tool_name)
        span.set_attribute("tool.source", "external" if "search" in tool_name else "internal")

        start = time.perf_counter()

        if "search" in tool_name:
            time.sleep(0.2)
            span.set_attribute("tool.input", "search query")
            span.set_attribute("tool.output", truncate("Results found"))
        elif "calc" in tool_name:
            time.sleep(0.01)
            span.set_attribute("tool.input", "sqrt(144)")
            span.set_attribute("tool.output", "12.0")
        else:
            time.sleep(0.05)
            span.set_attribute("tool.input", "generic input")
            span.set_attribute("tool.output", "generic output")

        duration_ms = (time.perf_counter() - start) * 1000
        span.set_attribute("tool.duration_ms", round(duration_ms, 2))
        span.set_attribute("tool.status", "success")
        span.set_status(StatusCode.OK)


def agent_request(user_query: str) -> str:
    """Process a complete agent request with all steps instrumented."""
    with tracer.start_as_current_span(
        "agent.request",
        kind=SpanKind.SERVER,
    ) as span:
        span.set_attribute("agent.query", truncate(user_query))
        span.set_attribute("agent.pipeline", "rag_with_tools")

        start = time.perf_counter()

        query_embedding = simulate_embedding(user_query)

        context_docs = simulate_vector_search(query_embedding)

        with tracer.start_as_current_span("prompt.construction") as build_span:
            context_text = "\n".join(doc["text"] for doc in context_docs)
            full_prompt = f"Context:\n{context_text}\n\nQuestion: {user_query}"
            build_span.set_attribute("prompt.context_docs", len(context_docs))
            build_span.set_attribute("prompt.total_length", len(full_prompt))

        tools = ["web_search", "calculator"]
        response = simulate_llm_call(full_prompt, tools)

        duration_ms = (time.perf_counter() - start) * 1000
        span.set_attribute("agent.total_duration_ms", round(duration_ms, 2))
        span.set_attribute("agent.steps_completed", 4)
        span.set_status(StatusCode.OK)

        return response


result = agent_request("How much does it cost to implement observability with OpenTelemetry?")
print(f"\nResult: {result}")

When you run this, you'll see in the console (or in Jaeger) a hierarchical trace:

agent.request                              ~1000ms
├── gen_ai.embeddings                      ~150ms
├── vector_search                          ~50ms
├── prompt.construction                    ~1ms
└── gen_ai.chat                            ~800ms
    ├── tool.web_search                    ~200ms
    └── tool.calculator                    ~10ms

The key is that start_as_current_span automatically establishes the parent-child relationship. When you create a span inside another active span, OTel makes it a child. You don't need to pass trace IDs manually — OTel's context management does it for you.

Comparison: Flat Spans vs Hierarchical Spans

The hierarchy isn't optional for AI systems. A typical agent request has 5-15 nested operations. Without a hierarchy, debugging is looking for a needle in a haystack. With a hierarchy, it's following a story from start to finish.

Connection with the Project

The spans you learned to create here — embeddings, tool calls, and the parent-child hierarchy — are exactly the ones you're going to implement in the module's project (capsule 08). Your OTel Instrumented AI App will have a RAG pipeline where each step (embed query → search → construct prompt → LLM call → respond) is a child span of the main trace. The attributes you capture in each span (model, tokens, dimensions, cost) feed the module 4 dashboards and the module 5 alerts.

Troubleshooting

"The spans appear but without a parent-child relationship"

Verify that you're using start_as_current_span and not start_span. The first automatically establishes the parent context; the second creates an independent span. If you need to use start_span, pass the context explicitly:

parent_span = tracer.start_span("parent")
ctx = trace.set_span_in_context(parent_span)
child_span = tracer.start_span("child", context=ctx)

"The embedding span doesn't capture the tokens"

The OpenAI embeddings API returns the token count in response.usage.total_tokens. Make sure you access usage.total_tokens, not usage.prompt_tokens (which is chat completions terminology). For embeddings, only total_tokens exists.

"The tool calls appear outside the LLM span"

If the tool call executes outside the with block of the LLM span, OTel doesn't associate it as a child. Make sure execute_tool_in_agent() is called inside the with tracer.start_as_current_span("gen_ai.chat") block. Python's context management works by with scope.

"Should I capture the full input/output content of the tools?"

For debugging, yes — but truncate to a reasonable size (200-500 chars). Tracing backends have size limits per attribute (Jaeger: 1MB per span, but very long attributes degrade search performance). Use the truncate() function to keep the attributes informative but not excessive.

"When do I use SpanKind.CLIENT vs SpanKind.INTERNAL?"

Use CLIENT when the span represents a call to an external service (embedding API, LLM API, vector database). Use INTERNAL when it's logic within your service (local tool execution, prompt construction, formatting). The distinction helps backends like Jaeger differentiate your own latency vs external dependency latency.

Exercises

Exercise 1: Instrument an embedding with cost attributes (Easy)

Create a function embed_with_cost_tracking that generates an embedding for a given text and captures in the span: model, tokens, dimensions, cost in USD, and latency. Use the provided pricing table.

from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("exercises", "1.0.0")

PRICING = {
    "text-embedding-3-small": 0.00002,
    "text-embedding-3-large": 0.00013,
}

def embed_with_cost_tracking(text: str, model: str = "text-embedding-3-small"):
    # Your implementation here
    pass

# Test
result = embed_with_cost_tracking("What is observability in AI systems?")

import time
from openai import OpenAI
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.name": "exercise-embedding"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("exercises", "1.0.0")
client = OpenAI()

PRICING = {
    "text-embedding-3-small": 0.00002,
    "text-embedding-3-large": 0.00013,
}


def embed_with_cost_tracking(
    text: str,
    model: str = "text-embedding-3-small",
) -> dict:
    with tracer.start_as_current_span(
        "gen_ai.embeddings",
        kind=SpanKind.CLIENT,
    ) as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.operation.name", "embeddings")
        span.set_attribute("input.text_length", len(text))

        start = time.perf_counter()

        response = client.embeddings.create(model=model, input=text)

        duration_ms = (time.perf_counter() - start) * 1000
        embedding = response.data[0].embedding
        tokens = response.usage.total_tokens
        price = PRICING.get(model, 0.0001)
        cost = tokens / 1000 * price

        span.set_attribute("gen_ai.usage.input_tokens", tokens)
        span.set_attribute("embedding.dimensions", len(embedding))
        span.set_attribute("gen_ai.usage.cost_usd", round(cost, 8))
        span.set_attribute("duration_ms", round(duration_ms, 2))
        span.set_status(StatusCode.OK)

        return {
            "embedding": embedding,
            "tokens": tokens,
            "cost_usd": round(cost, 8),
            "dimensions": len(embedding),
            "duration_ms": round(duration_ms, 2),
        }


result = embed_with_cost_tracking("What is observability in AI systems?")
print(f"Tokens: {result['tokens']}")
print(f"Dimensions: {result['dimensions']}")
print(f"Cost: ${result['cost_usd']}")
print(f"Latency: {result['duration_ms']}ms")

Exercise 2: Instrument a tool call with error handling (Medium)

Create a decorator @traced_tool that wraps any tool function and automatically generates a span with tool.name, tool.input, tool.output, tool.status, and tool.duration_ms. If the tool fails, it must capture the exception in the span.

import functools

def traced_tool(tool_name: str):
    # Your decorator implementation here
    pass

@traced_tool("weather_api")
def get_weather(city: str) -> dict:
    if city == "Atlantis":
        raise ValueError("City not found")
    return {"city": city, "temp_c": 22, "condition": "sunny"}

# Test
print(get_weather("Mexico City"))
print(get_weather("Atlantis"))  # Must capture the error in the span

import time
import json
import functools
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.name": "exercise-tools"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("exercises", "1.0.0")


def truncate(text: str, max_len: int = 200) -> str:
    return text[:max_len] + "..." if len(text) > max_len else text


def traced_tool(tool_name: str):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span(
                f"tool.{tool_name}",
                kind=SpanKind.INTERNAL,
            ) as span:
                span.set_attribute("tool.name", tool_name)
                input_repr = json.dumps(
                    {"args": [str(a) for a in args], "kwargs": kwargs},
                    ensure_ascii=False,
                )
                span.set_attribute("tool.input", truncate(input_repr))

                start = time.perf_counter()

                try:
                    result = fn(*args, **kwargs)
                    duration_ms = (time.perf_counter() - start) * 1000

                    output_repr = json.dumps(result, ensure_ascii=False)
                    span.set_attribute("tool.output", truncate(output_repr))
                    span.set_attribute("tool.status", "success")
                    span.set_attribute("tool.duration_ms", round(duration_ms, 2))
                    span.set_status(StatusCode.OK)

                    return result

                except Exception as e:
                    duration_ms = (time.perf_counter() - start) * 1000
                    span.set_attribute("tool.status", "error")
                    span.set_attribute("tool.duration_ms", round(duration_ms, 2))
                    span.set_attribute("error.type", type(e).__name__)
                    span.set_attribute("error.message", str(e))
                    span.set_status(StatusCode.ERROR, str(e))
                    span.record_exception(e)

                    return {"error": str(e), "error_type": type(e).__name__}

        return wrapper
    return decorator


@traced_tool("weather_api")
def get_weather(city: str) -> dict:
    if city == "Atlantis":
        raise ValueError("City not found")
    return {"city": city, "temp_c": 22, "condition": "sunny"}


@traced_tool("calculator")
def calculate(expression: str) -> dict:
    result = eval(expression, {"__builtins__": {}}, {"abs": abs, "round": round})
    return {"expression": expression, "result": result}


print("Test 1 (success):", get_weather("Mexico City"))
print("Test 2 (error):", get_weather("Atlantis"))
print("Test 3 (success):", calculate("abs(-42) + round(3.7)"))

Exercise 3: Create a complete hierarchical trace (Medium)

Implement a function rag_pipeline that executes a complete RAG pipeline with nested spans: embed query → vector search → construct prompt → LLM call. Each step must be a child span of the main span rag.pipeline. Use simulated functions (no real API calls).

def rag_pipeline(query: str) -> str:
    # Your implementation here
    # It must create:
    #   rag.pipeline (parent)
    #   ├── gen_ai.embeddings (child)
    #   ├── vector_search (child)
    #   ├── prompt.construction (child)
    #   └── gen_ai.chat (child)
    pass

import time
import json
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.name": "exercise-rag-pipeline"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("exercises", "1.0.0")


def rag_pipeline(query: str) -> str:
    with tracer.start_as_current_span(
        "rag.pipeline",
        kind=SpanKind.SERVER,
    ) as root_span:
        root_span.set_attribute("rag.query", query)
        root_span.set_attribute("rag.pipeline_version", "1.0")
        pipeline_start = time.perf_counter()

        with tracer.start_as_current_span(
            "gen_ai.embeddings",
            kind=SpanKind.CLIENT,
        ) as emb_span:
            emb_span.set_attribute("gen_ai.system", "openai")
            emb_span.set_attribute("gen_ai.request.model", "text-embedding-3-small")
            emb_span.set_attribute("input.text_length", len(query))
            time.sleep(0.1)
            emb_span.set_attribute("gen_ai.usage.input_tokens", len(query.split()) * 2)
            emb_span.set_attribute("embedding.dimensions", 1536)
            emb_span.set_status(StatusCode.OK)
            query_embedding = [0.1] * 1536

        with tracer.start_as_current_span(
            "vector_search",
            kind=SpanKind.CLIENT,
        ) as vs_span:
            vs_span.set_attribute("vector_db.system", "pinecone")
            vs_span.set_attribute("vector_db.top_k", 3)
            time.sleep(0.05)
            context_docs = [
                "OTel is the observability standard for cloud native.",
                "Traces let you follow a request across services.",
                "Metrics quantify the system's behavior.",
            ]
            vs_span.set_attribute("vector_db.results_count", len(context_docs))
            vs_span.set_attribute("vector_db.top_score", 0.92)
            vs_span.set_status(StatusCode.OK)

        with tracer.start_as_current_span(
            "prompt.construction",
        ) as pc_span:
            context = "\n".join(f"- {doc}" for doc in context_docs)
            prompt = (
                f"Based on the following context:\n{context}\n\n"
                f"Answer: {query}"
            )
            pc_span.set_attribute("prompt.context_docs", len(context_docs))
            pc_span.set_attribute("prompt.total_length", len(prompt))
            pc_span.set_status(StatusCode.OK)

        with tracer.start_as_current_span(
            "gen_ai.chat",
            kind=SpanKind.CLIENT,
        ) as llm_span:
            llm_span.set_attribute("gen_ai.system", "openai")
            llm_span.set_attribute("gen_ai.request.model", "gpt-4o-mini")
            llm_span.set_attribute("gen_ai.request.max_tokens", 500)
            llm_span.set_attribute("gen_ai.usage.prompt_tokens", len(prompt.split()) * 2)
            time.sleep(0.3)
            response = (
                f"Based on the provided context, I can tell you that "
                f"OpenTelemetry is the observability standard that lets you "
                f"instrument applications for tracing, metrics, and logs."
            )
            llm_span.set_attribute("gen_ai.usage.completion_tokens", len(response.split()) * 2)
            llm_span.set_attribute("gen_ai.response.finish_reason", "stop")
            llm_span.set_status(StatusCode.OK)

        pipeline_ms = (time.perf_counter() - pipeline_start) * 1000
        root_span.set_attribute("rag.total_duration_ms", round(pipeline_ms, 2))
        root_span.set_attribute("rag.steps_completed", 4)
        root_span.set_status(StatusCode.OK)

        return response


output = rag_pipeline("What is OpenTelemetry and what is it for?")
print(f"\nResponse: {output}")

Exercise 4: Instrument an agent with multiple tool calls (Hard)

Create an InstrumentedAgent that processes user queries. The agent has 3 available tools (weather, calculator, translator). Each query generates a trace with the pattern: agent.request → gen_ai.chat → tool calls (0-N) → response. The agent decides which tools to use based on keywords in the query.

class InstrumentedAgent:
    def __init__(self):
        # Your implementation here
        pass

    def process(self, query: str) -> str:
        # Must generate a complete hierarchical trace
        pass

# Test
agent = InstrumentedAgent()
agent.process("What temperature is it in Madrid? Convert to Fahrenheit")
agent.process("Translate 'observability' to Spanish")
agent.process("How much is 2^10 + sqrt(256)?")

import time
import json
import math
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.name": "exercise-agent"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("exercises", "1.0.0")


class InstrumentedAgent:
    def __init__(self):
        self.tools = {
            "weather": self._tool_weather,
            "calculator": self._tool_calculator,
            "translator": self._tool_translator,
        }
        self.tool_keywords = {
            "weather": ["temperature", "climate", "weather", "rain", "is it in"],
            "calculator": ["calculate", "how much is", "sqrt", "sum", "2^", "pow"],
            "translator": ["translate", "translation", "english", "spanish"],
        }

    def _tool_weather(self, input_data: str) -> dict:
        time.sleep(0.2)
        return {"city": input_data, "temp_c": 18, "condition": "partly_cloudy"}

    def _tool_calculator(self, input_data: str) -> dict:
        time.sleep(0.01)
        allowed = {
            "abs": abs, "round": round, "pow": pow,
            "sqrt": math.sqrt, "pi": math.pi,
        }
        try:
            result = eval(input_data, {"__builtins__": {}}, allowed)
            return {"expression": input_data, "result": result}
        except Exception as e:
            return {"expression": input_data, "error": str(e)}

    def _tool_translator(self, input_data: str) -> dict:
        time.sleep(0.1)
        translations = {
            "observability": "observabilidad",
            "monitoring": "monitoreo",
            "tracing": "rastreo",
        }
        word = input_data.lower().strip("'\"")
        translated = translations.get(word, f"[translation of '{word}']")
        return {"original": word, "translated": translated, "lang": "es"}

    def _detect_tools(self, query: str) -> list[str]:
        query_lower = query.lower()
        needed = []
        for tool_name, keywords in self.tool_keywords.items():
            if any(kw in query_lower for kw in keywords):
                needed.append(tool_name)
        return needed

    def _execute_tool(self, tool_name: str, input_data: str) -> dict:
        with tracer.start_as_current_span(
            f"tool.{tool_name}",
            kind=SpanKind.INTERNAL,
        ) as span:
            span.set_attribute("tool.name", tool_name)
            span.set_attribute("tool.input", input_data[:200])

            start = time.perf_counter()
            tool_fn = self.tools[tool_name]

            try:
                result = tool_fn(input_data)
                duration_ms = (time.perf_counter() - start) * 1000

                output_str = json.dumps(result, ensure_ascii=False)
                span.set_attribute("tool.output", output_str[:200])
                span.set_attribute("tool.status", "success")
                span.set_attribute("tool.duration_ms", round(duration_ms, 2))
                span.set_status(StatusCode.OK)

                return result

            except Exception as e:
                duration_ms = (time.perf_counter() - start) * 1000
                span.set_attribute("tool.status", "error")
                span.set_attribute("tool.duration_ms", round(duration_ms, 2))
                span.set_attribute("error.type", type(e).__name__)
                span.set_status(StatusCode.ERROR, str(e))
                span.record_exception(e)
                return {"error": str(e)}

    def process(self, query: str) -> str:
        with tracer.start_as_current_span(
            "agent.request",
            kind=SpanKind.SERVER,
        ) as root_span:
            root_span.set_attribute("agent.query", query[:300])
            pipeline_start = time.perf_counter()

            needed_tools = self._detect_tools(query)
            root_span.set_attribute("agent.tools_detected", json.dumps(needed_tools))

            with tracer.start_as_current_span(
                "gen_ai.chat",
                kind=SpanKind.CLIENT,
            ) as llm_span:
                llm_span.set_attribute("gen_ai.system", "openai")
                llm_span.set_attribute("gen_ai.request.model", "gpt-4o-mini")
                llm_span.set_attribute("tools.available", json.dumps(list(self.tools.keys())))
                llm_span.set_attribute("tools.needed", json.dumps(needed_tools))

                time.sleep(0.2)

                tool_results = {}
                for tool_name in needed_tools:
                    tool_input = self._extract_tool_input(query, tool_name)
                    tool_results[tool_name] = self._execute_tool(tool_name, tool_input)

                fake_prompt_tokens = len(query.split()) * 3
                fake_completion_tokens = 80
                llm_span.set_attribute("gen_ai.usage.prompt_tokens", fake_prompt_tokens)
                llm_span.set_attribute("gen_ai.usage.completion_tokens", fake_completion_tokens)
                llm_span.set_attribute("gen_ai.response.finish_reason", "stop")
                llm_span.set_attribute("tools.called_count", len(needed_tools))
                llm_span.set_status(StatusCode.OK)

            response = self._format_response(query, tool_results)

            pipeline_ms = (time.perf_counter() - pipeline_start) * 1000
            root_span.set_attribute("agent.duration_ms", round(pipeline_ms, 2))
            root_span.set_attribute("agent.tools_used", len(needed_tools))
            root_span.set_status(StatusCode.OK)

            return response

    def _extract_tool_input(self, query: str, tool_name: str) -> str:
        if tool_name == "weather":
            for word in ["Madrid", "Mexico", "Barcelona", "London"]:
                if word.lower() in query.lower():
                    return word
            return "Madrid"
        elif tool_name == "calculator":
            import re
            match = re.search(r'[\d\w\+\-\*/\^()sqrt.]+', query)
            return match.group() if match else "0"
        elif tool_name == "translator":
            import re
            match = re.search(r"'([^']+)'", query)
            return match.group(1) if match else query.split()[-1]
        return query

    def _format_response(self, query: str, tool_results: dict) -> str:
        parts = [f"Response for: {query}"]
        for tool, result in tool_results.items():
            parts.append(f"  [{tool}]: {json.dumps(result, ensure_ascii=False)}")
        return "\n".join(parts)


agent = InstrumentedAgent()

print("=" * 60)
print("Test 1: Weather + Calculator")
r1 = agent.process("What temperature is it in Madrid? Calculate what it is in Fahrenheit")
print(r1)

print("\nTest 2: Translator")
r2 = agent.process("Translate 'observability' to Spanish")
print(r2)

print("\nTest 3: Calculator")
r3 = agent.process("How much is pow(2, 10) + sqrt(256)?")
print(r3)
print("=" * 60)

Exercise 5: Compare flat vs hierarchical traces (Medium)

Implement the same operation (embedding + search + LLM) in two ways: one with flat spans (no parent-child) and another with hierarchical spans. Print a summary that shows the difference in the information each trace captures.

import time
from opentelemetry import trace
from opentelemetry.trace import SpanKind, StatusCode, NonRecordingSpan
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource
from opentelemetry.context import Context

resource = Resource.create({"service.name": "exercise-comparison"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("exercises", "1.0.0")


def flat_pipeline(query: str):
    """All spans at the same level, no parent-child."""
    empty_ctx = trace.set_span_in_context(NonRecordingSpan(
        trace.INVALID_SPAN_CONTEXT
    ))

    span1 = tracer.start_span("embedding", context=empty_ctx)
    span1.set_attribute("query", query)
    time.sleep(0.1)
    span1.set_attribute("tokens", 15)
    span1.end()

    span2 = tracer.start_span("search", context=empty_ctx)
    span2.set_attribute("top_k", 3)
    time.sleep(0.05)
    span2.set_attribute("results", 3)
    span2.end()

    span3 = tracer.start_span("llm_call", context=empty_ctx)
    span3.set_attribute("model", "gpt-4o-mini")
    time.sleep(0.3)
    span3.set_attribute("completion_tokens", 100)
    span3.end()

    return "flat response"


def hierarchical_pipeline(query: str):
    """Nested spans with parent-child relationships."""
    with tracer.start_as_current_span(
        "rag.pipeline",
        kind=SpanKind.SERVER,
    ) as root:
        root.set_attribute("query", query)
        start = time.perf_counter()

        with tracer.start_as_current_span(
            "gen_ai.embeddings",
            kind=SpanKind.CLIENT,
        ) as emb:
            emb.set_attribute("gen_ai.request.model", "text-embedding-3-small")
            time.sleep(0.1)
            emb.set_attribute("gen_ai.usage.input_tokens", 15)
            emb.set_attribute("embedding.dimensions", 1536)

        with tracer.start_as_current_span(
            "vector_search",
            kind=SpanKind.CLIENT,
        ) as search:
            search.set_attribute("vector_db.top_k", 3)
            time.sleep(0.05)
            search.set_attribute("vector_db.results_count", 3)

        with tracer.start_as_current_span(
            "gen_ai.chat",
            kind=SpanKind.CLIENT,
        ) as llm:
            llm.set_attribute("gen_ai.request.model", "gpt-4o-mini")
            time.sleep(0.3)
            llm.set_attribute("gen_ai.usage.completion_tokens", 100)
            llm.set_attribute("gen_ai.response.finish_reason", "stop")

        total_ms = (time.perf_counter() - start) * 1000
        root.set_attribute("total_duration_ms", round(total_ms, 2))

    return "hierarchical response"


print("=" * 60)
print("FLAT PIPELINE (no hierarchy)")
print("-" * 60)
flat_pipeline("What is OTel?")

print("\n" + "=" * 60)
print("HIERARCHICAL PIPELINE (with hierarchy)")
print("-" * 60)
hierarchical_pipeline("What is OTel?")

print("\n" + "=" * 60)
print("COMPARISON")
print("-" * 60)
comparison = [
    ("Trace IDs", "3 different (not correlated)", "1 shared"),
    ("Parent-child", "None", "pipeline → embedding → search → llm"),
    ("Total duration", "Must be summed manually", "Automatic in root span"),
    ("Debugging", "'It was slow' — where?", "'The LLM took 300ms of 450ms total'"),
    ("Visualization", "3 loose lines", "Tree with waterfall"),
    ("Error tracking", "Isolated error", "Error propagated to the parent"),
]
for dim, flat_val, hier_val in comparison:
    print(f"\n  {dim}:")
    print(f"    Flat:          {flat_val}")
    print(f"    Hierarchical:  {hier_val}")
print("=" * 60)

Summary

• Embedding spans capture model, tokens, dimensions, cost, and latency. They're critical in RAG pipelines where the embedding is the first step and its latency impacts the whole request.
• Tool call spans capture tool name, input, output, status, and duration. In agents, each tool execution is an operation with its own risk of failure and latency.
• The span hierarchy (parent-child) is what makes tracing useful for debugging. Without a hierarchy, you have a list of events. With a hierarchy, you have a story you can follow.
• start_as_current_span is your main tool: it automatically establishes parent-child relationships based on Python's scope. Whatever is inside the with block becomes a child.
• Truncate inputs/outputs in the span attributes. Tracing backends have limits and long attributes degrade performance.
• SpanKind matters: use CLIENT for external calls (APIs, databases), INTERNAL for local logic (tools, formatting). It helps backends calculate your own latency vs dependencies.
• The spans you built here (embeddings, tools, hierarchy) are the foundation of the project (capsule 08) where you'll instrument a complete RAG pipeline visible in Jaeger.

Additional Resources

1. OpenTelemetry Python — Manual Instrumentation — Official guide for creating manual spans with the Python SDK
2. OpenTelemetry Semantic Conventions — GenAI — Standard conventions for generative AI operations (embeddings, chat, completions)
3. OpenTelemetry — Context Propagation — How the context management that enables automatic parent-child works
4. Jaeger — Getting Started — To visualize the hierarchical traces you created
5. OpenAI Embeddings Guide — Reference for the embeddings API to understand the response fields
6. Arize AI — LLM Tracing with OpenTelemetry — Practical example of tracing for LLM applications
7. OpenTelemetry — SpanKind — Documentation on when to use CLIENT, SERVER, INTERNAL, PRODUCER, CONSUMER
8. Grafana Tempo — Trace Visualization — Alternative to Jaeger for trace visualization