Logging OpenTelemetry Data with W&B Weave

Why Send OpenTelemetry Data to W&B Weave?

If you’re already instrumenting your Python applications with OpenTelemetry, you might wonder why you’d want to send that telemetry data to [W&B Weave](http://Logging OpenTelemetry Data with W&B Weave) instead of (or in addition to) traditional observability backends like Jaeger, Grafana, or Datadog.

The Challenge with Traditional Observability Stacks

Traditional observability platforms excel at operational monitoring but often fall short when you’re building AI-powered applications:

• Disconnected workflows: Your production traces live in one tool, your model experiments in another, and your evaluation metrics somewhere else entirely
• Limited AI context: Standard tracing tools don’t understand LLM calls, prompt versions, or model outputs
• Poor reproducibility: When something breaks in production, recreating the exact conditions that led to the issue is difficult
• Siloed teams: ML engineers work in notebooks, backend engineers work in APM dashboards, and nobody has a shared view

The Weave Advantage

W&B Weave bridges this gap by combining OpenTelemetry’s standardized instrumentation with purpose-built AI tooling:

1. Unified observability: See your service-level traces, LLM calls, and model evaluations in one place
2. Experiment integration: Every trace can be linked to W&B runs, making it easy to correlate production behavior with training experiments
3. Rich AI context: Automatic capture of prompts, completions, token counts, and model parameters
4. Collaborative debugging: Share annotated traces with your team, add comments, and track issues
5. Evaluation at scale: Run systematic evaluations on production traces to catch regressions

The best part? You don’t have to choose. Send OpenTelemetry data to Weave AND your existing backends simultaneously.

Architecture Overview

Here’s how the pieces fit together:

┌─────────────────┐
│  Your Python    │
│  Application    │
│  (FastAPI, etc) │
└────────┬────────┘
         │
         ├─→ OpenTelemetry SDK
         │   ├─→ Traces (spans)
         │   ├─→ Metrics (counters, histograms)
         │   └─→ Logs (structured events)
         │
         ├─→ W&B Weave Decorator
         │   └─→ High-level function traces
         │
         └─→ Custom Exporter Bridge
             └─→ W&B Backend
                 ├─→ Trace visualization
                 ├─→ Metrics dashboards
                 └─→ Evaluation pipelines

Tutorial: Building an Instrumented Service

Let’s build a practical example: an AI-powered product recommendation API that we’ll fully instrument with OpenTelemetry and Weave.

Step 1: Environment Setup

First, install the necessary packages:

python -m venv .venv
source .venv/bin/activate

# Core OpenTelemetry
pip install opentelemetry-api opentelemetry-sdk

# OTLP exporters (optional, for dual export)
pip install opentelemetry-exporter-otlp-proto-http

# Auto-instrumentation
pip install opentelemetry-instrumentation-fastapi
pip install opentelemetry-instrumentation-requests

# Application dependencies
pip install fastapi uvicorn requests openai

# W&B and Weave
pip install wandb weave

Authenticate with W&B:

wandb login

Step 2: Create the Weave-Native OpenTelemetry Exporter

This is the key integration piece. It converts OpenTelemetry spans into Weave traces:

weave_otel_exporter.py

python

import time
import weave
from typing import Sequence, Optional
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.trace import StatusCode

class WeaveSpanExporter(SpanExporter):
    """
    Exports OpenTelemetry spans to W&B Weave for unified observability.

    This exporter bridges the gap between OTel's standardized instrumentation
    and Weave's AI-focused tracing capabilities.
    """

    def __init__(self, weave_project: str):
        """
        Args:
            weave_project: W&B project name (format: "entity/project")
        """
        self.weave_project = weave_project
        self._weave_initialized = False

    def export_spans(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        """Convert OTel spans to Weave trace format and log them."""

        # Lazy initialization of Weave
        if not self._weave_initialized:
            weave.init(self.weave_project)
            self._weave_initialized = True

        for span in spans:
            self._export_single_span(span)

        return SpanExportResult.SUCCESS

    def _export_single_span(self, span: ReadableSpan):
        """Convert a single OTel span to Weave format."""

        ctx = span.get_span_context()

        # Build the trace record
        trace_data = {
            "trace_id": f"{ctx.trace_id:032x}",
            "span_id": f"{ctx.span_id:016x}",
            "parent_span_id": f"{span.parent.span_id:016x}" if span.parent else None,
            "name": span.name,
            "start_time": span.start_time / 1e9,  # Convert to seconds
            "end_time": span.end_time / 1e9 if span.end_time else time.time(),
            "duration_ms": (span.end_time - span.start_time) / 1e6 if span.end_time else 0,
            "status": span.status.status_code.name if span.status else "UNSET",
            "attributes": dict(span.attributes) if span.attributes else {},
            "events": [
                {
                    "name": event.name,
                    "timestamp": event.timestamp / 1e9,
                    "attributes": dict(event.attributes) if event.attributes else {}
                }
                for event in span.events
            ] if span.events else [],
            "resource": dict(span.resource.attributes) if span.resource else {},
        }

        # Log to Weave
        weave.log({
            "otel_span": trace_data,
            "service_name": trace_data["resource"].get("service.name", "unknown"),
            "operation": span.name,
            "duration_ms": trace_data["duration_ms"],
            "success": trace_data["status"] in ["OK", "UNSET"],
        })

    def shutdown(self) -> None:
        """Cleanup on shutdown."""
        pass

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        """Force flush of pending spans."""
        return True

Step 3: Initialize OpenTelemetry with Weave Export

otel_weave_setup.py

import logging
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    PeriodicExportingMetricReader,
    ConsoleMetricExporter,
)
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from weave_otel_exporter import WeaveSpanExporter

def setup_otel_with_weave(
    service_name: str,
    weave_project: str,
    enable_console: bool = True,
    enable_otlp: bool = False,
    otlp_endpoint: str = "http://localhost:4318/v1/traces"
):
    """
    Initialize OpenTelemetry with Weave export and optional additional backends.

    Args:
        service_name: Name of your service
        weave_project: W&B project for Weave (format: "entity/project")
        enable_console: Also print spans to console for debugging
        enable_otlp: Also send to OTLP endpoint (e.g., local collector)
        otlp_endpoint: OTLP HTTP endpoint URL
    """

    # Define service resource
    resource = Resource.create({
        SERVICE_NAME: service_name,
        "service.version": "1.0.0",
        "deployment.environment": "production",
    })

    # Setup Tracing
    tracer_provider = TracerProvider(resource=resource)

    # Add Weave exporter (primary)
    weave_exporter = WeaveSpanExporter(weave_project=weave_project)
    tracer_provider.add_span_processor(BatchSpanProcessor(weave_exporter))

    # Optionally add console exporter for debugging
    if enable_console:
        tracer_provider.add_span_processor(
            BatchSpanProcessor(ConsoleSpanExporter())
        )

    # Optionally add OTLP exporter for dual export
    if enable_otlp:
        otlp_exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
        tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

    trace.set_tracer_provider(tracer_provider)

    # Setup Metrics (simplified for this example)
    meter_provider = MeterProvider(resource=resource)
    metrics.set_meter_provider(meter_provider)

    print(f"✓ OpenTelemetry initialized")
    print(f"✓ Exporting to W&B Weave: {weave_project}")
    if enable_otlp:
        print(f"✓ Also exporting to OTLP: {otlp_endpoint}")

    return {
        "tracer": trace.get_tracer(__name__),
        "meter": metrics.get_meter(__name__),
    }

Step 4: Build an AI-Powered Service

recommendation_service.py

import time
import weave
from fastapi import FastAPI, HTTPException
from openai import OpenAI
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.trace import Status, StatusCode
from otel_weave_setup import setup_otel_with_weave

# Initialize OpenTelemetry with Weave
otel = setup_otel_with_weave(
    service_name="recommendation-api",
    weave_project="my-org/recommendations",
    enable_console=True  # For debugging
)

tracer = otel["tracer"]
meter = otel["meter"]

# Create metrics
request_counter = meter.create_counter(
    "api.requests.total",
    unit="1",
    description="Total API requests"
)

recommendation_latency = meter.create_histogram(
    "api.recommendation.duration",
    unit="ms",
    description="Recommendation generation latency"
)

# Initialize FastAPI
app = FastAPI(title="Product Recommendation API")
FastAPIInstrumentor.instrument_app(app)
# OpenAI client (set OPENAI_API_KEY env var)
openai_client = OpenAI()

@weave.op()
def generate_recommendations(user_context: dict, product_catalog: list) -> dict:
    """
    Generate AI-powered product recommendations.

    This function is decorated with @weave.op() to create high-level
    traces that link to the underlying OpenTelemetry spans.
    """

    with tracer.start_as_current_span("llm.generate_recommendations") as span:
        # Add rich context to the span
        span.set_attribute("user.id", user_context.get("user_id"))
        span.set_attribute("user.preferences", str(user_context.get("preferences", [])))
        span.set_attribute("catalog.size", len(product_catalog))

        try:
            # Build the prompt
            prompt = f"""Given a user with preferences: {user_context.get('preferences', [])},
            recommend 3 products from this catalog: {product_catalog[:10]}

            Return as JSON with: product_id, reason"""

            span.add_event("prompt.created", {
                "prompt.length": len(prompt),
                "prompt.preview": prompt[:100]
            })

            # Call OpenAI
            start = time.perf_counter()

            with tracer.start_as_current_span("llm.openai_call") as llm_span:
                response = openai_client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[
                        {"role": "system", "content": "You are a product recommendation expert."},
                        {"role": "user", "content": prompt}
                    ],
                    temperature=0.7,
                    max_tokens=500
                )

                # Record LLM metadata
                llm_span.set_attribute("llm.model", "gpt-4o-mini")
                llm_span.set_attribute("llm.tokens.prompt", response.usage.prompt_tokens)
                llm_span.set_attribute("llm.tokens.completion", response.usage.completion_tokens)
                llm_span.set_attribute("llm.tokens.total", response.usage.total_tokens)

            llm_duration = (time.perf_counter() - start) * 1000

            result = {
                "recommendations": response.choices[0].message.content,
                "model": "gpt-4o-mini",
                "tokens_used": response.usage.total_tokens,
                "llm_latency_ms": llm_duration
            }

            span.set_status(Status(StatusCode.OK))
            span.set_attribute("result.token_count", response.usage.total_tokens)

            return result

        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

@weave.op()
def fetch_user_history(user_id: str) -> dict:
    """Simulate fetching user purchase history."""

    with tracer.start_as_current_span("db.fetch_user_history") as span:
        span.set_attribute("db.user_id", user_id)

        # Simulate database call
        time.sleep(0.05)

        history = {
            "user_id": user_id,
            "preferences": ["electronics", "books"],
            "past_purchases": ["laptop", "python-book"]
        }

        span.set_attribute("db.records_returned", len(history["past_purchases"]))
        return history

@app.post("/recommend/{user_id}")
async def recommend(user_id: str):
    """
    Main API endpoint for product recommendations.

    This endpoint demonstrates full instrumentation:
    - Automatic FastAPI span from OpenTelemetry
    - Manual spans for key operations
    - Weave ops for AI-specific tracing
    - Metrics for observability
    """

    start_time = time.perf_counter()

    with tracer.start_as_current_span("api.recommend") as span:
        span.set_attribute("api.user_id", user_id)
        span.set_attribute("api.endpoint", "/recommend")

        try:
            # Fetch user context
            user_context = fetch_user_history(user_id)

            # Mock product catalog
            product_catalog = [
                "smartphone", "tablet", "headphones", "smartwatch",
                "novel", "cookbook", "camera", "speaker"
            ]

            # Generate recommendations using AI
            recommendations = generate_recommendations(user_context, product_catalog)

            # Record metrics
            duration_ms = (time.perf_counter() - start_time) * 1000
            request_counter.add(1, {"endpoint": "/recommend", "status": "success"})
            recommendation_latency.record(duration_ms, {"endpoint": "/recommend"})

            span.set_attribute("response.duration_ms", duration_ms)
            span.set_status(Status(StatusCode.OK))

            return {
                "user_id": user_id,
                "recommendations": recommendations,
                "duration_ms": duration_ms,
                "trace_id": f"{span.get_span_context().trace_id:032x}"
            }

        except Exception as e:
            request_counter.add(1, {"endpoint": "/recommend", "status": "error"})
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "healthy", "service": "recommendation-api"}

Step 5: Run and Observe

Start your service:

uvicorn recommendation_service:app --reload --port 8000

Make some requests:

# Generate recommendations
curl -X POST "http://localhost:8000/recommend/user123"

# Response includes trace_id for correlation
{
  "user_id": "user123",
  "recommendations": {...},
  "duration_ms": 1234.5,
  "trace_id": "abc123..."
}

Step 6: Explore Your Data in Weave

Navigate to your W&B project and open the Weave interface. You’ll see:

Trace Timeline: Every API request appears as a trace with nested spans:

• GET /recommend/user123 (auto-instrumented by FastAPI)
• api.recommend (your manual span)
• db.fetch_user_history (database operation)
• llm.generate_recommendations (AI logic)
• llm.openai_call (LLM API call)

Rich Metadata: Click any span to see:

• User ID, preferences, and context
• Prompt text and LLM parameters
• Token counts and costs
• Error traces with full stack traces
• Resource attributes (service name, version, environment)

Weave Operations: The @weave.op() decorated functions appear as high-level operations with:

• Input parameters captured automatically
• Output values logged
• Linked to underlying OTel spans via trace ID

Advanced Patterns

Pattern 1: Correlating Traces with Experiments

Link production traces to training runs:

import wandb
import weave

# During model training
run = wandb.init(project="my-org/recommendations", job_type="train")
run.log({"model_version": "v2.3", "accuracy": 0.94})
model_artifact = run.log_model(path="model.pkl", name="recommendation-model")
run.finish()
# In production service
@weave.op()
def generate_recommendations(user_context: dict):
    with tracer.start_as_current_span("llm.generate") as span:
        # Tag spans with model version
        span.set_attribute("model.version", "v2.3")
        span.set_attribute("model.artifact", model_artifact.name)

        # Now you can filter Weave traces by model version
        # and correlate production behavior with training metrics
        ...

Pattern 2: Distributed Tracing Across Services

When your system has multiple services, OpenTelemetry automatically propagates context:

# Service A (recommendation-api)
@app.post("/recommend/{user_id}")
async def recommend(user_id: str):
    with tracer.start_as_current_span("api.recommend"):
        # Call downstream service - context automatically propagated
        response = requests.get(f"http://inventory-service:9000/stock/{user_id}")
        ...

# Service B (inventory-service) 
from opentelemetry.instrumentation.requests import RequestsInstrumentor
RequestsInstrumentor().instrument()  # Auto-propagates context
@app.get("/stock/{product_id}")
async def check_stock(product_id: str):
    # This span will be a child of the parent span from Service A
    with tracer.start_as_current_span("inventory.check"):
        ...

Both services export to Weave, and you’ll see a single unified trace showing the full request flow across service boundaries.

Pattern 3: Real-time Evaluation on Production Traces

Use Weave evaluations to systematically assess production behavior:

import weave

# Define an evaluation
class RecommendationQuality(weave.Evaluation):
    @weave.op()
    def score(self, recommendation: dict) -> dict:
        """Score recommendation quality based on criteria."""

        # Access the full trace context
        trace_id = recommendation.get("trace_id")

        # Evaluate based on business logic
        has_reasoning = "reason" in recommendation.get("recommendations", "")
        token_efficiency = recommendation.get("tokens_used", 0) < 1000
        latency_ok = recommendation.get("duration_ms", 0) < 2000

        return {
            "has_reasoning": has_reasoning,
            "token_efficient": token_efficiency,
            "latency_acceptable": latency_ok,
            "overall_score": sum([has_reasoning, token_efficiency, latency_ok]) / 3
        }
# Run evaluation on recent production traces
evaluation = RecommendationQuality()
results = evaluation.evaluate(
    weave.get_recent_traces(
        project="my-org/recommendations",
        operation="generate_recommendations",
        limit=100
    )
)
# View results in Weave UI
print(f"Average quality score: {results.mean_score}")

Pattern 4: Custom Metrics Integration

Send OpenTelemetry metrics to Weave alongside traces:

python

from opentelemetry.sdk.metrics.export import MetricExporter

class WeaveMetricExporter(MetricExporter):
    """Export OTel metrics to Weave for unified dashboards."""

    def export(self, metrics_data):
        for resource_metrics in metrics_data.resource_metrics:
            for scope_metrics in resource_metrics.scope_metrics:
                for metric in scope_metrics.metrics:
                    weave.log({
                        "metric_name": metric.name,
                        "metric_unit": metric.unit,
                        "metric_data": self._serialize_data_points(metric.data),
                        "timestamp": time.time()
                    })
        return MetricExportResult.SUCCESS
# Register the exporter
metric_reader = PeriodicExportingMetricReader(
    exporter=WeaveMetricExporter(),
    export_interval_millis=60000  # Export every minute
)

Best Practices

1. Span Naming Conventions

Use consistent, hierarchical naming:

# Good
"api.recommend"
"db.fetch_user"
"llm.generate_recommendations"
"cache.get_product"

# Avoid
"recommend"  # Too generic
"get_user_from_database_table"  # Too verbose

2. Attribute Naming

Follow OpenTelemetry semantic conventions:

python

span.set_attribute("http.method", "POST")
span.set_attribute("http.status_code", 200)
span.set_attribute("db.system", "postgresql")
span.set_attribute("db.statement", query)

# Custom attributes
span.set_attribute("app.user_id", user_id)
span.set_attribute("app.experiment_id", experiment_id)

3. Error Handling

Always record exceptions and set error status:

try:
    result = risky_operation()
except Exception as e:
    span.set_status(Status(StatusCode.ERROR, str(e)))
    span.record_exception(e)
    raise  # Re-raise after recording

4. Sampling Strategy

For high-traffic services, use sampling:

from opentelemetry.sdk.trace.sampling import ParentBasedTraceIdRatioBased

# Sample 10% of traces
sampler = ParentBasedTraceIdRatioBased(0.1)
tracer_provider = TracerProvider(resource=resource, sampler=sampler)

5. Cost Management

Monitor Weave usage to control costs:

# Tag expensive operations
span.set_attribute("cost.tokens", token_count)
span.set_attribute("cost.estimated_usd", token_count * 0.00002)

# Create alerts in Weave when costs exceed thresholds

Troubleshooting

Spans not appearing in Weave

Check initialization order:

# Initialize Weave BEFORE setting up OTel
weave.init("my-org/project")
setup_otel_with_weave(...)

Missing trace correlation

Ensure context propagation is working:

from opentelemetry.propagate import inject, extract

# Outgoing request
headers = {}
inject(headers)  # Injects W3C Trace Context
requests.get(url, headers=headers)
# Incoming request
context = extract(request.headers)
with tracer.start_as_current_span("handler", context=context):
    ...

High latency from export

Use async export and batching:

tracer_provider.add_span_processor(
    BatchSpanProcessor(
        WeaveSpanExporter(...),
        max_queue_size=2048,
        max_export_batch_size=512,
        schedule_delay_millis=5000
    )
)

Conclusion

By sending OpenTelemetry data to W&B Weave, you get the best of both worlds: standardized instrumentation with AI-native tooling. This integration enables:

• Unified observability: See infrastructure, application, and AI layers together
• Reproducible debugging: Every production issue is linked to experiments and model versions
• Systematic evaluation: Run evaluations on real production traces at scale
• Team collaboration: Share insights across ML and engineering teams

Start with the patterns in this tutorial, then extend them to match your specific architecture and requirements. The key is establishing the export bridge early, then layering on evaluation and experiment integration as your needs evolve.

Next Steps

• Add more services: Instrument your entire service mesh with consistent OTel + Weave export
• Build evaluation suites: Create Weave evaluations that run automatically on production traces
• Set up alerts: Use Weave to trigger notifications when quality metrics degrade
• Integrate with CI/CD: Run evaluations on staging traces before promoting to production