Під час розробки складних production-grade AI-агентів ми часто стикаємося з ситуацією, коли лінійні ланцюжки LangChain Expression Language (LCEL) перестають бути ефективними. Коли потрібна циклічна обробка, умовні переходи залежно від результатів проміжних кроків або переривання для підтвердження людиною, LCEL не підходить. Саме для таких сценаріїв існує LangGraph — бібліотека, що розширює LangChain і дозволяє будувати агентів у вигляді спрямованих графів із явним станом (state machine). Наш досвід впровадження LangGraph у production включає проекти для фінтеху, логістики та юридичних сервісів. Ми гарантуємо відмовостійку архітектуру та повну документацію. На сьогодні реалізовано понад 50 успішних проектів із впровадження LangGraph.
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("Оплати рахунок на 50000 грн")]}, 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 хв (у 11.25 разів швидше)
- Автоматичне схвалення без юриста: 61% договорів
- Пропущених нестандартних умов: 0 (vs ~3% вручну при втомі)
- Навантаження на юридичний відділ: -58%
- Економія: ~$35 000 на місяць (за 40 договорів/день × 20 днів)
Порівняння 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 днів (від $5 000).
- Мульти-агентна система з supervisor: від 2 до 3 тижнів (від $15 000).
- Human-in-the-loop workflow з persistence: від 1 до 2 тижнів (від $8 000).
- Інтеграція в production з PostgreSQL checkpoint: +3–5 днів.
Вартість розраховується індивідуально — ми оцінюємо проект безкоштовно за один робочий день. Наша команда з 5+ роками досвіду в AI виконує роботи під ключ. Пишіть на пошту або замовте попередню оцінку, і ми підготуємо комерційну пропозицію з гарантією якості та прозорими термінами.