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)

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

```python
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
```

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

```python
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)
```

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

```python
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

```python
from langgraph.types import interrupt

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

### Subgraphs

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

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

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

```python
from langgraph.store.memory import InMemoryStore

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

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

### Streaming

```python
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

```python
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

```python
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 (роутинг по содержимому)

```python
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

```python
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 (параллельные ветки)

```python
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

```python
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 для долгосрочной памяти

```python
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

```python
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`:

```typescript
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) — отдельная тема.