Агентный цикл своими руками

Что строим

Агент из урока a2-1, но руками. Модель подменяем fake model (поддельной моделью) — обычной функцией, которая читает context (историю сообщений) и возвращает либо вызов tool (инструмента), либо итоговый ответ. Так код запускается без ключей, без сети и без пакетов, а поведение каждый раз одинаковое. Всё остальное — настоящее: так же устроен harness (обвязка) в любом агенте.

Части программы:

• Tool registry (реестр инструментов): словарь имя → функция, описание, параметры с типами, флаг «нужно подтверждение».
• Session (сессия): личность клиента после входа. Модель её не видит и не заполняет.
• run_tool: проверка, что tool есть, аргументы ровно те, что в schema (описании параметров), типы верные, подтверждение получено. Любая ошибка — не падение, а Observation (наблюдение) для модели.
• agent: цикл с лимитом шагов.

Код целиком

Сохрани как agent.py и запусти python3 agent.py.

import json

SESSION = {"customer_id": "C-001"}  # identity comes from login, never from the model
TXNS = {"C-001": [("taxi", "2026-09", 3200), ("taxi", "2026-09", 1800), ("food", "2026-09", 5400)]}
TOOLS = {}

def tool(name, description, params, confirm=False):
    def wrap(fn):
        TOOLS[name] = {"fn": fn, "description": description, "params": params, "confirm": confirm}
        return fn
    return wrap

@tool("txn_sum", "Customer's spending for one category and month (YYYY-MM).",
      {"category": str, "month": str})
def txn_sum(category, month):
    rows = [a for c, m, a in TXNS[SESSION["customer_id"]] if c == category and m == month]
    return {"category": category, "month": month, "count": len(rows), "total_kzt": sum(rows)}

@tool("card_block", "Block the customer's card. Changes state.", {}, confirm=True)
def card_block():
    return {"status": "blocked"}

def fake_model(messages, tool_specs):
    """Stands in for an LLM: reads the context, returns a tool call or a final answer."""
    results = [json.loads(m["content"]) for m in messages if m["role"] == "tool"]
    if not results:
        return {"tool": "txn_sum", "args": {"category": "taxi", "month": "2026-09"}}
    last = results[-1]
    if "error" in last:
        return {"final": "Не получилось: " + last["error"]}
    return {"final": f"Такси в сентябре: {last['count']} поездки, {last['total_kzt']} ₸."}

def run_tool(name, args, confirmed=False):
    spec = TOOLS.get(name)
    if spec is None:
        return {"error": f"unknown tool '{name}', available: {sorted(TOOLS)}"}
    if set(args) != set(spec["params"]):
        return {"error": f"arguments must be exactly {sorted(spec['params'])}"}
    for p, t in spec["params"].items():
        if not isinstance(args[p], t):
            return {"error": f"'{p}' must be {t.__name__}"}
    if spec["confirm"] and not confirmed:
        return {"error": "needs customer confirmation: ask the customer first"}
    try:
        return spec["fn"](**args)
    except Exception as e:  # a tool failure becomes an observation, not a crash
        return {"error": f"{type(e).__name__}: {e}"}

def agent(question, model=fake_model, max_steps=5):
    specs = [{"name": n, "description": s["description"], "params": list(s["params"])}
             for n, s in TOOLS.items()]
    messages = [{"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        reply = model(messages, specs)
        if "final" in reply:
            return reply["final"], step
        result = run_tool(reply["tool"], reply.get("args", {}))
        print(f"step {step}: {reply['tool']}({reply.get('args', {})}) -> {result}")
        messages.append({"role": "assistant", "content": json.dumps(reply)})
        messages.append({"role": "tool", "content": json.dumps(result, ensure_ascii=False)})
    return "Лимит шагов исчерпан: передаю оператору.", max_steps

print(agent("Сколько я потратил на такси в сентябре?"))

Вывод (проверен запуском 10.10.2026, Python 3.12):

step 1: txn_sum({'category': 'taxi', 'month': '2026-09'}) -> {'category': 'taxi', 'month': '2026-09', 'count': 2, 'total_kzt': 5000}
('Такси в сентябре: 2 поездки, 5000 ₸.', 2)

Модель вызвана два раза: первый раз она попросила tool, второй — прочитала Observation и ответила. Сумма 3 200 + 1 800 = 5 000 ₸ посчитана кодом tool, а не моделью.

Где здесь гарантии

Пройди по коду глазами инженера безопасности. У txn_sum нет параметра customer_id: клиент берётся из SESSION, поэтому модель не может попросить чужие операции. Это poka-yoke (защита конструкцией). Если модель всё же передаст лишний аргумент customer_id, run_tool вернёт ошибку «arguments must be exactly…». card_block помечен confirm=True, и без явного подтверждения код откажет, как бы уверенно модель ни просила. Лимит max_steps гарантирует, что цикл кончится, и говорит, что видит клиент при срабатывании. Каждый шаг печатается: это зачаток trace (журнала шагов).

Как подставить настоящую модель

Замени fake_model функцией, которая отправляет messages и tool_specs в API модели. У Claude Messages API та же форма. Tools передаются полем tools (имя, описание, input_schema в формате JSON Schema). Если модель хочет вызвать tool, ответ кончается с stop_reason: "tool_use" и содержит блок tool_use с id, name и input. Ты выполняешь tool своим run_tool и отправляешь результат следующим сообщением пользователя, блоком tool_result с тем же tool_use_id. Цикл, проверки и лимиты остаются твоими: API лишь возвращает предложение модели. Для текстовой модели без native tool calling (вызова tools на уровне API) работает старый способ ReAct (схема Thought → Action → Observation из урока a2-2): формат Thought/Action в prompt (инструкции для модели) и stop sequence (строка, на которой API обрывает генерацию) Observation:, чтобы модель не дописала результат сама.

Что дают framework и no-code

Библиотеки агентов и no-code конструкторы (визуальные сборщики без кода) дают готовые адаптеры моделей, память, повторы и интерфейс. Платишь ты непрозрачностью: Anthropic в «Building effective agents» предупреждает, что слои абстракции прячут настоящие prompts и ответы и усложняют отладку, и советует начинать с прямых вызовов API и понимать код framework, если берёшь его. Теперь ты знаешь, что внутри любого из них: этот же цикл.

Одно, что framework даёт сверх цикла и что стоит знать по имени, — checkpointing (сохранение состояния между шагами). В LangGraph checkpointer хранит состояние графа по thread_id (идентификатору разговора); на этом держатся human-in-the-loop, продолжение после сбоя и time travel (повтор с любого сохранённого шага для отладки). Функция interrupt() останавливает граф и ждёт человека, пока граф не вызовут снова с Command(resume=...) и тем же thread_id. Тонкость из документации: после resume узел выполняется с начала, и код до interrupt() идёт второй раз. Поэтому действие, которое что-то меняет, ставят после подтверждения или делают idempotent (повтор безопасен, урок a2-3). В нашем цикле аналог — записывать messages после каждого шага в файл или базу: упавший запуск можно продолжить, а провал — воспроизвести шаг за шагом.