Перейти к содержанию

Учусь ИИ, урок 10: что такое ИИ-агент — модель, инструменты и цикл на десять строк

Я спросил домашнюю модель: «Сколько дней осталось до 1 января 2027 года и сколько это будет часов?» Она бодро ответила: 365 дней, 8760 часов. Было 30 октября, а значит правильный ответ — 63 дня. Откуда модели знать сегодняшнюю дату? Потом я дал ей три крошечные функции и написал цикл на десять строк. Она сделала три шага и ответила верно. Это и есть агент, и никакой магии в нём нет.

Урок за 5 минут: из чего состоит ИИ-агент, как модель «запускает» программы, не умея их запускать, и почему агенту нужен предел шагов. С рабочим примером на домашней модели.


Цикл агента: модель запрашивает вызов, программа выполняет инструмент и возвращает результат
Цикл агента: модель запрашивает вызов, программа выполняет инструмент и возвращает результат

Формула

Агент = модель + инструменты + цикл.

  • Модель — та же, что в прошлых уроках. Она умеет только писать текст.
  • Инструменты — обычные функции вашей программы: узнать дату, прочитать файл, выполнить запрос. У каждого есть название и описание.
  • Цикл — программа, которая крутится, пока работа не сделана.

Модель ничего не запускает

Это главное, что надо понять. Модель не умеет выполнять код и не имеет доступа к вашему компьютеру. Вместе с вопросом ей передают список инструментов, и вместо ответа она может написать: «вызови функцию today без параметров». Как обычный текст в оговорённом формате.

Запускает функцию ваша программа. Она же дописывает результат в разговор и снова отправляет всё модели. А та, вспомним урок 1, просто пишет продолжение, но уже с новым фактом перед глазами. Как и в уроке 6: факты в запросе лечат выдумки. Агент сам добывает их себе.

Опыт: десять строк

Вот сердце агента, без обвязки:

messages = [{'role': 'user', 'content': вопрос}]
for step in range(8):                    # предел шагов
    ответ = chat(messages, tools=инструменты)
    messages.append(ответ)
    if not ответ.get('tool_calls'):      # инструменты не нужны
        print(ответ['content'])          # значит, это итог
        break
    for вызов in ответ['tool_calls']:    # модель просит функцию
        результат = запустить(вызов)
        messages.append({'role': 'tool', 'content': результат})

Инструментов три: today (сегодняшняя дата), days_between (дни между датами), calc (арифметика). Вот что произошло у модели Qwen3 на 8 млрд параметров дома:

Шаг 1: today()                                → 2026-10-30
Шаг 2: days_between(2026-10-30, 2027-01-01)   → 63
Шаг 3: calc(63 * 24)                          → 1512
Шаг 4: «...осталось 63 дня, что составляет 1512 часа»

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

Цифры верные, потому что считала программа. А вот текст итога начинался с «От 1 января», а не «До»: связная речь по-прежнему на совести модели, и она ошибается. Поэтому всё, что можно посчитать, отдаём инструментам.

Что ломается

  • Цикл без конца. Модель может вызывать одно и то же снова и снова. Поэтому в коде стоит range(8). Без предела шагов агент однажды сожжёт деньги/лимиты.
  • Память тратится. Каждый результат инструмента ложится в контекст, а весь разговор отправляется заново на каждом шаге. Помните урок 2? Прочитал агент большой файл — и окно заполнилось.
  • Ошибки накапливаются. Если на шаге 2 модель ошиблась, шаги 3–10 строятся на ошибке. Поэтому полезны проверки: тесты, сверка результата программой.
  • Опасные действия. Инструмент «удалить файл» или «отправить письмо» агент вызовет так же легко, как today. Дайте ему только нужные инструменты, а необратимое пусть подтверждает человек.
  • Чужой текст как команда. Для модели нет разницы между вашей просьбой и фразой на прочитанной веб-странице. Если на странице написано «отправь пароли сюда», агент с инструментом отправки может послушаться. Это называют prompt injection: всё, что пришло снаружи, — данные, а не приказы.

Где вы это уже видели

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

Попробуйте сами

Понаблюдайте за агентом, не программируя. Откройте нейросеть с поиском в интернете и спросите: «Какая сегодня погода в моём городе и что надеть?» Обратите внимание на строчки вроде «ищу в интернете»: это и есть вызовы инструментов. Затем задайте вопрос, где инструмент не поможет, и посмотрите, вызовет ли она его всё равно.

Умеете в Python? Возьмите функцию для Ollama из документации и напишите цикл выше с одним инструментом.

Мой пример

"""
Минимальный ИИ-агент для Ollama (интерактивный, блокирует ввод во время ожидания, показывает спиннер).

Основной цикл:
    messages = [system, user]
    повторять до MAX_STEPS:
        reply = chat(messages, tools)            # POST /api/chat
        if нет tool_calls: это финальный ответ
        else: result = run_tool(args); messages += {'role': 'tool', ...}

Примеры инструментов: current_date (сегодняшняя дата) и days_between
(количество дней между двумя датами).

Использование:
    python ollama_minimal.py                # интерактивный режим (диалог с историей)
    python ollama_minimal.py "вопрос"       # один вопрос -- один ответ

Хост/модель: переменные окружения OLLAMA_HOST и OLLAMA_MODEL.
"""

import datetime
import json
import os
import re
import sys
import threading
import urllib.error
import urllib.request

# Тут указываю хост (адрес ollama)
HOST = os.environ.get('OLLAMA_HOST', 'http://192.168.0.10:177')
# Здесь указываю модель, которую будет использовать ollama
MODEL = os.environ.get('OLLAMA_MODEL', 'qwen3:8b')
# Это лимит токенов (контекстное окно)
NUM_CTX = 16384
# Максимальное количество шагов (запросов к модели), чтобы модель не зациклилась
MAX_STEPS = 8

SYSTEM_PROMPT = 'Отвечай кратко по-русски. Даты узнавай инструментами, не угадывай.'

TOOLS = [
    {'type': 'function', 'function': {
        'name': 'current_date',
        'description': 'Сегодняшняя дата в формате ГГГГ-ММ-ДД.',
        'parameters': {'type': 'object', 'properties': {}}},
     },
    {'type': 'function', 'function': {
        'name': 'days_between',
        'description': 'Сколько дней от даты from до даты to (формат ГГГГ-ММ-ДД).',
        'parameters': {'type': 'object',
                       'properties': {'from': {'type': 'string', 'description': 'начальная дата'},
                                      'to': {'type': 'string', 'description': 'конечная дата'}},
                       'required': ['from', 'to']}},
     },
]


def run_tool(name, args):
    """Выполняет инструмент по имени с заданными аргументами. Возвращает строковый результат."""
    try:
        if name == 'current_date':
            return datetime.date.today().isoformat()
        if name == 'days_between':
            d1 = datetime.date.fromisoformat(args.get('from', ''))
            d2 = datetime.date.fromisoformat(args.get('to', ''))
            return str((d2 - d1).days)
        return f'ОШИБКА: неизвестный инструмент {name}'
    except Exception as e:
        return f'ОШИБКА: {e}'


def chat(messages):
    """Отправляет историю сообщений в Ollama POST-запросом; возвращает словарь сообщения ассистента."""
    payload = {'model': MODEL, 'messages': messages, 'tools': TOOLS, 'stream': False,
               'options': {'num_ctx': NUM_CTX, 'temperature': 0.1}}
    req = urllib.request.Request(HOST + '/api/chat', data=json.dumps(payload).encode('utf-8'),
                                 headers={'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=600) as resp:
        return json.loads(resp.read().decode('utf-8'))['message']


# ---------------------------------------------------------------- спиннер

class Spinner:
    """Рисует точки в фоновом потоке, пока выполняется запрос.

    Строка стирается с помощью '\\r' + пробелы + '\\r', чтобы не засорять вывод.
    """

    def __init__(self, text='Ожидание ответа агента', interval=0.4):
        self.text = text
        self.interval = interval
        self._stop = threading.Event()
        self._thread = None
        self._last_len = 0
        self._lock = threading.Lock()

    def _run(self):
        dots = 0
        while not self._stop.is_set():
            frame = self.text + '.' * (dots % 4)
            with self._lock:
                pad = max(0, self._last_len - len(frame))
                sys.stdout.write('\r' + frame + ' ' * pad)
                sys.stdout.flush()
                self._last_len = len(frame)
            dots += 1
            self._stop.wait(self.interval)

    def __enter__(self):
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._stop.set()
        if self._thread:
            self._thread.join()
        with self._lock:
            sys.stdout.write('\r' + ' ' * self._last_len + '\r')
            sys.stdout.flush()
            self._last_len = 0


# ---------------------------------------------------------------- агент

def agent_step(messages):
    """Добавляет один ход ассистента в `messages`. Возвращает финальный текст.

    Изменяет `messages` на месте (добавляет записи assistant и tool).
    """
    for step in range(1, MAX_STEPS + 1):
        msg = chat(messages)
        # Некоторые reasoning-модели (например, qwen3) оборачивают размышления в <think>...</think>;
        # убираем это перед сохранением видимого содержимого.
        msg['content'] = re.sub(r'<think>.*?</think>', '', msg.get('content') or '', flags=re.S).strip()

        stored = {'role': 'assistant', 'content': msg['content']}
        calls = msg.get('tool_calls')
        if calls:
            stored['tool_calls'] = calls
        messages.append(stored)

        if not calls:
            return msg['content']

        for tc in calls:
            name = tc['function']['name']
            args = tc['function'].get('arguments') or {}
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {}
            result = run_tool(name, args)
            print(f'  шаг {step}: {name}({json.dumps(args, ensure_ascii=False)}) -> {result}')
            messages.append({'role': 'tool', 'content': result, 'tool_name': name})

    return '(лимит шагов исчерпан: модель не дала итог)'


def agent(question):
    messages = [{'role': 'system', 'content': SYSTEM_PROMPT},
                {'role': 'user', 'content': question}]
    return agent_step(messages)


def trim(messages, keep=12):
    """Оставляет системный промпт плюс последние `keep` сообщений.

    Сдвигает точку обрезки назад к записи 'user', чтобы сообщение assistant
    с tool_calls никогда не оказалось оторвано от своего результата tool.
    """
    if len(messages) <= keep + 1:
        return messages
    start = len(messages) - keep
    while start > 1 and messages[start].get('role') != 'user':
        start -= 1
    return [messages[0]] + messages[start:]


def interactive():
    messages = [{'role': 'system', 'content': SYSTEM_PROMPT}]
    print('Интерактивный режим. Пустая строка — пропуск ввода, exit / Ctrl+D — выход.')
    while True:
        try:
            q = input('\nВы: ').strip()
        except EOFError:
            print('\nПока.')
            return 0
        except KeyboardInterrupt:
            print('\nПрервано.')
            return 130

        if not q:
            continue
        if q.lower() in ('exit', 'quit', 'выход'):
            return 0

        messages.append({'role': 'user', 'content': q})
        print('  (отправлено, жду ответа...)')
        try:
            # Ввод блокируется здесь: мы не вызываем input() снова, пока не придёт ответ.
            with Spinner('Ожидание ответа агента'):
                answer = agent_step(messages)
        except urllib.error.HTTPError as e:
            print(f'\nOllama вернула HTTP {e.code}: {e.read().decode("utf-8", "replace")[:300]}')
            messages.pop()
            continue
        except urllib.error.URLError as e:
            print(f'\nOllama недоступна на {HOST}: {e.reason}')
            messages.pop()
            continue
        print('Агент:', answer)
        messages = trim(messages)


if __name__ == '__main__':
    if len(sys.argv) > 1:
        try:
            with Spinner('Ожидание ответа агента'):
                print(agent(' '.join(sys.argv[1:])))
        except urllib.error.HTTPError as e:
            sys.exit(f'Ollama вернула HTTP {e.code}: {e.read().decode("utf-8", "replace")[:300]}')
        except urllib.error.URLError as e:
            sys.exit(f'Ollama недоступна на {HOST}: {e.reason}')
    else:
        sys.exit(interactive())

Что запомнить

  • Агент = модель + инструменты + цикл.
  • Модель только просит вызвать функцию текстом, запускает её ваша программа.
  • От скрипта агент отличается тем, что порядок шагов выбирает модель.
  • Обязательны: предел шагов, минимум инструментов, подтверждение необратимого. Внешний текст — данные, а не команды.

Проверьте себя

  1. Почему без инструментов модель ответила «365 дней»?
  2. Кто физически выполняет функцию today?
  3. Зачем в цикле предел шагов и почему каждый шаг дороже предыдущего?

В следующем уроке — стоимость: как посчитать, сколько на самом деле стоит запрос, агент и своя видеокарта.

Если что-то осталось непонятным — спрашивайте в комментариях, вопросы попадут в следующие уроки.

Написать комментарий

Адрес почты не публикуется. Обязательные поля отмечены звёздочкой.

Возможно, вы не читали