Structured logging, распределённый трейсинг, аудит LLM-вызовов и real-time мониторинг для AI-агентов в продакшене. Отладка, compliance и cost tracking в одном флаконе.
Классические текстовые логи (print/console.log) бесполезны для отладки AI-агентов. Когда агент выполняет цепочку из 15 шагов — retrieval, вызов LLM, вызов tool, повторный retrieval, финальный ответ — вам нужно видеть не только что произошло, но и контекст: session_id, trace_id, latency каждого шага, промпты и ответы LLM, расход токенов. Всё это требует structured logging.
Structured logging (JSON или logfmt) позволяет grep-ать логи по полям, строить дашборды в Grafana Loki, и анализировать паттерны ошибок. Для AI-агентов критичны поля: trace_id (сквозной идентификатор цепочки), span_id (шаг цепочки), model, tokens_used, latency_ms, error_type. Без них расследование инцидентов превращается в гадание.
# ПЛОХО: текстовые логи — бесполезны для поиска print(f"Agent error: {e}") print(f"LLM call took {time}s") # ХОРОШО: structured logging с контекстом import structlog import uuid logger = structlog.get_logger() def agent_run(session_id: str, task: str): trace_id = str(uuid.uuid4()) log = logger.bind(session_id=session_id, trace_id=trace_id) log.info("agent.start", task=task[:100]) try: result = execute_chain(task) log.info( "agent.complete", status="success", tokens_used=result.token_usage.total_tokens, latency_ms=round(result.latency * 1000), model=result.model_name ) return result except Exception as e: log.error( "agent.error", error_type=type(e).__name__, error_message=str(e)[:500], exc_info=True ) raise
OpenTelemetry (OTel) — стандарт для распределённого трейсинга. Для AI-агентов он особенно полезен, потому что агент — это цепочка распределённых вызовов: ваш сервис → LLM API → векторная БД → tool API → снова LLM. OTel объединяет все эти вызовы в один трейс с вложенными спанами, показывая где теряется время.
Настройте экспорт трейсов в Jaeger или Grafana Tempo. Добавьте кастомные спаны на каждый шаг агента: retrieval, llm_call, tool_execution, final_response. Атрибуты спанов: model, prompt_hash, tokens_in, tokens_out, tool_name. Это даёт полную картину производительности и позволяет находить узкие места.
from opentelemetry import trace from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.sdk.resources import Resource # Настройка OTel exporter в Jaeger/Tempo resource = Resource.create({"service.name": "ai-agent"}) provider = TracerProvider(resource=resource) exporter = OTLPSpanExporter(endpoint="http://jaeger:4317", insecure=True) provider.add_span_processor(BatchSpanProcessor(exporter)) trace.set_tracer_provider(provider) tracer = trace.get_tracer(__name__) def instrumented_agent_chain(task: str, session_id: str): with tracer.start_as_current_span("agent.chain") as root_span: root_span.set_attribute("session_id", session_id) root_span.set_attribute("task_hash", hash(task)) # Span: Retrieval with tracer.start_as_current_span("retrieval.search") as span: span.set_attribute("db.type", "qdrant") docs = vectorstore.search(task, k=5) span.set_attribute("docs_found", len(docs)) # Span: LLM Call with tracer.start_as_current_span("llm.generate") as span: span.set_attribute("gen_ai.model", "gpt-4o") response = llm.invoke(prompt) span.set_attribute("gen_ai.tokens_in", response.usage_metadata["input_tokens"]) span.set_attribute("gen_ai.tokens_out", response.usage_metadata["output_tokens"]) return response
Для compliance и отладки необходимо логировать все LLM-вызовы: полный промпт (system + user + контекст), ответ модели, usage (токены), latency и model. Храните это в отдельной таблице PostgreSQL или в Loki с retention-политикой 90 дней. Это критично для: расследования галлюцинаций, cost tracking, сравнения качества разных моделей.
Реализуйте callback на каждый LLM-вызов через LangChain/LlamaIndex callback систему. Callback перехватывает промпты ДО отправки и ответы ПОСЛЕ получения — идеальное место для аудита. Не логируйте чувствительные данные (PII, ключи) — фильтруйте их через regex или пресеты (spacy NER).
from langchain.callbacks.base import BaseCallbackHandler from datetime import datetime import json, hashlib, re import asyncpg class LLMAuditCallback(BaseCallbackHandler): def __init__(self, pg_pool, pii_filter=True): self.pool = pg_pool self.pii_pattern = re.compile( r'\b[\w.-]+@[\w.-]+\.\w+\b|\b\d{3}-\d{2}-\d{4}\b' ) def _sanitize(self, text: str) -> str: return self.pii_pattern.sub("[REDACTED]", text) async def on_llm_start(self, serialized, prompts, **kwargs): self._start_time = datetime.utcnow() self._prompts = [self._sanitize(p) for p in prompts] async def on_llm_end(self, response, **kwargs): elapsed = (datetime.utcnow() - self._start_time).total_seconds() prompt_hash = hashlib.sha256( "".join(self._prompts).encode() ).hexdigest()[:16] async with self.pool.acquire() as conn: await conn.execute(""" INSERT INTO llm_audit_log (ts, model, prompt_hash, prompts, response, tokens_in, tokens_out, latency_ms, session_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) """, datetime.utcnow(), response.llm_output.get("model_name", "unknown"), prompt_hash, json.dumps(self._prompts), self._sanitize(response.generations[0][0].text), response.llm_output["token_usage"]["prompt_tokens"], response.llm_output["token_usage"]["completion_tokens"], int(elapsed * 1000), kwargs.get("run_id") )
LangSmith (от создателей LangChain) и LangFuse (open-source альтернатива) — это специализированные платформы для observability LLM-приложений. Они автоматически трейсируют все вызовы LangChain/LlamaIndex, показывают дашборды с latency, стоимостью, качеством ответов и позволяют дебажить конкретные трейсы.
LangFuse особенно хорош для self-hosted сценариев: вы поднимаете его в Docker, и он собирает все трейсы в вашу PostgreSQL. Поддерживает scoring (оценку качества ответов), A/B сравнение промптов, и экспорт данных для fine-tuning. Интеграция через переменные окружения — ни строчки кода менять не нужно.
# LangFuse self-hosted через Docker Compose # docker-compose.yaml version: "3.8" services: langfuse-server: image: ghcr.io/langfuse/langfuse:2 ports: - "3000:3000" environment: DATABASE_URL: postgresql://langfuse:pass@postgres:5432/langfuse NEXTAUTH_SECRET: "your-secret-key" SALT: "your-salt" ENCRYPTION_KEY: "your-encryption-key" depends_on: - postgres # Подключение LangFuse в коде агента # .env LANGFUSE_PUBLIC_KEY=pk-lf-... LANGFUSE_SECRET_KEY=sk-lf-... LANGFUSE_HOST=http://localhost:3000 # Python: LangChain callback — автоматический трейсинг from langfuse.langchain import CallbackHandler langfuse_handler = CallbackHandler( session_id="user-123", user_id="acme-corp", tags=["production", "gpt-4o"] ) # Просто передаём handler в цепочку chain = prompt | llm | output_parser result = chain.invoke( {"input": "..."}, config={"callbacks": [langfuse_handler]} ) # Все трейсы автоматически появятся в LangFuse UI
LLM-вызовы стоят денег, и без cost tracking вы рискуете получить счёт на тысячи долларов. Отслеживайте расходы в real-time: количество токенов × цена за 1K токенов для каждой модели. Агрегируйте по session_id (клиенту), по endpoint-у, по модели. Настройте алерты: если дневной бюджет превышен на 80% — уведомление в Slack; если превышен на 100% — автоматическая блокировка вызовов (circuit breaker).
Для мульти-модельных агентов считайте blended cost: GPT-4o для сложных задач ($5/1M input), Claude 3.5 Haiku для простых ($0.25/1M input). Оптимизируйте: кешируйте частые запросы, используйте меньшие модели для простых интентов, агрегируйте запросы где возможно.
import time, json from datetime import datetime from collections import defaultdict import redis # Цены за 1M токенов (input / output) PRICING = { "gpt-4o": (5.00, 15.00), "gpt-4o-mini": (0.15, 0.60), "claude-3.5": (3.00, 15.00), } class CostTracker: def __init__(self, redis_client, daily_budget_usd=500.00): self.redis = redis_client self.daily_budget = daily_budget_usd self.daily_key = f"cost:daily:{datetime.utcnow().strftime('%Y-%m-%d')}" def track(self, model: str, tokens_in: int, tokens_out: int, session_id: str): if model not in PRICING: return 0.0 price_in, price_out = PRICING[model] cost = (tokens_in * price_in + tokens_out * price_out) / 1_000_000 # Инкрементируем счётчики в Redis pipe = self.redis.pipeline() pipe.incrbyfloat(self.daily_key, cost) pipe.incrbyfloat(f"cost:session:{session_id}", cost) pipe.incrbyfloat(f"cost:model:{model}", cost) pipe.expire(self.daily_key, 86400) pipe.execute() # Проверка бюджета daily_total = float(self.redis.get(self.daily_key) or 0) if daily_total > self.daily_budget * 0.8: alert(f"Budget at {daily_total/daily_budget*100:.0f}%!") if daily_total > self.daily_budget: raise RuntimeError("Daily budget exceeded — LLM calls blocked") return round(cost, 6)
Соберите все метрики на Grafana-дашборде. Ключевые панели: общее количество запросов (rate), latency по перцентилям (heatmap), error rate по типам ошибок, cost per minute/hour, active sessions, cache hit ratio. Отдельная панель для LLM-метрик: tokens per request, cost per model, HTTP 429 rate (rate limit).
Настройте алерты в Grafana Alerting: если p95 latency > 10s в течение 5 минут, error rate > 5%, или cost per hour превышает норму в 2 раза. Каналы уведомлений: Slack (детали), PagerDuty (критические), email (daily digest). Для LLM-специфичных алертов добавьте проверку rate limit exhausted (429 ошибки) и уведомление о необходимости перехода на fallback-модель.
# prometheus-alerts.yml — алерты для AI-агента groups: - name: ai_agent_alerts rules: - alert: HighLatency expr: histogram_quantile(0.95, rate(agent_request_latency_seconds_bucket[5m])) > 10 for: 5m labels: severity: warning annotations: summary: "AI Agent p95 latency > 10s" description: "Latency p95 is {{ $value }}s for the last 5 minutes" - alert: HighErrorRate expr: rate(agent_llm_calls_total{status="error"}[5m]) / rate(agent_llm_calls_total[5m]) > 0.05 for: 3m labels: severity: critical annotations: summary: "AI Agent error rate > 5%" - alert: LLMRateLimitExhausted expr: rate(agent_llm_calls_total{status="429"}[5m]) > 1 for: 2m labels: severity: critical annotations: summary: "LLM rate limit hit — switch to fallback model" - alert: DailyBudgetWarning expr: increase(agent_llm_cost_usd_total[24h]) > 400 for: 5m labels: severity: warning annotations: summary: "LLM cost exceeded $400/day"
Итоговая архитектура observability для AI-агента включает: агент → structlog (JSON-логи) → Loki (хранение) → Grafana (визуализация); агент → OpenTelemetry → Tempo/Jaeger (трейсы); агент → Prometheus (метрики); агент → LangFuse (LLM-трейсинг). Все компоненты работают параллельно, не блокируя основной поток выполнения.
Ключевые практики: всегда включайте trace_id и session_id в каждый лог; структурируйте логи в JSON/logfmt; настройте sampling трейсов (100% для ошибок, 10% для успешных); храните LLM-аудит отдельно от operational-логов; автоматически ротируйте и архивируйте старые данные в S3. Помните: хорошее логирование — это не роскошь, а необходимость для продакшен AI-агентов, где цена ошибки измеряется в долларах и репутации.
# Полная конфигурация structlog для агента import structlog import logging import sys # Настройка structured logging в JSON structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer(ensure_ascii=False), ], wrapper_class=structlog.stdlib.BoundLogger, context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), cache_logger_on_first_use=True, ) logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.INFO) log = structlog.get_logger() log.info("agent.ready", version="2.1.0", python="3.12") # Вывод: {"version": "2.1.0", "python": "3.12", "event": "agent.ready", "level": "info", "timestamp": "..."}
Продакшен без observability — это полёт на радаре в грозу. Внедрите structured logging (structlog/logfmt), распределённый трейсинг (OpenTelemetry → Jaeger/Tempo), аудит LLM-вызовов (LangFuse/LangSmith), cost tracking (Redis + Prometheus) и дашборды с алертами (Grafana + AlertManager). Ключевое правило: каждый запрос пользователя должен иметь trace_id, каждый LLM-вызов — залогирован и учтён в стоимости, каждая аномалия latency — заалерчена. Это превращает «чёрный ящик» AI-агента в прозрачную, управляемую и надёжную систему. Начните с structlog и Prometheus — это даст 80% observability за 20% усилий.