Śledzenie agentów LangGraph w Vertex AI Agent Engine przy użyciu OpenTelemetry

Rozmieść agenta LangGraph w Agent Engine Google Cloud, otwórz Cloud Trace i zobaczysz pusty ekran. Agent Engine to zarządzany środowisko wykonywania, które zajmuje się automatycznym skalowaniem, tworzeniem punktów kontrolnych oraz wykonywaniem bez serwerów po spakowaniu agenta i wywołaniu API create. Nie umożliwia jednak instrumentacji agenta – nie ma drzewa span, nie ma informacji o opóźnieniach w wywołaniach LLM, nic, co mogłoby pomóc przy badaniu przyczyn, dlaczego prośba zajęła 30 sekund lub dlaczego odpowiedź była błędna.

Rozwiązaniem jest powtarzalna konfiguracja: uruchom OpenTelemetry przy starcie agenta, otocz ważne funkcje specjalnymi elementami span i przesyłaj wszystko do Cloud Trace, bez konieczności proszenia autora pliku YAML agenta o jakiekolwiek zmiany.

Włączanie śledzenia przy starcie

W środowisku niestandardowego silnika agentów nie ma cyklu życia serwera webowego, do którego można by się podłączyć. Proces rozpoczyna się od zserializowanego pliku wykonywalnego, więc kod inicjalizacyjny musi sam wykonać konfigurację, a pierwszą funkcją, którą powinien wywołać, jest funkcja przyjmująca nazwę usługi:

set_opentelemetry("my-agent-service")

To jedno wywołanie łączy wszystkie trzy sygnały OpenTelemetry. Ślady trafiają do Cloud Trace przez OTLP/gRPC, logi do Cloud Logging, a metryki do Cloud Monitoring.

Ważne są dwie decyzje projektowe. Funkcja jest idempotentna, więc jej wywołanie z kilku ścieżek kodu nie spowoduje ponownego budowania stosu dostawców. Ponadto została zaprojektowana tak, by awarie występowały w łagodny sposób. Jeśli brakuje uprawnień lub roli IAM, system zapisuje ostrzeżenie, a agent nadal obsługuje ruch, ponieważ błędna konfiguracja telemetrii nie może wyłączyć agenta produkcyjnego. Uprawnienia pochodzą z Application Default Credentials, co oznacza, że nie jest potrzebne nic dodatkowego, jeśli środowisko wykonawcze ma już przypisaną konto usługowe.

Co robi pełna konfiguracja

Każdy sygnał ma własne zarządzanie błędami, dzięki czemu awaria w obsłudze metryk nie może powstrzymać działania śladów:

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)

Ślady trafiają do telemetry.googleapis.com, a nie do standardowego zbieracza danych. API Telemetry firmy Google jest zalecanym punktem wejścia OTLP dla Cloud Trace. Przyjmuje format OTLP/gRPC, autoryzuje się za pomocą Google OAuth2 i zapisuje dane bezpośrednio do Cloud Trace. Ponieważ nie wymaga uruchamiania dodatkowego zbieracza, pasuje do bezserwerowego modelu Agent Engine.

Zabezpieczanie poufnych danych przed dostępem w ramach spanów

Narzędzie OpenInference w LangGraph serializuje cały słownik stanu agenta do atrybutów spanów, a ten stan może zawierać dane uwierzytelniające takie jak klucze API lub tokeny do narzędzi korporacyjnych. Specjalny procesor spanów usuwa je przed tym, jak jakikolwiek span opuści proces:

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

Zarejestruj to przed BatchSpanProcessor. Taka kolejność gwarantuje, że dane uwierzytelniające nigdy nie trafią do kolejki na eksport.

Spans dostosowane z użyciem dekoratora

Automatyczna instrumentacja tworzy spans HTTP i Vertex AI, ale nie może przejrzeć logiki biznesowej aplikacji. Pytania takie jak to, jak długo trwało wdrożenie, który agent potrzebował 45 sekund lub czy zapytanie zakończyło się sukcesem, wymagają stosowania dostosowanych spans. Dekorator do_trace obejmuje zarówno funkcje synchroniczne, jak i asynchroniczne:

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)

Gdy nie podano nazwy span, dekorator pobiera agent_id, resource_id oraz version z argumentów funkcji i tworzy znaczącą nazwę:

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

Wynikiem jest element o nazwie research-agent.v2.invoke zamiast zwykłego invoke, dzięki czemu w Cloud Trace jest oczywiste, który agent i która wersja działają wolno. Dekorator rejestruje również dane wejściowe zapytania oraz odpowiedź w tym elemencie, przycięte do bezpiecznych długości:

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

Stosowanie elementów span na każdej warstwie

Na warstwie usługi API nazwany element span oznacza punkt wejścia:

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

W egzekutorze agenta dekorator otacza metodę invoke, a aktywny obszar jest wzbogacany o metadane agenta:

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

W Cloud Trace wynikiem jest zagnieżdżona struktura drzewiasta, która rozciąga się od wywołania API, przez elementy wykonawcze, aż po wywołania modeli i narzędzi.

Dostarczanie OpenTelemetry razem z agentem

Instalacja w Agent Engine tworzy nowe środowisko Pythona, dlatego biblioteka pomocnicza oraz jej zależności OTel muszą być przekazywane wraz z definicją agenta. Usługa implementacji kopiuje tę bibliotekę do tymczasowego katalogu i umieszcza ją jako 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

Jest jednak haczyk. Agent Engine nie instaluje zależności z lokalnych kopii katalogów, więc każda z nich musi również występować jako wyraźne wymaganie pakietu:

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

Następnie agent jest tworzony z ustawieniem enable_tracing=True, co włącza narzędzia OpenInference LangGraph:

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

Wewnątrz uruchamialnego budowniczego wykonawca jest ładowany z włączonym śledzeniem. Konieczne jest również użycie use_sync_nodes=True, ponieważ model wykonywania Agent Engine jest synchroniczny:

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

Lekcje z konfiguracji

1. Agent Engine dostarczany jest bez funkcji śledzenia dla LangGraph. Wszystko trzeba dodać ręcznie.
2. Należy wywołać funkcję konfiguracji OpenTelemetry podczas inicjalizacji agenta. To jest punkt wejścia, ponieważ nie istnieje cykl życia serwera webowego.
3. Należy wysyłać dane OTLP/gRPC pod adres telemetry.googleapis.com:443 i pominąć kolektor.
4. Konfigurację należy uczynić niefatalną – brak roli IAM nie powinien powodować wyłączenia agenta.
5. Należy usunąć dane uwierzytelniające przed eksportem, ponieważ stan LangGraph zserwowany do elementów typu spans może zawierać poufne informacje.
6. Zakres zależności OTel należy wyraźnie wymienić w wymaganiach.
7. Należy połączyć dekorator z nazwanymi elementami spans, aby ślady działania były dostępne do wyszukiwania w formie drzew, a nie zbiorem ogólnych nazw funkcji.

Role IAM dla konta usługowego

Pełna możliwość obserwacji wymaga trzech ról w koncie usługowym Agent Engine:

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

Kod konfiguracyjny sprawdza uprawnienia do Cloud Monitoring przed włączeniem eksportera metryk. Brak monitoring.metricWriter powoduje jedynie wyłączenie metryk, natomiast ślady działania nadal są przekazywane.