LangChain with_structured_output: типизированный ответ модели

LangChain with_structured_output решает проблему, с которой сталкивается каждый, кто разбирает ответы модели регулярками: вместо текста вы получаете типизированный объект, а вместо json.loads с надеждой на лучшее — проверку схемы. В статье покажем, как задать схему через Pydantic, какой метод выбрать при работе через OpenAI-совместимый шлюз, что делать при ошибках парсинга и как проверить несколько моделей на своих данных. Код следует документации LangChain (разделы Models и Structured output); версии пакетов сверяйте с ней.

LangChain with_structured_output: базовый пример с Pydantic

Схема описывается обычным классом Pydantic. Документация LangChain отмечает, что Pydantic даёт самый богатый набор возможностей: валидацию полей, описания, вложенные структуры. Описания полей (description) попадают в запрос к модели, поэтому пишите их внятно.

pip install langchain langchain-openai pydantic
export ZUKKO_API_KEY="fetch_ваш_ключ"

import os
from typing import Literal

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI


class Ticket(BaseModel):
    """Разбор обращения клиента."""

    category: Literal["оплата", "доступ", "ошибка", "другое"] = Field(
        description="Категория обращения"
    )
    urgency: int = Field(ge=1, le=5, description="Срочность от 1 (низкая) до 5 (высокая)")

Результат это экземпляр Ticket, а не словарь. Для ChatOpenAI адрес шлюза должен заканчиваться на /v1. Если у вас нет Pydantic, with_structured_output принимает также TypedDict и JSON Schema, но по документации проверку значений в этих случаях придётся писать самим.

Три метода: json_schema, function_calling, json_mode

Параметр method определяет, как LangChain просит модель вернуть структуру. По документации:

• json_schema использует родные возможности структурированного вывода у провайдера;
• function_calling получает структуру через принудительный вызов инструмента, соответствующего схеме;
• json_mode генерирует корректный JSON, но схему нужно описать в промпте.

Какой метод по умолчанию у ChatOpenAI, смотрите в документации интеграции для вашей версии. Здесь есть нюанс, из-за которого стоит выбирать метод явно. Для LangChain ChatOpenAI это «OpenAI», поэтому он может использовать родной режим OpenAI. А за шлюзом стоят разные модели (Claude, GPT, DeepSeek, GLM и другие), и мы не гарантируем, что каждая из них одинаково принимает такой режим. Самый переносимый вариант это function_calling: он опирается на вызов инструментов, который в каталоге умеют многие модели.

structured = llm.with_structured_output(Ticket, method="function_calling")

Если модель не поддерживает нужный режим, вы получите ошибку 400 от провайдера или None вместо объекта. Поэтому не угадывайте, а проверяйте на своих данных (ниже есть скрипт). Сравнение поведения моделей при вызове функций разобрано в статье function calling у Claude, GPT и Qwen↗.

Что делать при ошибках парсинга

Модель может вернуть значение вне Literal, срочность 7 вместо диапазона 1 до 5 или вообще не вызвать нужный инструмент. Чтобы не получать исключение в середине конвейера, включите include_raw=True. Тогда метод возвращает словарь с тремя ключами: raw (сообщение модели), parsed (объект или None) и parsing_error.

structured = llm.with_structured_output(Ticket, method="function_calling", include_raw=True)


def classify(text: str, retries: int = 2) -> Ticket:
    last_error = None
    for _ in range(retries + 1):
        out = structured.invoke(text)
        if out["parsed"] is not None:
            return out["parsed"]
        last_error = out["parsing_error"]
    raise ValueError(f"не удалось получить структуру: {last_error}")


print(classify("Не могу войти в кабинет, пишет неверный пароль"))

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

• Обрезанный ответ. При слишком малом max_tokens JSON не закрывается. Увеличьте лимит выходных токенов.
• Слишком строгая схема. Literal с десятком редких значений или вложенность в несколько уровней. Упростите или разбейте на два шага.
• Размытые описания полей. Модель опирается на description, и расплывчатое описание даёт случайные значения.
• Модель или метод не подходят. Смените метод или модель.

Структурированный ответ внутри агента

Если у вас агент на create_agent, структуру можно получить в конце цикла через response_format. По документации LangChain можно передать схему напрямую (тогда выбор стратегии автоматический) или обернуть её в ToolStrategy. У ToolStrategy по умолчанию включена обработка ошибок валидации: модель получает сообщение об ошибке и пробует снова.

from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

agent = create_agent(model=llm, tools=[], response_format=ToolStrategy(Ticket))
result = agent.invoke({"messages": [{"role": "user", "content": "Ошибка 402 при запросе, что делать?"}]})
print(result["structured_response"])

Для связки с инструментами модель должна уметь вызывать инструменты и одновременно возвращать структуру. Проверяйте такую связку на выбранной модели отдельно.

Без LangChain: OpenAI SDK

Запрос «python openai structured output» часто означает тот же результат без LangChain. В SDK openai есть вспомогательный метод parse, который принимает Pydantic-класс (ниже используется Ticket из первого примера):

import os
from openai import OpenAI

client = OpenAI(base_url="https://api.zukko.pro/v1", api_key=os.environ["ZUKKO_API_KEY"])

completion = client.chat.completions.parse(
    model="gpt-6.1-sol",
    messages=[{"role": "user", "content": "Оплатил вчера, баланс не пополнился"}],
    response_format=Ticket,
)
print(completion.choices[0].message.parsed)

Поддержка этого режима зависит от модели и провайдера за шлюзом: проверяйте на своей. Запасной путь для любой модели: попросить JSON в промпте и проверить его вручную через Ticket.model_validate_json(text). Этот способ разобран в статье строго структурированный JSON от модели↗.

Проверка моделей на своих данных и цена

Заведите 20–30 реальных примеров и прогоните через каждую модель, считая долю parsed is None:

models = ["gpt-6.1-sol", "claude-sonnet-5-5", "deepseek-v4.1-flash"]
samples = ["Оплатил, баланс не пополнился", "Не приходит письмо", "Хочу сменить e-mail"]

for name in models:
    model = ChatOpenAI(model=name, base_url="https://api.zukko.pro/v1", api_key=os.environ["ZUKKO_API_KEY"])
    chain = model.with_structured_output(Ticket, method="function_calling", include_raw=True)
    failed = sum(chain.invoke(s)["parsed"] is None for s in samples)
    print(name, "ошибок:", failed, "из", len(samples))

Сколько это стоит. Пусть запрос с определением схемы занимает 300 входных токенов, ответ 60 выходных. Формула: вход × цена входа + выход × цена выхода, цены на 18 октября 2026, в миллионных долях доллара:

Определение схемы добавляется к каждому запросу, так что длинные описания полей увеличивают входные токены. Для Claude через OpenAI-формат весь вход идёт по цене входа. Список моделей и цен: GET https://api.zukko.pro/v1/models.

Частые вопросы

Какие модели поддерживают структурированный вывод?

Универсального списка мы не даём: поддержка зависит от модели и выбранного метода. Надёжнее всего начать с function_calling и проверить свои 20–30 примеров на нескольких моделях из каталога.

Что вернётся, если модель не смогла заполнить схему?

С include_raw=True вы получите parsed=None и описание ошибки в parsing_error. Без этого параметра возможно исключение.

Можно ли использовать вложенные структуры?

Да, Pydantic-модели могут содержать списки и другие модели, как показывает документация. Но чем сложнее схема, тем выше риск ошибок, поэтому начинайте с плоской.

Влияет ли схема на цену?

Да: описание схемы отправляется на вход каждого запроса и оплачивается как входные токены.

Создайте ключ для тестов в кабинете↗ и поставьте на него дневной лимит. Справочник клиента на Python: Python↗.