Вызов моделей через invoke, stream и batch, типы сообщений, content_blocks, мультимодальный ввод, bind_tools, with_structured_output, учёт токенов.
Прежде чем строить агентов, нужно уверенно работать с уровнем ниже: как вызывается модель, из чего состоит диалог и как получить от модели данные, а не текст.
init_chat_model возвращает объект модели с единым набором методов независимо от провайдера. Три базовых способа вызова покрывают почти все случаи.
invoke делает один запрос и возвращает AIMessage целиком:
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-4o-mini")
response = model.invoke("Почему попугаи разговаривают?")
print(response.text)stream отдаёт ответ по частям, и это то, что видит пользователь в чате как «печатающийся» текст:
for chunk in model.stream("Почему попугаи разговаривают?"):
print(chunk.text, end="", flush=True)batch обрабатывает список запросов параллельно. Это не то же самое, что цикл с invoke: запросы уходят одновременно, поэтому на пакете из 20 промптов разница во времени обычно кратная:
responses = model.batch(["Столица Франции?", "Столица Японии?", "Столица Перу?"])Параметры задаются при создании: temperature управляет разбросом (0 — максимально предсказуемо), max_tokens ограничивает длину ответа, timeout — время ожидания, max_retries — число повторов при сетевых сбоях.
model = init_chat_model(
"openai:gpt-4o-mini",
temperature=0,
max_tokens=1024,
timeout=60,
max_retries=10,
)LangChain сам повторяет запросы с экспоненциальной задержкой при сетевых ошибках, ответах 429 и 5xx. Отдельный обработчик писать не нужно, но стоит понимать: max_retries=10 при таймауте 60 секунд в худшем случае означает долгое ожидание вместо быстрой ошибки.
Диалог это список сообщений. Каждое имеет роль, и роль определяет, как модель его трактует.
SystemMessage задаёт правила и роль ассистента, идёт первым. HumanMessage — то, что написал пользователь. AIMessage — ответ модели, в нём же приезжают запросы на вызов инструментов. ToolMessage — результат выполнения инструмента, обязательно связанный с конкретным вызовом через tool_call_id.
from langchain.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
messages = [
SystemMessage("Ты лаконичный помощник. Отвечай одним предложением."),
HumanMessage("Что такое контрольная точка?"),
]
response = model.invoke(messages)Тот же диалог можно записать словарями в формате чат-совместимого API — LangChain принимает оба варианта:
messages = [
{"role": "system", "content": "Ты лаконичный помощник."},
{"role": "user", "content": "Что такое контрольная точка?"},
]
response = model.invoke(messages)Словари короче и удобны в примерах; классы дают автодополнение и защиту от опечаток в роли. В агентах вход обычно записывают словарями ({"messages": [{"role": "user", ...}]}), а на выходе получают объекты сообщений.
У сообщения есть три способа добраться до содержимого, и путаница между ними — источник неожиданных ошибок.
content — «сырое» поле: у одних провайдеров это строка, у других список словарей в их собственном формате. Работать с ним напрямую значит привязать код к провайдеру.
content_blocks — нормализованное представление: список типизированных блоков (text, reasoning, image, tool_call), одинаковый для всех провайдеров. Именно его используют, когда нужно разобрать составной ответ.
.text — просто текст ответа, склеенный из текстовых блоков. Это то, что нужно в 90% случаев.
response = model.invoke("Посчитай 2+2 и объясни ход мысли")
print(response.text) # только текст
for block in response.content_blocks:
print(block["type"]) # 'reasoning', 'text', ...Рассуждающие модели возвращают блоки типа reasoning с ходом мысли. У Anthropic они называются thinking, у OpenAI — reasoning, но content_blocks приводит их к общему виду. Практическое следствие: если вы печатаете response.content и видите список словарей вместо строки — переключитесь на .text или content_blocks.
Изображения, аудио и файлы передаются блоками внутри сообщения пользователя. Источник указывается ссылкой, base64-строкой или идентификатором файла у провайдера:
message = {
"role": "user",
"content": [
{"type": "text", "text": "Что изображено на схеме?"},
{"type": "image", "url": "https://example.com/diagram.png"},
],
}
response = model.invoke([message])Поддержка зависит от модели: текстовая модель на такой ввод ответит ошибкой. Проверяйте возможности конкретной модели до того, как закладываться на мультимодальность в архитектуре.
Агент — это цикл вокруг модели с инструментами. Но сам механизм доступен и напрямую: bind_tools возвращает модель, которая знает о переданных инструментах:
from langchain.tools import tool
@tool
def get_weather(location: str) -> str:
"""Получить погоду в указанном месте."""
return f"В {location} солнечно."
model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("Какая погода в Бостоне?")
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'location': 'Бостон'}, 'id': 'call_abc'}]Обратите внимание: инструмент не выполнен. bind_tools даёт только намерение — выполнять и добавлять ToolMessage придётся вручную. Именно эту рутину и берёт на себя create_agent, поэтому писать цикл самому обычно не нужно; но понимание, что внутри, помогает при отладке.
Когда нужен не текст, а данные, применяют with_structured_output со схемой. Модель обязана вернуть объект, соответствующий схеме, а LangChain провалидирует его:
from pydantic import BaseModel, Field
class Movie(BaseModel):
title: str = Field(description="Название фильма")
year: int = Field(description="Год выхода")
genres: list[str] = Field(description="Жанры, от одного до трёх")
structured_model = model.with_structured_output(Movie)
movie = structured_model.invoke("Расскажи про фильм «Начало»")
print(movie.year) # 2010Схемой может быть Pydantic-модель, TypedDict или JSON Schema. Pydantic предпочтительнее: он и валидирует, и даёт типы в IDE.
Разница с агентным response_format: with_structured_output — это одиночный вызов модели, а response_format в create_agent применяется к финальному ответу агента, который до этого мог сходить в инструменты. Для задачи «извлечь поля из текста» достаточно первого, агент здесь избыточен.
Каждый AIMessage несёт метаданные использования — по ним считают стоимость и ловят распухание контекста:
response = model.invoke("Привет")
print(response.usage_metadata)
# {'input_tokens': 8, 'output_tokens': 10, 'total_tokens': 18}В агенте стоит смотреть на input_tokens по шагам: если он растёт от вызова к вызову быстрее, чем добавляется полезной информации, значит в контекст попадают слишком объёмные результаты инструментов.
Когда провайдер отдаёт 429, помогает не увеличение числа повторов, а ограничитель частоты — он не даёт превысить лимит вообще:
from langchain.rate_limiters import InMemoryRateLimiter
limiter = InMemoryRateLimiter(requests_per_second=0.5)
model = init_chat_model("openai:gpt-4o-mini", rate_limiter=limiter)InMemoryRateLimiter работает в пределах процесса. Для нескольких воркеров этого мало: там лимит должен быть общим (через внешнее хранилище) или заданным с запасом на количество процессов.
Провайдеры OpenAI и Anthropic умеют кэшировать повторяющееся начало запроса и берут за него меньше. Это важно именно для агентов: системный промпт и описания инструментов отправляются на каждом шаге цикла заново.
Отсюда практический вывод по структуре промпта — стабильная часть (роль, правила, описания инструментов) идёт в начало, изменяемая (запрос пользователя, свежие данные) в конец. Тогда неизменный префикс попадает в кэш.
Ноль не делает модель «точнее», он делает её предсказуемее: при одинаковом входе ответ будет практически тем же. Для извлечения данных, классификации и вызова инструментов — это то, что нужно. Для генерации текстов, где повторяемость не требуется, разброс выше нуля даёт более живой результат. Если агент зацикливается на одном и том же неверном вызове инструмента, снижение температуры не поможет — проблема в описаниях инструментов или в модели.
Далее: Инструменты и первый агент