← Back to DevBytes

Observability and Tracing with AutoGen: Complete Guide

Observability and Tracing with AutoGen: Complete Guide

Building multi-agent systems with AutoGen is powerful, but as you chain together multiple agents, tools, and LLM calls, understanding what actually happens during a conversation becomes increasingly difficult. Observability and tracing solve this problem by giving you deep visibility into every step your agents take, every token they consume, and every decision they make. In this guide, we'll explore how to instrument your AutoGen applications for production-grade observability.

What Is Observability in AutoGen?

Observability is the ability to understand the internal state of a system based on its external outputs. In the context of AutoGen, observability means you can inspect every LLM call, tool invocation, message exchange, and agent handoff that occurs during a multi-agent conversation. Tracing is the specific technique of recording these events as a hierarchical timeline of spans, each representing a discrete unit of work.

AutoGen integrates with OpenTelemetry, the industry-standard observability framework, through dedicated instrumentation packages. This means you can send traces to any compatible backend — Jaeger, Zipkin, Datadog, Honeycomb, Langfuse, or others — without vendor lock-in.

Why Observability Matters for Multi-Agent Systems

Multi-agent systems introduce complexity that traditional debugging cannot handle. Here's why observability is essential:

Setting Up Tracing in AutoGen

AutoGen provides the autogen-ext package with OpenTelemetry instrumentation. Let's walk through a complete setup from scratch.

Installing Dependencies

First, install AutoGen along with its observability extensions and an OpenTelemetry exporter. For this tutorial, we'll use the console exporter for local development and mention how to switch to a real backend later.

pip install "autogen-agentchat" "autogen-ext[openai,otel]" opentelemetry-sdk opentelemetry-exporter-otlp

If you plan to send traces to Jaeger or another OTLP-compatible backend, also install the OTLP exporter package:

pip install opentelemetry-exporter-otlp-proto-grpc

Configuring the Tracer Provider

Before you create any agents, you must configure the OpenTelemetry tracer provider. This is the foundation that collects spans and exports them to your chosen backend. The key is to set this up early — before any AutoGen code runs.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import ConsoleSpanExporter

# Define a resource that identifies your service
resource = Resource.create({
    "service.name": "autogen-multi-agent",
    "service.version": "1.0.0",
    "deployment.environment": "development",
})

# Create and set the tracer provider
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

# Option 1: Console exporter for local debugging
console_exporter = ConsoleSpanExporter()
provider.add_span_processor(BatchSpanProcessor(console_exporter))

# Option 2: OTLP exporter for sending to Jaeger, Datadog, etc.
# otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
# provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

The Resource object attaches metadata to every trace. This metadata helps you filter and group traces in your observability backend. The BatchSpanProcessor batches spans before exporting, which improves performance in production.

Instrumenting AutoGen Agents

AutoGen's OpenTelemetry instrumentation is applied through a runtime tracing extension. Once configured, it automatically wraps agent conversations, LLM calls, and tool invocations in spans. You do not need to manually add spans to every function.

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.instrumentation.runtime import enable_telemetry

# Enable AutoGen's built-in telemetry instrumentation
enable_telemetry()

# Create the model client
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="...",  # Set via OPENAI_API_KEY environment variable
)

# Create two agents that will collaborate
writer = AssistantAgent(
    name="Writer",
    model_client=model_client,
    system_message=(
        "You are a technical writer. Write clear, concise documentation. "
        "When you are satisfied with your draft, end your message with 'APPROVED'."
    ),
)

reviewer = AssistantAgent(
    name="Reviewer",
    model_client=model_client,
    system_message=(
        "You are a technical reviewer. Review the writer's draft for accuracy "
        "and clarity. Provide specific feedback. If the draft is good, say 'APPROVED'."
    ),
)

# Create a team with a round-robin conversation pattern
termination = TextMentionTermination("APPROVED") | MaxMessageTermination(10)
team = RoundRobinGroupChat([writer, reviewer], termination_condition=termination)

async def main():
    result = await team.run(task="Write a short guide on Python list comprehensions.")
    print(result)

asyncio.run(main())

When you run this code, AutoGen's telemetry instrumentation automatically creates spans for each agent's turn, each LLM call, and each tool invocation. With the console exporter enabled, you'll see the span data printed to your terminal.

Adding Custom Spans

While AutoGen's automatic instrumentation covers most agent operations, you'll often want to add custom spans for application-specific logic. This helps you correlate agent behavior with your own business logic.

from opentelemetry import trace

tracer = trace.get_tracer("my-app")

async def process_user_request(user_input: str):
    with tracer.start_as_current_span("process_user_request") as span:
        span.set_attribute("user.input.length", len(user_input))
        span.set_attribute("user.input.preview", user_input[:100])
        
        # Validate input
        with tracer.start_as_current_span("validate_input"):
            if not user_input.strip():
                span.set_status(trace.Status(trace.StatusCode.ERROR, "Empty input"))
                raise ValueError("Input cannot be empty")
            span.set_attribute("validation.passed", True)
        
        # Run the agent team
        with tracer.start_as_current_span("run_agent_team"):
            result = await team.run(task=user_input)
            span.set_attribute("team.messages_count", len(result.messages))
            span.set_attribute("team.termination_reason", result.stop_reason)
        
        # Post-process the result
        with tracer.start_as_current_span("post_process"):
            final_output = result.messages[-1].content
            span.set_attribute("output.length", len(final_output))
            return final_output

Custom spans nest automatically within the current trace context. This means your run_agent_team span will contain all the child spans that AutoGen creates for individual agent turns and LLM calls, giving you a clean hierarchical view.

Tracing Tool Calls

Tool calls are a critical part of agent observability. When an agent calls a function — whether it's a web search, a database query, or a calculation — you need to see the inputs, outputs, and duration. AutoGen's instrumentation captures this automatically, but you can enrich it further.

from autogen_core.tools import FunctionTool
import json

async def search_database(query: str, limit: int = 10) -> str:
    """Search the internal database for matching records."""
    with tracer.start_as_current_span("search_database") as span:
        span.set_attribute("db.query", query)
        span.set_attribute("db.limit", limit)
        
        try:
            # Simulate a database query
            results = await execute_db_query(query, limit)
            span.set_attribute("db.results_count", len(results))
            span.set_attribute("db.success", True)
            return json.dumps(results)
        except Exception as e:
            span.set_attribute("db.success", False)
            span.set_attribute("db.error", str(e))
            span.record_exception(e)
            raise

# Wrap the function as an AutoGen tool
search_tool = FunctionTool(
    search_database,
    name="search_database",
    description="Search the internal database for matching records.",
)

# Attach the tool to an agent
researcher = AssistantAgent(
    name="Researcher",
    model_client=model_client,
    tools=[search_tool],
    system_message="You are a research agent. Use the search_database tool to find information.",
)

By wrapping your tool logic in a custom span, you add business-relevant attributes that the automatic instrumentation cannot infer. This makes it much easier to query and filter traces in your observability backend.

Adding Baggage for Cross-Cutting Context

OpenTelemetry baggage lets you attach key-value pairs that propagate across all spans in a trace. This is useful for tracking things like user IDs, session IDs, or request origins across an entire multi-agent conversation.

from opentelemetry.baggage import set_baggage, get_baggage
from opentelemetry.context import attach, detach

def start_traced_session(user_id: str, session_id: str, request_id: str):
    """Attach baggage that will propagate to all spans in this trace."""
    context = set_baggage("user.id", user_id)
    context = set_baggage("session.id", session_id, context=context)
    context = set_baggage("request.id", request_id, context=context)
    return attach(context)

def end_traced_session(token):
    """Detach the baggage context when done."""
    detach(token)

# Usage
token = start_traced_session(
    user_id="user-12345",
    session_id="sess-abcde",
    request_id="req-67890",
)
try:
    asyncio.run(main())
finally:
    end_traced_session(token)

With baggage set, every span created during the session — including those AutoGen generates internally — will carry these attributes. This is invaluable for filtering traces by user or session in your observability dashboard.

Sending Traces to a Real Backend

For production use, you'll want to send traces to a dedicated backend rather than the console. Here's how to configure AutoGen tracing with Jaeger, one of the most popular open-source options.

First, start Jaeger locally using Docker:

docker run -d --name jaeger \
  -p 4317:4317 \
  -p 16686:16686 \
  jaegertracing/all-in-one:latest

Then update your tracer provider configuration to use the OTLP gRPC exporter pointing at Jaeger:

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",
    insecure=True,
)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

After running your AutoGen application, open http://localhost:16686 in your browser. Select your service name (autogen-multi-agent) and you'll see a visual timeline of every span in each trace, including the hierarchical relationship between agent turns, LLM calls, and tool invocations.

Using Langfuse for LLM-Specific Observability

Langfuse is a popular open-source observability platform specifically designed for LLM applications. It provides cost tracking, prompt management, and evaluation features on top of standard tracing. AutoGen can send traces to Langfuse through its OpenTelemetry endpoint.

import os
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Set Langfuse credentials via environment variables
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://cloud.langfuse.com/api/public/otel"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = (
    f"Authorization=Basic {os.environ['LANGFUSE_PUBLIC_KEY']},"
    f"x-langfuse-project-id={os.environ['LANGFUSE_SECRET_KEY']}"
)

langfuse_exporter = OTLPSpanExporter()
provider.add_span_processor(BatchSpanProcessor(langfuse_exporter))

Langfuse automatically recognizes LLM-related spans and presents them in a specialized UI that shows token counts, costs, and prompt versions alongside the standard trace timeline.

Best Practices for AutoGen Observability

Configuring Sampling for Production

In production, you typically don't want to trace every single request. Here's how to configure sampling to balance visibility with cost:

from opentelemetry.sdk.trace.sampling import (
    ParentBased,
    TraceIdRatioBased,
    ALWAYS_ON,
)

# Sample 10% of root traces, but always follow parent trace decisions
sampler = ParentBased(
    root=TraceIdRatioBased(0.1),
)

provider = TracerProvider(resource=resource, sampler=sampler)
trace.set_tracer_provider(provider)

The ParentBased sampler ensures that if a trace is sampled, all its child spans are also captured. The TraceIdRatioBased(0.1) root sampler means 10% of new traces will be recorded. For development, use ALWAYS_ON to capture everything.

Putting It All Together

Here's a complete, production-ready example that combines everything we've covered — tracer setup, sampling, custom spans, tool tracing, and baggage propagation:

import asyncio
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.baggage import set_baggage
from opentelemetry.context import attach, detach
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.instrumentation.runtime import enable_telemetry

# 1. Configure the tracer provider with sampling
resource = Resource.create({
    "service.name": "autogen-production",
    "service.version": os.getenv("APP_VERSION", "1.0.0"),
    "deployment.environment": os.getenv("ENV", "production"),
})

sampler = ParentBased(root=TraceIdRatioBased(float(os.getenv("TRACE_SAMPLE_RATIO", "0.1"))))
provider = TracerProvider(resource=resource, sampler=sampler)
trace.set_tracer_provider(provider)

# 2. Add the OTLP exporter
otlp_endpoint = os.getenv("OTLP_ENDPOINT", "http://localhost:4317")
exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))

# 3. Enable AutoGen telemetry
enable_telemetry()

tracer = trace.get_tracer("autogen-app")

# 4. Define a traced tool
async def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result."""
    with tracer.start_as_current_span("calculate") as span:
        span.set_attribute("calc.expression", expression)
        try:
            result = eval(expression, {"__builtins__": {}}, {})
            span.set_attribute("calc.result", str(result))
            return str(result)
        except Exception as e:
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            return f"Error: {e}"

# 5. Create agents
model_client = OpenAIChatCompletionClient(model="gpt-4o")

analyst = AssistantAgent(
    name="Analyst",
    model_client=model_client,
    system_message="You analyze data. Use the calculate tool for math. End with 'DONE' when finished.",
)

writer = AssistantAgent(
    name="Writer",
    model_client=model_client,
    system_message="You write summaries of the analyst's findings. End with 'APPROVED' when satisfied.",
)

# 6. Create the team
termination = TextMentionTermination("APPROVED") | MaxMessageTermination(8)
team = RoundRobinGroupChat([analyst, writer], termination_condition=termination)

# 7. Run with full tracing context
async def run_conversation(task: str, user_id: str, session_id: str):
    ctx = set_baggage("user.id", user_id)
    ctx = set_baggage("session.id", session_id, context=ctx)
    token = attach(ctx)
    
    try:
        with tracer.start_as_current_span("conversation") as span:
            span.set_attribute("conversation.task", task[:200])
            result = await team.run(task=task)
            span.set_attribute("conversation.message_count", len(result.messages))
            span.set_attribute("conversation.stop_reason", str(result.stop_reason))
            return result
    finally:
        detach(token)

if __name__ == "__main__":
    result = asyncio.run(run_conversation(
        task="Calculate the compound interest on $10,000 at 5% for 3 years, then summarize.",
        user_id="user-001",
        session_id="sess-001",
    ))
    print(result)

Conclusion

Observability is not an afterthought for multi-agent systems — it is a fundamental requirement for building reliable, cost-effective, and debuggable AutoGen applications. By leveraging OpenTelemetry integration, you gain visibility into every LLM call, tool invocation, and agent handoff, enabling you to identify performance bottlenecks, track costs, debug conversation loops, and audit agent behavior in production. Start with automatic instrumentation through enable_telemetry(), enrich it with custom spans and baggage for your business context, choose a sampling strategy that fits your traffic volume, and send traces to a backend that matches your team's workflow. With these practices in place, you can confidently deploy AutoGen agents to production knowing that you have the visibility needed to understand and improve their behavior over time.

— Ad —

Google AdSense will appear here after approval

← Back to all articles