Потокові глибокі агенти 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, проблема простору імен у багатокористувацьких сховищах та шаблони сандбоксів.
- Beta-версія Managed Deep Agents: спосіб розгортання агентів LangChain через каталоги — Функція Managed Deep Agents від LangChain перетворює каталог проекту на хостованого агента. Що робить кожен файл, як працює механізм розгортання mda та які недоліки бета-версії варто знати.