Rastreo de agentes LangGraph en Vertex AI Agent Engine con OpenTelemetry
Por defecto, Agent Engine no rastrea a los agentes de LangGraph. Agregue OpenTelemetry al iniciar, redacte las credenciales, decore las funciones clave y envíe los spans a Cloud Trace.
Despliegue un agente LangGraph en el Agent Engine de Google Cloud, abra Cloud Trace y encontrará una pantalla vacía. Agent Engine es un entorno de ejecución gestionado que se encarga del escalado automático, de la creación de puntos de control y de la ejecución sin servidores una vez que ha empacado el agente y llamado a la API de creación. Lo que no hace es instrumentar su agente. No existe un árbol de spans, ni información sobre la latencia de las llamadas a LLM, ni nada con lo que investigar cuando un usuario pregunta por qué una solicitud tardó 30 segundos o por qué una respuesta fue incorrecta.
La solución es una configuración repetible: inicie OpenTelemetry cuando arranque el agente, envuelva las funciones importantes en spans personalizados y envíe todo a Cloud Trace, sin pedirle al autor del archivo YAML del agente que cambie nada.
Activar el seguimiento al iniciar
Dentro de un entorno personalizado del Agent Engine no existe un ciclo de vida de servidor web al que conectarse. El proceso comienza desde cero a partir de un archivo ejecutable serializado, por lo que el código de inicialización debe realizar la configuración por sí mismo, y lo primero que debe llamar es una función que recibe el nombre de un servicio:
set_opentelemetry("my-agent-service")
Esa única llamada conecta los tres canales de OpenTelemetry. Las trazas se envían a Cloud Trace a través de OTLP/gRPC, los registros a Cloud Logging y las métricas a Cloud Monitoring.
Hay dos decisiones de diseño importantes. La función es idempotente, por lo que llamarla desde varios caminos de código no volverá a crear la pila de proveedores. Además, está diseñada para fallar de manera suave: si faltan credenciales o un rol IAM, se escribe una advertencia y el agente sigue procesando el tráfico, ya que una mala configuración de telemetría nunca debe hacer que deje de funcionar un agente en producción. Las credenciales provienen de las Credenciales Predeterminadas de la Aplicación, lo que significa que no se necesita nada adicional cuando el entorno de ejecución ya cuenta con una cuenta de servicio asociada.
Qué hace la configuración completa
Cada canal cuenta con su propio manejo de errores, por lo que un fallo en las métricas no puede impedir que las trazas funcionen:
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)
Las trazas se envían a telemetry.googleapis.com y no a un colector estándar. La API de Telemetría de Google es el punto de entrada OTLP recomendado para Cloud Trace. Acepta formato OTLP/gRPC puro, se autentica mediante Google OAuth2 y escribe directamente en Cloud Trace. Al no requerir un sidecar de colector, se adapta al modelo sin servidor de Agent Engine.
Evitar que los secretos aparezcan en los spans
La instrumentación OpenInference de LangGraph serializa todo el diccionario de estado del agente en atributos de span, y ese estado puede contener credenciales en tiempo de ejecución como claves API o tokens para herramientas empresariales. Un procesador de spans personalizado las elimina antes de que cualquier span abandone el proceso:
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
Regístrelo antes que BatchSpanProcessor. Ese orden garantiza que las credenciales nunca se coloquen en cola para su exportación.
Intervalos personalizados con un decorador
La instrumentación automática genera intervalos HTTP y de Vertex AI, pero no puede ver su lógica de negocio. Preguntas como cuánto tiempo tomó una implementación, qué agente necesitó 45 segundos o si una consulta tuvo éxito requieren intervalos personalizados. Un decorador do_trace cubre tanto funciones síncronas como asíncronas:
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)
Cuando no se proporciona un nombre de span, el decorador toma agent_id, resource_id y version de los argumentos de la función y forma un nombre significativo:
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__}"
La ventaja es que se utiliza un elemento llamado research-agent.v2.invoke en lugar de un simple invoke, por lo que en Cloud Trace resulta evidente qué agente y versión son lentos. El decorador también registra la entrada de la consulta y la respuesta en dicho elemento, truncándolas a longitudes seguras:
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))
Aplicación de spans en cada capa
En la capa de servicio API, un span con nombre marca el punto de entrada:
@do_trace(span_name="api.install_agent")
async def install_agent(agent_id: str, request: DeploymentRequest):
En el ejecutor del agente, el decorador envuelve a invoke, y el span activo se enriquece con los metadatos del agente:
@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)
...
En Cloud Trace, el resultado es un árbol anidado que va desde la llamada a la API, pasando por el ejecutor, hasta las llamadas al modelo y a las herramientas.
Envío de OpenTelemetry con el agente
Al instalarlo en Agent Engine se crea un entorno de Python nuevo, por lo que la biblioteca de utilidades y sus dependencias de OTel deben ir junto con la definición del agente. El servicio de despliegue copia la biblioteca en un directorio temporal y la lista como 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
Hay un problema. Agent Engine no instala las dependencias a partir de copias del directorio local, por lo que cada una también debe aparecer como un requisito de paquete explícito:
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
Luego se crea el agente con enable_tracing=True, lo que activa la instrumentación OpenInference LangGraph:
agent = LanggraphAgent(
model="gemini-2.0-flash",
runnable_builder=runnable_builder_fn,
enable_tracing=True,
)
Dentro del constructor ejecutable, el ejecutor se carga con el rastreo activado. También es necesario especificar use_sync_nodes=True, ya que el modelo de ejecución de Agent Engine es síncrono:
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
Lecciones de la configuración
- Agent Engine se entrega sin funcionalidad de seguimiento para LangGraph. Todo debe añadirse manualmente.
- Llame a la función de configuración de OpenTelemetry al inicializar el agente. Ese es el punto de entrada, ya que no existe un ciclo de vida de servidor web.
- Envíe datos en formato OTLP/gRPC a
telemetry.googleapis.com:443y omita el recolector. - Haga que la configuración no sea fatal. La falta de un rol IAM no debería impedir que el agente funcione.
- Redacte las credenciales antes de exportarlas, ya que el estado de LangGraph serializado en spans puede contener información confidencial.
- Enumere explícitamente las dependencias de OTel en los requisitos.
- Asocie el decorador con spans nombrados para que los rastreos formen árboles buscables y no solo un conjunto de nombres de funciones genéricos.
Roles IAM para la cuenta de servicio
La observabilidad completa requiere tres roles en la cuenta de servicio de Agent Engine:
roles/cloudtrace.agentroles/logging.logWriterroles/monitoring.metricWriter
El código de configuración verifica si existen permisos de Cloud Monitoring antes de habilitar el exportador de métricas. Por lo tanto, la falta de monitoring.metricWriter solo desactiva las métricas, mientras que los rastreos siguen fluyendo.
Leer más
- LangGraph, OpenAI Agents SDK o Claude Agent SDK: Elegir el framework adecuado según la tarea — Cuatro cargas de trabajo de ejemplo, voz, reembolsos duraderos, migración de código y operaciones MCP, muestran qué framework de agentes es el más adecuado en cada caso, con notas sobre el efecto de bloqueo, durabilidad y tiempo real.
- Conceptos de LangGraph en lenguaje sencillo: estado, memoria y orquestación — Antes de escribir código: qué significan la orquestación, el tiempo de ejecución, los agentes con estado, la memoria a corto y largo plazo, los sistemas multiagente y la intervención humana en LangGraph.