Відстеження агентів LangGraph у Vertex AI Agent Engine за допомогою OpenTelemetry
Agent Engine за замовчуванням не відстежує агентів LangGraph. Додайте OpenTelemetry під час запуску, зашифруйте облікові дані, оформіть ключові функції та надсилайте дані до Cloud Trace.
Розгорніть агента LangGraph у Engine Agent Google Cloud, відкрийте Cloud Trace — і ви побачите порожній екран. Engine Agent — це керований середовище виконання, яке займається автомасштабуванням, створенням точок контролю та безсерверним виконанням після того, як ви запакуєте агента та викличете API create. Однак воно не забезпечує вбудовування інструментів у ваш агент. Немає дерева спанів, немає інформації про затримку виклику LLM, немає нічого, що дозволило б з’ясувати причини, чому запит виконався за 30 секунд або чому відповідь була неправильною.
Рішенням є стандартизована налаштування: запускайте OpenTelemetry під час запуску агента, обгортайте важливі функції у власні спани та надсилайте все до 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, серіалізований у спанах, може містити конфіденційну інформацію.
- Чітко вказуйте залежності від OTel у вимогах.
- Поєднуйте декоратор із іменованими спанами, щоб трекінги були структурованими деревами, а не просто набором загальних назв функцій.
Ролі IAM для облікового запису сервісу
Для повної можливості спостереження обліковому запису сервісу Agent Engine потрібні три ролі:
roles/cloudtrace.agentroles/logging.logWriterroles/monitoring.metricWriter
Код налаштування перевіряє наявність дозволів Cloud Monitoring перед увімкненням експортера метрик. Відсутність monitoring.metricWriter тому вимикає лише метрики, а трекування продовжує працювати.
Пов’язані матеріали
- LangGraph, OpenAI Agents SDK чи Claude Agent SDK: підбір інструменту під завдання — Чотири приклади завдань: робота з голосом, стійкі повернення коштів, міграція коду та операції MCP, демонструють, яка платформа підходить у кожному випадку, з примітками щодо залежності, стійкості та роботи в реальному часі.