langgraph opentelemetry google-cloud cloud-trace observability

使用 OpenTelemetry 在 Vertex AI Agent Engine 上追踪 LangGraph Agent

Agent Engine 默认不会追踪 LangGraph 代理。需在启动时添加 OpenTelemetry，隐藏凭证信息，为关键函数添加装饰，并将时间跨度发送至 Cloud Trace。

发布者：REACTAPP.TOP加入本站 2026年10月28日阅读约 6 分钟1446 词

来源: Tracing LangGraph Agents on Agent Engine

将 LangGraph 代理部署到 Google Cloud 的 Agent Engine 后打开 Cloud Trace，你会看到一个空页面。Agent Engine 是一种托管型运行时环境，在你打包代理并调用 create API 后，它会负责自动扩展、检查点以及无服务器执行等功能。但它不会对代理进行性能监控，没有跨度树，也没有 LLM 调用延迟数据，因此当用户询问为何某个请求需要 30 秒才能完成或为何答案有误时，无法从中获取相关线索。

解决办法是采用可重复的配置方式：在代理启动时即启动 OpenTelemetry，将重要函数封装在自定义跨度中，再将所有数据发送到 Cloud Trace，这样无需让编写代理 YAML 文件的人做任何修改。

在启动时开启追踪功能

在自定义的 Agent Engine 环境中，没有可接入的 Web 服务器生命周期。进程是从序列化的可执行文件直接启动的，因此初始化代码必须自行完成各项设置，而它首先应该调用的就是一个接受服务名称作为参数的函数：

set_opentelemetry("my-agent-service")

那次调用会同时启用 OpenTelemetry 的三种信号传输方式：通过 OTLP/gRPC 将追踪数据发送到 Cloud Trace，将日志发送到 Cloud Logging，将指标发送到 Cloud Monitoring。

有两个设计要点值得注意。该函数是幂等的，因此从多个代码路径调用它也不会重新构建提供程序栈。此外它还被设计为能够优雅地处理故障——如果缺少凭证或 IAM 角色，它会记录警告信息，代理仍会继续处理请求，因为 telemetry 配置错误绝不能导致生产环境中的代理停止运行。凭证来自应用程序默认凭证，这意味着当运行时环境已绑定服务账户时无需额外配置。

完整设置的功能

每种信号都有独立的错误处理机制，因此指标传输失败不会影响追踪功能的正常运行：

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，而非标准的收集器。Google 的 Telemetry API 是 Cloud Trace 推荐的 OTLP 数据接入端点，它支持普通的 OTLP/gRPC 格式，通过 Google OAuth2 进行身份验证，然后直接将数据写入 Cloud Trace。由于无需运行收集器侧车程序，因此非常适合 Agent Engine 的无服务器架构。

防止机密信息出现在追踪数据中

LangGraph 的 OpenInference 工具会将整个代理状态字典序列化为追踪数据属性，而这些状态中可能包含运行时的凭证，如 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 环境，因此工具库及其 OpenTelemetry 相关依赖项必须随代理定义一同传输。部署服务会将该库复制到临时目录中，并将其标记为 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 的配置函数，因为这里不存在 Web 服务器的生命周期机制，该函数是配置的入口点。
将 OTLP/gRPC 数据发送到 telemetry.googleapis.com:443，无需使用收集器。
确保配置失败不会导致系统崩溃，缺失的 IAM 角色不应使代理停止运行。
在导出数据前需对凭证进行脱敏处理，因为序列化到追踪事件中的 LangGraph 状态可能包含敏感信息。
在需求清单中明确列出 OpenTelemetry 的依赖项。
将装饰器与带名称的追踪事件结合使用，这样追踪数据就能以可搜索的树形结构呈现，而非一堆通用的函数名称。

服务账户的 IAM 角色

实现全面的可观测性需要在 Agent Engine 服务账户上配置三个角色：

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

配置代码会在启用指标导出功能之前检查云监控权限。如果缺少monitoring.metricWriter，则仅会关闭指标功能，而追踪数据仍会继续传输。