Потоковые глубокие агенты LangChain: реальные планы, вызовы инструментов и токены.
Замените блокирующий вызов на поток, чтобы пользователи могли наблюдать работу глубокого агента: три режима потока, пространства имен для подагентов и фильтрация токенов для краткого изложения.
Источники: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II
Немногие вещи так быстро разрушают доверие, как замороженный экран. Когда пользователь не может понять, думает ли агент, застрял ли он или уже не работает, он предполагает худшее. Технология стриминга заменяет это молчание живым отображением процесса: план, вызовы инструментов и текст появляются в реальном времени, и люди могут видеть, что что-то происходит.
Почему функция invoke заставляет глубокого агента казаться неактивным
Попробуйте запустить глубокого агента для выполнения реальной задачи при наблюдении кого-то ещё. Запрос проходит через функцию invoke, после чего отображение замирает на сорок секунд. Внутри агент читает файлы, составляет план, использует инструменты и принимает решения. Пользователь видит лишь неподвижный квадрат и индикатор загрузки, что может означать интенсивную работу или тихую ошибку. Поскольку эти два состояния выглядят одинаково, люди перезагружают страницу и подают заявку с указанием на зависание программы.
Ничего не было размещено. У агента просто не было способа для самовыражения. Блоки invoke созданы именно так: они выполняют весь цикл незаметно для пользователя и возвращают единый готовый ответ, что идеально подходит для скриптов или заданий cron, за которыми никто не наблюдает. Однако это неподходящий вариант, когда с другой стороны находится человек.
Решение заключается в том, чтобы оставить цикл без изменений. Вместо блокирующего вызова используется поток, благодаря чему планы, вызовы инструментов и токены отображаются в реальном времени.
От invoke к stream
Стриминг тесно связан с invoke. Вместо ожидания окончательного состояния вы циклически обрабатываете объект agent.stream(...), и данные поступают каждый раз, когда выполняется шаг цикла. Асинхронная версия agent.astream(...) имеет такую же структуру и используется веб-серверами.
Ментальная модель проста. Под капотом глубокого агента находится надежный LangGraph, который генерирует событие на каждом шаге, а потоковая обработка передает эти события по мере их возникновения. Это не отдельный агент и не отдельный цикл — просто вид на тот агент, который вам уже знаком.
Сначала необходимо уточнить один момент, от которого зависят все примеры: формат потока v2, активируемый с помощью version="v2". Каждый фрагмент представляет собой небольшой словарь с тремя ключами:
typeуказывает на тип события.ns, пространство имён, указывает на того, кто его сгенерировал.dataсодержит полезную нагрузку.
Эти три ключа остаются неизменными независимо от того, что вы передаёте в поток, что делает код более читаемым, а не запутанным из-за разбора кортежей.
Три режима, три уровня
Стрим — это не один длинный шланг. Параметр stream_mode определяет тип событий, которые вы получаете, причем три распространённых варианта соответствуют трем уровням детализации.
updates — это режим пошагового отображения. Каждый раз, когда узел графа завершает свою работу, поступает одно событие, что на практике означает, что агент совершил один шаг или запустил какой-то инструмент. Этот уровень подходит для журнала прогресса: он достаточен для отображения действий агента без избытка информации, и является разумным стандартным вариантом для краткого отчёта вида «вот что делает агент».
messages — это режим отображения по токенам. Результаты работы модели поступают по частям по мере их генерации, точно так же, как при вводе текста чат-ассистентом. Используйте его, когда человек должен видеть текст, генерируемый агентом, в момент его создания. Сюда также поступают вызовы потоковых инструментов, поэтому вы можете объявить «вызов read_file» в тот момент, когда модель принимает решение о вызове, а не после его завершения.
values — это режим отображения кадра. После каждого шага вы получаете полное состояние агента, а не только информацию о произошедших изменениях. Этот способ требует больше ресурсов и используется реже, но он подходит тогда, когда важен полный обзор на каждом этапе, например для отображения актуального списка задач или рабочих файлов агента.
Предоставление списка режимов позволяет получить несколько результатов сразу, и именно так обычно работают отлаженные интерфейсы: updates управляет индикатором шагов, в то время как messages формируют видимый текст. Пример грубого варианта — чистый журнал прогресса от updates:
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Summarize the repo’s failing tests”}]},
stream_mode=”updates”,
version=”v2”,
):
if chunk[”type”] == “updates”:
for node_name in chunk[”data”]:
print(f”step: {node_name}”)
Выводится список шагов графа по мере их выполнения: по одной строке на каждый ход модели и каждую операцию инструмента. Экран, казавшийся замороженным, теперь видимо движется вперед. Если вместо этого установить stream_mode в значение "messages", вы получите сам текст, готовый к выводу в момент, когда агент его пишет.
Различение главного агента и его подагентов
Глубокий агент редко работает в одиночку в течение длительного времени. Как только он передает подзадачу подагенту, сразу несколько агентов начинают генерировать результаты. Если отправлять всё в один поток, получится беспорядок: фрагмент предложения от координатора, несколько токенов от исследователя, вызов инструмента из неизвестного источника. Поле ns решает эту проблему. При включении потоковой передачи подграфов с параметром subgraphs=True каждый фрагмент содержит такое значение.
Это правило легко применять. Если пространство имён пустое, событие генерировало основное агентство. Если в нём есть сегмент tools:, значит, его создало подагентство, причём указывается название этого подагентства. Затем результаты можно отсортировать по источнику: текст координатора здесь, а результаты работы каждого подагентства — по отдельности.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Research and summarize quantum computing”}]},
stream_mode=”messages”,
subgraphs=True, # ①
version=”v2”,
):
if chunk[”type”] == “messages”: # ②
token, metadata = chunk[”data”] # ③
is_subagent = any(s.startswith(”tools:”) for s in chunk[”ns”]) # ④
source = “subagent” if is_subagent else “main” # ⑤
if token.content:
print(f”[{source}] {token.content}”, end=”“, flush=True) # ⑥
В этом фрагменте важны шесть деталей. Во-первых, именно параметр subgraphs=True придает значение параметру ns, поскольку без него события подагента не имеют отдельного пространства имён. Во-вторых, ветка сохраняет только события messages, так как этот цикл обрабатывает токены, а не шаги. В-третьих, данные каждого фрагмента распаковываются в токен вместе с его метаданными. В-четвертых, именно проверка пространства имён выполняет основную работу: наличие сегмента tools: где-либо в ns указывает на то, что токен происходит от подагента. В-пятых, полученный булево значение преобразуется в читаемый источниковый меткой. В-шестых, вывод с этой меткой в качестве префикса превращает один запутанный поток в два отдельных потока с атрибутами.
Перед выводом каждый токен маркируется именем его генератора, поэтому данные, передаваемые координатором исследователю, представляют собой два потока с указанными именами. Полноценный веб-интерфейс требует больше усилий, но пространство имён каждого фрагмента данных является тем же сигналом, который считывает фронтенд на базе React.
Отдельные токены, не принадлежащие говорящему агенту
Это явление удивляет пользователей при первом длительном потоковом взаимодействии. Чтобы управлять собственным контекстом, система сжимает более старые сообщения, когда объём диалога приближается к лимиту модели. Процесс сжатия также представляет собой вызов модели, в результате чего генерируются токены, которые попадают в поток messages рядом с настоящим ответом агента. Если этот поток напрямую передаётся пользователю, тот увидит, как агент вдруг зачитывает краткий обзор собственной истории диалога, что может показаться ошибкой.
К счастью, токены имеют метки. Каждое событие messages содержит метаданные, а токены подведения итогов имеют значение lc_source равное "summarization". Пропустите их и оставьте остальные.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “...”}]},
stream_mode=”messages”,
version=”v2”,
):
token, metadata = chunk[”data”]
if metadata.get(”lc_source”) == “summarization”:
continue
# ... handle real agent tokens here
Достаточно одной проверки метаданных, чтобы отличить агента, общающегося с пользователем, от системы, общающейся сама с собой. Отображается только первый вариант. Это незначительная работа, но она значительно улучшает качество результата.
Наблюдение за процессом выполнения vs поддержание диалога
Стриминг и диалог часто путают, поэтому необходимо провести четкое разграничение. Все, что было сказано выше, касается наблюдения за процессом выполнения задачи. Каждый вызов stream, как и каждый вызов invoke, по умолчанию является одноразовым. Агент выполняет задачу, заканчивает ее и забывает о ней, поэтому последующий вызов начинается без какой-либо памяти о предыдущем обмене сообщениями.
Поддержание открытого диалога на протяжении нескольких этапов, чтобы фраза «исправьте и другой тест» сработала, поскольку агент помнит первую правку, представляет собой отдельный механизм. Он основан на потоках выполнения и указателе состояния — слое сохранения, в котором постоянно обновляемая структура хранит свое состояние. Важно различать: потоки обеспечивают видимость происходящего внутри одной сессии, а потоки выполнения — преемственность между сессиями. Они независимы друг от друга, и часто требуется использовать оба.
Небольшой интерфейс терминала для отображения процесса
Далее необходимо собрать всё в консоль, которую можно будет запустить на практике. Она должна выводить одну строку за каждым шагом по мере продвижения агента и транслировать его текст по мере написания; поэтому требуется задать оба режима одновременно и принимать решения в зависимости от значения поля type.
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="You are a code-maintenance assistant for a small Python repo.",
) # ①
prompt = "List the files you'd look at first to find a failing test, and explain why."
for chunk in agent.stream(
{"messages": [{"role": "user", "content": prompt}]},
stream_mode=["updates", "messages"], # ②
version="v2",
):
if chunk["type"] == "updates": # ③
for node_name in chunk["data"]:
print(f"\n[step] {node_name}")
elif chunk["type"] == "messages": # ④
token, metadata = chunk["data"]
if metadata.get("lc_source") == "summarization": # ⑤
continue
if token.content:
print(token.content, end="", flush=True) # ⑥
Код короткий, но у каждой его части есть своя функция. Структура остаётся прежней, поскольку функция create_deep_agent создаёт агента точно так же, как и раньше. Запрос двух режимов в одном вызове позволяет одной петле одновременно обрабатывать индикатор шага и поток текста. Когда данные представляют собой событие updates, петля выводит строку [step] для узла, который только что завершил работу. Если это событие messages, петля выводит сгенерированный текст. Проверка на краткость из предыдущего раздела выполняется внутри этой ветки, поэтому вспомогательные процедуры никогда не достигают терминала, а пустые токены пропускаются перед выводом.
В результате каждый раз, когда агент делает шаг вперед, формируется строка вида [step], которая чередуется с текстом, поступающим по одному символу за раз; при этом уже применяется фильтр, чтобы ничего не утекало. Внешне это не очень красиво, но результат честный. Пользователь может видеть, что агент активен, и наблюдать за его мышлением.
В новых версиях Deep Agents предоставляется более продвинутый API для этой цели — agent.stream_events(...) в режиме version="v3" — с типизированными проекциями вроде stream.subagents. В документации рекомендуется использовать именно этот подход, поскольку каждая проекция представляет собой отдельный упорядоченный итератор, и нет необходимости самостоятельно обрабатывать параметр stream_mode или декодировать пространства имен. Тем не менее стоит сначала изучить способ работы с stream_mode, так как проекции основаны на тех же событиях. Перед тем как полностью перейти на новый интерфейс, ознакомьтесь с актуальной документацией.
Чек-лист для применения сегодня
- Начните с поиска всех элементов
invoke, в которых есть ожидающий человек, и замените их наagent.stream(...)в режимеupdates. Даже примитивный журнал шагов лучше, чем пустой экран. - Рассмотрите один реальный фрагмент данных. Запустите процесс с параметром
version="v2", выведите результат и прочитайте поляtype,nsиdata, чтобы понять их структуру. - Прежде чем реальные пользователи увидят результат, удаляйте все фрагменты
messages, метаданные которых указывают наlc_source == "summarization". - Если агент передает задачи подагентам, установите параметр
subgraphs=Trueи отсортируйте результат по пространству имён, причем каждый агент будет иметь свой отдельный результат. - Сохраняйте в репозитории простую консоль, подобную приведенной выше. Она позволяет наблюдать за выполнением каждой новой функции вместо того, чтобы догадываться на основе предыдущего сообщения.
Что изменилось, а что нет
Итог: блокирующий вызов превратился в прямую трансляцию. Параметр stream_mode определяет уровень детализации информации, ns указывает, кто генерировал каждое событие, а фильтр обобщения удаляет лишний шум от самого инструмента; теперь должно быть ясно, что наблюдение за процессом выполнения и запоминание его результатов — это разные задачи.
Ничего не изменилось ни в самом агенте, ни в его цикле работы. Единственное добавление — это окно, которое разделяет инструменты, которым люди доверяют, и те, которые они в спешке обновляют. Когда агент показывает результаты своей работы, пользователи считают, что всё работает корректно.
Поэтому в следующий раз, когда кто-то сообщит, что агент «завис», пропустите профилирование и сначала попробуйте использовать прямую трансляцию. Агент всё это время «разговаривал»; ему был нужен канал связи.
Связанные материалы
- Файловая система вашего агента — это решение: объяснение бэкендов Deep Agents — Как бэкенды LangChain Deep Agents определяют, что сохраняется: пять встроенных решений, механизм маршрутизации CompositeBackend, проблема пространства имён при многопользовательских хранилищах и шаблоны сандбоксов.
- Бета-версия Managed Deep Agents: способ развертывания агентов LangChain на основе каталогов — Функция Managed Deep Agents от LangChain превращает папку проекта в хостингового агента. Что делает каждый файл, как работает механизм развертывания mda и какие недостатки бета-версии стоит знать.