При разработке сложных AI-агентов мы часто сталкиваемся с ситуацией, когда линейные цепочки LangChain Expression Language (LCEL) перестают быть эффективными. Когда требуется цикличная обработка, условные переходы в зависимости от результатов промежуточных шагов или прерывание для подтверждения человеком, LCEL не подходит. Именно для таких сценариев существует LangGraph — библиотека, расширяющая LangChain и позволяющая строить агентов в виде направленных графов с явным состоянием. Наш опыт внедрения LangGraph в production включает проекты для финтеха, логистики и юридических сервисов. Мы гарантируем отказоустойчивую архитектуру и полную документацию.
LangGraph реализует концепцию StateGraph, где каждый узел — это функция, а ребра — переходы. Состояние агента описывается TypedDict и может автоматически объединять сообщения. Это позволяет легко реализовывать циклы и мульти-агентные системы.
Почему для агентов нужен граф, а не цепочка?
Линейные цепочки LCEL хороши для простых пайплайнов: взяли вход, применили последовательность шагов, получили выход. Но в реальных агентных сценариях часто требуется вернуться к предыдущему шагу, запустить параллельные проверки или приостановить выполнение для ручного контроля. Графовая модель LangGraph решает эти проблемы естественным образом: циклы — это ребра, ведущие назад; параллелизм — несколько узлов, выполняющихся одновременно; human-in-the-loop — встроенные прерывания.
Базовая структура графа
from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # Автоматически конкатенируются user_id: str iteration_count: int llm = ChatOpenAI(model="gpt-4o") def agent_node(state: AgentState) -> AgentState: response = llm.bind_tools(tools).invoke(state["messages"]) return {"messages": [response], "iteration_count": state["iteration_count"] + 1} def should_continue(state: AgentState) -> str: last_msg = state["messages"][-1] if last_msg.tool_calls: return "tools" return END # Сборка графа graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) graph.add_edge("tools", "agent") # Цикл: после инструментов — снова к агенту app = graph.compile(checkpointer=MemorySaver()) Персистентное состояние и прерывания
LangGraph поддерживает checkpoint-сохранение состояния между запусками и паузы для human approval:
from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection # Persistence в PostgreSQL conn = Connection.connect("postgresql://user:pass@localhost/langgraph_db") checkpointer = PostgresSaver(conn) # Interrupt: граф останавливается перед указанной нодой app = graph.compile( checkpointer=checkpointer, interrupt_before=["execute_payment"], # Требует человеческого подтверждения ) config = {"configurable": {"thread_id": "order_12345"}} # Запуск до точки прерывания result = app.invoke({"messages": [HumanMessage("Оплати счёт на $450–650")]}, config) # Граф остановился перед execute_payment # После проверки человеком — продолжение app.invoke(None, config) # None = продолжить с текущего состояния Multi-agent: Supervisor pattern
from langgraph.graph import StateGraph, END from typing import Literal class SupervisorState(TypedDict): messages: Annotated[list, operator.add] next_agent: str AGENTS = ["researcher", "analyst", "writer"] supervisor_prompt = f"""Ты — супервайзер мульти-агентной системы. На основе запроса и текущего прогресса выбери следующего агента: {AGENTS} Или верни FINISH если задача выполнена. """ def supervisor_node(state: SupervisorState): response = llm.with_structured_output( {"next": {"type": "string", "enum": AGENTS + ["FINISH"]}} ).invoke([{"role": "system", "content": supervisor_prompt}] + state["messages"]) return {"next_agent": response["next"]} def route_to_agent(state: SupervisorState) -> str: if state["next_agent"] == "FINISH": return END return state["next_agent"] # Создаём агентов def make_agent_node(name: str, system_prompt: str): agent_llm = ChatOpenAI(model="gpt-4o").bind_tools(get_tools_for(name)) def node(state): result = agent_llm.invoke( [{"role": "system", "content": system_prompt}] + state["messages"] ) return {"messages": [result]} return node graph = StateGraph(SupervisorState) graph.add_node("supervisor", supervisor_node) graph.add_node("researcher", make_agent_node("researcher", "Исследуй тему и найди факты")) graph.add_node("analyst", make_agent_node("analyst", "Анализируй данные и делай выводы")) graph.add_node("writer", make_agent_node("writer", "Формулируй финальный ответ")) graph.set_entry_point("supervisor") graph.add_conditional_edges("supervisor", route_to_agent) for agent in AGENTS: graph.add_edge(agent, "supervisor") multi_agent = graph.compile() Streaming и потоковый вывод
# Стриминг событий из графа async for event in app.astream_events( {"messages": [HumanMessage("Проанализируй продажи за Q1")]}, config={"configurable": {"thread_id": "analysis_001"}}, version="v2", ): kind = event["event"] if kind == "on_chat_model_stream": print(event["data"]["chunk"].content, end="", flush=True) elif kind == "on_tool_start": print(f"\n[Вызов инструмента: {event['name']}]") elif kind == "on_tool_end": print(f"[Результат инструмента получен]") Как вложенные графы упрощают модульность?
LangGraph поддерживает SubGraphs — вложенные графы. Вы можете определить граф для обработки документа, а затем включить его как обычную ноду в родительский граф. Это позволяет декомпозировать сложные системы на переиспользуемые компоненты.
# Подграф для обработки документа doc_graph = StateGraph(DocumentState) doc_graph.add_node("extract", extract_text) doc_graph.add_node("classify", classify_document) doc_graph.add_node("validate", validate_structure) # ... построение подграфа doc_subgraph = doc_graph.compile() # Включаем подграф в родительский main_graph = StateGraph(MainState) main_graph.add_node("process_document", doc_subgraph) # Подграф как нода main_graph.add_node("send_result", send_to_crm) main_graph.add_edge("process_document", "send_result") Практический кейс: система проверки контракта (из нашей практики)
Задача: один из наших клиентов — юридический отдел крупной компании — получал 30–50 договоров ежедневно. Каждый договор требовал 1–2 часа работы юриста. Мы построили графового агента на LangGraph, который автоматизировал проверку.
Граф:
- extract_node — парсинг PDF, извлечение структуры
- classify_node — тип договора (поставка, услуги, аренда, NDA)
- risk_check_node — параллельные проверки: финансовые условия, срок, ответственность
- legal_rules_node — проверка против корпоративного списка запрещённых условий
- human_review — interrupt для договоров с
risk_score > 7 - finalize_node — генерация заключения и рекомендаций
app = graph.compile( checkpointer=PostgresSaver(conn), interrupt_before=["human_review"], # Пауза только для рискованных ) Маршрутизация: низкий риск → автоматическое одобрение; высокий риск → пауза для юриста с готовым заключением агента.
Результаты:
- Время проверки стандартного договора: 90 мин → 8 мин
- Автоматическое одобрение без юриста: 61% договоров
- Пропущенных нестандартных условий: 0 (vs ~3% вручную при усталости)
- Нагрузка на юридический отдел: -58%
Сравнение LangGraph и LCEL
| Критерий | LCEL | LangGraph |
|---|---|---|
| Структура | Линейная цепочка | Произвольный граф |
| Циклы | Нет | Да |
| Состояние | Передаётся через pipe | TypedDict с merge-стратегией |
| Checkpoint | Нет | PostgreSQL, Redis, SQLite |
| Human-in-the-loop | Нет | interrupt_before/after |
| Использование | Простые пайплайны | Агенты, мульти-агенты |
Компоненты LangGraph
| Компонент | Назначение | Пример |
|---|---|---|
| StateGraph | Определяет состояние и узлы | StateGraph(AgentState) |
| Node | Функция обработки | agent_node |
| Edge | Связь между узлами | graph.add_edge("tools", "agent") |
| Conditional Edge | Условный переход | should_continue |
| Checkpointer | Сохранение состояния | MemorySaver, PostgresSaver |
| Interrupt | Прерывание для HITL | interrupt_before=["node"] |
Что входит в нашу работу по интеграции LangGraph
- Архитектурная сессия: анализ задачи, выбор топологии графа, определение точек human-in-the-loop.
- Разработка кода: реализация узлов, рёбер, состояния, интеграция с вашей LLM и инструментами.
- Настройка персистентности: PostgreSQL, Redis или SQLite для checkpoint'ов.
- Интеграция с окружением: деплой через Docker/Kubernetes, подключение мониторинга (LangSmith).
- Тестирование: юнит-тесты, нагрузочное тестирование с рекордом latency (p99).
- Документация: полное описание графа, API, руководство по эксплуатации.
- Обучение команды: воркшоп по LangGraph для ваших разработчиков.
Сроки ориентировочно
- Базовый ReAct-агент на LangGraph: от 3 до 5 дней.
- Мульти-агентная система с supervisor: от 2 до 3 недель.
- Human-in-the-loop workflow с persistence: от 1 до 2 недель.
- Интеграция в production с PostgreSQL checkpoint: +3–5 дней.
Стоимость рассчитывается индивидуально — мы оцениваем проект бесплатно за один рабочий день. Свяжитесь с нами для консультации: наши сертифицированные инженеры помогут подобрать архитектуру под вашу задачу. Закажите предварительную оценку, и мы подготовим коммерческое предложение с гарантией качества и прозрачными сроками.