Шість примітивів LangGraph та способи приховування проблем у кожному з них

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

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

Чому саме граф, а не ланцюг

Синтаксис трубки LangChain, prompt | llm | parser, зручний для однократного оброблення даних через модель. Він перестає бути ефективним, як тільки агент мусить прийняти рішення: шукати чи відповідати безпосередньо, спробувати знову чи здатися, запитати людину чи продовжити. Ланцюг не має поняття „це залежить“, тому розробники обгортають виклики ланцюга у оператори if, і незабаром створюють власну, недокументовану та складнішу для дебаггингу машину станів.

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

1. Стан: один спільний об’єкт та важливі функції зміни стану

Стан — це єдиний об’єкт, з яким кожен вузол читає та записує дані. Без нього контекст часто передається як аргументи функцій, і стає складно визначити, що саме знав кожен конкретний крок. Наведене нижче визначення — це TypedDict, який містить запитання, відповідь та список повідомлень; оновлення цього списку об’єднуються за допомогою редуктора add_messages.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add не є редуктором повідомлень

Багато навчальних посібників замість цього позначають поле повідомлень як operator.add. Це здається правильним: add додає елемент до списку, а не перезаписує його, що саме потрібно для розвитку діалогу. Проблема полягає у тому, що він об’єднує елементи без перевірки. Як тільки з’являється потреба оновити або видалити наявне повідомлення, наприклад під час скорочення історії розмови або заміни результату виклику інструменту, він додає копію замість цього, і історія розмови заповнюється застарілими записами без жодної помилки.

add_messages створений саме для цієї мети. Він знаходить повідомлення за ID та замінює наявне повідомлення, якщо такий ID вже існує, додаючи лише справді нові повідомлення. Правило просте: використовуйте add_messages для полів, які містять об’єкти HumanMessage та AIMessage, а operator.add — для звичайних списків, які накопичують інформацію, наприклад, список інструментів, які використовувалися.

Зберігайте мінімальну структуру стану

Другою поширеною помилкою є проектування стану як схеми бази даних, з полем для кожної потреби, яка може виникнути згодом. Додавайте поле лише тоді, коли вузол дійсно його читає або записує. Наслідки ігнорування цього принципу є конкретними: уявіть граф обробки документів, який зберігає повні необроблені відповіді LLM разом із метаданими про використання токенів у стані. Обробка 50 документів у циклі призводила до того, що розмір кожного контрольного пункту становив близько 180 КБ, а час запису в Postgres перевищував 400 мс, що є достатньо повільним для того, щоб користувачі, які чекають на відповідь, це помітили. Рішення було простим: скоротити обсяг стану до трьох полів, які дійсно використовуються вузлами на наступних етапах. Пам’ятайте, що за наявності контрольного пункту все, що знаходиться у стані, серіалізується та зберігається на кожному кроці.

2. Вузли: повертайте лише те, що змінилося

Вузол — це звичайна функція Python. Вона отримує стан, виконує свою роботу та повертає словник, який містить лише ті поля, які вона змінила. Це і є весь контракт. У першому прикладі викликається модель чату OpenAI з запитанням, а відповідь записується у змінну answer.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

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

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Ця версія вимагає локальної роботи Ollama з завантаженою моделлю (ollama pull llama3.1) та встановленим пакетом інтеграції (pip install langchain-ollama). Встановлення значення temperature=0 також робить виконання більш відтворюваним, що допомагає під час тестування логіки маршрутизації.

Повернення всього стану заважає іншим оновленням

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

3. Прямі краї: завжди підключайте вихід

Краї визначають, що буде виконуватися далі. Прямий край є безумовним: коли завершується вузол A, починає працювати вузол B. Наведена нижче діаграма містить два вузли, пов’язує answer з refine, пов’язує refine з END, визначає точку входу та компілює код.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

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

4. Умовні краї: де агент насправді приймає рішення

Умовні ребра є тим, що перетворює граф на агента замість фіксованої схеми обробки. Функція маршрутизації перевіряє стан та повертає мітку; функція відповідності перетворює кожну мітку на наступний вузол. У цьому прикладі коротка відповідь (менше 50 символів) надсилається до refine, а все інше — до END.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Зауважте, що це умовне ребро замінює пряме ребро answer до refine з попереднього фрагмента. Якщо ви зареєструєте обидва, будуть обрані обидві схеми, що рідко є бажаним.

Невідповідні мітки маршрутів спричиняють помилки яскраво, але незрозуміло

Частою помилкою тут є функція маршрутизації, яка повертає рядок, що відсутній у словнику мапування. Результатом є досить загальна помилка ключа, прихована на кількох рівнях у стек-трейсі, і навіть така дрібниця, як зайва пробіл, може коштувати значної кількості часу. Надійна звичка: спочатку написати словник мапування, а потім створити маршрутизатор, скопіювавши з нього точні ключі. Ще краще — визначити мітки один раз як константи або позначити тип повернення маршрутизатора за допомогою Literal["refine", "done"], щоб перевірювачі типів та читачі могли одразу побачити дозволені значення.

5. Пункти контролю: пам’ять, яка зберігається між викликами

Чекпоінтер перетворює виклик функції без стану на „розмову“ з пам’яттю. Без нього кожен виклик app.invoke() починається з нуля. З його допомогою стан зберігається для кожної потоку, і будь-який виклик, який передає у своїй конфігурації той самий thread_id, продовжує роботу з того місця, де зупинився попередній. У прикладі другий виклик на потоці user-session-42 пам’ятає перше запитання.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver підходить лише для локальної розробки. Він знаходиться у пам’яті процесу, тому перезапуск сервера стирає всі „розмови“. Усе, від чого залежать справжні користувачі, потребує постійного бекенду: SQLite для одного сервера або Postgres, коли кілька інстанцій мають ділитися станом.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Кожен бекенд постачається як окремий пакет, як показано у коментарях до встановлення. У поточних версіях ці зберігачі зазвичай створюються за допомогою рядка підключення (наприклад, через from_conn_string), а Postgres потребує одноразового виклику setup() для створення своїх таблиць, тому перевірте документацію до checkpointer щодо точної ініціалізації у вашій версії.

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

6. Участь людини: статичні точки зупинки проти динамічних переривань

Шаблон, який показують більшість навчальних матеріалів, — це interrupt_before, список імен вузлів, де скомпільована структура зупиняється перед виконанням:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Цей підхід працює та його легко пояснити, але він є статичним. Точка зупинки визначається ім’ям вузла; її не можна зробити умовною, а також не можна додати інформацію про те, що потрібно перевірити. Реальні вимоги швидко перевершують можливості цього підходу, адже „зупинитися перед цим вузлом“ та „зупинитися лише тоді, коли сума повернення перевищує 500 доларів“ — це різні правила, і лише перше можна сформулювати таким чином.

Зупинка зсередини вузла за допомогою interrupt()

Більш гнучким підходом є виклик interrupt() безпосередньо зсередини вузла. Наведений нижче вузол перевіряє суму повернення грошей; якщо вона перевищує 500 доларів, він зупиняється та передає інформацію про проект та суму людині, яка приймає рішення. Перший виклик invoke триває до моменту цієї зупинки. Другий виклик передає Command(resume="approve") у тому ж потоці, і значення, передане до resume, стає значенням, яке повертає interrupt(); тому вузел або продовжує процес, або повертає статус скасування. Потрібен показник стану, оскільки стан зупинки має бути збережений де-небудь під час очікування.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

У прикладному стані використовуються такі поля, як refund_amount, draft та task, яких немає у попередньому AgentState; у реальній схемі їх потрібно було б оголосити там.

Відновлення запускає весь вузол заново

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

Модель, яка схвалює саму себе, не є системою з участю людини

Який би механізм ви не обрали, запит у моделі «Чи слід мені продовжувати?» та довіра до відповіді не є людським наглядом, як би це не позначали. Це агент, який підтверджує власне рішення. Справжній крок схвалення передає контроль особі, яка знаходиться поза структурою, та чекає на її відповідь.

Шість примітивів у короткому огляді

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

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Розумний порядок складання

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

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

Основні висновки

• Використовуйте add_messages для історії чату та operator.add лише для звичайних списків, і тримайте стан мінімальним, оскільки він зберігається на кожному кроці.
• Повертайте лише змінені поля вузлів; повне повернення стану мовчки перезапише паралельні або попередні оновлення.
• Надайте кожному шляху чіткий маршрут до END, та використовуйте обмежені цикли спроб, замість того щоб покладатися на ліміт рекурсії.
• Отримуйте мітки маршрутизації зі схеми, щоб вони не відхилялися одна від одної.
• Вважайте InMemorySaver інструментом лише для розробки; заміна контрольного показника є дешевою, тому робіть це до того, як користувачі почнуть покладатися на граф.
• Віддавайте перевагу динамічному interrupt() для умовних схвалень, і зберігайте код до того, як він стане ідемпотентним, оскільки продовження роботи змусить перезапустити вузол.

Документація для посилання: Документація до Graph API для LangGraph, Посібник з переривань та interrupt() — довідник API.