Отслеживание агентов LangGraph в Vertex AI Agent Engine с использованием OpenTelemetry

Разверните агента LangGraph в Agent Engine от Google Cloud, откройте Cloud Trace — и вы увидите пустой экран. Agent Engine — это управляемая среда выполнения, которая обеспечивает автомасштабирование, сохранение состояния и серверless-выполнение после того, как вы запакуете агента и вызовете API create. Однако она не оснащена инструментами для отслеживания действий вашего агента. Здесь нет дерева таймспанов, нет информации о задержках вызовов LLM, нет ничего, что позволило бы разобраться, почему запрос занял 30 секунд или почему ответ был неверным.

Решение заключается в создании воспроизводимой конфигурации: запускайте OpenTelemetry при старте агента, оборачивайте важные функции в пользовательские таймспаны и отправляйте всё в Cloud Trace, не требуя от автора YAML-конфигурации агента вносить какие-либо изменения.

Включение отслеживания при старте

Внутри собственной среды Agent Engine нет жизненного цикла веб-сервера, к которому можно было бы подключиться. Процесс запускается нулевым состоянием из сериализованного исполняемого файла, поэтому код инициализации должен сам выполнять настройку, причем первым делом он должен вызвать функцию, принимающую имя сервиса:

set_opentelemetry("my-agent-service")

Этот вызов подключает все три сигнала OpenTelemetry. Трейсы передаются в Cloud Trace по протоколам OTLP/gRPC, логи — в Cloud Logging, а метрики — в Cloud Monitoring.

Важны два аспекта проектирования. Функция является идемпотентной, поэтому её вызов с нескольких путей кода не приведёт к повторной настройке стека поставщиков данных. Кроме того, она реализована так, чтобы мягко справляться с сбоями. Если отсутствуют учётные данные или роль IAM, записывается предупреждение, и агент продолжает обрабатывать трафик, поскольку неправильная настройка телеметрии ни в коем случае не должна приводить к остановке агента в производственной среде. Учётные данные берутся из Application Default Credentials, что означает, что ничего дополнительного не требуется, если у среды выполнения уже есть привязанный учётный аккаунт сервиса.

Что делает полная настройка

Каждый сигнал имеет собственную обработку ошибок, поэтому сбой в работе метрик не может помешать корректной работе трейсов:

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)

Сигналы направляются на telemetry.googleapis.com, а не на стандартный сборщик данных. API Telemetry от Google является рекомендуемым конечным пунктом ввода данных в формате OTLP для Cloud Trace. Он принимает данные в формате OTLP/gRPC, аутентифицируется с помощью Google OAuth2 и записывает их непосредственно в Cloud Trace. Поскольку не требуется запуск дополнительного компонента-сборщика, это подходит к безсерверной модели Agent Engine.

Изоляция конфиденциальной информации от элементов трекинга

Инструментарий OpenInference от LangGraph сериализует весь словарь состояния агента в атрибуты элементов трекинга, и в этом состоянии могут храниться учетные данные, такие как ключи API или токены для корпоративных инструментов. Специальный обработчик элементов трекинга удаляет их до того, как элемент трекинга покидает процесс:

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

Зарегистрируйте его до использования BatchSpanProcessor. Такой порядок гарантирует, что учетные данные никогда не попадут в очередь на экспорт.

Персонализированные интервалы с декоратором

Автоматическая инструментализация создает интервалы HTTP и Vertex AI, но она не может отслеживать вашу бизнес-логику. Для ответов на вопросы о том, сколько времени заняла развертка, какому агенту потребовалось 45 секунд или успешно ли был выполнен запрос, необходимы персонализированные интервалы. Декоратор do_trace подходит как для синхронных, так и для асинхронных функций:

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)

Если не указано имя span, декоратор берет agent_id, resource_id и version из аргументов функции и формирует осмысленное имя:

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

Результатом является элемент с именем research-agent.v2.invoke вместо простого invoke, поэтому в Cloud Trace сразу становится ясно, какой агент и какая версия работают медленно. Декоратор также записывает входные данные запроса и ответ в этот элемент, сокращая их до безопасной длины:

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

Применение спанов на каждом уровне

На уровне службы API именованный спан обозначает точку входа:

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

У исполнителя агента декоратор оборачивает метод invoke, и активный интервал обогащается метаданными агента:

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

В Cloud Trace результат представляет собой вложенную структуру дерева, идущую от вызова API через исполнитель до вызовов модели и инструментов.

Развертывание OpenTelemetry с агентом

Установка в Agent Engine создает новую среду Python, поэтому библиотека утилит и связанные с ней зависимости OTel должны передаваться вместе с определением агента. Служба развертывания копирует библиотеку в временный каталог и указывает её как 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

Есть нюанс. Agent Engine не устанавливает зависимости из копий локальных каталогов, поэтому каждая из них также должна указываться как явное требование к пакету:

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

Затем агент создается с параметром enable_tracing=True, что включает инструментальное обеспечение OpenInference LangGraph:

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

Внутри запускаемого конструктора исполнитель загружается с включенным отслеживанием. Также требуется параметр use_sync_nodes=True, поскольку модель выполнения Agent Engine является синхронной:

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

Уроки, полученные в процессе настройки

1. В составе Agent Engine отсутствует поддержка трейсинга для LangGraph; всё необходимое нужно добавлять отдельно.
2. Вызывайте функцию настройки OpenTelemetry при инициализации агента — это входная точка, поскольку у системы нет жизненного цикла веб-сервера.
3. Отправляйте данные в формате OTLP/gRPC на адрес telemetry.googleapis.com:443, игнорируя компонент сбора данных.
4. Сделайте процесс настройки нефатальным: отсутствие роли IAM не должно приводить к остановке работы агента.
5. Зашифруйте учетные данные перед экспортом, поскольку данные состояния LangGraph, сериализованные в формате spans, могут содержать конфиденциальную информацию.
6. Явно указывайте зависимости от OTel в списке требований.
7. Сопоставляйте декораторы с именованными spans, чтобы трейсы представляли собой структурированные деревья, а не просто набор общих названий функций.

Роли IAM для учетной записи сервиса

Для полноценной возможности наблюдаемости необходимы три роли для учетной записи сервиса Agent Engine:

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

Код настройки проверяет наличие разрешений Cloud Monitoring перед включением экспортера метрик. Отсутствие monitoring.metricWriter приводит лишь к отключению отчетности по метрикам, при этом трекинг остается активным.

Связанные материалы

• LangGraph, OpenAI Agents SDK или Claude Agent SDK: выбор подходящей платформы — Четыре примера задач: обработка голосовых команд, гарантированные возвраты средств, миграция кода и операции через MCP, позволяют определить, какая платформа подходит в каждом случае, с учетом факторов привязки, надежности и реального времени.

Концепции LangGraph объяснены простым языком: состояние, память и оркестрация — Прежде чем писать код: что означают оркестрация, время выполнения, агенты с состоянием, краткосрочная и долгосрочная память, многопроцессные агенты и участие человека в работе в LangGraph.