5. Spans para Embeddings y Tool Calls

Descripción de la cápsula

En las cápsulas anteriores configuraste el SDK de OpenTelemetry, creaste tu primer tracer, y aprendiste a instrumentar llamadas al LLM con spans que capturan prompt tokens, completion tokens, modelo y latencia. Esos spans representan una parte del trabajo que hace tu sistema — pero solo una parte. Un sistema AI en producción no solo llama al LLM. Genera embeddings para buscar contexto. Ejecuta tools como búsquedas web, calculadoras, o consultas a bases de datos. Cada una de esas operaciones necesita su propio span si quieres entender qué pasó realmente en un request.

El problema con instrumentar solo la llamada al LLM es que te deja ciego ante el resto del pipeline. Si tu sistema RAG tarda 3 segundos, ¿cuánto es la generación de embeddings? ¿Cuánto es la búsqueda vectorial? ¿Cuánto es la construcción del prompt? Sin spans para cada paso, solo sabes que "fue lento" — no dónde ni por qué.

Esta cápsula te enseña a crear spans para las dos operaciones más comunes después de las LLM calls: embedding API calls y tool calls en agentes. Vas a aprender a capturar los atributos relevantes de cada operación, a anidar spans como hijos de un trace principal, y a entender por qué la jerarquía de spans es lo que hace que el tracing sea útil para debugging.

Spans para Embedding API Calls

Qué capturar en un span de embedding

Cuando llamas a un embedding API (OpenAI, Cohere, o cualquier provider), hay atributos específicos que necesitas registrar para que el span sea útil operacionalmente:

Atributo                      ¿Por qué?
──────────────────────────────────────────────────────────────────
gen_ai.system                 Identificar el provider (openai, cohere)
gen_ai.request.model          El modelo de embedding usado
gen_ai.operation.name         "embeddings" — tipo de operación
input_text_length             Tamaño del texto de entrada (chars)
input_token_count             Tokens del input (para costo)
embedding_dimensions          Dimensiones del vector resultante
embedding_count               Cantidad de embeddings generados
latency_ms                    Duración de la llamada

Instrumentar una llamada de embedding

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("¿Qué es OpenTelemetry?")
print(f"Embedding generado: {len(embedding)} dimensiones")
print(f"Primeros 5 valores: {embedding[:5]}")

El span resultante tiene toda la información que necesitas para debugging y cost tracking: qué modelo se usó, cuántos tokens consumió, cuánto costó, cuánto tardó, y las dimensiones del vector resultante.

Instrumentar múltiples embeddings en batch

En un sistema RAG, a veces generas embeddings para múltiples textos en una sola llamada. El span debe reflejar eso.

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 es el estándar de observabilidad",
    "Los embeddings convierten texto en vectores numéricos",
    "FastAPI es un framework web moderno para Python",
]
results = create_embeddings_batch(documents)
print(f"Embeddings generados: {len(results)}")
for i, emb in enumerate(results):
    print(f"  Doc {i+1}: {len(emb)} dimensiones")

La diferencia clave con el span individual: capturas batch_size y tokens_per_text_avg. Esto te permite detectar llamadas batch con textos excesivamente largos (que consumen más tokens y cuestan más) y optimizar el tamaño del batch.

Spans para Tool Calls en Agentes

Qué capturar en un span de tool call

Cuando un agente ejecuta un tool (búsqueda web, consulta SQL, calculadora, llamada a API externa), cada ejecución merece su propio span:

Atributo                      ¿Por qué?
──────────────────────────────────────────────────────────────────
tool.name                     Qué tool se ejecutó
tool.input                    Qué recibió como input (truncado)
tool.output                   Qué devolvió (truncado)
tool.status                   success / error / timeout
tool.duration_ms              Cuánto tardó
tool.source                   internal / external
error.type                    Si falló, qué tipo de error

Instrumentar un tool call individual

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:
    """Simula una búsqueda web."""
    results = {
        "query": query,
        "results": [
            {
                "title": f"Resultado sobre: {query}",
                "snippet": f"Información relevante sobre {query} encontrada en la 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:
    """Evalúa una expresión matemática de forma segura."""
    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}")

Cada tool call produce un span con nombre tool.<nombre>, el input que recibió, el output que produjo, y cuánto tardó. Cuando un tool falla, el span captura el error con record_exception — esto aparece como un evento dentro del span en Jaeger.

Nested Spans: La Jerarquía que Importa

El problema de los spans planos

Si todas las operaciones de un request son spans al mismo nivel (sin parent-child), tu trace se ve así:

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

Cinco spans sueltos. ¿Cuál depende de cuál? ¿El tool call fue antes o después del LLM? ¿El embedding fue para el query o para algo más? Sin jerarquía, pierdes la narrativa.

Spans jerárquicos cuentan una historia

Con parent-child relationships, el mismo request se ve así:

[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

Ahora la historia es clara: el request del agente generó un embedding, buscó en vectores, construyó el prompt, llamó al LLM, el LLM decidió usar dos tools, y finalmente se formateó la respuesta. Si el request fue lento, puedes ver exactamente dónde: la llamada al LLM tomó 1200ms, y de esos, 300ms fueron el tool de búsqueda web.

Implementar un trace completo con 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"Contexto relevante #{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 "Respuesta generada por el agente con información de 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", "query de búsqueda")
            span.set_attribute("tool.output", truncate("Resultados encontrados"))
        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", "input genérico")
            span.set_attribute("tool.output", "output genérico")

        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:
    """Procesa un request completo del agente con todos los pasos instrumentados."""
    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"Contexto:\n{context_text}\n\nPregunta: {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("¿Cuánto cuesta implementar observabilidad con OpenTelemetry?")
print(f"\nResultado: {result}")

Al ejecutar esto, verás en la consola (o en Jaeger) un trace jerárquico:

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

La clave es que start_as_current_span automáticamente establece el parent-child relationship. Cuando creas un span dentro de otro span activo, OTel lo convierte en hijo. No necesitas pasar trace IDs manualmente — el context management de OTel lo hace por ti.

Comparación: Flat Spans vs Hierarchical Spans

La jerarquía no es opcional para sistemas AI. Un request típico de un agente tiene 5-15 operaciones anidadas. Sin jerarquía, debugging es buscar una aguja en un pajar. Con jerarquía, es seguir una historia de principio a fin.

Conexión con Proyecto

Los spans que aprendiste a crear aquí — embeddings, tool calls, y la jerarquía parent-child — son exactamente los que vas a implementar en el proyecto del módulo (cápsula 08). Tu OTel Instrumented AI App tendrá un pipeline RAG donde cada paso (embed query → search → construct prompt → LLM call → respond) es un span hijo del trace principal. Los atributos que captures en cada span (model, tokens, dimensions, cost) alimentan los dashboards del módulo 4 y las alertas del módulo 5.

Troubleshooting

"Los spans aparecen pero sin parent-child relationship"

Verifica que estás usando start_as_current_span y no start_span. El primero establece automáticamente el parent context; el segundo crea un span independiente. Si necesitas usar start_span, pasa el context explícitamente:

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

"El span de embedding no captura los tokens"

La API de embeddings de OpenAI devuelve el token count en response.usage.total_tokens. Asegúrate de acceder a usage.total_tokens, no a usage.prompt_tokens (que es la terminología de chat completions). Para embeddings, solo existe total_tokens.

"Los tool calls aparecen fuera del span del LLM"

Si el tool call se ejecuta fuera del with block del span del LLM, OTel no lo asocia como hijo. Asegúrate de que execute_tool_in_agent() se llama dentro del bloque with tracer.start_as_current_span("gen_ai.chat"). El context management de Python funciona por scope de with.

"¿Debo capturar el contenido completo del input/output de los tools?"

Para debugging, sí — pero trunca a un tamaño razonable (200-500 chars). Los backends de tracing tienen límites de tamaño por atributo (Jaeger: 1MB por span, pero atributos muy largos degradan performance de búsqueda). Usa la función truncate() para mantener los atributos informativos pero no excesivos.

"¿Cuándo uso SpanKind.CLIENT vs SpanKind.INTERNAL?"

Usa CLIENT cuando el span representa una llamada a un servicio externo (embedding API, LLM API, vector database). Usa INTERNAL cuando es lógica dentro de tu servicio (tool execution local, prompt construction, formatting). La distinción ayuda a backends como Jaeger a diferenciar latencia propia vs latencia de dependencias externas.

Ejercicios

Ejercicio 1: Instrumentar un embedding con atributos de costo (Fácil)

Crea una función embed_with_cost_tracking que genere un embedding para un texto dado y capture en el span: model, tokens, dimensiones, costo en USD, y latencia. Usa la tabla de precios proporcionada.

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"):
    # Tu implementación aquí
    pass

# Test
result = embed_with_cost_tracking("¿Qué es observabilidad en sistemas AI?")

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("¿Qué es observabilidad en sistemas AI?")
print(f"Tokens: {result['tokens']}")
print(f"Dimensiones: {result['dimensions']}")
print(f"Costo: ${result['cost_usd']}")
print(f"Latencia: {result['duration_ms']}ms")

Ejercicio 2: Instrumentar un tool call con manejo de errores (Medio)

Crea un decorator @traced_tool que envuelva cualquier función de tool y genere automáticamente un span con tool.name, tool.input, tool.output, tool.status, y tool.duration_ms. Si el tool falla, debe capturar la excepción en el span.

import functools

def traced_tool(tool_name: str):
    # Tu implementación del decorator aquí
    pass

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

# Test
print(get_weather("Ciudad de México"))
print(get_weather("Atlantis"))  # Debe capturar el error en el 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("Ciudad no encontrada")
    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("Ciudad de México"))
print("Test 2 (error):", get_weather("Atlantis"))
print("Test 3 (success):", calculate("abs(-42) + round(3.7)"))

Ejercicio 3: Crear un trace jerárquico completo (Medio)

Implementa una función rag_pipeline que ejecute un pipeline RAG completo con spans anidados: embed query → vector search → construct prompt → LLM call. Cada paso debe ser un child span del span principal rag.pipeline. Usa funciones simuladas (sin llamadas reales a APIs).

def rag_pipeline(query: str) -> str:
    # Tu implementación aquí
    # Debe crear:
    #   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 es el estándar de observabilidad para cloud native.",
                "Los traces permiten seguir un request a través de servicios.",
                "Las métricas cuantifican el comportamiento del sistema.",
            ]
            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"Basándote en el siguiente contexto:\n{context}\n\n"
                f"Responde: {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"Basándome en el contexto proporcionado, puedo decirte que "
                f"OpenTelemetry es el estándar de observabilidad que permite "
                f"instrumentar aplicaciones para tracing, métricas y 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("¿Qué es OpenTelemetry y para qué sirve?")
print(f"\nRespuesta: {output}")

Ejercicio 4: Instrumentar un agente con múltiples tool calls (Difícil)

Crea un InstrumentedAgent que procese queries del usuario. El agente tiene 3 tools disponibles (weather, calculator, translator). Cada query genera un trace con el patrón: agent.request → gen_ai.chat → tool calls (0-N) → response. El agente decide qué tools usar basándose en keywords del query.

class InstrumentedAgent:
    def __init__(self):
        # Tu implementación aquí
        pass

    def process(self, query: str) -> str:
        # Debe generar un trace jerárquico completo
        pass

# Test
agent = InstrumentedAgent()
agent.process("¿Qué temperatura hace en Madrid? Convierte a Fahrenheit")
agent.process("Traduce 'observability' al español")
agent.process("¿Cuánto es 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": ["temperatura", "clima", "weather", "lluvia", "hace en"],
            "calculator": ["calcula", "cuánto es", "sqrt", "suma", "2^", "pow"],
            "translator": ["traduce", "traducir", "translate", "inglés", "español"],
        }

    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"[traducción de '{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", "México", "Barcelona", "Londres"]:
                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"Respuesta para: {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("¿Qué temperatura hace en Madrid? Calcula cuánto es en Fahrenheit")
print(r1)

print("\nTest 2: Translator")
r2 = agent.process("Traduce 'observability' al español")
print(r2)

print("\nTest 3: Calculator")
r3 = agent.process("¿Cuánto es pow(2, 10) + sqrt(256)?")
print(r3)
print("=" * 60)

Ejercicio 5: Comparar traces planos vs jerárquicos (Medio)

Implementa la misma operación (embedding + search + LLM) de dos formas: una con spans planos (sin parent-child) y otra con spans jerárquicos. Imprime un resumen que muestre la diferencia en la información que cada trace captura.

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):
    """Todos los spans al mismo nivel, sin 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 "respuesta flat"


def hierarchical_pipeline(query: str):
    """Spans anidados con 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 "respuesta jerárquica"


print("=" * 60)
print("FLAT PIPELINE (sin jerarquía)")
print("-" * 60)
flat_pipeline("¿Qué es OTel?")

print("\n" + "=" * 60)
print("HIERARCHICAL PIPELINE (con jerarquía)")
print("-" * 60)
hierarchical_pipeline("¿Qué es OTel?")

print("\n" + "=" * 60)
print("COMPARACIÓN")
print("-" * 60)
comparison = [
    ("Trace IDs", "3 diferentes (no correlacionados)", "1 compartido"),
    ("Parent-child", "Ninguno", "pipeline → embedding → search → llm"),
    ("Duración total", "Hay que sumar manualmente", "Automática en root span"),
    ("Debugging", "'Fue lento' — ¿dónde?", "'El LLM tomó 300ms de 450ms total'"),
    ("Visualización", "3 líneas sueltas", "Árbol con waterfall"),
    ("Error tracking", "Error aislado", "Error propagado al parent"),
]
for dim, flat_val, hier_val in comparison:
    print(f"\n  {dim}:")
    print(f"    Flat:         {flat_val}")
    print(f"    Jerárquico:   {hier_val}")
print("=" * 60)

Resumen

• Los spans de embedding capturan model, tokens, dimensions, cost y latency. Son críticos en pipelines RAG donde el embedding es el primer paso y su latencia impacta el request completo.
• Los spans de tool calls capturan tool name, input, output, status y duration. En agentes, cada tool execution es una operación con su propio riesgo de fallo y latencia.
• La jerarquía de spans (parent-child) es lo que hace que el tracing sea útil para debugging. Sin jerarquía, tienes una lista de eventos. Con jerarquía, tienes una historia que puedes seguir.
• start_as_current_span es tu herramienta principal: automáticamente establece relationships parent-child basándose en el scope de Python. Lo que esté dentro del with block se convierte en child.
• Trunca inputs/outputs en los atributos del span. Los backends de tracing tienen límites y los atributos largos degradan performance.
• SpanKind importa: usa CLIENT para llamadas externas (APIs, databases), INTERNAL para lógica local (tools, formatting). Ayuda a los backends a calcular latencia propia vs dependencias.
• Los spans que construiste aquí (embeddings, tools, jerarquía) son la base del proyecto (cápsula 08) donde instrumentarás un pipeline RAG completo visible en Jaeger.

Recursos Adicionales

1. OpenTelemetry Python — Manual Instrumentation — Guía oficial para crear spans manuales con el SDK de Python
2. OpenTelemetry Semantic Conventions — GenAI — Convenciones estándar para operaciones de AI generativa (embeddings, chat, completions)
3. OpenTelemetry — Context Propagation — Cómo funciona el context management que permite parent-child automático
4. Jaeger — Getting Started — Para visualizar los traces jerárquicos que creaste
5. OpenAI Embeddings Guide — Referencia de la API de embeddings para entender los campos de respuesta
6. Arize AI — LLM Tracing with OpenTelemetry — Ejemplo práctico de tracing para LLM applications
7. OpenTelemetry — SpanKind — Documentación sobre cuándo usar CLIENT, SERVER, INTERNAL, PRODUCER, CONSUMER
8. Grafana Tempo — Trace Visualization — Alternativa a Jaeger para visualización de traces