Map-reduce паттерн:
параллельная обработка через Send

from langgraph.types import Send     # (в старых версиях: from langgraph.constants import Send)

def dispatch_to_summaries(state: OverallState):
    """Возвращает СПИСОК Send — по одной ветке на главу."""
    return [
        Send("summarize", {"chapter": chapter})   # узел "summarize" + его личное состояние
        for chapter in state["chapters"]
    ]

# Подключаем как условное ребро: из узла-источника, функция, список целей
builder.add_conditional_edges("split", dispatch_to_summaries, ["summarize"])

from typing import Annotated, TypedDict
import operator

class OverallState(TypedDict):
    chapters: list[str]
    summaries: Annotated[list[str], operator.add]   # ← сюда стекаются ветки
    digest: str                                     # итог reduce

# Состояние ОДНОЙ ветки — своя маленькая схема
class ChapterState(TypedDict):
    chapter: str

def summarize(state: ChapterState) -> dict:
    # обрабатываем один элемент; возвращаем список из одного результата
    return {"summaries": [f"конспект главы: {state['chapter'][:30]}..."]}

def reduce(state: OverallState) -> dict:
    # все ветки завершились — собираем итог
    return {"digest": f"Сводка из {len(state['summaries'])} глав"}

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

# --- Состояние ---
class OverallState(TypedDict):
    text: str
    chapters: list[str]
    summaries: Annotated[list[str], operator.add]
    digest: str

class ChapterState(TypedDict):
    chapter: str

# --- Узлы ---
def split(state: OverallState) -> dict:
    chapters = state["text"].split("\n\n")          # бьём текст на главы
    return {"chapters": chapters}

def summarize(state: ChapterState) -> dict:          # MAP-воркер
    return {"summaries": [f"конспект: {state['chapter'][:40]}"]}

def reduce(state: OverallState) -> dict:             # REDUCE
    return {"digest": " | ".join(state["summaries"])}

# --- Функция fan-out: по одному Send на главу ---
def to_summaries(state: OverallState):
    return [Send("summarize", {"chapter": c}) for c in state["chapters"]]

# --- Сборка ---
builder = StateGraph(OverallState)
builder.add_node("split", split)
builder.add_node("summarize", summarize)
builder.add_node("reduce", reduce)

builder.add_edge(START, "split")
builder.add_conditional_edges("split", to_summaries, ["summarize"])  # MAP
builder.add_edge("summarize", "reduce")              # fan-in
builder.add_edge("reduce", END)

graph = builder.compile()

result = graph.invoke({"text": "Глава 1...\n\nГлава 2...\n\nГлава 3...",
                       "chapters": [], "summaries": [], "digest": ""})
print(result["digest"])
# конспект: Глава 1... | конспект: Глава 2... | конспект: Глава 3...
# Все три главы обработаны параллельно, reduce собрал их в сводку

from langgraph.types import Send
from typing import Annotated, TypedDict
import operator

# 1. Поле сбора — ОБЯЗАТЕЛЬНО с reducer
class State(TypedDict):
    items: list
    results: Annotated[list, operator.add]   # сюда стекаются ветки map
    final: str

# 2. MAP-воркер: обрабатывает ОДИН элемент, возвращает список
def worker(s) -> dict:
    return {"results": [process(s["item"])]}

# 3. Fan-out: функция ребра возвращает СПИСОК Send
def fan_out(state):
    return [Send("worker", {"item": x}) for x in state["items"]]

builder.add_conditional_edges("split", fan_out, ["worker"])  # MAP
builder.add_edge("worker", "reduce")                         # fan-in
builder.add_edge("reduce", END)

# 4. Контроль параллелизма (тяжёлые ветки)
graph.invoke(inp, {"max_concurrency": 5})

# Правила:
#  • Send — из функции ребра, не из узла
#  • число Send = число параллельных веток (динамически)
#  • у каждого Send своё состояние (своя схема)
#  • поле сбора с operator.add; воркер возвращает СПИСОК