Агенты на API без цензуры LLM: вызов функций, фреймворки и защита

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

Уберите фреймворки, и любой агент делает одно и то же четыре вещи. Вы отправляете разговор плюс список tools. Модель отвечает либо текстом, что означает завершение, либо tool_calls, что означает желание что-то выполнить. Вы запускаете каждый вызов и добавляете одно сообщение tool для каждого вызова, цитируя id вызова. Затем вы снова спрашиваете модель с более длинным разговором.

Сигнал для ветвления — finish_reason. Когда он равен tool_calls, сообщение ассистента имеет content равным null и массив tool_calls, каждая запись которого содержит id, имя функции и строку arguments в формате JSON. На странице справки показана точная структура. Два момента часто вызывают затруднения: аргументы — это строка, которую нужно распарсить, и сообщение ассистента с вызовами должно оставаться в истории перед результатами ваших инструментов в правильном порядке.

Полный цикл с нуля

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

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

Обратите внимание, что делает цикл при сбое: ошибки становятся результатами инструментов в виде строк. Это позволяет модели прочитать проблему и попробовать снова, например, исправив неверный id тикета, вместо того чтобы ваша программа падала. Оператор for ... else срабатывает только если лимит достигнут без окончательного ответа, что позволяет обнаружить зацикливание модели.

Держите промпты сфокусированными. Короткое системное сообщение, указывающее, когда вызывать инструмент, а когда отвечать напрямую, снижает количество лишних вызовов, и каждый пропущенный вызов экономит раунд, токены и один слот вашего лимита запросов.

Защита выполнения инструментов

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

• Белый список по имени. Ищите функцию в реестре, который вы создали. Никогда не используйте строку для импорта или выполнения кода.
• Разделяйте чтение и запись. Инструменты только для чтения работают свободно. Всё, что отправляет почту, тратит деньги, удаляет данные или работает с файлами, требует явного подтверждения человеком.
• Валидируйте аргументы по схеме, которую вы опубликовали: обязательные ключи, отсутствие лишних ключей, типы и ограничения по длине.
• Ограничивайте время и вывод. Устанавливайте таймауты и обрезайте результаты перед возвратом, чтобы огромный вывод инструмента не съел ваше контекстное окно на 100 000 токенов.
• Лимитируйте итерации и стоимость. Ограничивайте шаги на задачу и токены на сессию, используя блок usage.
• Логируйте каждый вызов с именем, аргументами и результатом, чтобы можно было проверить поведение.

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

Относитесь к результатам инструментов как к недоверенным тоже. Если инструмент возвращает текст с инструкциями, модель может им следовать. Оберните такой контент четкими разделителями в сообщении tool и укажите в системном промпте, что вывод инструмента — это данные, а не команды.

Написание схем инструментов для модели

Схема — единственная документация, которую видит модель, поэтому она заслуживает такого же внимания, как публичный API. Называйте функции глаголами и существительными, соответствующими задаче, например ticket_status или lookup_invoice, а не общими метками вроде run или helper. Пишите description как инструкцию: что делает функция, что возвращает и когда её не следует использовать.

• Держите параметры немногочисленными и плоскими. Два или три простых поля вызываются правильно гораздо чаще, чем вложенный объект с десятью полями.
• Используйте перечисления для закрытых наборов, например "enum": ["open", "closed"], чтобы модель не могла придумать значение.
• Описывайте каждое свойство с единицей измерения или примером: "ID тикета, например T-101" лучше, чем просто строка.
• Возвращайте компактные результаты. Строки или небольшие объекты JSON работают лучше всего; длинные дампы тратят контекст без особой пользы.
• Предпочитайте несколько узких инструментов одному универсальному инструменту с флагом режима.

Когда вызов не проходит валидацию, возвращайте сообщение об ошибке валидации как результат инструмента, а не вызывайте исключение. Модель обычно исправляет вызов в следующем раунде, так как она может прочитать, что было не так. Если она терпит неудачу тем же образом дважды, перепишите описание, а не цикл.

Конфигурация LangChain

Класс ChatOpenAI LangChain принимает base_url, поэтому пакет пользовательского провайдера не нужен. Установите model в uncensored, передайте ключ из вашей среды и свяжите инструменты с помощью bind_tools. Установите langchain-openai и langchain-core.

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

Результат вызова invoke — это AIMessage, в списке tool_calls которого содержатся распарсенные имена и аргументы, поэтому вы пропускаете ручной шаг с JSON. Чтобы получить полный автономный цикл, передайте тот же объект llm в конструкторы агентов LangChain; эндпоинту остаётся только поддерживать вызов функций, что он и делает. Если агент, кажется, игнорирует свои инструменты, убедитесь, что присутствуют docstring и аннотации типов, поскольку они становятся схемой, которую видит модель.

Конфигурация LlamaIndex

LlamaIndex поставляется с OpenAILike для серверов, следующих формату завершения чата. Важны три настройки. api_base указывает на API, is_chat_model=True выбирает маршрут чата, а is_function_calling_model=True сообщает фреймворку, что он может отправлять инструменты. Установите context_window в 100000, чтобы его сплиттеры текста и буферы памяти правильно настраивали размер. Установите llama-index-llms-openai-like.

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

Отсюда вы можете передать тот же llm рабочему агенту или движку запросов. Помните, что конвейеры поиска могут помещать много фрагментов в один промпт; с окном в 100 000 токенов, общим для промпта и завершения, ограничьте количество извлеченных фрагментов и держите max_tokens умеренным.

Управление агентами с учетом бюджета

Агент превращает одно сообщение пользователя в несколько запросов, поэтому планируйте мощность. Задача с пятью раундами инструментов стоит шесть вызовов модели. Десять одновременных задач поэтому создают шестьдесят запросов за короткий промежуток времени, и потолок в 300 запросов в минуту достигается быстрее, чем в приложении чата. Поставьте ограничитель скорости с ведрами токенов из руководства по ошибкам и лимитам запросов перед вашим клиентом.

Контекст растёт с каждым раундом, так как каждый результат инструмента добавляется в конец. Сокращайте или удаляйте старые результаты инструментов после их использования и следите за usage.prompt_tokens; когда он приближается к 50 000, сжимайте историю. Для оценки стоимости примите примерную задачу из 6 вызовов со средним объёмом 3 000 промпт-токенов и 200 токенов завершения: это 18 000 входных токенов (около $0,0045 по тарифу $0,25 за миллион) и 1 200 выходных токенов (около $0,0012 по тарифу $1,00 за миллион). Эти подсчёты токенов — предположения для сумм, а не измерения. Если вы новичок в эндпоинте, начните с docs и бесплатного пробного баланса.

Тестирование и отладка агента

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

1. Подделайте модель. Заглушите клиент, чтобы он возвращал сценарные сообщения tool_calls. Это тестирует ваш реестр, валидацию и завершение цикла без сети и без расходов.
2. Выборка по температуре. Используйте низкую temperature, например 0.2, для задач с вызовом инструментов. Меньшая случайность даёт более воспроизводимый JSON аргументов, а позже вы можете повысить её для этапов, связанных с генерацией текста.
3. Проверка траекторий. Для небольшого набора фиксированных промптов проверяйте, какие инструменты были вызваны и в каком порядке, а не только итоговый текст.

Обратите также внимание на finish_reason. Значение length в середине вызова инструмента означает, что аргументы были обрезаны; увеличьте max_tokens (до 32 000), и JSON будет завершён. Потоковая передача также работает с инструментами, но накапливайте фрагменты аргументов по каждому индексу вызова и анализируйте их только после закрытия потока. Для повторения материала о таких фрагментах см. раздел потоковой передачи в документации.