Стриминг: поток вместо ожидания | Курс LangChain урок 7

"""Общая сборка модели для примеров урока 7.

Тот же модуль, что course_model.py уроков 4 и 5, без изменений: у build_model есть
необязательный первый аргумент с именем модели, а рядом лежит gateway_kwargs()
для примеров, которые собирают модель сами.

Файл .env берётся тот же, что в уроке 0. Положите его рядом с этой папкой или
выше по дереву: load_dotenv() ищет файл начиная с папки этого модуля и поднимается
вверх.
"""

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()


def gateway_kwargs():
    """Возвращает аргументы доступа к провайдеру: имя, адрес, ключ.

    Нужны примерам, которые вызывают init_chat_model сами. У настраиваемой модели
    имени модели при создании нет, поэтому build_model ей не подходит.
    """
    base_url = os.getenv("MODEL_BASE_URL")

    if base_url:
        return {
            "model_provider": "openai",
            "base_url": base_url,
            "api_key": os.environ["OPENAI_API_KEY"],
        }

    return {}


def build_model(model_name=None, **kwargs):
    """Собирает модель курса.

    model_name без значения означает модель из переменной MODEL_NAME. Явное имя
    нужно примерам, где моделей в приложении больше одной.

    Все именованные аргументы уходят в init_chat_model как есть: temperature,
    max_tokens, timeout, max_retries, rate_limiter, profile и прочее из раздела
    Parameters.
    """
    model_name = model_name or os.environ["MODEL_NAME"]
    access = gateway_kwargs()

    if access:
        # Путь для любого адреса, совместимого с OpenAI Chat Completions API.
        return init_chat_model(model=model_name, **access, **kwargs)

    # Путь напрямую к провайдеру.
    return init_chat_model(model_name, **kwargs)

"""Пример 1 урока 7: поток у самой модели и куда девается расход токенов.

Метод stream() возвращает итератор chunks. Chunks складываются оператором + в одно
сообщение, и только у собранного сообщения есть цельный текст. Первые chunks
приходят без текста: модель курса сначала рассуждает, а ChatOpenAI текст
рассуждения не извлекает. Расход токенов в потоке приходит не всегда: у Chat
Completions его надо попросить отдельно.
"""

from course_model import build_model

QUESTION = "Назовите три причины использовать очередь задач, по одному предложению на причину."


def run(title: str, **extra) -> None:
    """Прогоняет один поток и печатает первые chunks с текстом, счётчики и сборку."""
    print(title)
    model = build_model(temperature=0, max_tokens=1024, **extra)

    full = None
    count = 0
    empty = 0
    shown = 0
    with_usage = 0

    for chunk in model.stream(QUESTION):
        count += 1
        if not chunk.text:
            empty += 1
        elif shown < 3:
            shown += 1
            print(f"  chunk {count}: {type(chunk).__name__}, text={chunk.text!r}")
        if chunk.usage_metadata:
            with_usage += 1
        full = chunk if full is None else full + chunk

    print(f"  всего chunks: {count}, из них без текста: {empty}")
    print(f"  chunks с расходом токенов: {with_usage}")

    if full is None:
        # Защитная ветка: провайдер закрыл поток, не прислав ни одного chunk.
        print("  провайдер не прислал ни одного chunk")
        return

    print(f"  тип собранного: {type(full).__name__}")
    print(f"  длина текста: {len(full.text)} знаков")
    print(f"  расход у собранного: {full.usage_metadata}")


run("ПОТОК ПО УМОЛЧАНИЮ")
print()
run("ПОТОК С stream_usage=True", stream_usage=True)

# Вывод:
# ПОТОК ПО УМОЛЧАНИЮ
#   chunk 60: AIMessageChunk, text='В'
#   chunk 61: AIMessageChunk, text='от три причины использовать очередь задач'
#   chunk 62: AIMessageChunk, text=':1. '
#   всего chunks: 104, из них без текста: 63
#   chunks с расходом токенов: 1
#   тип собранного: AIMessageChunk
#   длина текста: 684 знаков
#   расход у собранного: {'input_tokens': 23, 'output_tokens': 461, 'total_tokens': 484, 'input_token_details': {}, 'output_token_details': {}}
#
# ПОТОК С stream_usage=True
#   chunk 51: AIMessageChunk, text='1. **Раз'
#   chunk 52: AIMessageChunk, text='деление ответственности**:'
#   chunk 53: AIMessageChunk, text=' Очередь задач'
#   всего chunks: 90, из них без текста: 54
#   chunks с расходом токенов: 1
#   тип собранного: AIMessageChunk
#   длина текста: 617 знаков
#   расход у собранного: {'input_tokens': 23, 'output_tokens': 398, 'total_tokens': 421, 'input_token_details': {}, 'output_token_details': {}}

"""Пример 2 урока 7: прогресс агента, режим потока updates.

Режим updates отдаёт по событию после каждого шага агента: что вернул узел model
и что вернул узел tools. Токенов в этом режиме нет, видно только, из каких
шагов состоит прогон.
"""

from langchain.agents import create_agent

from course_model import build_model


def get_weather(city: str) -> str:
    """Возвращает погоду в городе."""
    return f"В городе {city} всегда солнечно!"


def describe(message) -> str:
    """Одна строка на сообщение: тип, вызовы инструментов, начало текста."""
    calls = getattr(message, "tool_calls", None)
    if calls:
        names = [call["name"] for call in calls]
        return f"{type(message).__name__} вызывает {names}"

    text = message.text.replace("
", " ")
    if len(text) > 60:
        text = text[:57] + "..."
    return f"{type(message).__name__} {text!r}"


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[get_weather],
    system_prompt="Отвечайте по-русски, одним предложением.",
)

step = 0

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Какая погода в Сан-Франциско?"}]},
    stream_mode="updates",
    version="v2",
):
    if chunk["type"] != "updates":
        continue

    for node, update in chunk["data"].items():
        step += 1
        # У служебных ключей вроде __interrupt__ правка приходит не словарём.
        messages = update.get("messages", []) if isinstance(update, dict) else []
        if not messages:
            print(f"шаг {step}: узел {node}, правка без сообщений: {update!r}")
            continue
        for message in messages:
            print(f"шаг {step}: узел {node}, {describe(message)}")

print(f"всего шагов: {step}")

# Вывод:
# шаг 1: узел model, AIMessage вызывает ['get_weather']
# шаг 2: узел tools, ToolMessage 'В городе Сан-Франциско всегда солнечно!'
# шаг 3: узел model, AIMessage 'В Сан-Франциско всегда солнечно!'
# всего шагов: 3

"""Пример 3 урока 7: две формы chunk, v1 по умолчанию и v2 по требованию.

Один и тот же агент, один и тот же набор режимов, разная форма chunk. В v1 форма
зависит от того, сколько режимов вы запросили и нужны ли подграфы. В v2 форма
всегда одна: словарь с ключами type, ns и data.
"""

from langchain.agents import create_agent

from course_model import build_model


def get_weather(city: str) -> str:
    """Возвращает погоду в городе."""
    return f"В городе {city} всегда солнечно!"


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[get_weather],
    system_prompt="Отвечайте по-русски, одним предложением.",
)

PAYLOAD = {"messages": [{"role": "user", "content": "Какая погода в Сан-Франциско?"}]}
MODES = ["updates", "messages"]

print("ФОРМА v1, ПО УМОЛЧАНИЮ")
count = 0
for chunk in agent.stream(PAYLOAD, stream_mode=MODES):
    count += 1
    if count <= 3:
        head = chunk[0] if isinstance(chunk, tuple) else None
        print(f"  chunk {count}: {type(chunk).__name__}, первый элемент {head!r}")
print(f"  всего chunks: {count}")

print()
print("ФОРМА v2")
count = 0
for chunk in agent.stream(PAYLOAD, stream_mode=MODES, version="v2"):
    count += 1
    if count <= 3:
        print(
            f"  chunk {count}: {type(chunk).__name__}, ключи {sorted(chunk)}, "
            f"type={chunk['type']!r}, ns={chunk['ns']!r}"
        )
print(f"  всего chunks: {count}")

# Вывод:
# ФОРМА v1, ПО УМОЛЧАНИЮ
#   chunk 1: tuple, первый элемент 'messages'
#   chunk 2: tuple, первый элемент 'messages'
#   chunk 3: tuple, первый элемент 'messages'
#   всего chunks: 58
#
# ФОРМА v2
#   chunk 1: dict, ключи ['data', 'ns', 'type'], type='messages', ns=()
#   chunk 2: dict, ключи ['data', 'ns', 'type'], type='messages', ns=()
#   chunk 3: dict, ключи ['data', 'ns', 'type'], type='messages', ns=()
#   всего chunks: 55

"""Пример 4 урока 7: режим messages, токены модели и метаданные chunk.

Режим messages отдаёт пары (chunk сообщения, метаданные). Метаданные говорят, из
какого узла графа пришёл chunk. Аргументы инструмента приходят обрывками текста и
собираются в JSON только к концу вызова.
"""

from collections import Counter

from langchain.agents import create_agent
from langchain.messages import AIMessageChunk

from course_model import build_model


def get_weather(city: str) -> str:
    """Возвращает погоду в городе."""
    return f"В городе {city} всегда солнечно!"


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[get_weather],
    system_prompt="Отвечайте по-русски, одним предложением.",
)

by_node = Counter()
by_type = Counter()
arg_pieces = []
metadata_keys = []
shown = 0

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Какая погода в Сан-Франциско?"}]},
    stream_mode="messages",
    version="v2",
):
    if chunk["type"] != "messages":
        continue

    token, metadata = chunk["data"]
    node = metadata.get("langgraph_node", "?")
    by_node[node] += 1
    by_type[type(token).__name__] += 1
    if not metadata_keys:
        metadata_keys = sorted(metadata)

    blocks = [block["type"] for block in token.content_blocks]

    if isinstance(token, AIMessageChunk):
        for piece in token.tool_call_chunks:
            arg_pieces.append(piece["args"])

    # Пустые chunks рассуждения пропускаются, см. пример 1.
    if blocks and shown < 6:
        shown += 1
        print(f"  {shown}. узел {node}, {type(token).__name__}, блоки {blocks}, text={token.text!r}")

print()
print("CHUNKS ПО УЗЛАМ:", dict(by_node))
print("ТИПЫ СООБЩЕНИЙ В ПОТОКЕ:", dict(by_type))
print("КЛЮЧИ МЕТАДАННЫХ:", metadata_keys)

if arg_pieces:
    print("АРГУМЕНТЫ ИНСТРУМЕНТА ПО CHUNKS:", arg_pieces)
    print("ОНИ ЖЕ, СКЛЕЕННЫЕ:", "".join(piece or "" for piece in arg_pieces))
else:
    # Защитная ветка: модель обошлась без инструмента или отдала вызов целиком.
    print("АРГУМЕНТЫ ИНСТРУМЕНТА ПО CHUNKS: chunks не было")

# Вывод:
#   1. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#   2. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#   3. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#   4. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#   5. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#   6. узел model, AIMessageChunk, блоки ['tool_call_chunk'], text=''
#
# CHUNKS ПО УЗЛАМ: {'model': 49, 'tools': 1}
# ТИПЫ СООБЩЕНИЙ В ПОТОКЕ: {'AIMessageChunk': 49, 'ToolMessage': 1}
# КЛЮЧИ МЕТАДАННЫХ: ['checkpoint_ns', 'langgraph_checkpoint_ns', 'langgraph_node', 'langgraph_path', 'langgraph_step', 'langgraph_triggers', 'lc_versions', 'ls_integration', 'ls_max_tokens', 'ls_model_name', 'ls_model_type', 'ls_provider', 'ls_temperature']
# АРГУМЕНТЫ ИНСТРУМЕНТА ПО CHUNKS: [None, '{', '"city": "Сан', '-Ф', 'ранци', 'ско"}']
# ОНИ ЖЕ, СКЛЕЕННЫЕ: {"city": "Сан-Франциско"}

"""Пример 5 урока 7: свои сообщения о прогрессе из инструмента.

Пока работает инструмент, модель ничего не присылает, и для пользователя это
выглядит как зависание. Режим custom даёт инструменту собственный канал в поток: что угодно,
что вы передали в writer, приходит в цикл потока.
"""

from langchain.agents import create_agent
from langgraph.config import get_stream_writer

from course_model import build_model

PAGES = 3


def load_orders(day: str) -> str:
    """Загружает заказы за указанный день и возвращает сводку."""
    writer = get_stream_writer()

    loaded = 0
    for page in range(1, PAGES + 1):
        loaded += 40
        writer({"stage": "loading", "page": page, "of": PAGES, "rows": loaded})

    writer({"stage": "done", "rows": loaded})
    return f"за {day} заказов {loaded}, из них оплачено 97"


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[load_orders],
    system_prompt="Отвечайте по-русски, одним предложением.",
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Сколько заказов за вчера?"}]},
    stream_mode=["updates", "custom"],
    version="v2",
):
    if chunk["type"] == "custom":
        event = chunk["data"]
        if event["stage"] == "loading":
            print(f"[прогресс] страница {event['page']} из {event['of']}, строк {event['rows']}")
        else:
            print(f"[прогресс] загрузка закончена, строк {event['rows']}")

    elif chunk["type"] == "updates":
        for node, update in chunk["data"].items():
            messages = update.get("messages", []) if isinstance(update, dict) else []
            for message in messages:
                calls = getattr(message, "tool_calls", None)
                if calls:
                    print(f"[шаг] {node}: вызов {[call['name'] for call in calls]}")
                else:
                    text = message.text.replace("
", " ")
                    print(f"[шаг] {node}: {text[:70]!r}")

# Вывод:
# [шаг] model: вызов ['load_orders']
# [прогресс] страница 1 из 3, строк 40
# [прогресс] страница 2 из 3, строк 80
# [прогресс] страница 3 из 3, строк 120
# [прогресс] загрузка закончена, строк 120
# [шаг] tools: 'за вчера заказов 120, из них оплачено 97'
# [шаг] model: 'За вчера было 120 заказов.'

"""Пример 6 урока 7: chunk это не сообщение.

Разобрать вызов инструмента по одному chunk нельзя: в chunk лежит обрывок JSON.
Целое собирается двумя способами, и оба показаны здесь: сложением chunks в цикле
и чтением готового сообщения из режима updates.
"""

from langchain.agents import create_agent
from langchain.messages import AIMessage, AIMessageChunk, ToolMessage

from course_model import build_model


def get_weather(city: str) -> str:
    """Возвращает погоду в городе."""
    return f"В городе {city} всегда солнечно!"


def show_assembled(message: AIMessageChunk, reason: str) -> None:
    """Печатает то, что видно только у собранного сообщения."""
    text = message.text.replace("
", " ")
    print(f"СОБРАННОЕ СООБЩЕНИЕ ({reason})")
    print(f"  вызовы инструментов: {[call['name'] for call in message.tool_calls]}")
    print(f"  аргументы: {[call['args'] for call in message.tool_calls]}")
    print(f"  текст: {text[:70]!r}")
    print(f"  расход: {message.usage_metadata}")


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[get_weather],
    system_prompt="Отвечайте по-русски, одним предложением.",
)

full = None
partial_calls = 0

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Какая погода в Сан-Франциско?"}]},
    stream_mode=["messages", "updates"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, _metadata = chunk["data"]
        if not isinstance(token, AIMessageChunk):
            continue

        if token.tool_call_chunks:
            partial_calls += 1
            piece = token.tool_call_chunks[0]
            print(f"chunk вызова: name={piece['name']!r}, args={piece['args']!r}")

        full = token if full is None else full + token

        if token.chunk_position == "last":
            show_assembled(full, "по признаку chunk_position")
            full = None

    elif chunk["type"] == "updates":
        for node, update in chunk["data"].items():
            messages = update.get("messages", []) if isinstance(update, dict) else []
            for message in messages:
                if isinstance(message, AIMessage) and message.tool_calls:
                    print(f"из режима updates, узел {node}: {message.tool_calls}")
                elif isinstance(message, ToolMessage):
                    print(f"из режима updates, узел {node}: ответ {message.text!r}")

if full is not None:
    # Защитная ветка: класс модели обошёл базовый класс LangChain,
    # и последний chunk пришёл без признака chunk_position.
    show_assembled(full, "признака конца не было")

print(f"chunks с обрывками вызова: {partial_calls}")

# Вывод:
# chunk вызова: name='get_weather', args=None
# chunk вызова: name=None, args='{'
# chunk вызова: name=None, args='"city": "С'
# chunk вызова: name=None, args='ан-'
# chunk вызова: name=None, args='Фран'
# chunk вызова: name=None, args='циско"}'
# СОБРАННОЕ СООБЩЕНИЕ (по признаку chunk_position)
#   вызовы инструментов: ['get_weather']
#   аргументы: [{'city': 'Сан-Франциско'}]
#   текст: ''
#   расход: {'input_tokens': 300, 'output_tokens': 82, 'total_tokens': 382, 'input_token_details': {}, 'output_token_details': {}}
# из режима updates, узел model: [{'name': 'get_weather', 'args': {'city': 'Сан-Франциско'}, 'id': 'call_ea222d072a154d1885f9b0d7', 'type': 'tool_call'}]
# из режима updates, узел tools: ответ 'В городе Сан-Франциско всегда солнечно!'
# СОБРАННОЕ СООБЩЕНИЕ (по признаку chunk_position)
#   вызовы инструментов: []
#   аргументы: []
#   текст: 'В Сан-Франциско всегда солнечно!'
#   расход: {'input_tokens': 375, 'output_tokens': 35, 'total_tokens': 410, 'input_token_details': {}, 'output_token_details': {}}
# chunks с обрывками вызова: 6

"""Пример 7 урока 7: служебный вызов модели в потоке не нужен.

Внутри инструмента работает вторая модель, служебная. Её ответ пользователю не
нужен, а в поток режима messages она попадает наравне с основной. Метка nostream
убирает её из потока, не отменяя самого вызова.
"""

from collections import Counter

from langchain.agents import create_agent

from course_model import build_model

INTERNAL = {"model": None}


def classify(text: str) -> str:
    """Определяет тему обращения одним словом."""
    answer = INTERNAL["model"].invoke(
        [{"role": "user", "content": f"Одним словом назовите тему обращения: {text}"}]
    )
    return answer.text.strip()


def run(title: str, tagged: bool) -> None:
    """Прогоняет агента и считает chunks потока по узлам графа."""
    print(title)

    service = build_model(temperature=0, max_tokens=1024)
    INTERNAL["model"] = service.with_config({"tags": ["nostream"]}) if tagged else service

    agent = create_agent(
        model=build_model(temperature=0, max_tokens=1024),
        tools=[classify],
        system_prompt="Отвечайте по-русски, одним предложением. Тему определяйте инструментом.",
    )

    by_node = Counter()
    answer = ""

    for chunk in agent.stream(
        {"messages": [{"role": "user", "content": "У меня дважды списали деньги за заказ 4412."}]},
        stream_mode="messages",
        version="v2",
    ):
        if chunk["type"] != "messages":
            continue
        token, metadata = chunk["data"]
        by_node[metadata.get("langgraph_node", "?")] += 1
        if metadata.get("langgraph_node") == "model":
            answer += token.text

    print(f"  chunks по узлам: {dict(by_node)}")
    print(f"  ответ пользователю: {answer.strip()[:70]!r}")


run("СЛУЖЕБНАЯ МОДЕЛЬ БЕЗ МЕТКИ", tagged=False)
print()
run("СЛУЖЕБНАЯ МОДЕЛЬ С МЕТКОЙ nostream", tagged=True)

# Вывод:
# СЛУЖЕБНАЯ МОДЕЛЬ БЕЗ МЕТКИ
#   chunks по узлам: {'model': 82, 'tools': 138}
#   ответ пользователю: 'Ваш вопрос относится к переплате — ошибка двойного списания за заказ 4'
#
# СЛУЖЕБНАЯ МОДЕЛЬ С МЕТКОЙ nostream
#   chunks по узлам: {'model': 84, 'tools': 1}
#   ответ пользователю: 'По вашему заказу 4412 произошло двойное списание, мы проверим и вернем'

"""Пример 8 урока 7: что из потока показывают пользователю.

Три режима сразу: шаги агента, токены модели и прогресс инструмента. Пользователь
видит текст ответа по мере готовности, строку статуса на вызове инструмента и
прогресс загрузки. Обрывки JSON с аргументами в окно не попадают.

Заодно печатается то, ради чего поток и заводят: сколько секунд прошло до первого
показанного знака и сколько до конца ответа.
"""

import time

from langchain.agents import create_agent
from langchain.messages import AIMessage, AIMessageChunk
from langgraph.config import get_stream_writer

from course_model import build_model

PAGES = 3


def load_orders(day: str) -> str:
    """Загружает заказы за указанный день и возвращает сводку."""
    writer = get_stream_writer()

    loaded = 0
    for page in range(1, PAGES + 1):
        loaded += 40
        writer({"stage": "loading", "page": page, "of": PAGES, "rows": loaded})

    return f"за {day} заказов {loaded}, из них оплачено 97, отменено 6"


agent = create_agent(
    model=build_model(temperature=0, max_tokens=1024),
    tools=[load_orders],
    system_prompt="Вы помощник аналитика. Отвечайте по-русски, до двух предложений.",
)

started = time.perf_counter()
first_text_at = None

try:
    for chunk in agent.stream(
        {"messages": [{"role": "user", "content": "Что со вчерашними заказами?"}]},
        stream_mode=["updates", "messages", "custom"],
        version="v2",
    ):
        if chunk["type"] == "messages":
            token, metadata = chunk["data"]
            # В поток режима messages попадают и готовые ToolMessage, и обрывки
            # вызовов. Пользователю показывается только текст ответа модели.
            if not isinstance(token, AIMessageChunk) or not token.text:
                continue
            if metadata.get("langgraph_node") != "model":
                continue
            if first_text_at is None:
                first_text_at = time.perf_counter() - started
            print(token.text, end="", flush=True)

        elif chunk["type"] == "custom":
            event = chunk["data"]
            print(f"
[загрузка] страница {event['page']} из {event['of']}, строк {event['rows']}")

        elif chunk["type"] == "updates":
            for _node, update in chunk["data"].items():
                messages = update.get("messages", []) if isinstance(update, dict) else []
                for message in messages:
                    if isinstance(message, AIMessage) and message.tool_calls:
                        for call in message.tool_calls:
                            print(f"
[инструмент] {call['name']}({call['args']})")

except Exception as error:
    # Отказ провайдера посреди потока: часть ответа уже показана пользователю.
    print(f"
[сбой] {type(error).__name__}: {error}")

print()
if first_text_at is None:
    # Защитная ветка: текста в потоке не было вовсе, показывать было нечего.
    print("первый знак ответа: текста в потоке не было")
else:
    print(f"первый знак ответа через: {first_text_at:.2f} с")
print(f"весь ответ через: {time.perf_counter() - started:.2f} с")

# Вывод:
#
# [инструмент] load_orders({'day': 'вчера'})
#
# [загрузка] страница 1 из 3, строк 40
#
# [загрузка] страница 2 из 3, строк 80
#
# [загрузка] страница 3 из 3, строк 120
# Вчера было 120 заказов: 97 оплачено, 6 отменено.
# первый знак ответа через: 6.12 с
# весь ответ через: 6.26 с

# Неправильно: tool_calls у chunk собран из обрывка JSON
for chunk in agent.stream(payload, stream_mode="messages", version="v2"):
    token, metadata = chunk["data"]
    for call in token.tool_calls:      # {} у первых chunks, дальше пустой список
        run_side_effect(call["args"])  # сработает с неполными аргументами

# Правильно: собрать сообщение и разобрать собранное
full = None
for chunk in agent.stream(payload, stream_mode="messages", version="v2"):
    token, metadata = chunk["data"]
    if not isinstance(token, AIMessageChunk):
        continue
    full = token if full is None else full + token
    if token.chunk_position == "last":
        for call in full.tool_calls:
            run_side_effect(call["args"])
        full = None  # следующее сообщение модели собирается с нуля

# Было и работало: один режим, v1, приходят сырые данные
for chunk in agent.stream(payload, stream_mode="updates"):
    for node, update in chunk.items():
        ...

# Добавили второй режим, и тот же код падает:
# AttributeError: 'tuple' object has no attribute 'items'
for chunk in agent.stream(payload, stream_mode=["updates", "messages"]):
    for node, update in chunk.items():
        ...

# Правильно: форма v2 не зависит от числа режимов
for chunk in agent.stream(payload, stream_mode=["updates", "messages"], version="v2"):
    if chunk["type"] == "updates":
        for node, update in chunk["data"].items():
            ...

# Неправильно: тот же инструмент вызывают в тесте напрямую
def load_orders(day: str) -> str:
    writer = get_stream_writer()
    writer({"stage": "loading"})
    return "готово"

load_orders("вчера")  # RuntimeError: Called get_config outside of a runnable context

# Правильно: в тестах вызывать инструмент через агента,
# а получение данных вынести в функцию без писателя
def fetch_orders(day: str) -> str:
    return "готово"

def load_orders(day: str) -> str:
    writer = get_stream_writer()
    writer({"stage": "loading"})
    return fetch_orders(day)