Parallelization:
одновременное выполнение узлов

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

class State(TypedDict):
    query: str
    sources: Annotated[list[str], operator.add]   # ← ОБЯЗАТЕЛЬНО reducer
    answer: str

def split(state):   return {}                      # точка ветвления
def search(state):  return {"sources": ["веб: ..."]}
def vector(state):  return {"sources": ["база: ..."]}
def api(state):     return {"sources": ["api: ..."]}
def aggregate(state):
    return {"answer": f"Собрано из {len(state['sources'])} источников"}

builder = StateGraph(State)
for name, fn in [("split", split), ("search", search),
                 ("vector", vector), ("api", api), ("aggregate", aggregate)]:
    builder.add_node(name, fn)

builder.add_edge(START, "split")
# fan-out: из split в три ветки — они пойдут в ОДНОМ супершаге
builder.add_edge("split", "search")
builder.add_edge("split", "vector")
builder.add_edge("split", "api")
# fan-in: все три ведут в aggregate — он ждёт всех
builder.add_edge("search", "aggregate")
builder.add_edge("vector", "aggregate")
builder.add_edge("api", "aggregate")
builder.add_edge("aggregate", END)

graph = builder.compile()
print(graph.invoke({"query": "LangGraph", "sources": [], "answer": ""}))
# search/vector/api отработали параллельно; aggregate собрал их результаты

from collections import Counter

class State(TypedDict):
    text: str
    votes: Annotated[list[str], operator.add]   # safe / unsafe от каждого судьи
    verdict: str

def judge_a(state): return {"votes": [classify(state["text"], temp=0.0)]}
def judge_b(state): return {"votes": [classify(state["text"], temp=0.5)]}
def judge_c(state): return {"votes": [classify(state["text"], temp=1.0)]}

def tally(state):                                  # сборщик: большинство голосов
    winner = Counter(state["votes"]).most_common(1)[0][0]
    return {"verdict": winner}

# START → (judge_a | judge_b | judge_c параллельно) → tally → END

# Модель: СУПЕРШАГ — независимые узлы исполняются одновременно

# Поле сбора — ОБЯЗАТЕЛЬНО reducer
class State(TypedDict):
    results: Annotated[list, operator.add]

# Статический fan-out / fan-in рёбрами:
builder.add_edge("split", "a")     # ┐
builder.add_edge("split", "b")     # ├ a,b,c — один супершаг (параллельно)
builder.add_edge("split", "c")     # ┘
builder.add_edge("a", "agg")       # ┐
builder.add_edge("b", "agg")       # ├ agg ждёт ВСЕХ (fan-in)
builder.add_edge("c", "agg")       # ┘

# Паттерны:
#  sectioning — разные подзадачи параллельно → свести
#  voting     — одна задача N раз → агрегировать (голосование/лучший)

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

# Правила:
#  • параллелятся ТОЛЬКО независимые узлы (нет рёбер между ними)
#  • поле fan-in — с reducer
#  • сборщик ждёт все ветки; порядок завершения не гарантирован
#  • динамическое число веток → Send (урок map-reduce)