Інтеграція 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-базоване збереження стану між запусками та паузи для людського схвалення:

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import Connection

# Постійність у 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 = продовжити з поточного стану

Мульти-агент: 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()

Потокова передача та трансляція подій

# Потокова передача подій з графа
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 — переривання для контрактів з risk_score > 7
6. finalize_node — генерування висновку та рекомендацій

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

Маршрутизація: низький ризик → автоматичне схвалення; високий ризик → пауза з готовим висновком агента.

Результати:

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

LangGraph vs LangChain LCEL

Графік реалізації

• Базовий ReAct-агент на LangGraph: 3–5 днів
• Мульти-агентна система з supervisor: 2–3 тижні
• Workflow human-in-the-loop з постійністю: 1–2 тижні
• Production-інтеграція з PostgreSQL checkpoint: +3–5 днів