LangChain: как указать наш base_url вместо OpenAI

LangChain по умолчанию бьётся в api.openai.com или api.anthropic.com, и оба клиента — ChatOpenAI и ChatAnthropic — принимают параметр base_url, который эту логику полностью переопределяет. Ключ в обоих случаях один и тот же fetch_....

ChatOpenAI: GPT, GLM

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-5.6-sol",
    api_key="fetch_ваш_ключ",
    base_url="https://api.zukko.pro/v1",
)

response = llm.invoke("Объясни разницу между base_url и base URL клиента")
print(response.content)

Тот же класс без изменений подходит для остального OpenAI-совместимого каталога — меняется только model: gpt-5.6-luna, glm-5.3, glm-5.3-flash (актуальный список — GET https://api.zukko.pro/v1/models).

ChatAnthropic: Claude, без /v1 в base_url

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(
    model="claude-sonnet-5",
    api_key="fetch_ваш_ключ",
    base_url="https://api.zukko.pro",
)

response = llm.invoke("Сравни claude-opus-5-5 и claude-sonnet-5 по задачам")
print(response.content)

Разница с предыдущим примером не только в классе — base_url здесь без /v1, потому что ChatAnthropic сам достраивает путь до /v1/messages и добавляет заголовок anthropic-version. Если по привычке из OpenAI-кода дописать /v1 и сюда, каждый вызов будет улетать на несуществующий путь и возвращать 404, при этом сам ключ будет приниматься нормально — это самая частая ошибка при переносе кода между двумя классами.

Доступные модели Claude: claude-opus-5-5 и claude-sonnet-5; актуальный список — GET https://api.zukko.pro/v1/models.

Streaming работает как обычно

Стриминг не требует ничего специфичного для шлюза — работает через штатный .stream() у обоих классов:

for chunk in llm.stream("Посчитай факториал 10 и покажи шаги"):
    print(chunk.content, end="", flush=True)

Подсчёт токенов без реального запроса

Если нужно заранее прикинуть длину промпта для Claude, не тратя деньги на сам вызов, проще сходить напрямую в /v1/messages/count_tokens через requests, минуя LangChain — этот эндпоинт бесплатный и не запускает генерацию:

import requests

resp = requests.post(
    "https://api.zukko.pro/v1/messages/count_tokens",
    headers={
        "Authorization": "Bearer fetch_ваш_ключ",
        "anthropic-version": "2023-06-01",
        "Content-Type": "application/json",
    },
    json={
        "model": "claude-sonnet-5",
        "messages": [{"role": "user", "content": "текст для оценки длины"}],
    },
)
print(resp.json())

Обработка ошибок в цепочках

LangChain пробрасывает ошибку API как исключение от openai или anthropic SDK — оборачивайте вызов в try/except и смотрите код внутри тела ответа, а не только HTTP-статус:

from openai import APIStatusError

try:
    llm.invoke("...")
except APIStatusError as e:
    body = e.response.json()
    code = body.get("error", {}).get("code")
    if code == "insufficient_balance":
        # баланс на нуле — запрос до модели не дошёл, ничего не потрачено
        ...
    elif code == "model_disabled":
        # 503: модель выключена — переключитесь на модель из GET /v1/models
        ...

Коды из практики: 401 — неверный или обрезанный ключ; 402/insufficient_balance — пустой баланс, проверяется до похода к провайдеру; 404/model_not_found — опечатка в id, сверяйте с GET /v1/models; 429/key_daily_limit_reached или key_monthly_limit_reached — упёрлись в лимит конкретного ключа из кабинета; 403/model_not_allowed_for_key — модель запрещена политикой ключа; 503/model_disabled — модель выключена, замените её на модель из актуального списка; 503/provider_unavailable — модель временно недоступна, имеет смысл ретраить с задержкой.

Идемпотентность при ретраях в цепочках

Если оборачиваете вызов в собственный retry (например, через tenacity), передавайте заголовок Idempotency-Key в конкретном вызове, а не в конструкторе — например llm.invoke(prompt, extra_headers={"Idempotency-Key": key}) у ChatOpenAI. Ключ генерируется один раз на логический запрос до цикла ретраев и остаётся тем же на все его попытки — повтор в течение 24 часов вернёт тот же ответ и не спишет деньги дважды. Не кладите Idempotency-Key в default_headers: тогда один ключ уйдёт со всеми запросами клиента, и уже второй, другой запрос получит 422 idempotency_conflict.

Ключи и лимиты

Ключ создаётся один раз в кабинете https://zukko.pro/app/api-keys и там же показывается единственный раз — храните его в переменных окружения или секрет-хранилище, не в коде репозитория. На странице ключа можно задать дневной или месячный лимит расхода в долларах и список разрешённых моделей — полезно для сервисного ключа, который живёт в проде, отдельно от ключа для локальной разработки.

Кэш ответов при повторяющихся вызовах

Если в цепочке есть шаг с одинаковым запросом на одинаковых данных (классификация, извлечение сущностей по фиксированной схеме), включите кэш явным заголовком X-Zukko-Cache (temperature=0 в конструкторе тоже включает кэш, но только у моделей, которые принимают этот параметр; ChatAnthropic с claude-sonnet-5 или claude-opus-5-5 может получить 400 на temperature, отличную от 1, поэтому Claude-моделям его не передавайте):

llm = ChatOpenAI(
    model="gpt-5.6-sol",
    api_key="fetch_ваш_ключ",
    base_url="https://api.zukko.pro/v1",
    default_headers={"X-Zukko-Cache": "true"},
)

Повторный запрос с полностью совпадающим телом от того же пользователя вернётся мгновенно и бесплатно, в ответе будет заголовок X-Zukko-Cache: hit; «почти одинаковые» запросы или общий префикс в кэш не попадают, а сами попадания в /v1/usage не пишутся. Для чата кэш живёт до 7 суток — снижает счёт для цепочек, которые повторно гоняют в точности тот же запрос на тех же данных.

Свои инструменты (tools) через LangChain

Tool calling у ChatOpenAI и ChatAnthropic работает без доработок — bind_tools со списком Pydantic-схем или обычных функций уходит в теле запроса тем же способом, что и в оригинальных SDK, потому что /v1/chat/completions и /v1/messages принимают поле tools в том же формате:

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city"""
    return f"В {city} сейчас +18"

llm_with_tools = llm.bind_tools([get_weather])
response = llm_with_tools.invoke("Какая погода в Москве?")

Мониторинг расходов из кода

Если строите сервис на LangChain и хотите логировать стоимость каждого вызова, а не только смотреть баланс раз в день, читайте заголовки X-Zukko-Cost-Micros и X-Fetch-Balance-Cents (баланс в нём — в микродолларах, 1 000 000 = $1) из сырого HTTP-ответа — они есть только у обычных вызовов, у .stream() их нет; у ChatOpenAI это доступно через .invoke(...).response_metadata в части реализаций, но надёжнее дойти до заголовков напрямую через with_raw_response в клиенте openai, либо сходить в /v1/usage отдельным запросом за историей (она общая на весь аккаунт, все ключи вместе):

usage = requests.get(
    "https://api.zukko.pro/v1/usage",
    headers={"Authorization": "Bearer fetch_ваш_ключ"},
    params={"limit": 50},
).json()