День 1: Минимальный граф в LangGraph
Цель: Создать первый работающий граф с двумя узлами и понять основы LangGraph
Время изучения: 30-60 минут
Код: src/day1_min_graph.py
Тесты: tests/test_day1_min_graph.py
🤖 Связь с агентами
Важно понять: Каждый узел в графе — это будущий агент!
Что мы строим?
Зачем начинать с графов?
Граф — это структура для организации агентов:
- 🎯 Кто за что отвечает (узлы = агенты)
- 🔄 В каком порядке работают (рёбра = связи)
- 📦 Как передают данные (state = общая память)
Аналогия: Сначала учимся строить комнаты (узлы), потом строим дом (multi-agent систему).
Что такое LangGraph?
LangGraph — это фреймворк для создания графов вычислений с состоянием. Представьте конвейер, где:
- State (состояние) — данные, которые передаются между узлами
- Nodes (узлы) — функции, которые обрабатывают и обновляют состояние
- Edges (рёбра) — связи между узлами, определяющие порядок выполнения
В контексте агентов:
- State = общая память всех агентов
- Node = отдельный агент со сво��й задачей
- Edge = передача управления от агента к агенту
Мин��мальный пример: A → B
Создадим простой граф из двух узлов, где каждый узел добавляет свою пометку в сообщение.
Шаг 1: Определяем состояние
from typing import TypedDict
class SimpleState(TypedDict, total=False):
message: str # Накапливаем сообщения
count: int # Счётчик проходов
Зачем TypedDict?
- ✅ Строгая типизация — IDE подсказывает поля
- ✅ Проверка типов — меньше ошибок
- ✅ Документация — явно видно, что в состоянии
total=False — можно не инициализировать все поля сразу
Шаг 2: Создаём узлы
def node_a(state: SimpleState) -> SimpleState:
"""Узел A: увеличивает счётчик и добавляет пометку."""
new_state = dict(state) # ✓ Создаём копию!
new_state["count"] = (new_state.get("count") or 0) + 1
new_state["message"] = (new_state.get("message") or "") + " → A"
return new_state
def node_b(state: SimpleState) -> SimpleState:
"""Узел B: добавляет свою пометку."""
new_state = dict(state)
new_state["message"] = (new_state.get("message") or "") + " → B"
return new_state
Важные правила:
- Не мутируйте входной state — создавайте копию
- Используйте
.get()с дефолтами — защита от KeyError - Возвращайте полное состояние — даже если изменили одно поле
Шаг 3: Собираем граф
from langgraph.graph import StateGraph
def build_graph():
graph = StateGraph(SimpleState)
# Добавляем узлы
graph.add_node("A", node_a)
graph.add_node("B", node_b)
# Соединяем узлы
graph.add_edge("A", "B")
# Определяем вход и выход
graph.set_entry_point("A")
graph.set_finish_point("B")
# Компилируем в исполняемый workflow
return graph.compile()
Что происходит:
StateGraph(SimpleState)— создаём граф с типизированным состояниемadd_node(name, function)— регистрируем узлыadd_edge(from, to)— создаём связь между узламиset_entry_point/set_finish_point— обязательные точки входа/выходаcompile()— превращаем граф в исполняемый workflow
Шаг 4: Запускаем
def main():
workflow = build_graph()
# Начальное состояние
initial = {"message": "Start", "count": 0}
# Запускаем граф
result = workflow.invoke(initial)
print(result)
# {"message": "Start → A → B", "count": 1}
Полный код
from typing import TypedDict
from langgraph.graph import StateGraph
class SimpleState(TypedDict, total=False):
message: str
count: int
def node_a(state: SimpleState) -> SimpleState:
new_state = dict(state)
new_state["count"] = (new_state.get("count") or 0) + 1
new_state["message"] = (new_state.get("message") or "") + " → A"
print(f"Node A: {new_state}")
return new_state
def node_b(state: SimpleState) -> SimpleState:
new_state = dict(state)
new_state["message"] = (new_state.get("message") or "") + " → B"
print(f"Node B: {new_state}")
return new_state
def build_graph():
graph = StateGraph(SimpleState)
graph.add_node("A", node_a)
graph.add_node("B", node_b)
graph.add_edge("A", "B")
graph.set_entry_point("A")
graph.set_finish_point("B")
return graph.compile()
def main():
workflow = build_graph()
result = workflow.invoke({"message": "Start", "count": 0})
print(f"Final: {result}")
if __name__ == "__main__":
main()
Запуск
# Запустить пример
python src/day1_min_graph.py
# Запустить тесты
make test-day1
# Или напрямую
PYTHONPATH=. pytest tests/test_day1_min_graph.py -v
Вывод:
Отладка: stream() для пошагового выполнения
Если нужно видеть каждый шаг:
workflow = build_graph()
initial = {"message": "Start", "count": 0}
for step in workflow.stream(initial):
print(f"Step: {step}")
Вывод:
Типичные ошибки
❌ Ошибка 1: Мутация состояния
def bad_node(state: SimpleState) -> SimpleState:
state["message"] += " → Bad" # Мутирует оригинал!
return state
Проблема: Изменяет входной state, сложно отлаживать
✅ Правильно:
def good_node(state: SimpleState) -> SimpleState:
new_state = dict(state) # Копия
new_state["message"] = (new_state.get("message") or "") + " → Good"
return new_state
❌ Ошибка 2: Забыли entry/finish point
graph = StateGraph(SimpleState)
graph.add_node("A", node_a)
graph.compile() # ❌ Ошибка: нет entry_point!
✅ Правильно:
graph.set_entry_point("A")
graph.set_finish_point("A")
graph.compile() # ✅ Работает
❌ Ошибка 3: KeyError на несуществующем поле
def bad_node(state: SimpleState) -> SimpleState:
new_state = dict(state)
new_state["count"] = state["count"] + 1 # ❌ KeyError если count не задан!
return new_state
✅ Правильно:
def good_node(state: SimpleState) -> SimpleState:
new_state = dict(state)
new_state["count"] = (new_state.get("count") or 0) + 1 # ✅ Дефолт
return new_state
Тестирование
import pytest
from src.day1_min_graph import node_a, node_b, SimpleState
def test_node_a():
state: SimpleState = {"message": "Test", "count": 0}
result = node_a(state)
assert result["count"] == 1
assert "→ A" in result["message"]
def test_node_b():
state: SimpleState = {"message": "Test"}
result = node_b(state)
assert "→ B" in result["message"]
def test_full_graph():
from src.day1_min_graph import build_graph
workflow = build_graph()
result = workflow.invoke({"message": "Start", "count": 0})
assert result["count"] == 1
assert result["message"] == "Start → A → B"
Ключевые выводы
✅ StateGraph — основа LangGraph, граф с типизированным состоянием
✅ TypedDict — строгая типизация для state
✅ Неизменяемость — всегда создавайте копию state
✅ entry/finish points — обязательны для компиляции
✅ invoke() — запуск графа, возвращает финальное состояние
✅ stream() — пошаговое выполнение для отла��ки
🤖 Связь с агентами
Сегодня вы научились:
- ✅ Создавать узлы (будущие агенты)
- ✅ Передавать состояние (общая память агентов)
- ✅ Связывать узлы (координация агентов)
Следующий шаг: Научимся делать агентов умнее — добавим условную логику и циклы.
Следующий шаг: День 2
В следующей статье мы изучим:
- Conditional Edges — выбор следующего узла на основе state
- Циклы — как создать узел, который вызывает сам себя
- Защита от бесконечных циклов — MAX_ITERATIONS
👉 День 2: State и Conditional Edges
Ресурсы
Прогресс: 1/10 дней ✅