langgraph opentelemetry google-cloud cloud-trace observability

Трасіроўка агентаў LangGraph у Vertex AI Agent Engine праз OpenTelemetry

Agent Engine па змаўчанні не трасіруе агентаў LangGraph. Дадаём OpenTelemetry пры старце, хаваем уліковыя даныя, дэкаруем ключавыя функцыі і адпраўляем спаны ў Cloud Trace.

Выдавец: REACTAPP.TOPДададзена на сайт 28 кас 20266 хв чытання1446 слоў

Крыніцы: Tracing LangGraph Agents on Agent Engine

Ужывайце агента LangGraph у Agent Engine Google Cloud, адзьёўжыце Cloud Trace — і вы пабачыце порожній экран. Agent Engine — гэта кераваныя асяроддзе выканання, якая адпавядае за автаскалінг, стварэнне пунктавых контролі і выканання без сервера пасля таго, як вы запакавалі агента і вызвалі API create. Але ён не адбывае функцію інструментавання вашага агента. Не існуе дрэва span, няма затрымкі вызова LLM, няма нічога, што можна було бы аналізаваць, калі корыстнік спытаецца, чаму запіт зайняў 30 секунд чы рашэнне было некорэктным.

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

Увёмкнэнне трэйсінгу пад час запуску

У атмасферы спецыязначанага двайхарульніка агента няма жыццёвага цыклу веб-сервера, да якога можна было бы прыўязацца. Працэс запускаецца з нуля з серыяванага выконвальнага коду, таму код ініцыялізаціі сам павінен выконваць настройкі, а першым крокам яго ўвогуле ўжо ўтварэнне функцыі, якая прыме імя службы:

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

Урокі з налаштавання

Agent Engine прыходзіць без падтрымкі для LangGraph. Усё трэба дадаць самастаўна.
Выклікайце функцію налаштавання OpenTelemetry пад час запуску агента. Гэта ўвіходная точка, адколі не існуе жыццўнага цыклу веб-сервера.
Адправляйце даны OTLP/gRPC на telemetry.googleapis.com:443 і прыгнічвайце працу колектара.
Зробіце налаштавання нефатальным. Відсутнае IAM-ролі не павінна блакануць працу агента.
Змініце даны абмові перад экспортом, таму што стан LangGraph, сэrialізаваны ў spans, можа містыць секрэты.
Явна пераказайце залежнасці OTel у спісе трэбаванняў.
Спаруйце дэкаратар з іменаванымі 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.

← Усе артыкулы Даслаць брыф