Verfolgung von LangGraph-Agenten auf dem Vertex AI Agent Engine mit OpenTelemetry

Deployen Sie einen LangGraph-Agenten in Google Clouds Agent Engine, öffnen Sie Cloud Trace – und Sie sehen einen leeren Bildschirm. Agent Engine ist ein verwalteter Laufzeitumgebung, der das Automasch skalieren, das Erstellen von Checkpoints sowie die serverlose Ausführung nach dem Paketieren des Agenten und dem Aufruf der create-API übernimmt. Was er jedoch nicht tut, ist die Instrumentierung Ihres Agenten. Es gibt keinen Span-Tree, keine Latenzzeiten für LLM-Aufrufe – nichts, worauf man zurückgreifen kann, wenn ein Benutzer fragt, warum eine Anfrage 30 Sekunden gedauert hat oder warum eine Antwort falsch war.

Die Lösung besteht in einer wiederholbaren Einrichtung: Starten Sie OpenTelemetry beim Booten des Agenten, umhüllen Sie wichtige Funktionen mit benutzerdefinierten Spans und übertragen Sie alles an Cloud Trace – ohne vom Autor der Agent-YAML-Datei etwas ändern zu lassen.

Tracing beim Start aktivieren

In einer benutzerdefinierten Agent Engine-Umgebung gibt es keinen Webserver-Lebenszyklus, an den man sich anbinden könnte. Der Prozess startet direkt aus einer serialisierten Ausführbardatei, weshalb der Initialisierungscode die Einrichtung selbst vornehmen muss – und das Erste, was er aufrufen sollte, ist eine Funktion, die einen Service-Name entgegennimmt:

set_opentelemetry("my-agent-service")

Diese einzige Aufrufsequenz verbindet alle drei OpenTelemetry-Signale. Spuren werden über OTLP/gRPC an Cloud Trace gesendet, Logs an Cloud Logging und Metriken an Cloud Monitoring.

Zwei Gestaltungsentscheidungen sind wichtig. Die Funktion ist idempotent, sodass ihr Aufruf aus mehreren Codepfaden den Provider-Stack nicht erneut aufbaut. Zudem ist sie so konzipiert, dass Fehler sanft abgefangen werden. Fehlen Zugangsdaten oder eine IAM-Rolle, wird eine Warnung geschrieben, und der Agent setzt die Dienstleistung fort – schließlich darf eine Fehlkonfiguration der Telemetrie niemals dazu führen, dass ein Produktionsagent ausfällt. Die Zugangsdaten stammen aus den Application Default Credentials, was bedeutet, dass nichts Zusätzliches erforderlich ist, wenn der Laufzeitumgebung bereits ein Service-Konto zugeordnet ist.

Was die vollständige Einrichtung leistet

Jedes Signal verfügt über seine eigene Fehlerbehandlung, sodass ein Fehler bei den Metriken die Funktionsfähigkeit der Spuren nicht beeinträchtigt:

def set_opentelemetry(service_name: str) -> None:
    global _initialized
    if _initialized:
        return
    _initialized = True
    os.environ["OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED"] = "true"
    os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = "true"
    os.environ["OTEL_SERVICE_NAME"] = service_name
    gcp_creds, project_id = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    service_resource = Resource.create({
        SERVICE_NAME: service_name,
        "gcp.project_id": project_id,
    })
    auth_plugin = AuthMetadataPlugin(credentials=gcp_creds, request=Request())
    grpc_channel_creds = grpc.composite_channel_credentials(
        grpc.ssl_channel_credentials(),
        grpc.metadata_call_credentials(auth_plugin),
    )
    try:
        tracer_provider = TracerProvider(resource=service_resource)
        tracer_provider.add_span_processor(CredentialRedactingSpanProcessor())
        tracer_provider.add_span_processor(
            BatchSpanProcessor(
                OTLPSpanExporter(
                    credentials=grpc_channel_creds,
                    endpoint="https://telemetry.googleapis.com:443/v1/traces",
                )
            )
        )
        trace.set_tracer_provider(tracer_provider)
    except Exception as exc:
        logger.warning("Trace export setup failed: %s", exc)
        trace.set_tracer_provider(TracerProvider(resource=service_resource))
    try:
        log_provider = LoggerProvider(resource=service_resource)
        log_provider.add_log_record_processor(BatchLogRecordProcessor(CloudLoggingExporter()))
        logs.set_logger_provider(log_provider)
        event_log_provider = EventLoggerProvider(log_provider)
        events.set_event_logger_provider(event_log_provider)
    except Exception as exc:
        logger.warning("Cloud Logging export setup failed: %s", exc)
    try:
        metrics_exporter = CloudMonitoringMetricsExporter()
        metric_reader = PeriodicExportingMetricReader(metrics_exporter)
        meter_provider = MeterProvider(metric_readers=[metric_reader], resource=service_resource)
        metrics.set_meter_provider(meter_provider)
    except Exception as exc:
        logger.warning("Cloud Monitoring metrics setup failed: %s", exc)
    for cls, name in (
        (HTTPXClientInstrumentor, "httpx"),
        (SQLite3Instrumentor, "sqlite3"),
        (VertexAIInstrumentor, "vertexai"),
    ):
        try:
            cls().instrument()
        except Exception as exc:
            logger.warning("Instrumentation skipped for %s: %s", name, exc)

Die Spuren werden an telemetry.googleapis.com gesendet und nicht an einen Standard-Collector. Googles Telemetry API ist der empfohlene OTLP-Eingabepunkt für Cloud Trace. Sie akzeptiert reines OTLP/gRPC, authentifiziert sich mit Google OAuth2 und schreibt direkt in Cloud Trace. Da kein zusätzlicher Collector benötigt wird, eignet sie sich perfekt für das serverlose Modell von Agent Engine.

Vermeidung von Geheimnissen in den Spans

LangGraphs OpenInference-Instrumentierung serialisiert das gesamte Agent-Zustandsdictionary in Span-Attribute, wobei dieser Zustand Laufzeit-Zertifizierungsdaten wie API-Schlüssel oder Token für Unternehmenswerkzeuge enthalten kann. Ein benutzerdefinierter Span-Processor entfernt diese Daten, bevor ein Span den Prozess verlässt:

class CredentialRedactingSpanProcessor(SpanProcessor):
    _REDACT_KEYS = frozenset({"_runtime_credentials"})

    def on_end(self, span) -> None:
        attrs = getattr(span, "attributes", None)
        if not attrs:
            return
        for key, val in attrs.items():
            if not isinstance(val, str):
                continue
            if not any(f'"{k}"' in val for k in self._REDACT_KEYS):
                continue
            try:
                data = json.loads(val)
                if _redact_dict_recursive(data):
                    attrs[key] = json.dumps(data)
            except Exception:
                for k in self._REDACT_KEYS:
                    val = re.sub(
                        rf'"{re.escape(k)}"\s*:\s*(?:\{{[^{{}}]*\}}|"[^"]*")',
                        f'"{k}": "[REDACTED]"',
                        val,
                    )
                attrs[key] = val

Registrieren Sie es vor dem BatchSpanProcessor. Diese Reihenfolge garantiert, dass Anmeldeinformationen niemals zur Exportierung in die Warteschlange kommen.

Custom Spans mit einem Dekorator

Die automatische Instrumentierung erzeugt HTTP- und Vertex AI-Spans, kann aber Ihre Geschäftslogik nicht erkennen. Fragen wie die Dauer einer Bereitstellung, welcher Agent 45 Sekunden benötigte oder ob eine Abfrage erfolgreich war, erfordern Custom Spans. Ein do_trace-Dekorator deckt sowohl synchron als auch asynchrone Funktionen ab:

def do_trace(fn:callable | None = None, *, span_name: str | None = None) -> callable:
    def decorator(f:callable) -> callable:
        @functools.wraps(f)
        async def _async_wrap(*args, **kwargs):
            tracer = trace.get_tracer(f.__module__)
            active_span_name = span_name or _build_span_name(f, *args, **kwargs)
            with tracer.start_as_current_span(active_span_name) as current_span:
                _set_span_attributes(current_span, f, *args, **kwargs)
                try:
                    return_value = await f(*args, **kwargs)
                    current_span.set_attribute("execution.success", True)
                    current_span.set_status(trace.Status(trace.StatusCode.OK))
                    _set_span_response(current_span, return_value)
                    return return_value
                except Exception as exc:
                    current_span.set_attribute("execution.success", False)
                    current_span.record_exception(exc)
                    current_span.set_status(trace.Status(trace.StatusCode.ERROR))
                    raise

        @functools.wraps(f)
        def _sync_wrap(*args, **kwargs):
            tracer = trace.get_tracer(f.__module__)
            active_span_name = span_name or _build_span_name(f, *args, **kwargs)
            with tracer.start_as_current_span(active_span_name) as current_span:
                _set_span_attributes(current_span, f, *args, **kwargs)
                try:
                    return_value = f(*args, **kwargs)
                    current_span.set_attribute("execution.success", True)
                    current_span.set_status(trace.Status(trace.StatusCode.OK))
                    _set_span_response(current_span, return_value)
                    return return_value
                except Exception as exc:
                    current_span.set_attribute("execution.success", False)
                    current_span.record_exception(exc)
                    current_span.set_status(trace.Status(trace.StatusCode.ERROR))
                    raise

        if inspect.iscoroutinefunction(f):
            return _async_wrap
        return _sync_wrap
    if fn is None:
        return decorator
    return decorator(fn)

Wenn kein Span-Name übergeben wird, zieht der Dekorator agent_id, resource_id und version aus den Funktionsargumenten ab und erstellt einen aussagekräftigen Namen:

def _build_span_name(func, *args, **kwargs) -> str:
    bound = _bind_partial_safe(func, *args, **kwargs)
    parts = []
    agent_id = bound.arguments.get("agent_id")
    resource_id = bound.arguments.get("resource_id")
    version = bound.arguments.get("version")
    if agent_id:
        parts.append(str(agent_id))
    elif resource_id:
        parts.append(str(resource_id))
    if version:
        parts.append(str(version))
    parts.append(func.__name__)
    return ".".join(parts) if parts else f"LanggraphAgent.{func.__name__}"

Der Vorteil besteht darin, dass es sich um einen Span namens research-agent.v2.invoke handelt anstelle eines einfachen invoke; dadurch ist in Cloud Trace klar ersichtlich, welcher Agent und welche Version langsam sind. Der Dekorator protokolliert außerdem die Abfragedaten sowie die Antwort im Span, wobei diese auf sichere Längen gekürzt werden:

def _set_span_attributes(span, func, *args, **kwargs) -> None:
    bound = _bind_partial_safe(func, *args, **kwargs)
    span.set_attribute("code.function", func.__name__)
    span.set_attribute("code.module", func.__module__)
    for attr in ("agent_id", "resource_id", "version"):
        val = bound.arguments.get(attr)
        if val:
            span.set_attribute(f"execution.{attr}", str(val))
    request = bound.arguments.get("request")
    if request is not None:
        raw_input = getattr(request, "input", None)
        if raw_input is not None:
            span.set_attribute("query.input", str(raw_input)[:1000])
        session_id = getattr(request, "session_id", None)
        if session_id:
            span.set_attribute("query.session_id", str(session_id))

Anwendung von Spans in jeder Schicht

In der API-Serviceschicht markiert ein benannter Span den Eingangspunkt:

@do_trace(span_name="api.install_agent")
async def install_agent(agent_id: str, request: DeploymentRequest):

Beim Agent-Executor umhüllt der Dekorator die Funktion invoke, und der aktive Span wird mit den Metadaten des Agents angereichert:

@do_trace
def invoke(self, messages, config=None, thread_id="default", ...):
    active_span = trace.get_current_span()
    if active_span and hasattr(self.config, "metadata"):
        for key, value in self.config.metadata.model_dump().items():
            active_span.set_attribute(f"agent.metadata.{key}", str(value))
        active_span.set_attribute("agent.id", self.config.metadata.id)
    ...

In Cloud Trace ergibt sich ein verschachteltes Baumstrukturmodell, das von der API-Aufrufung ausgehend über den Ausführer bis zu den Aufrufen des Modells und der Tools reicht.

OpenTelemetry mit dem Agenten bereitstellen

Durch die Installation in Agent Engine entsteht eine neue Python-Umgebung, wodurch die Hilfsbibliothek sowie ihre OTel-Abhängigkeiten zusammen mit der Agentendefinition übertragen werden müssen. Der Deployment-Service kopiert die Bibliothek in einen temporären Ordner und listet sie als extra_packages auf:

def _setup_deployment_directory(self, agent_definition: str):
    temp_dir = tempfile.mkdtemp(prefix="agent_install_")
    temp_path = Path(temp_dir)
    yaml_path = temp_path / "agent_definition.yaml"
    yaml_path.write_text(agent_definition)
    import agent_utility
    lib_src = Path(agent_utility.__file__).parent
    shutil.copytree(lib_src, temp_path / "agent_utility")
    return temp_dir, "agent_definition.yaml", lib_src

Es gibt einen Haken: Agent Engine installiert keine Abhängigkeiten aus lokalen Verzeichniskopien, sodass jede davon auch als explizite Paketanforderung angegeben werden muss:

runtime_deps = self._get_library_dependencies(lib_src)
requirements = [
"google-cloud-aiplatform[agent_engines]>=1.120.0",
"langchain>=0.2.0",
"langchain-google-vertexai>=2.0.0",
] + runtime_deps

Dann wird der Agent mit enable_tracing=True erstellt, wodurch die OpenInference LangGraph-Instrumentierung aktiviert wird:

agent = LanggraphAgent(
    model="gemini-2.0-flash",
    runnable_builder=runnable_builder_fn,
    enable_tracing=True,
)

Innerhalb des ausführbaren Builders wird der Executor mit aktivierter Trace-Funktion geladen. use_sync_nodes=True ist ebenfalls erforderlich, da das Ausführungsmodell von Agent Engine synchron ist:

def runnable_builder_fn(model, *, tools=None, checkpointer=None, **kwargs):
    executor = load_agent_from_yaml(
        "agent.yaml",
        enable_tracing=True,
        use_sync_nodes=True,
    )
    return RunnableLambda(input_adapter) | executor.graph

Lektionen aus der Einrichtung

1. Agent Engine wird ohne Tracing für LangGraph ausgeliefert. Alles muss hinzugefügt werden.
2. Rufen Sie die OpenTelemetry-Einrichtungsfunktion bei der Initialisierung des Agents auf. Das ist der Eingangspunkt, da es keinen Lebenszyklus eines Web Servers gibt.
3. Senden Sie OTLP/gRPC an telemetry.googleapis.com:443 und überspringen Sie den Collector.
4. Machen Sie die Einrichtung nicht katastrophal – ein fehlender IAM-Rolle sollte den Agent nicht lahmlegen können.
5. Redigieren Sie die Anmeldeinformationen vor der Exportierung, da der in Spans serialisierte LangGraph-Zustand Geheimnisse enthalten kann.
6. Listen Sie die OTel-Abhängigkeiten ausdrücklich in den Anforderungen auf.
7. Koppeln Sie den Dekorator mit benannten Spans zusammen, damit die Traces suchbare Bäume sind und nicht nur eine Ansammlung allgemeiner Funktionsnamen.

IAM-Rollen für das Service-Konto

Für eine vollständige Überwachbarkeit sind drei Rollen auf dem Agent Engine Service-Konto erforderlich:

• roles/cloudtrace.agent
• roles/logging.logWriter
• roles/monitoring.metricWriter

Der Einrichtungskodet prüft vor der Aktivierung des Metriken-Exporters die Berechtigungen für Cloud Monitoring. Fehlt daher monitoring.metricWriter, werden lediglich die Metriken deaktiviert, während die Protokolle weiterhin übertragen werden.