Monitoring multi-agent reasoning latency & token costs: Distributed tracing with OpenTelemetry and Google Cloud Trace

Google Cloud Community Articles
, , , ,
1

In modern Generative AI architectures, multi-agent coordination (Agent-to-Agent / A2A) is rapidly becoming the standard design pattern. Instead of relying on a single monolith prompt, systems orchestrate networks of specialized agents—like Researcher, Analyst, and Writer agents—delegating tasks and synthesizing results.

However, moving to a distributed network of agents introduces severe observability gaps:

  • Reasoning latency bottlenecks: Which sub-agent or tool execution is slowing down the user response?
  • Hidden token costs: How can you track the cumulative token costs across multiple downstream LLM reasoning calls?
  • Distributed state isolation: How do you map a sequence of LLM calls across multiple network services to a single user request?

In this tutorial, we will build a practical, secure, and production-ready solution to these gaps. We’ll implement a multi-agent chain in Python, instrument it with OpenTelemetry, run it on Google Cloud Run, and visualize the entire reasoning chain—including tool latency and token metrics—as nested spans in Google Cloud Trace.

Architecture overview

Our system consists of three services deployed on Google Cloud Run:

  1. Orchestrator agent: Receives user queries, coordinates the workflow, calls sub-agents, and returns the compiled result.
  2. Researcher agent: Simulates running a search database query (tool call) and summarizing findings via an LLM.
  3. Writer agent: Receives summarized info and drafts a stylized response via another LLM call.
       [ User Request ]
              │
              ▼
   ┌──────────────────────┐
   │  Orchestrator Agent  │ (Service A)
   └──────────┬───────────┘
              │  (POST /research with traceparent & OIDC auth)
              ├───► ┌──────────────────┐
              │     │ Researcher Agent │ (Service B)
              │     │ ├── Search Tool  │ (Span)
              │     │ └── LLM Summary  │ (Span)
              │     └──────────────────┘
              │  (POST /write with traceparent & OIDC auth)
              └───► ┌──────────────────┐
                    │   Writer Agent   │ (Service C)
                    │   └── LLM Draft  │ (Span)
                    └──────────────────┘

Trace context is propagated automatically using the W3C traceparent HTTP header. Because the services are deployed securely (private endpoints), the Orchestrator will query the GCP metadata server to automatically obtain OIDC Identity Tokens for secure Service-to-Service communication.

1. Setup dependencies

Create a requirements.txt file containing the FastAPI server framework, HTTPX client, and the OpenTelemetry exporter libraries:

fastapi
uvicorn
httpx
opentelemetry-api
opentelemetry-sdk
opentelemetry-instrumentation-fastapi
opentelemetry-instrumentation-httpx
opentelemetry-exporter-gcp-trace

2. Common tracing setup

We configure a common initialization module (telemetry.py) to handle span exporting. When running locally, it defaults to printing spans directly to stdout. When deployed in GCP with the ENABLE_GCP_TRACE=true environment variable, it exports spans directly to the Google Cloud Trace API.

# telemetry.py
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter, SimpleSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter

def init_telemetry(service_name: str):
    resource = Resource.create(attributes={"service.name": service_name})
    provider = TracerProvider(resource=resource)

    enable_gcp_trace = os.environ.get("ENABLE_GCP_TRACE", "false").lower() == "true"
    
    if enable_gcp_trace:
        project_id = os.environ.get("GCP_PROJECT_ID")
        exporter = CloudTraceSpanExporter(project_id=project_id) if project_id else CloudTraceSpanExporter()
        processor = BatchSpanProcessor(exporter)
        provider.add_span_processor(processor)
        print(f"[{service_name}] OpenTelemetry: Exporting to Google Cloud Trace")
    else:
        exporter = ConsoleSpanExporter()
        processor = SimpleSpanProcessor(exporter)
        provider.add_span_processor(processor)
        print(f"[{service_name}] OpenTelemetry: Exporting to Console (stdout)")

    trace.set_tracer_provider(provider)

3. Creating the specialized sub-agents

Each sub-agent runs a FastAPI server. OpenTelemetry auto-instruments incoming routes, meaning it automatically parses the W3C traceparent header to identify if it is part of an ongoing trace, creating nested child spans.

Researcher sub-agent (researcher.py)

The researcher agent executes two simulated steps:

  1. A database/web search tool call (modeled as a span).
  2. An LLM reasoning call (modeled as a span) where it counts prompt and completion tokens and writes them as attributes.
# researcher.py
import time
import random
from fastapi import FastAPI
from pydantic import BaseModel
from telemetry import init_telemetry
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

init_telemetry("researcher-agent")
tracer = trace.get_tracer(__name__)
app = FastAPI(title="Researcher Sub-Agent")

class QueryPayload(BaseModel):
    query: str

@app.post("/research")
async def perform_research(payload: QueryPayload):
    query = payload.query
    
    # 1. Instrument Search Tool Execution as a Span
    with tracer.start_as_current_span("researcher_search_tool") as span:
        span.set_attribute("tool.name", "web_search")
        span.set_attribute("tool.query", query)
        
        sleep_time = random.uniform(0.4, 0.8)
        time.sleep(sleep_time) # Simulate network latency
        
        search_result = f"Key details gathered about {query}."
        span.set_attribute("tool.status", "success")
        span.set_attribute("tool.latency_seconds", sleep_time)

    # 2. Instrument LLM call and write token metadata
    with tracer.start_as_current_span("researcher_llm_call") as span:
        span.set_attribute("llm.model", "gemini-1.5-flash")
        time.sleep(random.uniform(0.6, 1.2)) # Simulate inference
        
        prompt_tokens = len(query) + 20
        completion_tokens = len(search_result) + 40
        span.set_attribute("llm.usage.prompt_tokens", prompt_tokens)
        span.set_attribute("llm.usage.completion_tokens", completion_tokens)
        span.set_attribute("llm.usage.total_tokens", prompt_tokens + completion_tokens)
        
        summary = f"[Researcher Summary] raw_data_processed: '{search_result}'"

    return {"summary": summary}

FastAPIInstrumentor.instrument_app(app)

Writer sub-agent (writer.py)

The writer sub-agent takes the summary and generates a stylized response using an LLM model, creating another child span.

# writer.py
import time
import random
from fastapi import FastAPI
from pydantic import BaseModel
from telemetry import init_telemetry
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

init_telemetry("writer-agent")
tracer = trace.get_tracer(__name__)
app = FastAPI(title="Writer Sub-Agent")

class WritePayload(BaseModel):
    summary: str

@app.post("/write")
async def perform_writing(payload: WritePayload):
    summary = payload.summary
    
    with tracer.start_as_current_span("writer_llm_call") as span:
        span.set_attribute("llm.model", "gemini-1.5-pro")
        time.sleep(random.uniform(0.8, 1.6))
        
        prompt_tokens = len(summary) + 15
        completion_tokens = len(summary) * 2 + 50
        span.set_attribute("llm.usage.prompt_tokens", prompt_tokens)
        span.set_attribute("llm.usage.completion_tokens", completion_tokens)
        span.set_attribute("llm.usage.total_tokens", prompt_tokens + completion_tokens)
        
        draft = f"Here is the final generated output:\n\n{summary}"

    return {"draft": draft}

FastAPIInstrumentor.instrument_app(app)

4. Building the orchestrator & injecting authentication

The orchestrator agent exposes a /query endpoint.

  • It auto-instruments HTTPX clients (HTTPXClientInstrumentor().instrument()), which injects W3C traceparent headers into outgoing requests automatically.
  • Security for Cloud Run: Because unauthenticated requests are blocked by corporate organizational policies in GCP, we add a helper get_auth_header to automatically fetch an OIDC ID token from GCP’s local instance Metadata Server.
# orchestrator.py
import os
import httpx
import urllib.request
import urllib.parse
from fastapi import FastAPI, Query
from telemetry import init_telemetry
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

init_telemetry("orchestrator-agent")
tracer = trace.get_tracer(__name__)
app = FastAPI(title="Orchestrator Agent")

RESEARCHER_URL = os.environ.get("RESEARCHER_URL", "http://localhost:8001/research")
WRITER_URL = os.environ.get("WRITER_URL", "http://localhost:8002/write")

# Fetch ID tokens from metadata server for service-to-service auth
def get_auth_header(target_url: str) -> dict:
    if os.environ.get("ENABLE_GCP_TRACE", "false").lower() != "true":
        return {}
    
    parsed = urllib.parse.urlparse(target_url)
    audience = f"{parsed.scheme}://{parsed.netloc}"
    
    metadata_url = f"http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience={audience}"
    req = urllib.request.Request(metadata_url)
    req.add_header("Metadata-Flavor", "Google")
    try:
        with urllib.request.urlopen(req, timeout=5.0) as response:
            token = response.read().decode("utf-8")
            return {"Authorization": f"Bearer {token}"}
    except Exception as e:
        print(f"Warning: Could not fetch OIDC identity token: {e}")
        return {}

@app.get("/query")
async def process_query(q: str = Query(..., description="Topic")):
    with tracer.start_as_current_span("orchestrator_run") as span:
        span.set_attribute("orchestrator.query", q)
        
        # 1. Call Researcher Agent (Headers automatically include tracing context)
        headers = get_auth_header(RESEARCHER_URL)
        async with httpx.AsyncClient(timeout=10.0) as client:
            research_resp = await client.post(RESEARCHER_URL, json={"query": q}, headers=headers)
            research_resp.raise_for_status()
            summary = research_resp.json().get("summary", "")

        # 2. Call Writer Agent
        headers = get_auth_header(WRITER_URL)
        async with httpx.AsyncClient(timeout=10.0) as client:
            writer_resp = await client.post(WRITER_URL, json={"summary": summary}, headers=headers)
            writer_resp.raise_for_status()
            final_draft = writer_resp.json().get("draft", "")

        span.set_attribute("orchestrator.status", "completed")
        
    return {"query": q, "result": final_draft}

FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()

5. Deploying to Google Cloud Run

To make deployment simple, package the application in a single Dockerfile and run the services by changing commands and ports.

Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PORT=8080
CMD uvicorn orchestrator:app --host 0.0.0.0 --port ${PORT}

Deployment commands

Deploy the Researcher and Writer sub-agents, gather their endpoints, and then deploy the Orchestrator with the environment variables set.

First, enable the required APIs:

gcloud services enable run.googleapis.com cloudtrace.googleapis.com

Grant your default Cloud Run service account access to write trace spans:

PROJECT_ID=$(gcloud config get-value project)
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)")
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
    --role="roles/cloudtrace.agent"

Now deploy the services:

# 1. Researcher
gcloud run deploy researcher-agent \
    --source . \
    --command "uvicorn" \
    --args "researcher:app,--host,0.0.0.0,--port,8080" \
    --set-env-vars "ENABLE_GCP_TRACE=true,GCP_PROJECT_ID=${PROJECT_ID}" \
    --region us-central1 --quiet

RESEARCHER_URL=$(gcloud run services describe researcher-agent --region us-central1 --format 'value(status.url)')

# 2. Writer
gcloud run deploy writer-agent \
    --source . \
    --command "uvicorn" \
    --args "writer:app,--host,0.0.0.0,--port,8080" \
    --set-env-vars "ENABLE_GCP_TRACE=true,GCP_PROJECT_ID=${PROJECT_ID}" \
    --region us-central1 --quiet

WRITER_URL=$(gcloud run services describe writer-agent --region us-central1 --format 'value(status.url)')

# 3. Orchestrator
gcloud run deploy orchestrator-agent \
    --source . \
    --command "uvicorn" \
    --args "orchestrator:app,--host,0.0.0.0,--port,8080" \
    --set-env-vars "ENABLE_GCP_TRACE=true,GCP_PROJECT_ID=${PROJECT_ID},RESEARCHER_URL=${RESEARCHER_URL}/research,WRITER_URL=${WRITER_URL}/write" \
    --region us-central1 --quiet

ORCHESTRATOR_URL=$(gcloud run services describe orchestrator-agent --region us-central1 --format 'value(status.url)')

6. Testing & visualizing in Cloud Trace

Invoke the orchestrator using a curl command. Since unauthenticated calls are blocked, pass an ID token:

curl -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
  "${ORCHESTRATOR_URL}/query?q=observability-in-gcp-ai"

Once the execution completes successfully, navigate to the Google Cloud Trace console list (console.cloud.google.com/traces/list).

Open the trace matching your request. You’ll see a clean Gantt chart outlining:

  • The overall duration of the orchestrator_run.
  • Two nested HTTP POST service boundaries (POST /research and POST /write).
  • Nested internal spans within the services:
    • researcher_search_tool (identifying query parameters and execution latency).
    • researcher_llm_call and writer_llm_call containing custom metadata.

Searching and filtering by token cost

One of the most powerful features of logging metadata directly to spans is the ability to isolate specific runs using custom filters in the Trace Explorer UI.

To search for spans matching a specific token threshold or custom attribute:

  1. Click the Filter box in the Trace Explorer dashboard.
  2. Scroll to the bottom of the dropdown menu and select Add attribute filter.
  3. In the fields that appear, enter the custom attribute key (e.g., llm.usage.total_tokens) and enter the exact value (e.g., 171 or 440).

GCP Trace Explorer will instantly filter the table below to show only the matching spans, such as isolating the specific researcher_llm_call span where the LLM generated that exact token count.

Conclusion

By standardizing on OpenTelemetry context propagation, we turned a silent black-box multi-agent chain into an observable network. Using Google Cloud Trace, developers can pinpoint exact sub-agent latency bottlenecks, log token costs per generation, and filter traces by metadata directly, leading to faster optimizations and predictable scaling.

Happy tracing!

6 Likes