langchain-evolution-deck/research/per-tech/langgraph.md

LangGraph (≥ 1.0)

Что это в одном абзаце

LangGraph — это низкоуровневый оркестрационный фреймворк LangChain Inc. для построения долгоживущих stateful-агентов. В отличие от LangChain (high-level create_agent), LangGraph даёт явный контроль над формой графа: узлы (add_node), рёбра (add_edge), условные переходы (add_conditional_edges), checkpointing, human-in-the-loop через interrupt, stream-режимы. С версии 1.0 (релиз 22 октября 2025) LangGraph — это production-ready durable runtime: состояние графа персистится автоматически, при падении сервера посреди long-running workflow он восстанавливается ровно с точки остановки. Вдохновлён Pregel и Apache Beam, public interface похож на NetworkX.

Метаданные на дату snapshot 2026-06-22:

• GitHub stars: ~35.4k
• Latest stable (Python): langgraph==1.2.6 (от 18.06.2026)
• License: MIT
• JS-аналог: @langchain/langgraph (npm)

Источники:

• README github.com/langchain-ai/langgraph
• https://changelog.langchain.com/announcements/langgraph-1-0-is-now-generally-available
• https://blog.langchain.com/langchain-langgraph-1dot0
• https://blog.langchain.com/fault-tolerance-in-langgraph

---

Ключевые API (≥ 1.0)

Импорты верхнего уровня

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command, interrupt, Send

Базовый граф с состоянием

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

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

def node_a(state: State):
    return {"messages": [{"role": "assistant", "content": "hi"}]}

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
builder.add_edge("a", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({"messages": []}, config=config)

Условные рёбра

def route(state: State) -> str:
    return "tool_node" if state.get("needs_tool") else END

builder.add_conditional_edges("agent", route, {
    "tool_node": "tool_node",
    END: END,
})

Human-in-the-loop через interrupt

from langgraph.types import interrupt

def approval_node(state: State):
    decision = interrupt({"question": "Approve?", "data": state["messages"]})
    return {"approved": decision == "yes"}

Subgraphs

sub_builder = StateGraph(SubState)
sub_builder.add_node("x", x_node)
sub_graph = sub_builder.compile()

# В родительском графе
parent_builder.add_node("sub", sub_graph)

Store (долгосрочная память)

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

# Внутри ноды
store.put(("user_123", "prefs"), "key", {"value": "dark"})

Streaming

for mode, chunk in graph.stream({"messages": []}, config, stream_mode=["values", "updates"]):
    print(mode, chunk)

---

Что нового в 1.0

1. Durable execution (стабилизировано) — автоматическая персистенция state, восстановление ровно с точки падения. Без своего DB-кода.
2. Built-in persistence как стабильное API — checkpointer теперь контракт, а не фича; Postgres / SQLite / memory — все first-class.
3. Human-in-the-loop first-class — interrupt() стал стабильным API, поддерживает multi-day approval workflows.
4. Graph-based execution как production pattern — смесь детерминированных узлов и агентных.
5. Deprecation: langgraph.prebuilt.create_react_agent → перенесён в langchain.agents.create_agent (LangChain 1.0).
6. API stability promise — без breaking changes до 2.0.
7. Middleware hooks (в 1.2) — fault tolerance: retries / timeouts / error handlers.

---

Что нужно раскрыть в презентации

• State, Channels, Reducers — что такое Annotated[list, add_messages] и зачем нужен reducer.
• Checkpointing — InMemorySaver для dev, PostgresSaver для prod. Что хранится в StateSnapshot.
• Threads — configurable.thread_id как ключ сессии.
• Human-in-the-loop через interrupt — не через callback, а через настоящий graph pause.
• Subgraphs — композитность графов, parent может заходить в subgraph целиком.
• Send / Map-reduce — параллельные ветки графа.
• Streaming modes — values / updates / events / messages / custom.
• Store vs Checkpointer — checkpoint для сессии, store для cross-session долгосрочной памяти.
• Pregel / Beam inspiration — почему именно «graph», а не «chain».

---

8 рабочих примеров кода Python (≥ 1.0)

1. StateGraph с message reducer

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

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

def echo(state: State):
    last = state["messages"][-1]
    return {"messages": [{"role": "assistant", "content": f"echo: {last.content}"}]}

g = StateGraph(State)
g.add_node("echo", echo)
g.add_edge(START, "echo")
g.add_edge("echo", END)
app = g.compile()
print(app.invoke({"messages": [{"role": "user", "content": "hi"}]}))

2. Checkpointing + thread_id

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = g.compile(checkpointer=checkpointer)

cfg = {"configurable": {"thread_id": "user-1"}}
app.invoke({"messages": [{"role": "user", "content": "hi"}]}, cfg)
app.invoke({"messages": [{"role": "user", "content": "again"}]}, cfg)
# state["messages"] содержит оба сообщения — thread persistence работает

3. Conditional edges (роутинг по содержимому)

def route(state: State) -> str:
    if "tool" in state["messages"][-1].content:
        return "tool_node"
    return END

builder.add_conditional_edges("agent", route, {"tool_node": "tool_node", END: END})

4. Human-in-the-loop через interrupt

from langgraph.types import interrupt

def approval(state: State):
    answer = interrupt({"prompt": "Approve?", "context": state})
    return {"approved": answer}

builder.add_node("approval", approval)
builder.add_edge(START, "approval")
app = builder.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "t1"}}
# Первый вызов упадёт в interrupt
try:
    app.invoke({}, cfg)
except Exception:
    pass

# Возобновляем с ответом пользователя
from langgraph.types import Command
result = app.invoke(Command(resume="yes"), cfg)

5. Send / Map-reduce (параллельные ветки)

from langgraph.types import Send

def fanout(state: State):
    return [Send("process", {"item": i}) for i in state["items"]]

def process(state: dict):
    return {"results": [state["item"] * 2]}

builder.add_conditional_edges("start", fanout)
builder.add_node("process", process)

6. Subgraphs

sub = StateGraph(SubState)
sub.add_node("inner", inner_fn)
sub.add_edge(START, "inner")
sub_compiled = sub.compile()

parent = StateGraph(ParentState)
parent.add_node("sub_block", sub_compiled)
parent.add_edge(START, "sub_block")

7. Store для долгосрочной памяти

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
app = builder.compile(checkpointer=InMemorySaver(), store=store)

def remember(state: State):
    store.put(("user-1", "facts"), "name", {"value": "Alice"})
    return {}

# В другом turn:
def recall(state: State):
    fact = store.get(("user-1", "facts"), "name")
    return {"user_name": fact.value["value"]}

8. Streaming

for event in app.stream({"messages": [{"role": "user", "content": "hi"}]}, stream_mode="values"):
    print(event)

# Кастомный streaming через writer
def node(state: State):
    writer = get_stream_writer()
    writer({"progress": "50%"})
    return {}

---

TypeScript-аналог

Все примеры имеют аналог в @langchain/langgraph:

import { StateGraph, START, END } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph-checkpoint";
import { Annotation, messagesStateReducer } from "@langchain/langgraph";

const State = Annotation.Root({
  messages: Annotation({ reducer: messagesStateReducer, default: () => [] }),
});

const g = new StateGraph(State)
  .addNode("echo", (s) => ({ messages: [{ role: "assistant", content: "hi" }] }))
  .addEdge(START, "echo")
  .addEdge("echo", END);

const app = g.compile({ checkpointer: new MemorySaver() });
const cfg = { configurable: { thread_id: "t1" } };
const result = await app.invoke({ messages: [{ role: "user", content: "hi" }] }, cfg);

Где аналог есть: весь базовый API (StateGraph, conditional edges, checkpoint, interrupt). Где нет / отличается: некоторые специфичные savers (PostgresSaver в JS требует отдельного пакета), Send API полностью паритетно.

---

Плюсы и минусы текущей версии (1.x)

Плюсы

• Durable execution из коробки — killer-фича для long-running агентов.
• HITL first-class API — interrupt() вместо костылей с callback-ами.
• Гибкость — можно построить любую топологию графа (циклы, ветки, параллелизм).
• Прозрачность — graph inspection в LangGraph Studio.
• Семантическая стабильность — semver до 2.0.

Минусы

• Кривая обучения — concepts (channels, reducers, send/receive) требуют времени.
• Boilerplate — базовый граф требует много кода по сравнению с create_agent.
• Checkpointing требует инфраструктуры — для prod нужен Postgres, настройка schema.
• Stream API многослойный — stream_mode (values / updates / events / messages / debug) сбивает с толку.
• Debugging сложных графов — без LangSmith Studio тяжело.

---

Заметки для презентации

• Подчеркнуть: LangGraph — runtime, не agent-harness. create_agent (LangChain) и create_deep_agent (Deep Agents) работают поверх LangGraph.
• Если есть HITL-сценарий — показать interrupt() как killer-фичу 1.0.
• Использовать аналогию: LangGraph = база данных для состояния агента, LangChain = ORM поверх.
• Упомянуть, что LangGraph вдохновлён Pregel (Google) и Apache Beam — это не новость из AI, это паттерн из распределённых систем.
• В 1.2 — fault tolerance (retries / timeouts / error handlers) — отдельная тема.