LangGraph — фреймворк от LangChain для создания агентов с контролем потока через графы состояний. Узлы = агенты/инструменты, Рёбра = переходы, Conditional Edges = ветвления и циклы. Полный контроль над каждым шагом агента.
LangGraph представляет агента как направленный граф, где каждый узел — это функция-обработчик (вызов LLM, инструмента, парсинг), а рёбра определяют поток данных. В отличие от CrewAI с фиксированной ролью агента, здесь вы задаёте логику переходов явно.
Архитектура LangGraph-агента: State хранит общие данные, Router направляет поток через conditional edges, Researcher/Writer/Tools — узлы-обработчики. Цикл обеспечивает итеративное улучшение ответа.
| Характеристика | LangGraph | CrewAI | AutoGen |
|---|---|---|---|
| Парадигма | State Graph | Role-play | Conversation |
| Контроль потока | ✅ Полный | ⚡ Частичный | ⚡ Через диалог |
| Циклы | ✅ Нативные | ❌ Нет | ✅ Через HITL |
| Checkpointing | ✅ MemorySaver | ❌ Нет | ⚡ Ограниченный |
| Сложность старта | ⚡ Средняя | ✅ Низкая | ⚡ Средняя |
| Продакшен | ✅ LangSmith | ⚡ Экспорт | ⚡ AutoGen Studio |
LangGraph — библиотека Python, устанавливается через pip. Зависит от langchain и langchain-openai для LLM-обёрток.
# Установка LangGraph pip install langgraph langchain langchain-openai # Ключевые импорты from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI
State — это TypedDict, который передаётся между узлами графа. Каждый узел читает из State и возвращает частичное обновление. LangGraph автоматически мёрджит возвращаемые словари. Поле с Annotated[list, "add"] добавляет элементы в список вместо перезаписи.
class AgentState(TypedDict): """Состояние — общий словарь, передаваемый между узлами.""" messages: Annotated[list, "add"] # "add" = append, не перезапись next_step: str research_notes: str iterations: int # Инициализация LLM llm = ChatOpenAI(model="gpt-4o", temperature=0)
Каждый узел — это функция, принимающая State и возвращающая dict с обновлениями. Researcher анализирует запрос и генерирует заметки. Writer пишет финальный ответ. Router определяет, к какому узлу идти дальше.
def researcher_node(state: AgentState) -> dict: """Узел-исследователь: анализирует запрос и генерирует заметки.""" last_msg = state["messages"][-1] prompt = f"Исследуй тему: {last_msg}. Дай 3 ключевых факта." response = llm.invoke(prompt) return { "research_notes": response.content, "messages": [response], "iterations": state.get("iterations", 0) + 1 } def writer_node(state: AgentState) -> dict: """Узел-райтер: пишет ответ на основе research_notes.""" prompt = f"Напиши краткий ответ на русском на основе заметок: {state['research_notes']}" response = llm.invoke(prompt) return {"messages": [response]}
Router — ключевой элемент LangGraph. Это функция, которая возвращает строку-имя следующего узла. Через неё реализуются циклы (researcher → writer → researcher → writer → ...), ветвления (если нужен поиск → tools, иначе → writer) и условия остановки (достигнут лимит итераций → END).
def router(state: AgentState) -> Literal["researcher", "writer", "tools", "END"]: """Определяем следующий шаг на основе состояния.""" iters = state.get("iterations", 0) if iters >= 3: # Лимит итераций — стоп return "END" if state.get("need_search"): # Нужен поиск → tools return "tools" if not state.get("research_notes"): # Нет заметок → исследовать return "researcher" return "writer" # Есть заметки → писать ответ
💡 Совет: Router с Literal-аннотацией даёт полную type-safety. LangGraph проверит, что все возвращаемые строки соответствуют реальным узлам графа — опечатка в имени узла = ошибка при компиляции графа, а не в runtime.
Финальный шаг: создаём StateGraph, добавляем узлы, настраиваем рёбра. add_conditional_edges подключает Router. compile() проверяет граф и возвращает готовый к вызову объект.
# Создаём граф builder = StateGraph(AgentState) # Добавляем узлы builder.add_node("researcher", researcher_node) builder.add_node("writer", writer_node) # Точка входа builder.set_entry_point("researcher") # Conditional edges: router определяет следующий шаг builder.add_conditional_edges("researcher", router, { "writer": "writer", "researcher": "researcher", "END": END }) # Обычное ребро (без условий) builder.add_edge("writer", END) # Компилируем и запускаем graph = builder.compile() result = graph.invoke({ "messages": ["Объясни принцип attention в трансформерах"], "iterations": 0 }) print(result["messages"][-1].content)
Для продакшена критично персистентное состояние. MemorySaver сохраняет состояние после каждого шага. Это даёт: возобновление прерванных цепочек, Human-in-the-Loop (пауза → человек проверяет → продолжить), отладку (посмотреть состояние на любом шаге), streaming промежуточных результатов.
# MemorySaver — персистентность между вызовами memory = MemorySaver() graph = builder.compile(checkpointer=memory) # Запуск с thread_id — состояние сохраняется между вызовами config = {"configurable": {"thread_id": "user-001"}} graph.invoke(initial_state, config) # Human-in-the-Loop: прервать перед опасным действием graph.invoke(state, config, interrupt_before=["tools"]) # Возобновить после проверки человеком graph.invoke(None, config) # продолжит с прерванного шага
Начните с туториала выше — скопируйте код и запустите за 5 минут.
📖 Официальные туториалы LangGraph