Потоковый вывод LangChain

Потоковый вывод заставляет ответы ИИ отображаться посимвольно, как при наборе текста, что значительно улучшает пользовательский опыт. Agent в LangChain имеет встроенную всестороннюю поддержку потокового вывода.

Зачем нужен потоковый вывод

Если используется invoke(), пользователям приходится ждать, пока Agent завершит все шаги (несколько вызовов модели + выполнение инструментов), прежде чем увидеть результат. Для сложных задач это может занять более десяти секунд или даже дольше.

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

Метод	Пользовательский опыт	Применимые сценарии
invoke()	Ожидание → увидеть полный результат сразу	Скрипты, API, пакетная обработка
stream()	Видеть каждый Token в реальном времени	Чат-интерфейсы, отображение в реальном времени

stream_mode="messages" — потоковая передача Token за Token

Это самый детализированный режим потоковой передачи, в котором каждый chunk соответствует одному Token:

Пример

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash")
agent = create_agent(
model=model,
system_prompt="Вы — ассистент EXAMPLE Tutorial.",
)

# stream_mode="messages" возвращает Token по одному
print("Потоковый вывод в реальном времени:")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Опишите Python одним предложением")]},
stream_mode="messages",
):
# msg_chunk — это AIMessageChunk
# каждый чанк содержит лишь небольшой фрагмент содержимого
if msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print() # перевод строки в конце

Результат выполнения:

Live streaming output:
Example is a free online learning platform for programming beginners, offering rich technical tutorials and hands-on examples.

Понимание метаданных

metadata содержит информацию об источнике этого чанка:

Пример

print("Просмотр информации о метаданных:"\n")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Привет, представься")]},
stream_mode="messages",
):
if msg_chunk.content and len(msg_chunk.content) > 5:
print(f"Содержимое: {msg_chunk.content}")
print(f"Исходный узел: {metadata.get('langgraph_node')}")
print(f"Тип сообщения: {type(msg_chunk).__name__}")
break # просмотреть только первый значимый фрагмент

Результат выполнения:

Inspect metadata:

Content: Hello! I am an AI assistant
Source node: model
Message type: AIMessageChunk

stream_mode="updates" — Пошаговый просмотр процесса выполнения агента

Этот режим очень полезен при создании интерфейсов, которым нужно отображать "процесс мышления":

Пример

from langchain.tools import tool

@tool
def search_course(keyword: str) -> str:
"""Поиск курсов на EXAMPLE"""
courses = {
"python": "Базовый учебник по Python3 (30 глав, 20 часов)",
"html": "Базовый учебник по HTML (25 глав, 15 часов)",
}
return courses.get(keyword.lower(), "Не найдено подходящих курсов")

agent = create_agent(
model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
tools=[search_course],
system_prompt="Вы консультант по курсам в EXAMPLE.",
)

# используйте режим updates для просмотра каждого шага
print("=== Процесс выполнения агента ==="\n")
for chunk in agent.stream(
{"messages": [HumanMessage(content=Помогите проверить курсы Python)]},
stream_mode="updates",
):
for node_name, update in chunk.items():
print(f"[{node_name}]", end=" ")
if "messages" in update:
for msg in update[messages"]:
if msg.type == "ai":
if hasattr(msg, 'tool_calls') and msg.tool_calls:
calls = [tc['name'] for tc in msg.tool_calls]
print(fЗапросы вызовов: {calls})
elif msg.content:
print(fОтвет: {msg.content[:80]})
elif msg.type == "tool":
print(fИнструмент вернул [{msg.name}]: {msg.content})

Результат выполнения:

=== Agent 执行过程 ===

[model] 请求调用: ['search_course']
[tools] 工具返回 [search_course]: Python3 基础教程（30章，20小时）
[model] Reply: Example has the Python3 basics tutorial, 30 chapters, about 20 hours of study, perfect for Python beginners.

stream_mode="custom" — Отправка пользовательских событий

С помощью runtime.stream_writer() Middleware вы можете отправлять пользовательские события в поток:

Пример

from langchain.agents.middleware import before_model, after_model

@before_model
def notify_before(state, runtime):
Отправка пользовательских событий перед вызовом модели
runtime.stream_writer({
"type": "status",
"message": Думаю...,
})
return None

@after_model
def notify_after(state, runtime):
Отправляйте пользовательские события после вызова модели
last_msg = state["messages"][-1] if state.get("messages") else None
has_tools = hasattr(last_msg, 'tool_calls') and last_msg.tool_calls

if has_tools:
tool_names = [tc['name'] for tc in last_msg.tool_calls]
runtime.stream_writer({
"type": "status",
"message": fВызов инструментов: {', '.join(tool_names)}...,
})
else:
runtime.stream_writer({
"type": "status",
"message": Ответ завершён,
})
return None

agent = create_agent(
model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
tools=[search_course],
middleware=[notify_before, notify_after],
system_prompt="Вы консультант по курсам в EXAMPLE.",
)

# Используйте stream_mode=["updates", "custom"], чтобы получать оба события одновременно
print(=== Смешанный потоковый вывод ===\n")
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content=Проверьте курсы Python)]},
stream_mode=["updates", "custom"],
):
if mode == "custom":
print(f[Пользовательское событие] Статус: {chunk['message']})
elif mode == "updates":
for node_name, update in chunk.items():
if "messages" in update:
for msg in update[messages"]:
if msg.type == "ai" and msg.content:
print(f[Ответ] {msg.content})

Результат выполнения:

=== 混合流式输出 ===

[custom event] Status: thinking...
[custom event] Status: calling tool: search_course...
[reply] Example has the Python3 basics tutorial, 30 chapters, about 20 hours of study.
[custom event] Status: thinking...
[custom event] Status: answer completed

Режимы stream_mode можно комбинировать, например stream_mode=["updates", "custom", "messages"]. Однако слишком много режимов увеличит количество событий в потоке, поэтому рекомендуется выбирать их по необходимости.

Асинхронный потоковый вывод

В веб-сервисах использование асинхронной потоковой передачи позволяет избежать блокировки цикла событий:

Пример

import asyncio

async def stream_agent():
Асинхронно запустить Agent в режиме потоковой передачи
agent = create_agent(
model=init_chat_model("deepseek:deepseek-v4-flash"),
system_prompt="Вы — ассистент EXAMPLE Tutorial.",
)

full_response = ""
async for msg_chunk, metadata in agent.astream(
{"messages": [HumanMessage(content=Расскажи о Python в одном предложении)]},
stream_mode="messages",
):
if msg_chunk.content:
full_response += msg_chunk.content
print(msg_chunk.content, end="", flush=True)

print(f"\n\nПолная длина ответа: {len(full_response)} символов)

# Запустить асинхронную функцию
asyncio.run(stream_agent())

Пример интеграции с FastAPI

Пример

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

app = FastAPI()
agent = create_agent(
model=init_chat_model("deepseek:deepseek-v4-flash"),
system_prompt="Вы — ассистент EXAMPLE Tutorial.",
)

@app.get("/chat")
async def chat(message: str):
Интерфейс чата, возвращает потоковый ответ SSE
async def generate():
async for msg_chunk, metadata in agent.astream(
{"messages": [HumanMessage(content=message)]},
stream_mode="messages",
):
if msg_chunk.content:
# Формат SSE: data: xxx
yield f"data: {msg_chunk.content}\n\n"
yield "data: [DONE]\n\n"

return StreamingResponse(
generate(),
media_type="text/event-stream",
)

# Запуск: uvicorn main:app --reload

В производственной среде рекомендуется создавать экземпляр Agent как глобальный singleton, чтобы не пересоздавать его при каждом запросе. Стоимость создания Agent невелика (в основном компиляция графа), но повторное использование экземпляра эффективнее.

stream_mode: краткая справка

Mode	Гранулярность	Объект итерации	Типичное использование
messages	Уровень токенов	(AIMessageChunk, metadata)	Эффект набора текста, чат в реальном времени
updates	Уровень узла	{node_name: state_update}	Отображение процесса мышления
values	Уровень узла (полный)	Полное состояние	Снимок состояния, отладка
custom	Индивидуальный	Произвольный dict	Уведомление о прогрессе, отправка статуса
debug	Подробный	Отладочная информация	Устранение неполадок во время разработки

Другие расширения

Agent Конфигурация разработки

Справочное руководство

Потоковый вывод LangChain

Зачем нужен потоковый вывод

stream_mode="messages" — потоковая передача Token за Token

Пример

Понимание метаданных

Пример

stream_mode="updates" — Пошаговый просмотр процесса выполнения агента

Пример

stream_mode="custom" — Отправка пользовательских событий

Пример

Асинхронный потоковый вывод

Пример

Пример интеграции с FastAPI

Пример

stream_mode: краткая справка