Suivi des agents LangGraph sur l’agent Engine de Vertex AI à l’aide d’OpenTelemetry

Déployez un agent LangGraph dans l’Agent Engine de Google Cloud, ouvrez Cloud Trace et vous verrez une interface vide. L’Agent Engine est un environnement d’exécution géré qui s’occupe de l’auto-échelle, du sauvegardage des points de contrôle et de l’exécution sans serveur après que vous ayez emballé l’agent et appelé l’API create. Ce qu’il ne fait pas, c’est instrumenter votre agent. Il n’y a ni arbre des spans, ni mesure de la latence des appels aux LLM, rien sur quoi s’appuyer lorsque un utilisateur demande pourquoi une requête a pris 30 secondes ou pourquoi une réponse est erronée.

La solution consiste en une configuration reproductible : lancez OpenTelemetry au démarrage de l’agent, enveloppez les fonctions importantes dans des spans personnalisés et envoyez tout vers Cloud Trace, sans demander à l’auteur du fichier YAML de l’agent de modifier quoi que ce soit.

Activation du suivi au démarrage

Dans un environnement d’Agent Engine personnalisé, il n’y a pas de cycle de vie de serveur web auquel s’attacher. Le processus démarre à partir d’un fichier exécutable serialisé, de sorte que le code d’initialisation doit effectuer lui-même les préparatifs, et la première chose qu’il doit appeler est une fonction qui prend en paramètre le nom d’un service :

set_opentelemetry("my-agent-service")

Cette seule appelation active les trois signaux d’OpenTelemetry. Les traces sont envoyées à Cloud Trace via OTLP/gRPC, les journaux à Cloud Logging et les métriques à Cloud Monitoring.

Deux choix de conception sont importants. La fonction est idempotente, donc son exécution depuis plusieurs chemins de code ne récréera pas l’ensemble des composants du fournisseur. De plus, elle est conçue pour échouer de manière douce : en l’absence de crédences ou d’un rôle IAM, elle enregistre un avertissement et l’agent continue de gérer le trafic, car une mauvaise configuration de la telemétrie ne doit en aucun cas mettre hors service un agent en production. Les crédences proviennent des Application Default Credentials, ce qui signifie qu’aucun élément supplémentaire n’est nécessaire lorsque le runtime dispose déjà d’un compte de service associé.

Ce que permet la configuration complète

Chaque signal dispose de son propre mécanisme de gestion des erreurs, de sorte qu’une panne liée aux métriques ne peut pas empêcher le fonctionnement des traces :

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)

Les traces sont envoyées vers telemetry.googleapis.com et non vers un collecteur standard. L’API Telemetry de Google est l’endpoint d’ingestion OTLP recommandé pour Cloud Trace. Elle accepte le format OTLP/gRPC pur, s’authentifie via Google OAuth2 et écrit directement dans Cloud Trace. Comme aucun sidecar de collecteur n’est nécessaire, elle convient au modèle serverless d’Agent Engine.

Éviter que les secrets ne figurent dans les spans

L’instrumentation OpenInference de LangGraph serialise tout le dictionnaire d’état de l’agent en attributs de span, et cet état peut contenir des identifiants en temps de exécution tels que des clés API ou des tokens pour des outils d’entreprise. Un processeur de span personnalisé les supprime avant que tout span ne quitte le processus :

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

Enregistrez-le avant le BatchSpanProcessor. Cet ordre garantit que les identifiants ne seront jamais mis en file d’attente pour l’exportation.

Spans personnalisés avec un décorateur

L’auto-instrumentation génère des spans HTTP et Vertex AI, mais elle ne peut pas prendre en compte votre logique métier. Des questions telles que le temps nécessaire pour un déploiement, le fait que tel agent ait nécessité 45 secondes ou encore si une requête a réussi exigent des spans personnalisés. Un décorateur do_trace couvre à la fois les fonctions synchrones et asynchrones :

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)

Lorsqu’aucun nom de span n’est fourni, le décorateur extrait agent_id, resource_id et version des arguments de la fonction et crée un nom significatif :

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__}"

Le résultat est une étiquette appelée research-agent.v2.invoke au lieu de simplement invoke ; ainsi, dans Cloud Trace, il est évident quel agent et quelle version sont lents. Le décorateur enregistre également les données d’entrée de la requête et la réponse sur l’étiquette, tronquées à des longueurs sûres :

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))

Application des spans à chaque couche

À la couche de service API, un span nommé marque le point d’entrée :

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

Au niveau de l’exécuteur d’agent, le décorateur enrobe la fonction invoke, et la période active est enrichie des métadonnées de l’agent :

@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)
    ...

Dans Cloud Trace, le résultat se présente sous forme d’un arbre imbriqué allant de l’appel API, en passant par l’exécuteur, jusqu’aux appels au modèle et aux outils.

Déploiement d’OpenTelemetry avec l’agent

L’installation dans Agent Engine crée un environnement Python vierge, ce qui oblige la bibliothèque d’outils ainsi que ses dépendances OTel à être incluses avec la définition de l’agent. Le service de déploiement copie la bibliothèque dans un répertoire temporaire et la liste sous le nom de extra_packages:

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

Il y a un piège. Agent Engine ne installe pas les dépendances à partir de copies de dossiers locaux, donc chacune d’elles doit également apparaître comme une exigence de package explicite :

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

L’agent est ensuite créé avec enable_tracing=True, ce qui active l’instrumentation OpenInference LangGraph :

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

Dans le constructeur exécutable, l’exécuteur se charge avec le suivi activé. use_sync_nodes=True est également requis, car le modèle d’exécution d’Agent Engine est synchrone :

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

Leçons tirées de la configuration

1. Agent Engine est fourni sans fonctionnalité de suivi pour LangGraph. Tout doit être ajouté manuellement.
2. Appeler la fonction de configuration d’OpenTelemetry lors de l’initialisation de l’agent. C’est le point d’entrée, étant donné qu’il n’existe pas de cycle de vie de serveur web.
3. Envoyer des données OTLP/gRPC vers telemetry.googleapis.com:443 et ignorer le collecteur.
4. Rendre la configuration non fatale : l’absence d’un rôle IAM ne doit pas entraîner l’arrêt de l’agent.
5. Masker les identifiants avant l’exportation, car l’état de LangGraph serialisé en spans peut contenir des secrets.
6. Indiquer explicitement les dépendances OTel dans les exigences.
7. Associer le décorateur à des spans nommés afin que les traces forment des arbres recherchables et non un amas de noms de fonctions génériques.

Rôles IAM pour le compte de service

Une observabilité complète nécessite trois rôles sur le compte de service d’Agent Engine :

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

Le code de configuration vérifie les permissions de suivi cloud avant d’activer l’exportateur de métriques. L’absence de monitoring.metricWriter désactive donc uniquement les métriques, tandis que les traces continuent d’être générées.