5 мар. 2026 г.

Отладка и Наблюдаемость в Системах Автономных Агентов

Автономный агент, который молча завершается с ошибкой, хуже, чем отсутствие агента вовсе. Когда традиционная функция выбрасывает исключение, вы получаете стек вызовов. Когда агент идёт по неверному пути через двадцать вызовов инструментов и три обращения к модели, вы получаете неправильный ответ — без очевидного объяснения.

Отладка агентов требует иной ментальной модели. Система не выполняет детерминированный путь; она принимает серию решений. Наблюдаемость означает захват этих решений — не только входных и выходных данных, но и рассуждений, которые их связывают.

Почему Традиционная Отладка Не Работает для Агентов

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

Режимы отказов тоже различаются:

Тихая галлюцинация: Агент уверенно выдаёт неверный ответ, не сигнализируя о неопределённости.
Дрейф решений: Каждый шаг выглядит разумным локально, но последовательность отклоняется от цели.
Неправильное использование инструментов: Агент вызывает правильный инструмент с незаметно неверными параметрами.
Бесконечные циклы: Агент застревает в повторных попытках неудачного подхода.
Отравление контекста: Плохой вывод на раннем шаге портит всё последующее рассуждение.

Ни один из этих случаев не генерирует исключение. Они производят неверное поведение, которое видно только при восстановлении полного трейса выполнения.

Структурированное Логирование для Решений Агента

Первый шаг — обернуть каждое взаимодействие агента в структурированные логи. Не логируйте сырые ответы API — логируйте семантические события.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic

client = anthropic.Anthropic()

@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float

def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # Замените своим хранилищем логов

class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []

    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })

    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )

        latency_ms = (time.monotonic() - start) * 1000

        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))

        return response

Построение Полного Трейса

Одна строка лога недостаточна — нужен полный трейс выполнения, связывающий каждое решение с его результатом:

from typing import Callable

def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)

    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0

    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)

        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })

                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"Ошибка: неизвестный инструмент '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Ошибка инструмента: {e}"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })

        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    else:
        trace["error"] = f"превышено максимальное количество шагов ({max_steps})"

    return trace

Обнаружение Циклов

Бесконечные циклы — распространённый режим отказа. Обнаруживайте их по отпечатку паттерна вызова инструментов каждого LLM-вызова:

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False

    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)

    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True

    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True

    return False

Метрики для Продакшена

from collections import Counter

def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]

    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1

    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

Ключевые сигналы для оповещений:

Частота циклов > 5% запусков — агент застревает
Частота ошибок по инструменту > порога — инструмент сломан
Среднее число шагов растёт — задачи усложняются
p99-латентность в пике — эндпоинт модели медленный

Интеграция с OpenTelemetry

Для команд, уже использующих OpenTelemetry, трейсы агента можно эмитировать как спаны:

from opentelemetry import trace as otel_trace

tracer = otel_trace.get_tracer("agent")

def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])

        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)

        messages = [{"role": "user", "content": task}]

        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)

                if response.stop_reason == "end_turn":
                    break

Редактирование PII в Логах

Логи агентов часто содержат чувствительные данные. Перед отправкой в любую внешнюю систему выполните редактирование:

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b\+?7[-\s]?\(?\d{3}\)?[-\s]?\d{3}[-\s]?\d{2}[-\s]?\d{2}\b'), '[ТЕЛЕФОН]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API_КЛЮЧ]'),
]

def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Три Важнейшие Метрики

Процент выполненных задач — какая доля запусков достигает final_answer против max_steps или ошибки. Установите базовую линию по типу задачи.

Стоимость токенов на задачу — сумма input_tokens + output_tokens по всем шагам. Рост стоимости на 20% без изменения процента выполнения обычно сигнализирует о деградации промпта.

Частота ошибок инструментов — error_steps / total_steps. Пики этой метрики напрямую указывают на сломанный инструмент или API.

Наблюдаемость в агентных системах — это не опция, а разница между системой, которую можно итерировать, и той, которую можно только перезапустить, когда она сломается. Начните со структурированных событий и идентификаторов трейсов. Добавьте обнаружение циклов. Публикуйте метрики. Инвестиция окупается при первом сбое в продакшене, когда вы можете точно восстановить произошедшее вместо того, чтобы гадать.