Интеграция LangGraph для графовых AI-агентов

Интеграция LangGraph для графовых AI-агентов

LangGraph — библиотека на базе LangChain для построения агентов и мульти-агентных систем в виде направленных графов со состоянием. В отличие от линейных цепочек LCEL, граф позволяет реализовывать циклы, условные переходы, параллельное выполнение и human-in-the-loop паузы. Это делает LangGraph основным инструментом для production-grade агентных систем.

Базовая структура графа

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"[Результат инструмента получен]")

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 часа юриста.

Граф:

1. extract_node — парсинг PDF, извлечение структуры
2. classify_node — тип договора (поставка, услуги, аренда, NDA)
3. risk_check_node — параллельные проверки: финансовые условия, срок, ответственность
4. legal_rules_node — проверка против корпоративного списка запрещённых условий
5. human_review — interrupt для договоров с risk_score > 7
6. finalize_node — генерация заключения и рекомендаций

app = graph.compile(
    checkpointer=PostgresSaver(conn),
    interrupt_before=["human_review"],  # Пауза только для рискованных
)

Маршрутизация: низкий риск → автоматическое одобрение; высокий риск → пауза для юриста с готовым заключением агента.

Результаты:

• Время проверки стандартного договора: 90 мин → 8 мин
• Автоматическое одобрение без юриста: 61% договоров
• Пропущенных нестандартных условий: 0 (vs ~3% вручную при усталости)
• Нагрузка на юридический отдел: -58%

LangGraph vs LangChain LCEL

Сроки

• Базовый ReAct-агент на LangGraph: 3–5 дней
• Мульти-агентная система с supervisor: 2–3 недели
• Human-in-the-loop workflow с persistence: 1–2 недели
• Интеграция в production с PostgreSQL checkpoint: +3–5 дней