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, а не звычны слоўнік чы ўпорадкованы клас? Ён адмаўляе статычную перапакоўку тыпаў, адтолькі як mypy і Pyright яго розумеюць, а таксама падтрымлівае падказкі пра тыпы, і ўтрымлівае додатковы навантажэння пад час выканання, якое ёсць у упорадкованых класах чы мадэлях 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-схему, якой трэба падпараджацца, і пераканальваецца, што адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна адпаведна а

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

Для практыкі перагляду з боку чалавека модуль email capstone выкарыстоўвае 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. Гэта фармат, які перэдаецца LLM як цэль для JSON-шэмы, і Pydantic перакантролюе адпаведнасць і негайна паводзіць паўтарэнне, калі адпаведнасць няма. EmailState — гэта кантэйнер графа, TypedDict, таму што LangGraph патрабуе сінтаксіса рэдуктара і эфектывае серыяванне.

classification зберагаецца як dict у EmailState, таму што пасля таго, як Pydantic перакантролюе адпаведнасць, метод .model_dump() ператварае яго ў звычны дыктарыян пры вводзе ў стан. Гэта дазволяе стану быць серыяваным будзь-якім checkpointerам, без патрэбы ў залежнасці ад 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 здаецца беспрагматычным, адколькі ён вяртае толькі тыя поля, які ужо ўсёчас знаходзяцца ў стане, але ён выступае як явны крок парсінгу. У версіі для практычнага вікорыстоўвання было б взятае необработанае паданне з webhook і выкарыстоўванае для выявлення поль, таму функцыя тут выполняе ролю такога выявлення.

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_info функцыя bug_tracking вяртае звычную строчку, у якой няма функціі для агрэгатавання, таму дастатньа звычная прызначэння.

Вузлом, які прыймае даныя з іншых вузлаў, ёсць write_response. Калі ён запускаецца, оба паралельныя вузлы вяршаюць свою працу, і ўсі ўсунутыя рэзультаты зберагаюцца ў стане, тады ён складае запіт і прадаўае адпаведны адпаведзень.

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

send_reply выкарыстоўвае state.get("approved", True) як захоўні значэнне за замовчаннем, таму калі значэнне approved чыго-такі не ў стане, ён яго адправляе. Гэта захоўнэ значэнне можа быць змененае на False, калі профіль рызыку ёсць аберэжным.

У рэальных умовах search_docs і bug_tracking будуць вызвалі рэальныя сістэмы, такія як база дадзейнаў вектараў, Jira чы GitHub Issues, а дэкоратар @tool з LangChain спакойна дапамагае абгортаваць гэтыя вызовы. Структура вузлаў застаецца той самый, пры чым вызов інструмента заменяе вызов 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() і зупініў працэс.

Усі етапы апрацоўкі, укладаючы ў сябе і паралельныя вызовы LLM, завершыліся да таго часу, калі была здыявлена пауза. Калі граф зупіняецца, стан абов’язкова зберагаецца, таму рэвізор можа адпаведзаць прыбліжна за хвіліны аб гадзіны, не втрачаючы нічога.

Потым рэвізор затверджвае:

# 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 праходзіць без адбыцця. Алгорытм выконваецца без перэрыва: чытанне, класыфікацыя, пошук паралельна, стварэнне праекта і адправка, пры чым перэрыв існуе толькі на тым шляху, які яго патрабуе. Гэта і ёсць перавага умовнага маршрутызавання.

Ішчэ адна паведамленне пра абераганне стосуецца пачатковага стану. TypedDict не задае значэнняя для відсутняйых полей, на вядзіманне з Pydantic. Калі write_response прабуе з’явіцца да state["search_results"], а ключа няма ў пачатковым стане, выйдзе KeyError. Кожна польна мусіць быць у словніку пачатковага стану, нават калі спачатку яна ўяўляе сабою порожняя строчка, порожній спіс або False.

Куды працягваць

Маршрут, які выкорыстоўваецца тут, пачынаецца з аднаго вузла і TypedDict, а заканчваецца рабочым працэсам у стылі прыменення, які можа адкрываць запыты до LLM паралельна, перакантроліваць структураваны выхід, робіць умовныя рашэнні, выжываць пасля збоев за дапамою пунктавання і чакаць на пераглядача. Звядзе там ёсць калькі напрамоў:

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

Короткая падсумовка

• Стан прэдставляецца як TypedDict. Кожны поле, заданае як Annotated[list, operator.add], скарабатывае значэнні праз усі змены, а поле без функцыі змены заменяецца пасля кожнай змены.
• Вузел — гэта проста функцыя: яна отрымае весь стан і вяртае слоўнік частковых змян, пры чым LangGraph адпаведна з’едначае іх.
• Канты выклікаюць тэкстуру прабегу. Іспользуйце серыяльныя канты там, дзе важна парадграфія, паралельныя канты для незалежных задач, якія выконваюцца адночасна, і умовныя канты, калі даныя павінны выбраць паводзь.
• Для стану графа іспользуйце TypedDict, а для контролю таго, што вяртае LLM, — Pydantic з .with_structured_output().
• Інструменты для стварэння чэкпоінтаў забезпечваюць стойкасць і можлівасць перарываў: 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, дэбагер і рэвізор саавтарсаваць і будуць працаваць па кругу, пакуль тэсты не пройдуць.

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