LLM-агент — это цикл, в котором модель выбирает вызов инструмента, ваш код его выполняет, а результат возвращается в контекст, пока задача не решена или не сработал ограничитель. Память, планирование и контроль — не отдельные «модули» фреймворка, а обычные структуры вокруг этого цикла. Anthropic описывает агентов так же: это LLM, которые используют инструменты на основе обратной связи от среды в цикле, а задача завершается по готовности или по условию остановки, например по максимуму итераций. Ниже разберём части цикла и соберём пример: агент создаёт тестовый тикет и проверяет результат.
Четыре части агента
| Часть | Что это в коде | Типичный сбой |
|---|---|---|
| Модель | Вызов chat.completions с tools | Выдумывает инструмент или аргумент |
| Инструменты | Функции плюс JSON Schema их аргументов | Неоднозначное описание, лишние права |
| Память | Список messages, журнал шагов, внешнее хранилище | Переполнение контекста, потеря фактов |
| Контроль | Лимит шагов, проверка результата, обработка ошибок | Зацикливание, ложное «готово» |
Модель в этой схеме ничего не исполняет: она возвращает текст с намерением, а исполнение и ответственность за последствия остаются в вашем коде.
Инструменты: схема как контракт
Инструмент для модели — это имя, описание и JSON Schema аргументов. В документации OpenAI по function calling есть флаг strict: при true вызов гарантированно соответствует схеме, но для этого в каждом объекте нужен additionalProperties: false, а все поля перечислены в required; необязательные поля помечают типом null. Там же есть parallel_tool_calls: значение false гарантирует ноль или один вызов за ход.
Не полагайтесь на строгий режим как на единственную защиту. Совместимые API у разных провайдеров поддерживают его по-разному, а часть ключевых слов схемы (например, pattern) в строгом режиме может быть отклонена. Поэтому проверку аргументов в примере выполняет собственный валидатор на jsonschema.
Anthropic в приложении к своей статье советует «poka-yoke» для инструментов: менять аргументы так, чтобы ошибиться было труднее. Их пример: после перехода на обязательные абсолютные пути модель перестала ошибаться с файлами в SWE-bench. В нашем примере то же делает правило «заголовок начинается с [TEST] », зашитое в инструмент, а не только в промпт.
Память: три уровня
- Рабочая память — сам список
messages. Всё, что агент помнит внутри запуска, оплачивается токенами на каждом шаге. - Журнал состояния — запись о каждом шаге вне контекста: инструмент, аргументы, успех, время. Это память для вас: по журналу восстанавливают ход событий и ловят повторы.
- Долговременная память — база или векторное хранилище, откуда агент достаёт факты инструментом; см. статью про RAG на своих документах.
Когда контекст растёт, сжимайте старые результаты инструментов до сводки в коде, а журнал оставляйте полным.
Планирование: когда достаточно цикла
Для задачи из трёх шагов отдельный планировщик не нужен: хватает цикла, в котором модель на каждом шаге видит результат предыдущего. Явный план оправдан, когда шагов много или порядок критичен. Тогда последовательность «поиск, создание, проверка» выносят в код, а модель только заполняет параметры. Это уже воркфлоу, и для многих продуктовых сценариев он надёжнее. В нашем примере порядок задан промптом, а правильность итога гарантирует проверка в коде.
Контроль выполнения
- Лимит шагов. Цикл имеет жёсткий верхний предел (
MAX_STEPS), после него возвращается статусstep_limit, а не очередной запрос к модели. - Детектор повторов. Если одна и та же пара «инструмент, аргументы» встретилась больше двух раз, агент остановлен: он застрял.
- Ошибки инструментов как данные. Исключение превращается в ответ
{"ok": false, "error": ...}, который модель видит и может исправить. Неизвестный инструмент, битый JSON и нарушение схемы обрабатываются так же. - Идемпотентность. Повторный
create_ticketс теми же аргументами возвращает прежний id, а не плодит дубли. - Проверка результата по состоянию системы. Фраза «тикет создан» от модели ничего не доказывает; функция
verifyсмотрит в хранилище. Если проверка провалена, модели один раз возвращается причина.
Код цикла агента
Пример работает с любым OpenAI-совместимым эндпоинтом: задайте LLM_BASE_URL, LLM_API_KEY и LLM_MODEL. Модель должна поддерживать вызов инструментов; бесплатные варианты и их лимиты собраны в гиде по бесплатным LLM API, а эндпоинт OpenRouter подходит для быстрой проверки. Зависимости: pip install openai jsonschema.
import hashlib
import json
import os
import time
from jsonschema import Draft202012Validator
MODEL = os.environ.get("LLM_MODEL", "your-model-id")
MAX_STEPS = 6
TICKETS: dict[str, dict] = {}
IDEMPOTENCY: dict[str, str] = {}
class ToolError(Exception):
"""Ошибка, которую полезно показать модели."""
def search_tickets(query: str) -> dict:
hits = [t for t in TICKETS.values() if query.lower() in t["title"].lower()]
return {"count": len(hits), "tickets": hits[:5]}
def create_ticket(title: str, body: str, priority: str) -> dict:
if not title.startswith("[TEST] "):
raise ToolError("title must start with '[TEST] ' in the sandbox")
key = hashlib.sha256(f"{title}|{body}|{priority}".encode()).hexdigest()
if key in IDEMPOTENCY: # повтор того же вызова не создаёт дубль
return {"id": IDEMPOTENCY[key], "duplicate": True}
ticket_id = f"T-{len(TICKETS) + 1}"
TICKETS[ticket_id] = {"id": ticket_id, "title": title, "body": body,
"priority": priority}
IDEMPOTENCY[key] = ticket_id
return {"id": ticket_id, "duplicate": False}
def get_ticket(ticket_id: str) -> dict:
if ticket_id not in TICKETS:
raise ToolError(f"ticket {ticket_id} not found")
return TICKETS[ticket_id]
def spec(name, description, properties, required):
return {"type": "function", "function": {
"name": name, "description": description, "strict": True,
"parameters": {"type": "object", "properties": properties,
"required": required, "additionalProperties": False}}}
TOOLS = {
"search_tickets": (search_tickets, spec(
"search_tickets", "Search tickets by a title substring.",
{"query": {"type": "string", "minLength": 2}}, ["query"])),
"create_ticket": (create_ticket, spec(
"create_ticket", "Create a TEST ticket. Title must start with '[TEST] '.",
{"title": {"type": "string", "maxLength": 120},
"body": {"type": "string", "maxLength": 2000},
"priority": {"type": "string", "enum": ["low", "normal", "high"]}},
["title", "body", "priority"])),
"get_ticket": (get_ticket, spec(
"get_ticket", "Read one ticket by id, e.g. T-1.",
{"ticket_id": {"type": "string", "pattern": "^T-[0-9]+$"}}, ["ticket_id"])),
}
TOOL_SPECS = [s for _, s in TOOLS.values()]
SYSTEM = (
"You create test tickets. First search for a duplicate, then create the "
"ticket, then read it back with get_ticket. Finish with one short sentence "
"that contains the ticket id. Do not call tools you are not given."
)
def execute(name: str, raw_args: str) -> dict:
"""Единая точка исполнения: ошибки превращаются в данные, а не в падение."""
if name not in TOOLS:
return {"ok": False, "error": f"unknown tool '{name}'",
"available": sorted(TOOLS)}
func, tool_spec = TOOLS[name]
try:
args = json.loads(raw_args or "{}")
except json.JSONDecodeError as e:
return {"ok": False, "error": f"arguments are not valid JSON: {e.msg}"}
schema = tool_spec["function"]["parameters"]
errors = sorted(Draft202012Validator(schema).iter_errors(args),
key=lambda e: list(e.path))
if errors:
return {"ok": False, "error": "; ".join(
f"{'/'.join(map(str, e.path)) or 'args'}: {e.message}" for e in errors[:3])}
try:
return {"ok": True, "result": func(**args)}
except ToolError as e:
return {"ok": False, "error": str(e)}
except Exception as e: # неожиданный сбой инструмента не должен валить цикл
return {"ok": False, "error": f"internal tool error: {type(e).__name__}"}
def verify(created_ids: list[str]) -> str | None:
"""Проверка по состоянию системы, а не по словам модели."""
if not created_ids:
return "No ticket was created."
t = TICKETS.get(created_ids[-1])
if not t or not t["title"].startswith("[TEST] "):
return "The created ticket is missing or has a wrong title."
return None
def run_agent(client, goal: str, max_steps: int = MAX_STEPS) -> dict:
messages = [{"role": "system", "content": SYSTEM},
{"role": "user", "content": goal}]
log, created, seen = [], [], {}
verify_retries = 1
for step in range(1, max_steps + 1):
resp = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOL_SPECS,
tool_choice="auto", temperature=0)
msg = resp.choices[0].message
entry = {"role": "assistant", "content": msg.content}
if msg.tool_calls:
entry["tool_calls"] = [
{"id": c.id, "type": "function",
"function": {"name": c.function.name,
"arguments": c.function.arguments}}
for c in msg.tool_calls]
messages.append(entry)
if not msg.tool_calls:
problem = verify(created)
if problem is None:
return {"status": "done", "answer": msg.content, "steps": step, "log": log}
if verify_retries == 0:
return {"status": "failed", "reason": problem, "steps": step, "log": log}
verify_retries -= 1
messages.append({"role": "user", "content": f"Check failed: {problem} Fix it."})
continue
for call in msg.tool_calls:
sig = (call.function.name, call.function.arguments)
seen[sig] = seen.get(sig, 0) + 1
if seen[sig] > 2: # зацикливание на одном и том же вызове
return {"status": "failed", "reason": "repeated call", "steps": step, "log": log}
t0 = time.monotonic()
out = execute(call.function.name, call.function.arguments)
if out["ok"] and call.function.name == "create_ticket":
created.append(out["result"]["id"])
log.append({"step": step, "tool": call.function.name,
"args": call.function.arguments, "ok": out["ok"],
"ms": round((time.monotonic() - t0) * 1000)})
messages.append({"role": "tool", "tool_call_id": call.id,
"content": json.dumps(out, ensure_ascii=False)})
return {"status": "step_limit", "steps": max_steps, "log": log}
if __name__ == "__main__":
from openai import OpenAI
client = OpenAI(base_url=os.environ.get("LLM_BASE_URL", "https://openrouter.ai/api/v1"),
api_key=os.environ["LLM_API_KEY"])
result = run_agent(client, "Create a test ticket about a broken login button, priority high.")
print(json.dumps(result, ensure_ascii=False, indent=2))Как читать этот код
Функция execute — единственное место, где модель касается реального мира: инструмент известен, JSON валиден, аргументы проходят схему, и только потом вызов функции. В run_agent три выхода, и каждый имеет свой статус: done (проверка пройдена), failed (повтор вызова или провал проверки после повторной попытки) и step_limit. Вызывающий код различает их и решает, что делать дальше.
Что добавить перед продакшеном
- Таймаут на каждый вызов и общий бюджет запуска. Для необратимых действий нужно подтверждение человека: руководство OpenAI советует не отключать подтверждения инструментов.
- Минимальные права. OWASP называет корнями избыточной агентности лишние функции, лишние разрешения и лишнюю автономность.
- Метрики: доля успешных запусков, число шагов, частота ошибок инструментов; подход описан в статье про LLMOps.
Итого
Агент — это цикл вокруг модели, а надёжность создают не умные промпты, а ограничители: схема аргументов, лимит шагов, ошибки как данные, идемпотентность и проверка итога по состоянию системы. Начните с трёх инструментов и жёсткого лимита, а планирование усложняйте потом. О том, как удешевить такой цикл маленькими моделями, читайте в материале про маршрутизацию в агентах, а о защите от внедрения инструкций — в статье про prompt injection.