Создание мульти-агентов и сложных AI-воркфлоу на Python. LangGraph от LangChain — самый мощный фреймворк для создания stateful AI-агентов. Разбираем архитектуру, графы состояний, ноды, рёбра, чекпоинты, интеграцию с LLM и продакшен-деплой.
▸ АРХИТЕКТУРА LANGGRAPH: ГРАФ СОСТОЯНИЙ АГЕНТА
LangGraph — это библиотека от команды LangChain, построенная поверх знакомой концепции графов состояний. Если LangChain даёт нам цепочки (chains) — линейные последовательности вызовов, то LangGraph открывает мир циклических, ветвящихся и само-корректирующихся AI-воркфлоу. Представьте агента, который не просто «спросил-ответил», а способен: проанализировать запрос, вызвать несколько инструментов, проверить результаты, вернуться и переспросить LLM, сохранить промежуточное состояние и продолжить с того же места после сбоя.
Ключевое отличие LangGraph от обычных цепочек — персистентное состояние (state). Каждый узел графа читает состояние, модифицирует его и передаёт дальше. Это позволяет реализовать паттерны вроде ReAct, Plan-and-Execute, Multi-Agent Supervisor и даже полностью автономных исследовательских агентов. В 2026 году LangGraph стал стандартом де-факто для продакшен-агентов на Python, обогнав CrewAI и AutoGen по гибкости и надёжности.
Архитектурно LangGraph состоит из четырёх ключевых компонентов: StateGraph (контейнер графа с типизированным состоянием), Nodes (Python-функции или runnable-объекты), Edges (обычные и условные переходы между узлами) и Checkpointer (механизм сохранения состояния после каждого шага). Именно чекпоинтер делает возможными такие продвинутые фичи как human-in-the-loop, тайм-тревел (откат к предыдущему состоянию) и стриминг промежуточных результатов.
Установка LangGraph тривиальна — один pip-пакет, который тянет за собой langchain-core и все необходимые зависимости. Для продакшена вам также понадобится langgraph-checkpoint-postgres или langgraph-checkpoint-sqlite для персистентности. Обратите внимание: LangGraph не зависит от конкретного LLM-провайдера — вы можете использовать OpenAI, Anthropic, Groq, локальные модели через Ollama или vLLM.
# Установка ядра LangGraph pip install langgraph langgraph-checkpoint-sqlite # Для работы с OpenAI pip install langchain-openai # Для Anthropic Claude pip install langchain-anthropic # Продакшен-чеекпоинтер (PostgreSQL) pip install langgraph-checkpoint-postgres psycopg[binary]
Ключевые концепции, которые нужно понять перед началом работы:
State (Состояние) — это словарь или Pydantic-модель, которая передаётся между узлами графа. Каждый узел получает текущее состояние и возвращает его обновлённую версию. Именно состояние делает агента «помнящим».
Node (Узел) — функция с сигнатурой (state: State) -> dict. Узел читает состояние, выполняет логику (вызов LLM, обращение к API, поиск в базе) и возвращает словарь с обновлёнными полями состояния. Возвращённый словарь мерджится с текущим состоянием.
Edge (Ребро) — определяет, куда переходит управление после выполнения узла. Обычное ребро ведёт строго к следующему узлу. Условное ребро (conditional edge) принимает решение на основе текущего состояния — например, продолжать цикл или завершить работу.
Checkpointer (Чекпоинтер) — сохраняет полное состояние после каждого шага (супер-шага). Благодаря этому граф можно прервать, откатить, продолжить с любого места, а также реализовать асинхронное взаимодействие с человеком.
Начнём с простейшего графа — агента, который получает сообщение, обрабатывает его и возвращает ответ. Это классический паттерн «узел-узел» с одним переходом. Состояние определим через TypedDict, узлы — как обычные Python-функции.
from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END # 1. Определяем состояние графа class AgentState(TypedDict): messages: list[str] # Список сообщений next_step: str # Куда идти дальше iteration_count: int # Счётчик итераций # 2. Создаём узлы графа def process_input(state: AgentState) -> dict: """Первый узел: получает ввод пользователя.""" user_msg = state["messages"][-1] print(f"[PROCESS] Получено: {user_msg}") return { "next_step": "respond", "iteration_count": state.get("iteration_count", 0) + 1 } def generate_response(state: AgentState) -> dict: """Второй узел: генерирует ответ.""" user_msg = state["messages"][-1] response = f"Ответ на: '{user_msg}' (итерация {state['iteration_count']})" new_messages = state["messages"] + [response] return {"messages": new_messages} # 3. Собираем граф builder = StateGraph(AgentState) builder.add_node("process", process_input) builder.add_node("respond", generate_response) # 4. Определяем рёбра builder.add_edge(START, "process") # Вход → process builder.add_edge("process", "respond") # process → respond builder.add_edge("respond", END) # respond → выход # 5. Компилируем и запускаем graph = builder.compile() result = graph.invoke({"messages": ["Привет, LangGraph!"]}) print(result["messages"]) # Вывод: # ['Привет, LangGraph!', "Ответ на: 'Привет, LangGraph!' (итерация 1)"]
Разберём ключевые моменты. add_node() регистрирует функцию как узел графа — первый аргумент это строковый идентификатор, второй — сама функция. add_edge() создаёт направленный переход: от узла-источника к узлу-назначения. Специальные константы START и END обозначают входную и выходную точки графа. После компиляции через .compile() граф становится вызываемым объектом — передаёте начальное состояние и получаете финальное.
Настоящая мощь LangGraph раскрывается при интеграции с LLM и инструментами. В этом примере мы создадим агента, который может искать информацию в интернете (через Tavily), выполнять Python-код и работать с калькулятором. Ключевой паттерн — ReAct (Reasoning + Acting): LLM думает, выбирает инструмент, получает результат, думает снова и либо вызывает ещё один инструмент, либо выдаёт финальный ответ.
from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import ToolNode from langgraph.graph import StateGraph, MessagesState # Инструмент 1: Поисковой агент (Tavily) from langchain_community.tools.tavily_search import TavilySearchResults tavily_tool = TavilySearchResults(max_results=3) # Инструмент 2: Выполнение Python-кода @tool def python_repl(code: str) -> str: """Выполняет Python-код и возвращает результат.""" try: result = eval(code, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"Ошибка: {e}" # Инструмент 3: Калькулятор @tool def calculator(expression: str) -> str: """Вычисляет математическое выражение.""" return str(eval(expression)) # Собираем все инструменты tools = [tavily_tool, python_repl, calculator] # Создаём LLM с привязкой инструментов llm = ChatOpenAI( model="gpt-4o", temperature=0 ).bind_tools(tools) # Определяем routing-функцию: куда идти после LLM? def should_continue(state: MessagesState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" # Если LLM хочет вызвать инструмент return END # Иначе завершаем # Собираем граф с LLM и инструментами workflow = StateGraph(MessagesState) workflow.add_node("agent", llm) workflow.add_node("tools", ToolNode(tools)) workflow.add_edge(START, "agent") workflow.add_conditional_edges("agent", should_continue) workflow.add_edge("tools", "agent") # Цикл: tools → agent app = workflow.compile()
Ключевой момент здесь — add_conditional_edges(). В отличие от обычного ребра, условное ребро вызывает routing-функцию, которая на основе состояния решает, к какому узлу перейти. В нашем случае: если LLM вернул tool_calls — идём к инструментам; если нет — завершаем работу. Обратите внимание на цикл tools → agent: после выполнения инструмента результат возвращается агенту, который может вызвать ещё один инструмент или выдать финальный ответ. Это и есть суть ReAct-паттерна.
Условные переходы — сердце гибкости LangGraph. Они позволяют создавать сложные деревья решений, где агент сам выбирает путь на основе контекста. Рассмотрим продвинутый пример: мульти-агентный супервайзер, который распределяет задачи между специализированными под-агентами (исследователь, кодер, аналитик) и агрегирует результаты.
from typing import Literal from langgraph.graph import StateGraph, END from typing_extensions import TypedDict class SupervisorState(TypedDict): task: str # Исходная задача research_result: str # Результат исследователя code_result: str # Результат кодера analysis_result: str # Результат аналитика next_agent: str # Кого вызвать следующим final_report: str # Финальный отчёт # Супервайзер: принимает решение на основе состояния def supervisor_router(state: SupervisorState) -> Literal["researcher", "coder", "analyst", "aggregator"]: task = state["task"].lower() # Логика роутинга на основе содержания задачи if "исследова" in task or "найди" in task: if not state.get("research_result"): return "researcher" elif "код" in task or "программ" in task: if not state.get("code_result"): return "coder" elif "анализ" in task or "данные" in task: if not state.get("analysis_result"): return "analyst" # Все подзадачи выполнены — агрегируем return "aggregator" # Сборка графа с условными переходами builder = StateGraph(SupervisorState) builder.add_node("supervisor", supervisor_router) builder.add_node("researcher", research_agent) builder.add_node("coder", coding_agent) builder.add_node("analyst", analysis_agent) builder.add_node("aggregator", aggregate_results) # Условные рёбра: супервайзер → нужный агент → обратно к супервайзеру builder.add_conditional_edges( "supervisor", supervisor_router, { "researcher": "researcher", "coder": "coder", "analyst": "analyst", "aggregator": "aggregator", } ) # Каждый под-агент возвращается к супервайзеру for agent in ["researcher", "coder", "analyst"]: builder.add_edge(agent, "supervisor") builder.add_edge("aggregator", END) builder.add_edge(START, "supervisor") graph = builder.compile()
Обратите внимание на паттерн «звезда» (hub-and-spoke): супервайзер в центре, под-агенты по краям. Каждый под-агент выполняет свою работу и возвращает управление супервайзеру, который решает, что делать дальше. Такой подход масштабируется на десятки агентов и позволяет реализовать сложные многоэтапные воркфлоу. Важно: LangGraph гарантирует, что состояние корректно передаётся при каждом переходе, даже в глубоко вложенных циклах.
Одна из самых мощных фич LangGraph — встроенная поддержка персистентности через чекпоинтеры. После каждого «супер-шага» (super-step) граф сохраняет полное состояние. Это открывает три критических возможности: (1) human-in-the-loop — прервать выполнение, показать результат человеку и продолжить после одобрения; (2) time-travel — откатиться к любому предыдущему состоянию и пойти по другому пути; (3) fault-tolerance — при сбое перезапустить граф с последнего чекпоинта.
from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command # 1. Создаём чекпоинтер (в памяти — для разработки) checkpointer = MemorySaver() # Для продакшена используйте PostgreSQL: # from langgraph.checkpoint.postgres import PostgresSaver # checkpointer = PostgresSaver.from_conn_string( # "postgresql://user:pass@localhost:5432/langgraph" # ) # 2. Компилируем граф с чекпоинтером app = workflow.compile(checkpointer=checkpointer) # 3. Конфигурация с thread_id (изолирует сессии) config = {"configurable": {"thread_id": "user-session-42"}} # 4. Запускаем граф — состояние сохраняется после каждого шага result = app.invoke( {"messages": ["Найди последние новости про AI"]}, config=config ) # 5. Получаем историю состояний (time-travel!) states = list(app.get_state_history(config)) print(f"Всего сохранённых состояний: {len(states)}") # Вывод: Всего сохранённых состояний: 5 # 6. Human-in-the-loop: прерываем и ждём одобрения def sensitive_operation(state): """Узел, требующий подтверждения человека.""" # interrupt() приостанавливает граф и ждёт внешнего ввода approved = interrupt({ "question": "Подтвердите отправку email?", "details": state["email_body"] }) if approved == "yes": send_email(state["email_body"]) return {"email_sent": approved == "yes"} # Возобновление с ответом человека: # app.invoke(Command(resume="yes"), config=config) # 7. Time-travel: откат к предыдущему состоянию previous_state = states[2] # Берём третье с конца состояние # Перезапускаем с этого состояния, но с другим вводом result_v2 = app.invoke( Command(resume="Попробуй другой подход"), config={**config, "configurable": { "thread_id": "user-session-42", "checkpoint_id": previous_state.config["configurable"]["checkpoint_id"] }} )
Функция interrupt() — это магия human-in-the-loop. Когда граф доходит до узла с interrupt(), он приостанавливается, сохраняет состояние и ждёт внешнего Command(resume=...). Это позволяет встроить человека в любой этап воркфлоу: одобрение платежа, проверка сгенерированного кода, модерация контента. В сочетании с get_state_history() вы получаете полный аудит-трейл всех действий агента — бесценно для compliance и отладки.
LangGraph Cloud (ранее LangGraph Platform) — это управляемый сервис от LangChain для деплоя графов в продакшен. Он предоставляет: автоматическое масштабирование, стриминг через Server-Sent Events, встроенную очередь задач, мониторинг и дашборд для отладки. Но вы также можете развернуть LangGraph самостоятельно через FastAPI — рассмотрим оба подхода.
# === langgraph.json — конфигурация для LangGraph Cloud === # Этот файл должен лежать в корне вашего проекта { "dependencies": ["."], "graphs": { "agent": "./src/graph.py:app" }, "env": ".env" } # === Самостоятельный деплой через FastAPI (server.py) === from fastapi import FastAPI from fastapi.responses import StreamingResponse from langgraph.checkpoint.postgres import PostgresSaver from langgraph.graph import StateGraph import json app = FastAPI(title="LangGraph Agent API") # Инициализация при старте async def lifespan(app: FastAPI): # Продакшен-чеекпоинтер с пулом соединений async with PostgresSaver.from_conn_string( "postgresql://user:pass@pg:5432/langgraph" ) as checkpointer: # Компилируем граф один раз app.state.graph = build_agent_graph().compile( checkpointer=checkpointer ) yield # Эндпоинт: инвокация графа (нестриминговая) @app.post("/invoke") async def invoke_agent(request: dict): thread_id = request.get("thread_id", "default") config = {"configurable": {"thread_id": thread_id}} result = await app.state.graph.ainvoke( {"messages": request["messages"]}, config=config ) return {"result": result} # Эндпоинт: стриминг через Server-Sent Events @app.post("/stream") async def stream_agent(request: dict): thread_id = request.get("thread_id", "default") config = {"configurable": {"thread_id": thread_id}} async def event_stream(): async for event in app.state.graph.astream_events( {"messages": request["messages"]}, config=config, version="v2" ): # Отправляем события в реальном времени yield f"data: {json.dumps(event)}\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream" ) # Human-in-the-loop эндпоинт: возобновление @app.post("/resume") async def resume_agent(request: dict): from langgraph.types import Command thread_id = request["thread_id"] config = {"configurable": {"thread_id": thread_id}} result = await app.state.graph.ainvoke( Command(resume=request["user_input"]), config=config ) return {"result": result} # Запуск: uvicorn server:app --host 0.0.0.0 --port 8000
Для продакшен-деплоя критически важно правильно настроить чекпоинтер. MemorySaver хорош только для разработки — все состояния живут в оперативной памяти и теряются при перезапуске. Для продакшена используйте PostgresSaver или SqliteSaver. PostgreSQL даёт горизонтальное масштабирование: несколько инстансов FastAPI могут разделять одну базу, и каждый запрос с уникальным thread_id будет изолирован.
Стриминг через astream_events() с версией "v2" — это лучший способ отдавать промежуточные результаты клиенту в реальном времени. Клиент (веб-интерфейс, мобильное приложение) видит, как агент «думает»: выбор инструмента, выполнение, результат, следующий шаг. Это критически важно для UX — пользователь не ждёт 30 секунд с пустым экраном, а наблюдает за прогрессом.
LangGraph в 2026 году — это зрелый, production-ready фреймворк для создания stateful AI-агентов. Мы разобрали полный путь от первого графа до продакшен-деплоя: определение состояния через TypedDict, создание узлов и рёбер, интеграцию LLM с инструментами через паттерн ReAct, условные переходы для мульти-агентных систем, чекпоинты для human-in-the-loop и time-travel, а также деплой через LangGraph Cloud или собственный FastAPI-сервер. Ключевые выводы: (1) состояние — это сердце LangGraph, оно делает агентов «помнящими» и отказоустойчивыми; (2) условные переходы позволяют строить сложные деревья решений без spaghetti-кода; (3) interrupt() и чекпоинты стирают грань между автоматическим и ручным выполнением; (4) LangGraph масштабируется от прототипа на ноутбуке до кластера из десятков инстансов. Если вы строите AI-агентов в 2026 году — LangGraph должен быть вашим инструментом номер один.
▸ ПОЛЕЗНЫЕ ССЫЛКИ