Шас прымітываў 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 створаны спецыяльна для гэтага. Яны падбіраюць прыемнікі за ідэнтыфікаторамі і заменяюць вялікі прыемнік на ўжо існуючы, калі такі ідэнтыфікатор відзначаецца, дадаючы даўедзьлі толькі сапраўдна новыя прыемнікі. Правіло проста: выкарыстоўваце add_messages для полаў, якія храняюць об’екты HumanMessage і AIMessage, а operator.add — для звычных списоў, якія накапліваюць інфармацыю, напрыклад, рэўісторію таго, якія інструменты былі выкліканы.

Зменшыце размер стану до мінімуму

Другая распаўсюджаная памылка — адляванне стану ў формате схемы базы дадзеных, калі для кожной можлівай пазнейшай потрэбы ствараецца адпаведны поле. Дадзейце поле толькі тады, калі вузел фактычна яго чытае або запішвае. Наследкі ігнаравання гэтага є рэальнымі: узьмімо граф для обробкі дакументаў, який зберагае ў стане цэлыя неапранутыя адпаведзі LLM, уключаючы метаданы аб викорыстоўванні токенаў. Обробка 50 дакументаў у цікле прыводзіла кожны пункт перапытку да розмеру або-альбо 180 KB, а час запісу ў 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. Пункты контролю: памяць, якая застаецца між вызывамі

Checkpointer ператварае вызов функцыі без стану ў абмен з памяцю. Без яго кожны 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 толькі для разработкі; замена checkpointer адбываецца дышэўна, таму робіце яе перад тым, как корыстувальнікі пачнуць апоўнювацца графам.
• Вярхуеце дынамічны interrupt() для умовных затверджэнняў, і заставаеце код нейтральным ў павторных запусках, адтакуль восстанавленне перзапускае вузел.

Даследжвальная дакументацыя: Дакументацыя Graph API для LangGraph, Інструкцыя па перарыванні і interrupt() — Даследжвальная інформацыя API.