Минимальная наблюдаемость AI-агента без тяжёлой платформы

Что именно нужно увидеть

Наблюдаемость — возможность восстановить состояние системы по данным, которые она оставляет во время работы. Для небольшого AI-агента не обязательно сразу разворачивать отдельный стек трассировки и мониторинга.

Начните с трёх элементов:

• корреляционный ID связывает все записи одного запуска;
• структурированные события показывают последовательность решений и вызовов;
• агрегируемые метрики отвечают, сколько длился и сколько стоил запуск.

Минимальный результат — один JSON-объект на строку в стандартном выводе. Такие записи можно читать вручную, фильтровать командной строкой или позднее отправить в выбранное хранилище без изменения формата событий.

1. Определите контракт события

Каждая запись должна иметь одинаковую основу. Не записывайте полный prompt, ответ модели, секреты, персональные данные или содержимое документов по умолчанию.

{
  "timestamp": "2026-01-15T10:30:00.000Z",
  "event": "model_call_finished",
  "run_id": "4b563e3e-72eb-4da3-8391-f8d4f35f88d7",
  "step": "plan",
  "duration_ms": 842,
  "status": "ok",
  "model": "example-model",
  "input_tokens": 640,
  "output_tokens": 118,
  "estimated_cost": 0.001234
}

Это пример формата, а не запись реального запуска. Поле estimated_cost является расчётной величиной: его точность зависит от актуальных тарифов, правил округления и того, какие категории токенов возвращает провайдер.

Минимальные поля

timestamp

Время события в UTC и формате ISO 8601.

event

Стабильное машинное имя события.

run_id

Случайный UUID, созданный на входе и переданный во все шаги.

step

Короткое имя этапа: plan, tool_search, answer.

duration_ms

Продолжительность завершившейся операции.

status

Ограниченный набор значений, например ok или error.

2. Добавьте маленький эмиттер событий

Следующий воспроизводимый пример использует только стандартную библиотеку Python. Он пишет JSON Lines в стандартный вывод, поэтому не создаёт файлы и не требует секретов.

import json
import sys
import time
import uuid
from datetime import datetime, timezone
from typing import Any


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")


def emit(event: str, run_id: str, **fields: Any) -> None:
    record = {
        "timestamp": utc_now(),
        "event": event,
        "run_id": run_id,
        **fields,
    }
    print(
        json.dumps(record, ensure_ascii=False, separators=(",", ":")),
        file=sys.stdout,
        flush=True,
    )


def run_agent(task: str) -> None:
    run_id = str(uuid.uuid4())
    started = time.perf_counter()
    emit("run_started", run_id, task_length=len(task), status="ok")

    try:
        step_started = time.perf_counter()

        # Пример шага. Здесь вызывается модель или инструмент.
        result = task.strip().upper()

        emit(
            "step_finished",
            run_id,
            step="transform",
            status="ok",
            duration_ms=round(
                (time.perf_counter() - step_started) * 1000
            ),
            result_length=len(result),
        )
        emit(
            "run_finished",
            run_id,
            status="ok",
            duration_ms=round((time.perf_counter() - started) * 1000),
        )
    except Exception as exc:
        emit(
            "run_finished",
            run_id,
            status="error",
            duration_ms=round((time.perf_counter() - started) * 1000),
            error_type=type(exc).__name__,
        )
        raise


run_agent("проверочный запуск")

Сохраняйте run_id в контексте запуска и явно передавайте его функциям. Глобальная переменная удобна только до появления параллельных задач: затем события разных запусков начинают смешиваться.

3. Оберните вызовы модели и инструментов

Фиксируйте начало и завершение внешней операции. Для метрик длительности используйте монотонный таймер, а календарное время оставьте для timestamp.

def observed_model_call(run_id, step, call_model, request):
    started = time.perf_counter()
    emit("model_call_started", run_id, step=step, status="ok")

    try:
        response = call_model(request)
        usage = response.usage

        emit(
            "model_call_finished",
            run_id,
            step=step,
            status="ok",
            duration_ms=round(
                (time.perf_counter() - started) * 1000
            ),
            model=response.model,
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
        )
        return response
    except Exception as exc:
        emit(
            "model_call_finished",
            run_id,
            step=step,
            status="error",
            duration_ms=round(
                (time.perf_counter() - started) * 1000
            ),
            error_type=type(exc).__name__,
        )
        raise

call_model, response и поля usage здесь условные. Сопоставьте их с интерфейсом используемого SDK. Не подставляйте примерные значения токенов, если API их не вернул: отсутствие данных честнее вымышленной точности.

4. Рассчитайте стоимость отдельно

Храните тарифы в конфигурации, а не внутри функции вызова. Ниже приведён пример структуры с условными числами — это не тарифы какого-либо провайдера.

{
  "example-model": {
    "input_per_million": 1.00,
    "output_per_million": 4.00,
    "currency": "USD"
  }
}

def estimate_cost(input_tokens, output_tokens, price):
    return round(
        input_tokens * price["input_per_million"] / 1_000_000
        + output_tokens * price["output_per_million"] / 1_000_000,
        8,
    )

Записывайте вместе с результатом валюту и версию тарифа, например pricing_version. Тогда исторические запуски останутся объяснимыми после изменения цен. Финансовым источником истины должен оставаться счёт провайдера, а не локальная оценка.

5. Соберите минимальные метрики

Для начала достаточно агрегировать завершённые события по часовому или дневному окну:

• число запусков по status;
• доля неуспешных запусков;
• медиана и высокий перцентиль duration_ms;
• сумма входных и выходных токенов;
• суммарная оценочная стоимость;
• число вызовов модели и инструментов на запуск.

Не превращайте run_id, текст ошибки или имя пользовательского файла в метку временного ряда. Такие значения имеют высокую кардинальность и быстро делают систему метрик дорогой. Они подходят для событий, но не для группировки метрик.

Проверка результата

Сохраните пример в agent_observed.py и выполните безопасную локальную команду:

python3 agent_observed.py

Ожидаемый результат — три JSON-строки: run_started, step_finished и run_finished. Конкретные UUID, отметки времени и длительности будут отличаться.

Если доступна утилита jq, проверьте синтаксис каждой строки без записи на диск:

python3 agent_observed.py | jq -c .

Для полной проверки убедитесь, что:

1. во всех строках одного запуска совпадает run_id;
2. последнее событие содержит итоговый status и duration_ms;
3. ошибка создаёт событие со значением status: error и типом исключения;
4. события не содержат prompt, ответ, ключ API или персональные данные;
5. сумма токенов отдельных вызовов согласуется с итогом запуска, если итог рассчитывается.

Типовые ошибки

Новый ID на каждом шаге

Так трасса распадается на несвязанные записи. Создавайте UUID один раз на границе запуска. При наличии входящего доверенного ID можно сохранить его отдельно как parent_run_id, предварительно проверив формат и длину.

Только сообщение об ошибке

Текст нестабилен и может содержать чувствительные данные. Записывайте безопасный error_type, этап, статус и длительность. Подробности исключения отправляйте только в защищённый канал с установленным сроком хранения.

Событие завершения теряется

Используйте try/except или try/finally, немедленно сбрасывайте буфер вывода и повторно возбуждайте исключение после записи. Не маскируйте исходную ошибку сбоем логирования.

Стоимость считается по устаревшей цене

Версионируйте таблицу тарифов и явно помечайте сумму как оценочную. Если модель неизвестна конфигурации, оставляйте стоимость пустой и создавайте отдельное событие о пропущенном расчёте.

Логи превращаются в копию пользовательских данных

Вместо содержимого записывайте длину, тип, количество элементов и разрешённые категориальные признаки. Хеш также может оставаться идентификатором чувствительных данных, поэтому его применение требует отдельного решения о рисках.

Ограничения минимального подхода

JSON Lines хорошо подходит для одного процесса и ранней эксплуатации, но сам по себе не обеспечивает долговременное хранение, поиск между узлами, контроль доступа, автоматические оповещения и гарантированную доставку.

При нескольких сервисах потребуется передавать корреляционный контекст через очереди и HTTP-запросы. При высокой нагрузке понадобятся буферизация, ограничение объёма, политика выборки и централизованное хранилище. Для строгого учёта стоимости необходимо учитывать повторные запросы, кэширование, специальные категории токенов и правила конкретного провайдера.

Переходить к более тяжёлой системе стоит тогда, когда локальный файл или стандартный вывод уже не позволяют быстро ответить на три вопроса: какой запуск пострадал, на каком шаге и каков его ресурсный итог.

Итоговая схема

Создайте run_id на входе, передайте его каждому шагу, пишите структурированные события начала и завершения, измеряйте длительность монотонным таймером и суммируйте фактические данные usage. Этого достаточно, чтобы локализовать большинство сбоев и оценить стоимость запуска без отдельной observability-платформы.

Продолжить настройку можно по материалам раздела «Руководства», а определения терминов сверить в глоссарии Agent Lab Journal.