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!

7 Likes
5

Great practical walkthrough. I really like the focus on making multi-agent latency and token usage observable rather than treating them as a black box.

@subhrajyoti_sau For cost-sensitive agentic systems, could the same tracing approach also be used to trigger a cost or token budget guardrail, for example stopping or switching to a cheaper model when a workflow exceeds a predefined threshold?

1 Like
6

Yes, but with an important architectural distinction: tracing telemetry is inherently post-execution (passive), so using it as an active guardrail requires an inline budget accumulator or interceptor.

While standard OpenTelemetry trace processors write spans asynchronously to a backend (like Google Cloud Trace) after an operation finishes, you can leverage the same distributed context propagation mechanisms (Baggage / Context) and span metrics to enforce real-time token and cost limits.

How to Implement Token/Cost Guardrails Using the Tracing Pattern

To turn passive tracing into an active circuit-breaker or dynamic model switcher, you can use these three core techniques:

1. W3C Baggage for Distributed Budget State

OpenTelemetry provides Baggage (a W3C standard key-value propagation mechanism) alongside traceparent. While trace context handles span IDs, Baggage travels across network boundaries in HTTP headers (baggage: budget_remaining=5000,total_spent=0.035).

  • Upstream: The Orchestrator initializes a token/cost budget in Baggage.

  • Downstream (Sub-Agents): Each sub-agent reads the remaining budget before making an LLM call. If remaining budget is below a threshold, it acts on it.

2. Pre-Execution Budget Interceptor (Circuit Breaker / Downgrade)

Before invoking an LLM or tool, an agent checks the current cumulative usage:

  • Threshold Exceeded → Circuit Breaker: Abort further sub-agent calls, stop recursive loops, and return an early synthesized response or a graceful degradation message.

  • Near Threshold → Model Downgrade: If prompt tokens + estimated completion tokens exceed remaining budget, dynamically switch the model

Custom Span Processors / Context Handlers

Instead of waiting for an external backend query, you can attach an in-process SpanProcessor or telemetry callback that updates the running request context:

# Conceptual inline token tracker
class BudgetEnforcer:
    def record_usage(self, context, cost, tokens):
        context["accumulated_cost"] += cost
        context["accumulated_tokens"] += tokens
        
        if context["accumulated_cost"] > MAX_COST_BUDGET:
            raise BudgetExceededException("Cost limit reached across agent chain")
1 Like