Как сделать AI-агента на Python без фреймворка: цикл, инструменты и память

Сначала оцените цену неверного действия

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

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

Цена ошибки определяет автономность. Неверный результат калькулятора заметит пользователь. Инструмент возврата в платёжной системе должен требовать подтверждения человека или быть недоступен модели. Агентный цикл не заменяет правила доступа.

Три способа собрать агента

Первый вариант: прямой OpenAI Python SDK и собственный цикл. Он хорош для узкой задачи и небольшого числа инструментов, когда нужны точный JSON, история и логи. Код не скрывает, кто вызвал функцию и почему цикл остановился.

Второй вариант: LangChain или LlamaIndex. LangChain удобен для адаптеров, трассировки и составных цепочек; LlamaIndex подходит, когда продукт строится вокруг индексирования и поиска по документам. За удобство платят дополнительными абстракциями, конфигурацией и местами, где поведение меняется вне вашего кода.

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

Критерии, которые меняют выбор

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

Порога по числу инструментов нет. Один обратимый вызов внутреннего API может годами жить в ручном цикле. Пять систем с разными правами, потоковой выдачей, retrieval и оценкой качества уже требуют столько поддержки, что разумнее взять библиотеку или отдельный оркестратор.

Из чего состоит цикл агента

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

У цикла два независимых условия остановки. Нормальное: нет function_call, но есть непустой текст пользователю. Предохранитель: достигнут MAX_TOOL_ROUNDS, даже если модель просит функции. Без лимита ошибка описания инструмента или неудачный результат могут породить бесконечный разговор модели с собой.

Описание функции работает как договор, а не как подсказка. additionalProperties: false запрещает неожиданные аргументы. Ваш код всё равно повторно проверяет имя функции, типы, права и допустимый диапазон. Формат вызовов и ответов описан в документации OpenAI по function calling.

Реестр инструментов и исполнимый пример

Ниже один файл agent.py. Нужны pip install openai и переменная OPENAI_API_KEY. Калькулятор ограничен арифметическим деревом ast: строка не попадает в eval, поэтому модель не получает путь к файловой системе или оболочке.

import ast
import json
import logging
import operator
import time
from openai import APIConnectionError, APIStatusError, OpenAI

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("agent")
client = OpenAI(max_retries=0, timeout=30.0)
MODEL = "gpt-4.1-mini"
MAX_TOOL_ROUNDS = 4

TOOLS = [{
    "type": "function", "name": "calculator",
    "description": "Считает арифметическое выражение из чисел, скобок и + - * /.",
    "parameters": {
        "type": "object",
        "properties": {"expression": {"type": "string"}},
        "required": ["expression"], "additionalProperties": False
    }
}]

OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
       ast.Mult: operator.mul, ast.Div: operator.truediv}

def calculator(expression):
    if len(expression) > 100:
        raise ValueError("слишком длинное выражение")
    tree = ast.parse(expression, mode="eval")

    def walk(node):
        if isinstance(node, ast.Constant) and type(node.value) in (int, float):
            return node.value
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -walk(node.operand)
        if isinstance(node, ast.BinOp) and type(node.op) in OPS:
            return OPS[type(node.op)](walk(node.left), walk(node.right))
        raise ValueError("разрешены только числа и арифметические операции")

    return {"value": walk(tree.body)}

HANDLERS = {"calculator": calculator}

def trim_memory(memory, max_messages=12):
    return memory[-max_messages:]

def create_response(input_items):
    for attempt in range(3):
        try:
            return client.responses.create(
                model=MODEL, input=input_items, tools=TOOLS
            )
        except (APIConnectionError, APIStatusError) as exc:
            transient = (isinstance(exc, APIConnectionError)
                         or exc.status_code in (408, 409, 429)
                         or exc.status_code >= 500)
            if not transient or attempt == 2:
                raise
            delay = 2 ** attempt
            log.warning("model_retry=%s error=%s", attempt + 1, type(exc).__name__)
            time.sleep(delay)

def run_agent(prompt, memory):
    context = trim_memory(memory) + [{"role": "user", "content": prompt}]
    try:
        response = create_response(context)
    except Exception as exc:
        log.exception("model_start_failed")
        return f"Не удалось получить ответ модели: {type(exc).__name__}"

    for step in range(1, MAX_TOOL_ROUNDS + 2):
        calls = [item for item in response.output if item.type == "function_call"]
        log.info("step=%s calls=%s", step, len(calls))

        if not calls:
            answer = response.output_text.strip()
            if not answer:
                return "Модель завершила ответ без текста. Повторите запрос."
            memory.extend([
                {"role": "user", "content": prompt},
                {"role": "assistant", "content": answer}
            ])
            return answer

        if step > MAX_TOOL_ROUNDS:
            break
        context.extend(response.output)
        for call in calls:
            try:
                args = json.loads(call.arguments)
                handler = HANDLERS[call.name]
                result = handler(**args)
                log.info("tool=%s status=ok", call.name)
            except Exception as exc:
                result = {"error": f"{type(exc).__name__}: {exc}"}
                log.warning("tool=%s status=error", call.name)

            context.append({
                "type": "function_call_output", "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False)
            })

        try:
            response = create_response(context)
        except Exception as exc:
            log.exception("model_continue_failed")
            return f"Инструмент выполнился, но модель не ответила: {type(exc).__name__}"

    return "Лимит вызовов инструментов исчерпан. Уточните запрос."

if __name__ == "__main__":
    memory = []
    while True:
        prompt = input("Вы: ").strip()
        if prompt.lower() in {"выход", "exit", "quit"}:
            break
        print("Агент:", run_agent(prompt, memory))

После сохранения проверка может выглядеть так:

from agent import run_agent
print(run_agent("Посчитай (18 * 7) / 3", []))

Реестр HANDLERS не строится из ответа модели: она выбирает только ключ заранее созданного словаря. Для HTTP-инструмента добавьте allowlist хостов, таймаут, лимит тела ответа и отдельные учётные данные с минимальными правами.

Память, лимиты и наблюдаемость

Долгая память и техническая трасса решают разные задачи. memory хранит пары «запрос пользователя и финальный ответ ассистента». Временный context содержит вызовы функций и результаты только до финального текста, поэтому обрезка не оставляет function_call без соответствующего function_call_output.

max_messages=12 лишь ограничивает размер контекста. Для долгого диалога заменяйте старые сообщения краткой сводкой подтверждённых пользователем фактов: целей, ограничений, выбранных сущностей. Не переносите туда сырые логи, токены, пароли и полный вывод внутренних API.

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

Повторяйте сбои сети, 429 и статусы 5xx. Повтор 400 с неверной схемой не исправит запрос. В примере задержка 2 ** attempt: первая повторная попытка ждёт 1 секунду, следующая 2 секунды. В рабочем коде добавьте случайный разброс (jitter), чтобы параллельные процессы не повторяли запрос одновременно.

Первый релиз: одна узкая задача

Условному внутреннему боту пишут: «Один заказ стоит 18 USD, заказов 7. Какая выручка?» Модель выбирает calculator с 18 * 7, получает 126 и формулирует ответ. Результат легко проверить: число вернул инструмент, модель не считала его сама. Боту не нужны платежи, поиск документов и автономные последующие действия.

Такой релиз проверяет промпт и схему. Журнал показывает, вызвала ли модель калькулятор, передала ли допустимое выражение и остановилась ли после наблюдения. Если задача влияет на бюджет, до запуска оцените стоимость AI-агента для бизнеса: расходы включают не только токены, но и интеграцию, проверку ответов, поддержку и обработку сбоев.

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

Где ручной агент перестаёт быть верным выбором

Фреймворк оправдан, когда код повторяет инфраструктурную работу: подключает много провайдеров, строит retrieval по документам, хранит трассы, запускает оценки качества и координирует несколько агентов. LangChain или LlamaIndex могут сократить клеевой код, но ограничения прав, схемы, лимит шагов и аудит остаются вашей обязанностью. Владельца агента стоит определить заранее, особенно если агент влияет на продуктовые эксперименты. Как распределять такую ответственность, разобрано в материале про организационный дизайн команд роста.

Не выбирайте ручной цикл, если нужны детерминированный результат, формальное согласование, необратимая операция или строгий аудит каждого бизнес-правила. Тогда модель может извлекать данные и предлагать черновик, а действие должен выполнять обычный сервис с явной последовательностью проверок. Если нужен один прозрачный маршрут через инструменты, собственный код на SDK проще контролировать, чем фреймворк, добавленный ради названия.