Агенты: чтение кода и цикла
Чтение реальных сниппетов agent-цикла — диспетч инструментов, лимиты шагов, обрезка контекста, error recovery — предскажите поведение и выберите фикс с наибольшим рычагом.
Agent-цикл — это место, где тратятся деньги и прячутся баги. Читайте каждый сниппет так, как читали бы его на код-ревью сервиса, который тарифицируется по токенам, и выбирайте фикс, который сеньор делает первым.
Отработать цикл, который вы прогоняете на каждом агенте: прочитать поток управления, предсказать, где он крутится вечно, переполняется или молотит (thrash), и взяться за структурный ограничитель прежде, чем винить модель.
Сниппет 1 — цикл с единственным выходом
def run_agent(task, tools):
messages = [SYSTEM, {"role": "user", "content": task}]
while True:
resp = model(messages, tools) # THINK
if not resp.tool_calls:
return resp.content # единственный выход: модель перестаёт звать инструменты
messages.append(resp)
for call in resp.tool_calls:
result = dispatch(call, tools) # ACT
messages.append(result) # OBSERVE
Это корректный ReAct, но небезопасный для продакшена. Какой один ограничитель важнее всего добавить и почему именно его первым?
Сниппет 2 — лимит шагов, прячущий баг
for step in range(MAX_STEPS): # MAX_STEPS = 100
resp = model(messages, tools)
if not resp.tool_calls:
return resp.content
messages.append(resp)
for call in resp.tool_calls:
messages.append(dispatch(call, tools))
# вывалились из цикла: лимит достигнут
return "Извините, не удалось это выполнить."
Телеметрия показывает, что ~30% прогонов вываливаются и возвращают извинение. Лимит делает свою работу — какой вывод делает сеньор?
Сниппет 3 — обрезка контекста
def trim(messages, budget=8000):
# держим самые свежие сообщения, пока укладываемся в бюджет токенов
kept = []
total = 0
for m in reversed(messages):
total += count_tokens(m)
if total > budget:
break
kept.append(m)
return list(reversed(kept))
Это держит цикл под окном, но на долгих задачах всплывает один режим отказа. Какой и каков фикс?
Сниппет 4 — error recovery
for step in range(MAX_STEPS):
resp = model(messages, tools)
if not resp.tool_calls:
return resp.content
messages.append(resp)
for call in resp.tool_calls:
try:
result = dispatch(call, tools)
except ToolError as e:
result = {"role": "tool", "content": f"Error: {e}"} # отдаём ошибку назад
messages.append(result)
Отдача ошибки назад позволяет модели адаптироваться — но нагрузочный тест показывает прогоны, где один и тот же инструмент падает идентично десятки раз. Какой ограничитель закрывает брешь?
Любой инцидент агента читается в цикле: единственный выход «модель остановилась» небезопасен, поэтому добавляют жёсткий лимит шагов плюс бюджет по wall-clock/токенам; лимит, срабатывающий регулярно, — это ремень, ловящий реальный баг, а не повод его ослабить; наивная обрезка контекста выселяет закреплённые системные/задачные сообщения и заставляет агента забыть задачу, поэтому инструкции закрепляют, а середину суммаризируют; а error-feedback без лимита retry/dedup на инструмент превращает recovery в бесконечный цикл из валидных вызовов. Прочитайте поток управления, найдите неограниченный путь, добавьте структурный ограничитель и перепрогоните под нагрузкой для подтверждения.
Что-то непонятно?
Задай вопрос по этому уроку. Вопросы анонимны и попадают напрямую автору — урок станет лучше.