Цепочка, DAG и автомат: чем описывают порядок шагов

В поддержку интернет-магазина приходит письмо: «Заказ 4412 не пришёл, хочу вернуть деньги». Что должно произойти дальше, понятно и без всякой архитектуры — разобрать обращение, поднять данные по заказу, свериться с правилами возврата, написать ответ. Непонятно другое: чем этот порядок шагов записать в коде. И ответ зависит не от количества шагов, а от того, есть ли в процессе возвраты и ожидания.

Различают три формы. Они не взаимозаменяемы, и выбор между ними делается один раз — до того, как написана первая строка.

Цепочка

Шаги известны заранее и идут подряд: разобрали обращение, написали ответ, отправили. Никаких развилок, ничего не повторяется. Такое пишется обычными вызовами функций одна за другой, и заворачивать это в граф — значит добавить слой, который ничего не решает. Мы уже отмечали это, когда собирали первый граф: если в схеме нет ни одной развилки и ни одного возврата, граф не окупается.

DAG — ориентированный ациклический граф

DAG (directed acyclic graph) описывает не порядок, а зависимости: какой шаг чьего результата ждёт. Данные по заказу, история обращений клиента и правила возврата друг от друга не зависят — значит, между ними связи нет, и порядок между ними не задан. Ответ зависит от всех трёх, и это единственное, что записано в графе.

Ацикличность при этом — не свойство, которое кто-то проверит за автора. Провести ребро от последнего узла обратно к первому технически ничто не мешает: LangGraph циклы разрешает, ради них он и сделан. Но граф с циклом — это уже не DAG, и вместе с буквой A из названия пропадает всё, ради чего DAG берут.

А берут его ради двух вещей. Первая: независимые шаги можно запустить одновременно, потому что ни один из них не ждёт другого. Вторая: процесс гарантированно закончится — вернуться назад в графе без циклов некуда, и бесконечно крутиться он не может. Обе держатся ровно на отсутствии рёбер назад: появилось хоть одно — гарантии завершения больше нет, и лимит числа шагов приходится ставить самому.

Автомат

Автомат (конечный автомат, в англоязычных источниках FSM — finite state machine) описывает третью вещь — текущее состояние и переходы из него. В любой момент процесс находится в одном состоянии; событие переводит его в следующее, и это следующее вполне может быть тем, где мы уже были.

Именно возврат назад и делает автомат нужным. В обращении не указан номер заказа — задаём уточняющий вопрос, ждём ответа клиента и возвращаемся к разбору. Ответ не прошёл проверку — отправляем на доработку и снова проверяем. Ни то, ни другое DAG выразить не может: там нет ни рёбер назад, ни состояния «стоим и ждём».

Плата за это тоже понятна. Раз активное состояние одно, само собой ничего не распараллелится. И раз возвраты разрешены, гарантии завершения больше нет — лимит числа шагов или раундов приходится ставить самому.

Чем они отличаются на самом деле

Разница не в том, что «DAG для сложного, а автомат для простого», и не в форме картинки. Разница в вопросе, на который отвечает описание. DAG отвечает на вопрос «что от чего зависит», и время в нём вторично: порядок вычисляется из зависимостей. Автомат отвечает на вопрос «что делать дальше, если сейчас мы здесь», и зависимости в нём вторичны: важен текущий момент.

	Цепочка	DAG	Автомат
Что записано	порядок шагов	зависимости между шагами	состояния и переходы
Возврат к пройденному шагу	нет	нет	есть
Параллельные ветви	нет	есть	по одной ветви за раз
Остановка в ожидании события	нет	нет	есть
Завершение	гарантировано	гарантировано	нужен свой лимит

Как выбрать, не гадая

Выбор делается по описанию процесса, а не по ощущению сложности. Работает простой приём: выписать процесс словами, как его формулирует заказчик или как он записан в регламенте, и поискать в тексте три группы слов.

Слова возврата и ожидания — «пока не», «повторить», «вернуть на доработку», «дождаться ответа», «если не подтвердил — спросить ещё раз». Любое из них означает ребро назад или паузу, и дальше выбора нет: нужен автомат.

Слова одновременности — «как только готовы и то, и другое», «параллельно», «собрать из нескольких источников». Они означают независимые ветви и точку сбора, то есть DAG.

Если ни того, ни другого в описании нет, а шаги просто перечислены через запятую — это цепочка, и городить граф не нужно.

Одно не отменяет другого. Реальный процесс поддержки — это автомат, внутри которого один из этапов собирает контекст параллельно. Такие сочетания LangGraph описывает штатно, и разбираться каждый уровень будет по своим правилам.

DAG: независимые ветви и точка сбора

Чтобы ответить клиенту по заказу 4412, нужны три вещи: статус самого заказа, история обращений этого клиента и правила возврата для категории товара. Три запроса в три разные системы, и ни один из них не ждёт результата другого. Если выполнять их подряд, ответ будет ждать суммы трёх задержек — при 0,6, 0,3 и 0,4 секунды это 1,3 секунды на пустом месте.

Записываем зависимости как есть: три узла ни от чего не зависят, четвёртый зависит от всех трёх.

import operator
import time
from typing import Annotated

from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

STARTED = time.perf_counter()


def took(text: str, started: float) -> str:
    return f"{text} — {time.perf_counter() - started:.1f}s"


class State(TypedDict):
    ticket: str
    order: str
    history: str
    policy: str
    log: Annotated[list[str], operator.add]
    context: str


def fetch_order(state: State) -> dict:
    started = time.perf_counter()
    time.sleep(0.6)
    print(took("order:   запрос в базу заказов", started))
    return {"order": "4412: отправлен 14.07, трек не обновлялся 6 дней",
            "log": ["order"]}


def fetch_history(state: State) -> dict:
    started = time.perf_counter()
    time.sleep(0.3)
    print(took("history: прошлые обращения клиента", started))
    return {"history": "2 обращения за год, оба закрыты без возврата",
            "log": ["history"]}


def fetch_policy(state: State) -> dict:
    started = time.perf_counter()
    time.sleep(0.4)
    print(took("policy:  правила возврата по категории", started))
    return {"policy": "электроника: возврат в течение 14 дней с доставки",
            "log": ["policy"]}


def collect(state: State) -> dict:
    started = time.perf_counter()
    context = "\n".join([
        f"Обращение: {state['ticket']}",
        f"Заказ: {state['order']}",
        f"Клиент: {state['history']}",
        f"Правила: {state['policy']}",
    ])
    print(took(f"collect: собрались ветви {state['log']}", started))
    return {"context": context}


builder = StateGraph(State)
builder.add_node("fetch_order", fetch_order)
builder.add_node("fetch_history", fetch_history)
builder.add_node("fetch_policy", fetch_policy)
builder.add_node("collect", collect)

builder.add_edge(START, "fetch_order")
builder.add_edge(START, "fetch_history")
builder.add_edge(START, "fetch_policy")
builder.add_edge(["fetch_order", "fetch_history", "fetch_policy"], "collect")
builder.add_edge("collect", END)

graph = builder.compile()

result = graph.invoke({
    "ticket": "Заказ 4412 не пришёл, хочу вернуть деньги",
    "order": "", "history": "", "policy": "", "log": [], "context": "",
})

print(took("весь прогон", STARTED))
print()
print(result["context"])

Три источника здесь заглушки со sleep — на их месте будут запрос в базу, обращение к сервису доставки и чтение справочника. Задержки настоящие, каждая ветвь печатает, сколько заняла, и последней строкой — сколько занял весь прогон:

Ветви заняли 0,3, 0,4 и 0,6 секунды, а весь прогон — те же 0,6: по самой долгой ветви, а не по сумме в 1,3. Больше ничего для этого делать не пришлось: ни потоков, ни asyncio, ни пула. Параллельность здесь не отдельная функция, а следствие того, что зависимости записаны честно.

Fan-out и fan-in

Расхождение на ветви — fan-out — это просто несколько рёбер из одной точки. Три строки add_edge(START, ...) и означают: у этих узлов нет входящих зависимостей, запускайте сразу.

Схождение — fan-in — записывается списком в первом аргументе:

builder.add_edge(["fetch_order", "fetch_history", "fetch_policy"], "collect")

Список означает «дождаться всех трёх и запустить collect один раз». Соблазн написать то же самое тремя строками велик, но смысл у них другой:

builder.add_edge("fetch_order", "collect")
builder.add_edge("fetch_history", "collect")
builder.add_edge("fetch_policy", "collect")

Это три независимых перехода: каждый из них сам по себе говорит «после меня иди в collect». Требования дождаться остальных здесь нет.

Что узел возвращает

Узел возвращает не всё состояние, а только изменённые поля — этим приёмом мы уже пользовались, когда собирали первый граф. В параллельных ветвях он перестаёт быть вопросом стиля. Пока fetch_order трогает только order, а fetch_policy только policy, пересечься им негде. Стоит вернуть из ветви всё состояние целиком — и каждая ветвь начнёт записывать поля, которых она не меняла, поверх работы соседа.

Отдельный случай — когда несколько ветвей пишут в один ключ намеренно. В нашем состоянии такой ключ есть: log, куда каждая ветвь добавляет свою отметку. Без указаний LangGraph не знает, что делать с тремя значениями для одного поля, и отказывается угадывать:

langgraph.errors.InvalidUpdateError
At key 'log': Can receive only one value per step. Use an Annotated key to handle multiple values.

Указание даётся прямо в схеме состояния: Annotated[list[str], operator.add]. Второй элемент — функция, которой LangGraph сведёт значения от разных ветвей; для списков это конкатенация. Ошибка исчезает, а в log оказываются отметки всех трёх ветвей.

На порядок элементов в таком списке опираться нельзя. В кадре выше это ['history', 'order', 'policy'] — не порядок объявления рёбер и не порядок завершения ветвей, а порядок, в котором LangGraph свёл записи. Если порядок важен, его кладут в сами значения, а не выводят из позиции в списке.

Как это выглядит на схеме

Схему рисует сам граф, руками её перебирать не нужно:

graph.get_graph().draw_mermaid_png(output_file_path="dag.png")

Ромб расхождения и схождения виден сразу, и это главное, ради чего на схему смотрят. Одного она при этом не показывает: точка сбора нарисована теми же тремя стрелками, что и три независимых перехода. Разница между двумя записями выше на картинке не видна — её надо смотреть в коде сборки.

Здесь число ветвей известно заранее — их ровно три, и они прописаны в графе. Когда ветвей столько, сколько пришло данных на прогоне, рёбрами это не описать: в LangGraph для такого случая есть Send из langgraph.types, который порождает ветви динамически.

Автомат: возврат назад и остановка в ожидании

Предыдущий граф молча предполагал, что номер заказа известен. В жизни это не так: половина обращений выглядит как «Здравствуйте! Заказ до сих пор не пришёл». Запросить по нему статус не получится — сначала надо спросить номер, дождаться ответа и разобрать обращение заново.

Ни одного из этих трёх действий DAG не выражает. Спросить и вернуться — это ребро назад, которого там нет по определению. Дождаться ответа — это остановка, а в DAG узел либо выполняется, либо нет.

from typing import Literal

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

MODEL = ChatOpenAI(model="gpt-5.6-luna")

PROMPT = (
    "Из обращения в поддержку интернет-магазина выпиши номер заказа — только "
    "цифры, без слов. Если номера в тексте нет, ответь одним словом: нет.\n\n"
    "Обращение: {ticket}"
)


class State(TypedDict):
    stage: str
    ticket: str
    order_id: str
    answer: str


def triage(state: State) -> dict:
    """Ищет в обращении номер заказа и решает, что с ним делать дальше."""
    reply = str(MODEL.invoke(PROMPT.format(ticket=state["ticket"])).content).strip()
    order_id = reply if reply.isdigit() else ""
    print(f"triage:  номер заказа — {order_id or 'не найден'}")
    return {"order_id": order_id, "stage": "resolve" if order_id else "ask_client"}


def ask_client(state: State) -> dict:
    """Останавливает граф и ждёт ответа клиента."""
    print("ask:     спрашиваем номер заказа и останавливаемся")
    answer = interrupt("Подскажите номер заказа — он в письме о доставке.")
    print(f"ask:     клиент ответил: {answer!r}")
    return {"ticket": f"{state['ticket']} Номер заказа: {answer}", "stage": "triage"}


def resolve(state: State) -> dict:
    print(f"resolve: работаем по заказу {state['order_id']}")
    return {"answer": f"Проверяем заказ {state['order_id']}, ответим в течение дня.",
            "stage": "done"}


def route(state: State) -> Literal["ask_client", "resolve"]:
    return "ask_client" if state["stage"] == "ask_client" else "resolve"


builder = StateGraph(State)
builder.add_node("triage", triage)
builder.add_node("ask_client", ask_client)
builder.add_node("resolve", resolve)

builder.add_edge(START, "triage")
builder.add_conditional_edges("triage", route, ["ask_client", "resolve"])
builder.add_edge("ask_client", "triage")
builder.add_edge("resolve", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ticket-8817"}}

state = graph.invoke(
    {"stage": "triage", "ticket": "Здравствуйте! Заказ до сих пор не пришёл.",
     "order_id": "", "answer": ""},
    config,
)

print(f"пауза:   граф стоит в {graph.get_state(config).next}")
print()

state = graph.invoke(Command(resume="4412"), config)

print()
print("ответ клиенту:", state["answer"])

Состояние здесь — поле, а не позиция в графе

Ключевая строка — stage в схеме состояния. В DAG узнать, где мы находимся, можно было только по тому, какие узлы уже отработали. Здесь текущее состояние записано явно, узел его меняет, а функция route читает и по нему выбирает переход. Условное ребро add_conditional_edges("triage", route, ["ask_client", "resolve"]) — это и есть таблица переходов автомата, записанная кодом.

Возврат при этом самый обычный: add_edge("ask_client", "triage"). Ребро назад ничем не отличается от ребра вперёд — отличается только тем, что в DAG его нельзя было бы провести.

Пунктиром на схеме идут условные переходы, сплошной линией — безусловные. Стрелка ask_client → triage замыкает цикл, и это видно на картинке раньше, чем в коде.

Остановка и возобновление

Строка interrupt("Подскажите номер заказа...") прерывает прогон целиком. Первый invoke возвращает управление, не дойдя до resolve; graph.get_state(config).next показывает, что граф стоит перед ask_client. Между этим моментом и ответом клиента может пройти минута или трое суток, процесс за это время может перезапуститься — состояние лежит в чекпойнтере, том же механизме, которым мы связывали письма клиента в один разговор. Без checkpointer остановиться граф ещё сможет, а вот продолжить — уже нет: попытка возобновиться падает с RuntimeError: Cannot use Command(resume=...) without checkpointer, потому что место остановки нигде не сохранено.

Ответ приходит вторым вызовом: graph.invoke(Command(resume="4412"), config). Тот же thread_id — то же место остановки.

В кадре строка «спрашиваем номер заказа и останавливаемся» напечатана дважды, хотя в коде она одна. Возобновление продолжает не с самой строки interrupt, а с начала узла: всё, что стояло до неё, выполняется ещё раз. Печать это переживёт, а вот запись в базу или отправка письма продублируются. Поэтому до interrupt в узле не должно быть ничего, кроме подготовки вопроса.

Цикл нужно ограничивать самому

У DAG завершение бесплатное: вернуться некуда, значит, прогон закончится. Как только появилось ребро назад, эта гарантия исчезла. Клиент может прислать в ответ «не помню», разбор снова не найдёт номер, и граф пойдёт по кругу.

Лечится это счётчиком в состоянии — ровно так же, как счётчик попыток в цикле «написал код, тест упал, переписал». Поле rounds, увеличение при каждом проходе и проверка в route:

def route(state: State) -> Literal["ask_client", "resolve", "escalate"]:
    if state["order_id"]:
        return "resolve"
    if state["rounds"] >= 2:
        return "escalate"
    return "ask_client"

Третий переход escalate здесь не украшение. Тупик в автомате должен вести в состояние, где кто-то принимает решение, — иначе граф остановится по внутреннему лимиту, а клиент останется без ответа. У LangGraph есть и собственный предохранитель: recursion_limit обрывает прогон после заданного числа шагов графа. Он спасает от бесконечного цикла, но обрывает работу исключением, а не переводит процесс в осмысленное состояние.

Отдельная тема, которой этот шаг не касается, — внешние эффекты, случившиеся до обрыва. Если резерв товара уже поставлен, а платёж не прошёл, резерв надо снимать отдельной операцией: изменить поле в состоянии недостаточно, потому что во внешней системе изменение уже произошло. Это вопрос надёжности, а не оркестрации, и решается он не рёбрами графа.

Автомат снаружи, DAG внутри

Выбирать между двумя предыдущими графами не нужно — настоящий процесс поддержки состоит из обоих. Снаружи это автомат: разобрать обращение, при нехватке данных спросить клиента и вернуться к разбору. Внутри, на этапе сбора контекста, — три независимые ветви и точка сбора. Один уровень отвечает за решения и ожидание, другой — за зависимости между запросами.

В LangGraph это записывается прямо: собранный граф можно поставить узлом в другой граф.

import operator
import time
from typing import Annotated, Literal

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

TRIAGE_MODEL = ChatOpenAI(model="gpt-5.6-luna")
DRAFT_MODEL = ChatOpenAI(model="gpt-5.6-terra")

STARTED = time.perf_counter()


def mark(text: str) -> str:
    return f"[{time.perf_counter() - STARTED:4.1f}s] {text}"


class State(TypedDict):
    stage: str
    ticket: str
    order_id: str
    order: str
    history: str
    policy: str
    log: Annotated[list[str], operator.add]
    answer: str


def fetch_order(state: State) -> dict:
    print(mark("  order:   запрос в базу заказов"))
    time.sleep(0.6)
    return {"order": f"{state['order_id']}: отправлен 14.07, трек стоит 6 дней",
            "log": ["order"]}


def fetch_history(state: State) -> dict:
    print(mark("  history: прошлые обращения клиента"))
    time.sleep(0.3)
    return {"history": "2 обращения за год, оба закрыты без возврата", "log": ["history"]}


def fetch_policy(state: State) -> dict:
    print(mark("  policy:  правила возврата по категории"))
    time.sleep(0.4)
    return {"policy": "электроника: возврат в течение 14 дней с доставки", "log": ["policy"]}


gather = StateGraph(State)
gather.add_node("fetch_order", fetch_order)
gather.add_node("fetch_history", fetch_history)
gather.add_node("fetch_policy", fetch_policy)
gather.add_edge(START, "fetch_order")
gather.add_edge(START, "fetch_history")
gather.add_edge(START, "fetch_policy")
gather.add_edge(["fetch_order", "fetch_history", "fetch_policy"], END)


def triage(state: State) -> dict:
    prompt = ("Из обращения в поддержку интернет-магазина выпиши номер заказа — "
              "только цифры. Если номера нет, ответь одним словом: нет.\n\n"
              f"Обращение: {state['ticket']}")
    reply = str(TRIAGE_MODEL.invoke(prompt).content).strip()
    order_id = reply if reply.isdigit() else ""
    print(mark(f"triage:  номер заказа — {order_id or 'не найден'}"))
    return {"order_id": order_id, "stage": "gather" if order_id else "ask_client"}


def ask_client(state: State) -> dict:
    answer = interrupt("Подскажите номер заказа — он в письме о доставке.")
    print(mark(f"ask:     клиент ответил: {answer!r}"))
    return {"ticket": f"{state['ticket']} Номер заказа: {answer}", "stage": "triage"}


def draft(state: State) -> dict:
    print(mark(f"draft:   контекст собран из {state['log']}"))
    prompt = ("Ты оператор поддержки интернет-магазина. Ответь клиенту одним "
              "абзацем, вежливо и по делу, опираясь только на данные ниже. "
              "Ничего не переспрашивай: все нужные данные уже есть.\n\n"
              f"Обращение: {state['ticket']}\nЗаказ: {state['order']}\n"
              f"Клиент: {state['history']}\nПравила: {state['policy']}")
    return {"answer": str(DRAFT_MODEL.invoke(prompt).content).strip(), "stage": "done"}


def route(state: State) -> Literal["ask_client", "gather"]:
    return "ask_client" if state["stage"] == "ask_client" else "gather"


builder = StateGraph(State)
builder.add_node("triage", triage)
builder.add_node("ask_client", ask_client)
builder.add_node("gather", gather.compile())
builder.add_node("draft", draft)

builder.add_edge(START, "triage")
builder.add_conditional_edges("triage", route, ["ask_client", "gather"])
builder.add_edge("ask_client", "triage")
builder.add_edge("gather", "draft")
builder.add_edge("draft", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ticket-8817"}}

graph.invoke(
    {"stage": "triage", "ticket": "Здравствуйте! Заказ до сих пор не пришёл.",
     "order_id": "", "order": "", "history": "", "policy": "", "log": [], "answer": ""},
    config,
)
print(mark(f"пауза:   граф стоит в {graph.get_state(config).next}"))

state = graph.invoke(Command(resume="4412"), config)

print(mark("готово"))
print()
print(state["answer"])

Весь путь виден по отметкам времени. Разбор не нашёл номер, граф остановился и дождался ответа. После возобновления разбор прошёл заново — уже успешно, — и три ветви сбора стартовали в один и тот же момент. Последний абзац написала модель по собранному контексту.

Подграф — обычный узел

Строка builder.add_node("gather", gather.compile()) ставит узлом целый скомпилированный граф. Отдельного механизма для этого нет: у собранного графа тот же интерфейс, что у функции-узла, — принимает состояние, возвращает обновления.

Связывает уровни общая схема состояния. Внешний граф и gather объявлены на одном State, поэтому fetch_order внутри читает order_id, который положил triage снаружи, а draft снаружи читает policy, который положила ветвь внутри. Если бы у подграфа была своя схема, поля пришлось бы перекладывать вручную на входе и выходе.

Reducer при этом работает по всей вложенности: log объявлен один раз в общем состоянии, и записи трёх внутренних ветвей сводятся в него так же, как если бы ветви лежали во внешнем графе.

Что видно на схеме

graph.get_graph(xray=True).draw_mermaid_png(output_file_path="both.png")

Без xray=True подграф нарисуется одним прямоугольником gather — снаружи он и есть один узел. С этим флагом рамка раскрывается, и обе конструкции видны разом: справа петля автомата triage → ask_client → triage, внутри рамки — расхождение на три ветви и сбор.

Где проходит граница

Вопрос при проектировании звучит не «DAG или автомат», а «на каком уровне что». Правило простое: наверх идёт то, что решает, вниз — то, что вычисляет.

Наверху остаются переходы, зависящие от результата: пойти к клиенту или к данным, принять ответ или отправить на доработку, остановиться и подождать. Там же считаются раунды и стоит лимит. Вниз уходит всё, у чего порядок определяется одними зависимостями: сбор данных из нескольких источников, независимые проверки, обогащение контекста.

Практический признак, что граница проведена неправильно, — условное ребро внутри подграфа сбора. Как только там появляется развилка «а если этого источника нет, спросим ещё раз», сбор перестал быть DAG и превратился во второй автомат, спрятанный от внешнего лимита раундов. Такой кусок либо поднимают наверх, либо он получает собственный счётчик.