Глубокое погружение в архитектурные паттерны LangGraph: параллельное выполнение, map-reduce цепочки, условное ветвление с памятью и human-in-the-loop. Эти приёмы превращают наивного агента в надёжную production-систему.
Архитектура: полный граф продвинутого LangGraph-агента с параллельным выполнением, памятью и human-in-the-loop
Базовый агент LangGraph часто пишут на коленке: одна нода LLM, одна нода инструментов, цикл. Но в production нужна типизированная модель состояния и умный роутер, который сам решает, куда направить поток — к агенту, к инструментам или сразу к финальному ответу. Определяем AgentState через TypedDict и базовый граф с условным рёбром.
from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_step: str tool_calls_count: int memory_context: str def router(state: AgentState) -> Literal["agent", "tools", "end"]: last_msg = state["messages"][-1] if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: return "tools" if state["tool_calls_count"] > 5: return "end" # Защита от зацикливания return "end"
Один из мощнейших паттернов — параллельная обработка подзадач. Когда пользователь спрашивает «сравни цены на трёх сайтах», агент должен одновременно отправить три запроса, дождаться всех и агрегировать. В LangGraph это достигается через Send API: граф динамически порождает параллельные «лучи» (fan-out), а затем собирает их в один узел (fan-in).
from langgraph.types import Send import operator class ParallelState(TypedDict): sources: list[str] results: Annotated[list, operator.add] # Аккумулятор final_summary: str def fan_out(state: ParallelState): # Создаём параллельные задания return [Send("fetch_source", {"source": src}) for src in state["sources"]] async def fetch_source(state): src = state["source"] # Параллельный HTTP-запрос к каждому источнику import httpx async with httpx.AsyncClient() as client: resp = await client.get(f"https://api.example.com/{src}") return {"results": [resp.json()]} def fan_in(state: ParallelState): combined = "\n".join(str(r) for r in state["results"]) state["final_summary"] = llm.invoke(f"Суммируй:\n{combined}") return state
В реальных системах агент не должен самовольно отправлять email на 10 000 адресов или удалять базу данных. Паттерн human-in-the-loop добавляет точки прерывания, где граф останавливается и ждёт одобрения человека. LangGraph поддерживает это нативно через interrupt() и Command API.
from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) def sensitive_action(state: AgentState): # Точка прерывания — ждём одобрения человека decision = interrupt({ "action": "send_bulk_email", "recipients": state["recipients"], "subject": state["email_subject"] }) if decision.get("approved"): send_emails(state["recipients"]) return state # Продолжение после одобрения человеком: # graph.invoke(Command(resume={"approved": True}), config)
LangGraph из коробки даёт кратковременную память через checkpoint (состояние одного треда). Но для кросс-сессионной памяти — чтобы агент помнил пользователя спустя неделю — нужен внешний Store. Паттерн: пишем важные факты в векторную базу, а при старте нового диалога инжектим релевантный контекст в system prompt.
from langgraph.store.memory import InMemoryStore from langgraph.store.base import BaseStore store = InMemoryStore() # В production: PostgresStore или RedisStore def remember_facts(state: AgentState, *, store: BaseStore): # Извлекаем факты через LLM и сохраняем в долгую память facts = extract_facts(state["messages"]) user_id = state.get("user_id") for fact in facts: store.put( ("user_facts", user_id, fact["key"]), {"value": fact["value"], "timestamp": now()} ) def recall_context(state: AgentState, *, store: BaseStore): user_id = state.get("user_id") items = store.search(("user_facts", user_id)) context = "\n".join(item.value["value"] for item in items) return {"memory_context": context}
Сложные агенты быстро превращаются в спагетти-граф на 50 нод. Решение — разбиение на подграфы (subgraphs). Каждый подграф инкапсулирует свою логику, состояние и скрывает детали от родительского графа. LangGraph позволяет вкладывать графы на любую глубину.
# Подграф: специализированный RAG-модуль rag_graph = StateGraph(RAGState) rag_graph.add_node("retrieve", retrieve_docs) rag_graph.add_node("generate", generate_answer) rag_graph.add_edge("retrieve", "generate") rag_graph.set_entry_point("retrieve") rag_graph.set_finish_point("generate") rag_compiled = rag_graph.compile() # Основной граф — использует подграф как ноду main_graph = StateGraph(MainState) main_graph.add_node("rag_module", rag_compiled) # Подграф как нода! main_graph.add_node("agent", agent_node) main_graph.add_conditional_edges( "agent", router, {"rag": "rag_module", "end": END} ) main_graph.set_entry_point("agent") app = main_graph.compile()
Пользователи не должны ждать 30 секунд, пока агент молча думает. LangGraph поддерживает несколько режимов стриминга: stream_mode="values" (полное состояние), stream_mode="updates" (дельты), и stream_mode="messages" (токены LLM в реальном времени). Комбинируя их, получаем живой UI.
async for mode, chunk in app.astream( {"messages": [HumanMessage(content="Сравни iPhone 16 и Galaxy S25")]}, stream_mode=["messages", "updates"], config={"configurable": {"thread_id": "user-42"}} ): if mode == "messages": token, metadata = chunk if metadata["langgraph_node"] == "agent": print(token.content, end="", flush=True) # Живой вывод токенов elif mode == "updates": print(f"\n[Node: {list(chunk.keys())[0]}]")
Мы разобрали пять продвинутых паттернов LangGraph, которые превращают прототип в надёжную production-систему: умная маршрутизация с TypedDict-состоянием, параллельное выполнение через Send (fan-out/fan-in), human-in-the-loop с interrupt, долговременная память через Store API и композиция подграфов. С этими инструментами вы можете строить агентов любой сложности. Дальше — интеграция с LangSmith для трассировки и деплой через LangGraph Cloud.