Главная / Статьи / Практические заметки: потоковые ответы в LangGraph: 3 практических шаблона, необходимых для использования.

langgraphllmagentsarchitecturetypescriptpython

Практические заметки: потоковые ответы в LangGraph: 3 практических шаблона, необходимых для использования.

Пошаговое руководство по практическим заметкам: потоковые ответы в LangGraph: 3 практических шаблона, необходимых для использования контрактов, проверок и готовых блоков кода командами, внедряющими этот шаблон.

3658 слов

В этом руководстве пошагово показан путь от сырья до готовой системы для темы: «Стриминг ответов в LangGraph: 3 практических подхода, которые должен знать каждый разработчик агентов». Основное внимание уделяется выполнимым шагам, явным проверкам и коду, который можно просто добавить в репозиторий без необходимости угадывать намерения автора. На этапе обзора необходимо определить входные данные, ответственного за шаг и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние. Необходимо документировать как успешный ход выполнения, так и пути восстановления после сбоев. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

Почему стриминг важнее, чем кажется

При работе над этапом «Почему стриминг важнее» сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Вносите контрольные точки после дорогостоящих шагов. Механизм возобновления работы не должен повторно оплачивать один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий узел.

Пример, над которым мы работаем

При работе над этапом «Пример, который мы реализуем», сначала запишите условия соглашения: необходимые входные данные, сигнал о успешном выполнении и то, что происходит при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как соглашение между входными данными и проверенными выходными результатами. Дайте названия элементам кода, определите критерии успешности и не допускайте молчаливого частичного выполнения задачи. Выполняйте контрольные точки после дорогостоящих операций. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий этап.

import asyncio
import operator
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph

load_dotenv()

# --- 1. STATE & GRAPH ---
class State(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)

def chatbot_node(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def dummy_node(state: State) -> State:
    return state

builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("dummy", dummy_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", "dummy")
builder.add_edge("dummy", END)
graph = builder.compile()

# --- 2A. stream_mode="updates" — one event per node ---
print("=== Method 1: stream_mode='updates' ===")
for event in graph.stream(
    {
        "messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
    },
    stream_mode="updates",
):
    for node_name, output in event.items():
        print(f"  [{node_name}] {output['messages']}")

# --- 2B. stream_mode="values" — full state after each node ---
print("\n=== Method 2: stream_mode='values' ===")
for snapshot in graph.stream(
    {
        "messages": [HumanMessage("Say hello in 3 languages")],
    },
    stream_mode="values",
):
    print(f"  State has {len(snapshot['messages'])} message(s) now")
    print(snapshot["messages"])

# --- 2C. astream_events — token-by-token (async) ---
async def token_stream():
    print("\n=== Method 3: Token-by-token streaming ===")
    print("🤖 Bot: ", end="", flush=True)
    async for event in graph.astream_events(
        {
            "messages": [HumanMessage("Count from 1 to 5 slowly, one per line")],
        },
        version="v2",
    ):
        if event["event"] == "on_chat_model_stream":
            chunk = event["data"]["chunk"].content
            if chunk:
                print(chunk, end="", flush=True)
    print()

asyncio.run(token_stream())

Шаг 1: Сначала разберитесь с проектированием состояния

При выполнении этапа 1 «Понимание структуры» сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Запишите также время выполнения операций и стоимость токенов или запросов рядом с результатами работы. Очевидность затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Выполняйте контрольные точки после дорогостоящих операций. Система должна не повторно взимать плату за один и тот же вызов ИИ-модели при повторной попытке обработки задачи. При выполнении этапа 1 «Понимание структуры» сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Одновременно задокументируйте успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки со стороны оператора и обработка некорректных сообщений являются частью продукта, а не элементами последующей доработки.

class State(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

{"messages": [some_new_message]}

Почему это важно для стриминга

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

Шаг 2: Сам граф специально разработан простым

Этап «Граф» на втором шаге работает наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успешного выполнения и не соглашайтесь на молчаливое частичное завершение работы. Сохраняйте состояние графа простым и типизированным. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.

chatbot_node

Этап chatbotnode работает наилучшим образом, когда его рассматривают как измеримую среду. Соберите один идеальный пример взаимодействия, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и могут нарушить возобновление работы после прерываний. Этап chatbotnode работает наилучшим образом, когда его рассматривают как измеримую среду. Соберите один идеальный пример взаимодействия, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

def chatbot_node(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

dummy_node

На этапе dummynode необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной структурой обработки. Внедрять утверждение человека там, где происходит трата средств или изменение данных в продакшене. Простая настройка во время компиляции не гарантирует полноты функционала системы.

def dummy_node(state: State) -> State:
    return state

START → chatbot → dummy → END

Метод 1: stream_mode="updates"

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

for event in graph.stream(..., stream_mode="updates"):

Что это делает

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

{
    "chatbot": {
        "messages": [...]
    }
}

{
    "dummy": {
        "messages": [...]
    }
}

Зачем полезен этот режим

При работе над этапом «Зачем полезен этот режим» сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список помогает сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые единицы кода, чем обширные скрипты. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Вносите контрольные точки после дорогостоящих шагов. Функция возобновления работы не должна повторно запускать один и тот же вызов большой языковой модели, когда оператор пытается выполнить более поздний элемент.

Лучшие сценарии применения

При работе над этапом определения наилучших случаев использования сначала запишите условия контракта: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте безусловного частичного завершения работы. Выполняйте контрольные точки после дорогостоящих операций. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.

Пример из реальной жизни

При работе над этапом примера из реального мира сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Выполняйте контрольные точки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов LLM при повторной попытке обработки последующего узла. При работе над этапом примера из реального мира сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Документируйте как успешный путь выполнения, так и пути восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

Метод 2: stream_mode="values"

Значения режима потока Method 2 в формате stage работают наилучшим образом, когда их рассматривают как измеримую поверхность. Сначала соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию, прежде чем расширять объем работы. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое какого-либо шага причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.

for snapshot in graph.stream(..., stream_mode="values"):

Что это делает

Этот этап «Что оно делает» работает наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успешного выполнения и не соглашайтесь на молчаливое частичное завершение работы. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.

print(f"  State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])

Почему это важно

Этап «Почему это важно» работает наилучшим образом, когда рассматривается как измеримая характеристика. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объема работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и могут нарушить возобновление работы после прерываний. Этап «Почему это важно» работает наилучшим образом, когда рассматривается как измеримая характеристика. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объема работ. Документируйте одновременно успешный и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

Лучшие сценарии применения

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

Пример из практики

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

Метод 3: astream_events() для потоковой обработки по токенам

Для astreamevents метода 3, предназначенных для использования на этапе реализации, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Регистрируйте время выполнения и стоимость токенов или запросов вместе с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. При следующем шаге, представляющем собой код или вызов инструмента, предпочтительнее использовать структурированные выходные данные с проверкой схемы вместо свободного текста. Для astreamevents метода 3, предназначенных для использования на этапе реализации, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами, добавляемыми позже.

async for event in graph.astream_events(..., version="v2"):

if event["event"] == "on_chat_model_stream":

chunk = event["data"]["chunk"].content

Что это делает

На этапе описания функционала сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые модули вместо обширных скриптов. При сбое какого-либо шага причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Выполняйте контрольные точки после дорогостоящих шагов. Система возобновления работы не должна повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий элемент работы.

Почему это важно

При работе над этапом «Почему это важно» сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и то, что происходит при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Укажите названия элементов, определите критерии успешности и не допускайте безусловного частичного завершения работы. Вносите контрольные точки после дорогостоящих операций. Система возобновления работы не должна снова взимать плату за один и тот же вызов LLM, когда оператор пытается выполнить следующий шаг.

Почему astream_events() является асинхронным

При работе над этапом «Почему astreamevents является асинхронным» сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным. Выполняйте контрольные точки после дорогостоящих операций. Система возобновления работы не должна снова взимать плату за один и тот же вызов LLM при повторной попытке обработки последующего узла. При работе над этапом «Почему astreamevents является асинхронным» сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки со стороны оператора и обработка некорректных сообщений являются частью продукта, а не элементами последующей доработки.

asyncio.run(token_stream())

Понимание фильтра событий

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

if event["event"] == "on_chat_model_stream":

chunk = event["data"]["chunk"].content

3 режима потоковой обработки простым языком

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

updates

Этап обновлений работает наилучшим образом, когда его рассматривают как измеримую среду. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил какое поле, и могут нарушить возобновление работы после прерываний. Этап обновлений работает наилучшим образом, когда его рассматривают как измеримую среду. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно успешный и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

values

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

astream_events

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

Какой режим стриминга следует использовать?

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

.

Используйте updates в следующих случаях:

При работе над этапом «Используйте обновления» сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Вносите контрольные точки после дорогостоящих шагов. Функция возобновления выполнения не должна повторно запрашивать один и тот же вызов LLM, когда оператор пытается выполнить более поздний узел.

Используйте values в следующих случаях:

При работе над разделом «Использование значений на этапе» сначала запишите контракт: необходимые входные данные, сигнал о успехе и что происходит при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными данными. Дайте названия результатам обработки, определите критерии успеха и не допускайте безусловного частичного завершения работы. Выполняйте контрольные точки после дорогостоящих операций. Механизм возобновления работы не должен снова взимать плату за один и тот же вызов LLM, когда оператор пытается выполнить следующий шаг.

Используйте astream_events в следующих случаях:

При работе над этапом «Использование astreamevents» сначала запишите условия взаимодействия: необходимые параметры входных данных, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Записывайте время выполнения операций, а также стоимость токенов или запросов рядом с результатами их работы. Отображение стоимости на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Выполняйте контрольные точки после дорогостоящих шагов. Механизм возобновления работы не должен снова взимать плату за один и тот же вызов большой языковой модели при повторной попытке обработки последующего узла. При работе над этапом «Использование astreamevents» сначала запишите условия взаимодействия: необходимые параметры входных данных, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки со стороны оператора и обработка некорректных сообщений являются частью продукта, а не элементами последующей доработки.

Лучшая когнитивная модель: стриминг для пользователей против стриминга для разработчиков

Этап «Лучшая когнитивная модель» работает наилучшим образом, если рассматривать его как измеримую основу. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работы.

Стриминг для пользователей

Стриминг для разработчиков

Шаблон работы в производственных условиях, который вам, вероятно, понадобится

Распространённые ошибки при стриминге ответов LangGraph

1. Использование стриминга токенов, когда на самом деле нужны обновления узлов

2. Ожидание того, что элементы типа values будут вести себя как токены при стриминге

3. Забывание о том, что функция astream_events() является асинхронной

4. Отсутствие фильтрации типов событий

5>Создание графов, которые поддерживают стриминг, но не предоставляют значимого состояния

Продвинутый совет: стриминг становится гораздо мощнее в многоэтапных агентских графах

updates

values

astream_events

for event in graph.stream(
    {
        "messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
    },
    stream_mode="updates",
):
    for node_name, output in event.items():
        print(f"\nNode: {node_name}")
        for message in output["messages"]:
            print(message)

  • Практические заметки: Anthropic Managed Agents против LangGraph против Google ADK: как выбрать один — Пошаговое руководство по использованию Практических заметок: Anthropic Managed Agents против LangGraph против Google ADK: как выбрать один, включающее шаблоны контрактов, проверок и блоки кода для команд, внедряющих эту архитектуру.
  • Практические заметки: 10 фреймворков ИИ-агентов, о которых стоит знать в 2026 году: LangGraph — Пошаговое руководство по использованию Практических заметок: 10 фреймворков ИИ-агентов, о которых стоит знать в 2026 году: LangGraph, включающее шаблоны контрактов, проверок и блоки кода для команд, внедряющих эту архитектуру.

    • Все статьиПрислать бриф