LangGraph у Python: стани, паралельні вузли, точки контролю та людський огляд

Кожен, хто створював щось більше, ніж просто іграшку, на основі LLM, знає про три постійні проблеми. Послідовні виклики призводять до зростання затримок. Довгий процес може зазнати збою та втратити весь досягнутий прогрес. А модель у найгірший можливий момент може надати впевнену, але абсолютно неправильну відповідь.

LangGraph вирішує всі ці проблеми. Це фреймворк оркестрації, заснований на графах, який складається з кількох простих елементів: стан представлений як TypedDict, вузли — це звичайні функції, а ребра — це логіка маршрутизації. Ці примітиви об’єднуються у процеси, які дозволяють виконувати вузли паралельно, зберігати стан для відновлення після збою та зупинятися, щоб людина могла переглянути результат у критичних ситуаціях.

Наступні розділи описують процес крок за кроком. Спочатку йде один окремий вузол, потім паралельна обробка, умовне спрямування та постійна пам’ять, далі — переривання з участю людини. У кінці наведений повний процес обробки електронних листів, який поєднує все це. Кожен приклад призначений для беззмінної роботи. SDK TypeScript надає майже такі самі API, тому там також застосовується той самий модель мислення.

Налаштування

Спочатку потрібні два пакети:

pip install langgraph langchain-openai

Потім змінні середовища для OpenAI та, за бажанням, відстеження діяльності LangSmith:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Приклади були протестовані з langgraph 1.1.3 (березень 2026 року) та langchain-openai 1.1.10, і вимагають Python 3.10 або новішої версії. Старі посібники іноді використовують LANGCHAIN_TRACING_V2 та LANGCHAIN_API_KEY; обидва все ще функціонують, але LANGSMITH_TRACING та LANGSMITH_API_KEY тепер є більш рекомендованими назвами.

Усе в LangGraph ґрунтується на двох основних ідеях: стан зберігає дані, а вузли — це функції, які їх перетворюють. Як тільки з цим стане зрозуміло, решта фреймворку автоматично впорядковується.

Визначення стану за допомогою TypedDict

Стан у LangGraph — це звичайний Python-об’єкт TypedDict, без спеціальних базових класів та декораторів, притаманних конкретному фреймворку, просто типові словники:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

Клас визначає форму даних, які проходять через граф, і кожен вузол читає з цієї спільної структури та записує в неї.

Чому саме TypedDict, а не звичайний словник чи dataclass? Він забезпечує статичну перевірку типів, оскільки його розуміють як mypy, так і Pyright, разом із підказками щодо типів, а також усуває додаткові витрати під час виконання, притаманні dataclass чи моделям Pydantic. LangGraph також може аналізувати анотації, щоб дізнатися схему, що має важливе значення для структурованого виведення та верифікації даних.

Рядок, який потрібно зрозуміти, — це Annotated[list[str], operator.add], який оголошує редуктор. Це найважливіша концепція у стані LangGraph, і майже кожен стикається з нею під час першого використання.

Без редуктора вузол, який повертає {"summary": "new value"}, просто замінює збережене значення. Для скалярних полів, таких як summary, це саме те, що потрібно. Проблеми виникають, коли два паралельні вузли записують дані у messages: зберігається лише той, хто завершить роботу останнім. Якщо замість цього оголосити operator.add, LangGraph буде додавати дані, тож кожен внесок у messages накопичується безпечно.

Більш складна логіка об’єднання вимагає використання власного редуктора:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

Цей редюсер об’єднує існуючий список із новими повідомленнями, а потім скорочує його до десяти найновіших записів, що допомагає у довгих розмовах, де кількість токенів інакше могла б значно зрости. operator.add використовується для простого додавання. Для видалення дублікатів, обмеження розміру списку чи будь-якого іншого об’єднання, крім простого додавання, потрібен власний редюсер.

Класичною помилкою є забування використання редюсера для поля, у яке записуються паралельні вузли. Без нього гілки мовчки перезаписують одна одну, перемагає та, що завершує роботу останньою, і жодної помилки не виникає.

TypedDict проти Pydantic

Обидва використовуються в LangGraph, і розділення між ними є навмисним. TypedDict призначений для стану графа, тобто контейнера, який переміщується по графу. LangGraph був розроблений навколо нього: його поля підтримують синтаксис редуктора Annotated, він чисто серіалізується для зберігання станів та не створює додаткового навантаження під час виконання.

Pydantic використовується для схем, які працюють з LLM. Виклик .with_structured_output(SomeModel) надає LLM JSON-схему, за якою потрібно формувати відповідь, та перевіряє, чи вона відповідає цій схемі. Відповідь, яка не відповідає, негайно викликає помилку та не дозволяє некоректним даним проходити через граф.

Основний робочий процес також ґрунтується саме на цьому розділенні: EmailState — це TypedDict, який виступає контейнером графа, а EmailClassification — це модель Pydantic, яка описує результат роботи LLM.

Створення першого вузла

Кожен вузол отримує повний стан як вхідні дані та повертає словник з частковими оновленнями. LangGraph об’єднує ці оновлення зі станом та застосовує всі оголошені функції зворотного обчислення:

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Розглянемо код:

• StateGraph(ChatState) створює граф, який використовує ChatState як свою схему, а LangGraph читає анотації, щоб дізнатися про редуктори та типи полів.
• graph.add_node("chatbot", chatbot) реєструє функцію під ім’ям "chatbot", до якого посилаються визначення ребер.
• graph.add_edge(START, "chatbot") робить "chatbot" першим вузлом, який виконується. START — це вбудований сенсор.
• graph.add_edge("chatbot", END) завершує виконання після того, як chatbot закінчить роботу. END — ще один вбудований сенсор.
• graph.compile() перевіряє структуру, шукаючи вільні ребра та недосяжні вузли, та повертає готовий до виконання CompiledGraph.

app.invoke({"messages": [...]}) виконує граф синхронно та повертає кінцевий стан.

Вузол обгортає свій рядок у список, оскільки messages використовує operator.add як редюсер. Повернення {"messages": [response.content]} додає список з одним елементом до існуючого, тоді як повернення простого рядка призведе до помилки під час виконання, оскільки operator.add не може додати рядок до списку.

Правило полягає у тому, щоб результат, який повертає вузол, відповідав типу поля, включаючи редюсер. Поле типу list[str] із operator.add приймає list[str], а звичайне поле типу str без редюсера приймає str.

Це повноцінний працюючий додаток LangGraph у приблизно 20 рядках. Наступне питання — що відбувається всередині invoke().

Суперкроки: як працює виконання

LangGraph виконує вузли у суперкроках. Кожен вузол, запланований до одного й того ж кроку, виконується одночасно, і всі вони повинні завершитися перед початком наступного кроку. У прикладі з одним вузлом існує один суперкрок, який містить цей вузол. Додавання паралельних ребер дозволяє розмістити кілька вузлів у одному суперкроці.

Це має значення з двох причин. Паралельне виконання є частиною моделі з самого початку, а не щось, що додається пізніше. Крім того, завдяки чекпоїнтеру LangGraph зберігає повну копію стану після завершення кожного суперкроку. Саме це уможливлює відновлення після збою: якщо процес зупиняється на півдорозі, виконання продовжується з останнього завершеного суперкроку.

Коли ми розуміємо, як стан передається через один вузол, наступне питання — як керувати порядком або виконувати вузли одночасно. Саме для цього існують ребра.

Ребра: послідовні, паралельні та умовні

Краї контролюють спосіб переходу виконання між вузлами. LangGraph пропонує три типи, які разом охоплюють усі можливі схеми маршрутизації. Найпростіша форма вже була показана у прикладі чат-бота.

Послідовні краї

У найпростішому випадку один вузол слідує за іншим.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft виконується спочатку. Як тільки воно завершується, а LangGraph об’єднує поле draft зі станом, виконується review_draft, який може отримати проект з state["draft"]. Послідовність виконання гарантована, оскільки крайка забезпечує залежність, і не потрібен жодний код для керування послідовністю. Послідовні крайки застосовуються тоді, коли вузол B залежить від результату роботи вузла A: спочатку генерація, потім перегляд, спочатку отримання даних, потім їх обробка, спочатку класифікація, потім направлення.

Випадкові цикли є поширеною проблемою. За замовчуванням LangGraph очікує орієнтований ациклічний граф, тому додавання крайки від review назад до generate спричиняє помилку під час компіляції, якщо тільки граф не налаштований явно як циклічний. Ітеративні цикли зазвичай створюються за допомогою умовних крайок, які можуть направляти потік назад до попереднього вузла.

Паралельні крайки

Коли дві грани виходять від одного вузла, їхні цілі обробляються одночасно у межах одного суперкроку. Тут стають необхідними редуктори:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Дві дуги від START планують завдання research та outline у одному суперкроці, і LangGraph виконує їх одночасно. Обидва також мають дугу до combine, і LangGraph не буде планувати combine, поки всі вузли, які на нього вказують, не завершать роботу, що функціонує як автоматичний бар’єр синхронізації.

findings має редуктор, хоча сюди записує дані лише research, тому що у реальній системі може бути чотири паралельні агенти дослідження, які додають дані до одного й того ж списку, і без редуктора залишився б лише один результат.

Розрахунок затримки простий. Якщо research займає 2 секунди, а outline — 3, то серійна версія потребує 5 секунд, а паралельна — 3, тобто стільки, скільки потрібно на виконання довшого завдання. Для незалежних викликів LLM паралельне виконання забезпечує майже повну економію часу.

Кандидати на паралелізацію — це всі пари вузлів, які не залежать від результатів роботи один одного. Навіть переміщення двох викликів тривалістю по одній секунді в один суперкрок має значення при тисячах запитів. Звертайте увагу на вузли, які здаються незалежними, але ділять стан способами, що не є очевидними. Якщо два паралельні вузли записують дані в одне поле без використання редуктора, це є умовою змагання, при якій значення визначається останнім, хто завершив роботу, і LangGraph не надає жодних попереджень.

Ця структура розгалуження та об’єднання є основою проекту з електронною поштою, який одночасно виконує пошук у базі знань та перевірку у системі відстеження помилок перед складанням відповіді.

Умовні ребра

Коли наступний вузол залежить від результатів роботи поточного, застосовується функція add_conditional_edges:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

Функція route_by_sentiment аналізує стан та повертає рядковий ключ, а словник-мапування, переданий функції add_conditional_edges, перетворює цей ключ на назву вузла. Функція маршрутизації, яка залишається „чистою“ (вхід – стан, вихід – рядок, без побічних ефектів), легко тестується, оскільки її можна перевіряти як модульний тест без використання LLM.

add_conditional_edges приймає три аргументи: вихідний вузол, функцію маршрутизації та словник, який мапує рядкові ключі на назви цільових вузлів. Кожен ключ, який може бути повернутий функцією маршрутизації, має бути присутнім у цьому словнику; інакше LangGraph видасть помилку під час виконання.

Альтернатива у вигляді Command

Коли вузлу потрібно оновити стан та визначити маршрут у одному кроці, він може безпосередньо повернути Command та пропустити використання add_conditional_edges:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Одне значення, яке повертає Command, містить як оновлення стану (update=), так і вибір маршруту (goto=), тож вузол записує дані та обирає наступну точку призначення в одному кроці.

add_conditional_edges є правильним вибором, коли логіка маршрутизації лише читає стан та не записує нічого нового, коли маршрутизація має залишатися окремою від логіки вузла, або коли кілька вихідних вузлів використовують одну й ту саму функцію маршрутизації.

Command є правильним вибором, коли вузол генерує нові дані, які впливають на його власну маршрутизацію, коли одноразовий випадок не вимагає окремої функції маршрутизації, або коли точка призначення залежить від чогось, що обчислюється всередині вузла та ще не є частиною стану.

Для людського перегляду електронний кейс базується на функції add_conditional_edges, оскільки маршрутизація там є простою: вона перевіряє лише пріоритет та рівень впевненості, які вже зберігаються у стані.

Пам’ять та чекпоїнти

Коли порядок виконання та розгалуження під контролем, наступною проблемою є стабільність: що відбувається, коли процес перезапускається посеред розмови?

До цього моменту кожен запуск починався з нуля. Розмови та багатокрокові робочі процеси потребують стану, який зберігатиметься між викликами. LangGraph вирішує це за допомогою чекпоїнтів, а API настільки простий, що перехід від розробки до продакшну з точки зору зберігання даних полягає у зміні однієї рядка коду.

MemorySaver для розробки

MemorySaver зберігає стан у пам’яті, що підходить для розробки та тестування:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

З параметром checkpointer=memory скомпільований граф зберігає повну копію стану після кожного „суперкроку“. Повторний виклик invoke() з тим самим thread_id змушує LangGraph спочатку завантажити найновіший чекпоїнт для цього потоку. thread_id функціонує як ідентифікатор сесії: використовуйте його знову, щоб продовжити роботу, або змініть його, щоб почати з чистого стану.

Другий виклик може не включати попередні повідомлення, оскільки чекпоїнт вже містить їх. Якщо Аліса викликає функцію з параметром {"messages": ["What should I learn first?"]}, LangGraph завантажує чекпоїнт "user-alice", знаходить збережений список та додає нову запис через функцію-редуктор operator.add. Після запуску chatbot поле state["messages"] містить весь історичний запис.

Повторне використання одного thread_id для непов’язаних розмов — це помилка. Якщо "user-alice" використовується знову для іншої теми, чат-бот отримує всю попередню розмову Аліси в контексті, тому кожна нова сесія має отримувати новий thread_id. Під час розробки app.get_state_history(config) відображає повну історію контрольних точок для певної розмови, і саме цей механізм використовується для виправлення помилок шляхом «подорожей у часі» у LangSmith.

Надійні інструменти зберігання стану для продакшну

MemorySaver знаходиться у пам’яті процесу, тому коли процес завершується, стан також втрачається. Для продакшну потрібен надійний бекенд:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() створює необхідні таблиці бази даних під час першого запуску та далі не виконує жодних дій, тому його потрібно викликати лише один раз під час ініціалізації. API залишається незмінним між різними бекендами, оскільки механізми контролю стану працюють через спільну абстракцію. MemorySaver, PostgresSaver та RedisSaver дотримуються одного протоколу, тому код графа не знає, з яким саме з них він взаємодіє.

Вибір між ними:

• MemorySaver підходить для розробки, тестування та демонстрацій, де не має значення збереження даних після перезапуску.
• PostgresSaver підходить для продакшн-середовищ, які потребують надійності та можливості звертатися до історії контрольних точок.
• RedisSaver з бібліотеки langgraph-redis підходить для сценаріїв з високою пропускною здатністю, де необхідні швидкі операції з читанням контрольних точок, і які готові до компромісів щодо стабільності, притаманних Redis.

Що зберігається

Повний знімок стану зберігається після кожного „суперкроку“. Це дозволяє „подорожувати в часі“: можна переглянути стан у будь-який момент історії, відтворити дії з обраної точки контролю чи виправити помилки, щоб з’ясувати, де саме виникла проблема. Той самий механізм використовується для переривань. Коли вузол викликає функцію interrupt(), LangGraph потребує місця для зберігання призупиненого стану, щоб згодом його відновити. Без точки контролю ніщо не може це зберегти, і схема обробки переривань руйнується.

Участь людини під час переривань

Деякі рішення занадто важливі, щоб їх автоматизувати. Перш ніж агент надішле електронний лист клієнту, схвалить повернення грошей чи розгорне код, людина повинна це переглянути та схвалити. LangGraph реалізує це за допомогою функції interrupt().

Функція interrupt()

Розміщення interrupt() всередині вузла зупиняє виконання, записує стан у checkpointer та повертає навантаження тому, хто його викликав:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

По черзі функція виклику виконує чотири дії:

1. Вона зупиняє виконання саме на тому рядку, де відбувся виклик.
2. Вона зберігає повний стан у checkpointer, тому він є необхідним.
3. Вона повертає переданий навантажувач даних, тут це проєкт тексту, тому, хто викликав invoke().
4. Вона чекає, залишаючи граф у призупиненому стані.

Змінна decision спочатку нічого не отримує, оскільки interrupt() не повертається, поки граф не буде відновлений за допомогою Command(resume=value). У цей момент decision отримує те, що було передано як параметр resume=.

Навантажувач даних інформує людину-рецензента про те, що вона бачить, тому він може містити все необхідне: проєкт, короткий огляд або структурований словник із контекстом. Той, хто викликав, отримує його як частину сповіщення про переривання.

Відновлення за допомогою Command

Продовження роботи після перерви означає знову викликати граф із Command(resume=value):

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

На першому кроці LangGraph виконує функцію generate, зберігає чернетку у стані та переходить до функції interrupt() у вузлі review. Він створює контрольну точку стану та повертається, причому result від цього першого виклику invoke() містить дані про переривання та чернетку для перегляду. На другому кроці функція Command(resume="approve") змушує LangGraph завантажити контрольну точку "email-review-1", продовжити виконання з моменту виклику interrupt() та повернути значення "approve" як результат. Значення decision у вузлі review стає "approve", встановлюється параметр approved=True, і виконання переходить до кроку send.

Для обох викликів має використовуватися та сама конфігурація config, оскільки LangGraph знаходить відповідний чекпоїнт за допомогою thread_id. Інший thread_id під час другого виклику призведе до пошуку чекпоїнта, якого немає, і виклик зазнає невдачі.

Чекпоїнт є обов’язковим для переривань. Без нього стан призупинки не має куди перейти, тому app.compile(checkpointer=memory) тут не є необов’язковим.

З версії LangGraph 1.1.0 функція invoke() приймає параметр version="v2" та повертає об’єкт GraphOutput із атрибутами .value та .interrupts, що забезпечує більш безпечне оброблення переривань. Цей підхід варто спробувати після того, як ви ознайомитеся з основами. Контент переривання може бути будь-якою серіалізованою величиною — рядком, словником чи списком. У продакшн-середовищах зазвичай передається структурований словник, який містить усе необхідне для перегляду: чернетку, класифікацію, відправника та URL для перегляду; ці дані може відобразити шар додатку у інтерфейсі користувача або надіслати через Slack чи електронну пошту.

Є ще одна підступність. Повторне викликання invoke() із повним початковим станом замість Command(resume=value) є помилкою: app.invoke({"messages": [...]}, config=config) у тому самому thread_id знову запускає граф для цього потоку та видаляє стан призупинення. Призупинений граф слід продовжувати лише за допомогою Command(resume=value).

Кульмінація: робочий процес сортування електронних листів

Коли всі елементи готові, їх можна об’єднати в систему сортування електронних листів у продакшн-режимі. Вона читає надісланий лист, класифікує його мету за допомогою структурованого виводу, одночасно шукає в базі знань та системі відстеження помилок, складає відповідь та може зупинитися на перевірку людиною перед надсиланням.

Кожна з попередніх концепцій відповідає певній частині цього робочого процесу:

• Стан TypedDict із редюсерами відображається як EmailState, з operator.add для search_results.
• Серійні краї з’єднують read_email з classify_intent.
• Паралельні краї дозволяють classify_intent розділятися на search_docs та bug_tracking.
• Синхронізація типу „фан-ін“ означає, що обидва паралельні вузли надсилають дані до write_response.
• Умовне маршрутизування здійснюється за допомогою route_by_priority, який обирає між людським переглядом та автоматичною відправкою.
• Структурований вихід у форматі Pydantic — це EmailClassification, який обмежує відповідь LLM.
• Інтерупція виконання — це human_review, яка призупиняє процес у очікуванні схвалення.
• Механізм контролю стану — це MemorySaver, який уможливлює цикл інтерупцій та відновлення роботи.

Стан та схеми

Спочатку йде держава та схема Pydantic для структурованої класифікації:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification — це модель Pydantic, а не частина EmailState, який є прикладом розділення TypedDict та Pydantic. Це формат даних, який передається ШІ як ціль JSON-схеми, причому Pydantic перевіряє відповідь та негайно викидає помилку, якщо вона не збігається. EmailState є контейнером для графу та є TypedDict, оскільки LangGraph потребує синтаксису редукторів та ефективної серіалізації.

classification зберігається у вигляді dict у EmailState, тому що після перевірки відповіді Pydantic функція .model_dump() перетворює її на звичайний словник перед збереженням у стані. Це забезпечує можливість серіалізації стану будь-яким інструментом перевірки, без необхідності використання Pydantic під час відновлення.

Найважливішим полем є search_results: Annotated[list[str], operator.add]. Оскільки search_docs та bug_tracking виконуються паралельно та обидва можуть зберігати результати, редюсер збирає їхні вихідні дані разом, тоді як без нього одна з гілок би тихо перезаписала іншу.

Створення вузлів

Вузли залишаються невеликими та виконують одну функцію. Кожен з них бере зі стану лише ті значення, які йому потрібні, виконує свою роботу та повертає часткове оновлення:

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

На перший погляд read_email здається марним, адже він повертає лише ті поля, які вже є у стані, але він виконує функцію явного аналізу. У продакшн-версії було б використано сирі дані вебхуку та витягнуто з них необхідні поля, тож ця функція тут замінює саме цей етап витягування даних.

classify_intent — це те місце, де Pydantic демонструє свою корисність. У structured_llm = llm.with_structured_output(EmailClassification) модель LLM обгортається так, щоб відповіді обов’язково відповідали схемі. Вона отримує JSON-схему, отриману з EmailClassification, і повинна повертати відповідний валідний JSON; LangChain автоматично намагається знову, перш ніж викинути помилку, якщо цього не відбувається. Потім .model_dump() перетворює валідований об’єкт на звичайний словник для зберігання у стані.

Два паралельні вузли записують дані у окремі поля — search_results та bug_info — тому їхні виклики LLM виконуються одночасно. Для search_results функція search_docs повертає список ([f"[Docs] {response.content}"]), оскільки operator.add вимагає вхідних даних у вигляді списку. Функція bug_tracking повертає звичайний рядок для bug_info, у якого немає функції об’єднання, тому достатньо звичайного присвоєння.

Вузлом з кількома вхідними зв’язками є write_response. Коли він починає роботу, обидва паралельні вузли вже завершили свою роботу, і їхні результати зберігаються у стані, тож він формує запит та складає відповідь на основі всіх цих даних.

human_review передає до interrupt() структурований словник, а не звичайний рядок, що є хорошою практикою для продакшну. У цьому словнику зберігається чернетка, класифікація та повідомлення для рецензента, тож будь-яка система, що знаходиться з іншого боку — чи то веб-інтерфейс, бот Slack чи електронне повідомлення — отримує весь контекст.

send_reply використовує state.get("approved", True) як захисний стандартний значення, тож якщо значення approved якимось чином відсутнє у стані, він все одно надсилає повідомлення. Це стандартне значення може бути змінене на False, якщо профіль ризику вимагає обережності.

У продакшені search_docs та bug_tracking будуть викликати справжні системи, такі як векторна база даних, Jira чи GitHub Issues, а декоратор LangChain @tool полегшує обгортання цих викликів. Структура вузлів залишається незмінною — виклик інструменту замінює виклик LLM. Ще одна проблема: write_response читає state["search_results"], тож якщо це все ще порожній список, доступ може зазнати невдачі; цього можна уникнути, ініціалізувавши search_results: [] у початковому пакеті виклику, щоб редюсер почав додавати дані саме звідти.

Підключення країв

Послідовні, паралельні та умовні краї з’єднують вузли:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority надсилає електронні листи високої пріоритетності на людську перевірку незалежно від рівня впевненості. Класифікації з рівнем впевненості нижче 0,8 також потрапляють на перевірку, оскільки якщо модель не впевнена щодо змісту листа, це має вирішити людина. Усе інше надсилається автоматично, що є розумним стандартним режимом у продакшені, який потрібно налаштовувати відповідно до реальних показників помилок.

Цей механізм маршрутизації працює через add_conditional_edges, а не через Command, оскільки route_by_priority є чистою функцією: вона лише аналізує стан та повертає ключ, не створюючи нових даних, що саме й призначено для add_conditional_edges.

Ось усі з’єднання в графі:

• Серійна обробка починається з START, проходить через read_email та потрапляє до classify_intent.
• Розгалуження від classify_intent одночасно запускає як search_docs, так і bug_tracking.
• Зворотне розгалуження надсилає результати цих двох вузлів до write_response.
• Після write_response умовний ребро обирає між human_review та send_reply.
• Після завершення перевірки human_review передає дані до send_reply, який закінчується на END.

Виконання робочого процесу від початку до кінця

Спочатку надходить зразок електронного листа, і перше викликання триває до моменту переривання:

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

До цього моменту граф вже виконав багато завдань:

1. read_email обробив повідомлення.
2. classify_intent позначив його як звіт про баг високої пріоритетності за допомогою структурованого вихідного даних Pydantic.
3. Пошук у документації та перевірка в системі відстеження багів відбулися паралельно протягом одного кроку.
4. write_response підготував відповідь, використавши всю цю інформацію.
5. Вузол human_review викликав interrupt() та призупинив процес.

Увесь процес обробки, включаючи обидва паралельні виклики ШІ, завершився до призупинення. Коли граф зупиняється, стан повністю зберігається, тож рецензент може відповісти протягом хвилин чи годин, не втрачаючи жодної інформації.

Потім рецензент схвалює:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph завантажив чекпоїнт для "email-001", продовжив процес human_review після виклику interrupt(), повернув значення "approve" як результат виконання, встановив approved=True у стані та перейшов до функції send_reply, яка повернула статус підтвердження.

Кожна з попередніх ідей знаходить своє відображення у цьому проекті: стан у форматі TypedDict, підтримуваний редюсером, послідовні та паралельні краї, умовне маршрутизування, структурований вихідний формат через Pydantic, людський огляд за допомогою механізму переривання та зберігання даних у чекпоїнті.

Для електронного листа низької пріоритетності з високим рівнем впевненості функція route_by_priority повертає значення "auto_send", і етап human_review пропускається. Процес виконується безперервно: одночасно відбувається читання, класифікація, пошук, створення проєкту та надсилання листа, причому переривання існує лише на тому шляху, де воно необхідне. Це і є перевагою умовного маршрутизування.

Останнє застереження стосується початкового стану. Відмінно від Pydantic, TypedDict не встановлює значень за замовчуванням для відсутніх полів. Якщо функція write_response намагається отримати доступ до state["search_results"], а ключа немає у початковому стані, виникає помилка KeyError. Кожне поле має бути присутнім у словнику початкового стану, навіть якщо спочатку це порожня стрічка, порожній список чи значення False.

Куди рухатися далі

Шлях, який описаний тут, починається з одного вузла та об’єкта TypedDict та закінчується робочим процесом у продакшн-стилі, який дозволяє одночасно виконувати запити до LLM, перевіряти структурований результат, умовно розгалужуватися, відновлюватися після збоїв за допомогою контрольних точок та чекати на перевірку. Звідти існує кілька напрямків розвитку:

• Platформа LangGraph забезпечує розгортання, кероване створення контрольних точок та горизонтальне масштабування, що знімає тягар керування інфраструктурою, коли метою є обробка справжнього трафіку.
• LangSmith відстежує та діагностує кожну експекурацію графа та показує точний стан на кожному кроці, коли вузол поводиться некоректно. Це стає важливим, коли графи містять п’ять або більше вузлів із умовними шляхами.
• Підграфи об’єднують менші графи в більші багатоагентні системи. Кожен з них керує власним станом та взаємодіє через батьківський граф, що є природним кроком після робочих процесів з одним графом.
• Інструмент, який викликається через LangChain за допомогою @tool та ToolNode, дозволяє агентам використовувати зовнішні API, бази даних та сервіси прямо всередині графа, причому вузли search_docs та bug_tracking є очевидними місцями для його випробування.
• Для збереження даних у продакшені необхідно встановити langgraph-checkpoint-postgres, щоб стан залишався стабільним після перезапуску, а заміна API відбувається за допомогою одного рядка коду.

Короткий огляд

• Стан представлений як TypedDict. Кожне поле, оголошене як Annotated[list, operator.add], збирає значення під час кожного запису, тоді як поле без функції обробки замінюється після кожного запису.
• Вузол — це просто функція: вона отримує повний стан та повертає словник з частковими оновленнями, а LangGraph здійснює їх об’єднання.
• Краї визначають потік. Використовуйте послідовні краї там, де важливий порядок, паралельні краї для незалежної роботи, що виконується одночасно, та умовні краї, коли дані мають обирати шлях.
• Використовуйте TypedDict для стану графа та Pydantic з методом .with_structured_output(), щоб чітко визначити, що повертає LLM.
• Інструменти для створення контрольних точок забезпечують постійність даних та можливість переривань: MemorySaver під час розробки, PostgresSaver у продакшені. Без контрольної точки — без переривань.
• Переривання — це просто збережений, призупинений стан. interrupt(payload) зупиняє роботу та створює контрольну точку, а Command(resume=value) запускає процес з саме того місця.
• Паралелізм вбудований: два краї, що виходять з одного вузла, виконують свої завдання одночасно, і кожне поле, яке записують ці вузли, потребує функції зворотного обчислення.

Питання для роздумів

1. MemorySaver підходить для розробки, а PostgresSaver — для продакшну. Які ще бекенди зберігання варто розглянути для системи з високою пропускною здатністю, та які компроміси допоможуть зробити вибір?
2. route_by_priority надсилає електронні листи високого пріоритету на перевірку та автоматично надсилає решту. Як можна налаштувати цей поріг у реальній системі, та які показники покажуть, чи він встановлений правильно?
3. Остаточне направлення відбувається після write_response за допомогою add_conditional_edges. Чи існує сценарій у робочому процесі електронних листів, де паттерн Command підійде краще, ніж окрема функція маршрутизації?

Пов’язана література

• Практичні нотатки: створення самокоригуючого AI-агента за допомогою Self-RAG та LangGraph — Детальний посібник з практичних нотаток: створення самокоригуючого AI-агента за допомогою Self-RAG та LangGraph, що включає контракти, перевірки та готові фрагменти коду для команд, які впроваджують цю модель.
• Створення агента для дослідження фондового ринку за допомогою ReAct та LangGraph — Агент на базі LangGraph та ReAct бере ідентифікатор акції, викликає інструменти для аналізу фундаментальних даних, новин, технічних показників та аналітичних засобів у визначеному порядку та складає короткий звіт з паузами для участі людини.

Робочий процес локального агента LLM, створений з папок та файлів Markdown — Створіть багатоетапну локальну систему агентів із контрольними точками, де кожен етап є папкою, запити знаходяться у форматі Markdown та людина перевіряє результат перед виконанням дорогого кроку.

Практичні нотатки: Агент Text-to-SQL у Python: посібник з використання інструментів LLM — Детальний посібник з практичних нотаток: Агент Text-to-SQL у Python: посібник з використання інструментів LLM, що включає контракти, перевірки та готові фрагменти коду для команд, які впроваджують цю схему.

Практичні нотатки: Чому ніхто не говорить про LangChain, LangGraph чи AutoAgent — Детальний огляд Практичних нотаток: Чому ніхто не говорить про LangChain, LangGraph чи AutoAgent: контракти, перевірки та готові фрагменти коду для команд, які використовують цю модель.

Команда багатьох агентів зі здатністю до самовідновлення в LangGraph: архітектор, тестувальник, дебагер — Створіть машину станів LangGraph на Python, де архітектор, розробник, среда pytest sandbox, дебагер та рецензент співпрацюють та повторюють процес до тих пір, поки тести не пройдуть.

Практичні нотатки: створення циклу перегляду коду з двома агентами за допомогою A2A, LangGraph та — детальний пошаговий огляд Практичних нотаток: створення циклу перегляду коду з двома агентами за допомогою A2A, LangGraph та: контрактів, перевірок та слотів для коду для команд, які використовують цю схему.