Skip to content
IRC-CodingIRC-Coding
langgraphmnogoagentnyye-sistemypythonagent-frameworktutoriallangchain

LangGraph: Многоагентные системы пошагово

Практическое введение в LangGraph. Изучите State Graphs, Nodes, Edges и координацию агентов с примерами Python.

I

IRC-Coding Team

14 min read
LangGraph: Многоагентные системы пошагово

Руководство LangGraph: Многоагентные системы шаг за шагом

LangGraph — это сейчас самый мощный фреймворк для разработки stateful приложений с несколькими агентами на базе LLM. Его создала компания LangChain. Он позволяет моделировать сложные workflow’и агентов как направленные графы с состояниями, условными переходами, циклами и параллельным исполнением.

Если ты когда-нибудь пробовал строить multi-agent workflow на простых LangChain цепочках, ты знаешь, как быстро это становится хаосом. LangGraph решает именно эту проблему: вместо того чтобы писать спагетти-код из цепочек промптов, ты определяешь чистый граф с узлами (агенты), рёбрами (переходы) и общим состоянием. Это куда больше похоже на архитектуру программного обеспечения, чем на инженерию промптов.

В этом руководстве я проведу тебя от базовых концепций к полноценной многоагентной системе с human-in-the-loop, памятью и обработкой ошибок. Все примеры кода работающие.

TL;DR — LangGraph за 90 секунд

LangGraph — это фреймворк для stateful многоагентных приложений, основанный на теории графов: агенты это узлы, переходы это рёбра, и общее состояние распределяется между всеми узлами.

---

4 ключевые концепции: State (общее состояние), Nodes (функции агентов), Edges (переходы), Conditional Edges (условные маршруты).

Главное преимущество: Ты можешь строить циклы — агент-рецензент может отправить агента-писателя переделывать, если качество недостаточно. На простых цепочках это невозможно.

Кривая обучения: Круче, чем у CrewAI, но ты получаешь fine-grained контроль над каждым шагом. Если строишь сложные workflow’и, LangGraph лучший выбор.

Конец краткого объяснения!

Что такое LangGraph и зачем он нужен

Проблема с простыми цепочками

Представь, что ты хочешь построить pipeline создания контента: один агент исследует, один пишет, один проверяет. На простых LangChain цепочках это выглядит так:

# Наивный подход — работает, но плохо
research = llm.invoke("Recherchiere Thema X")
draft = llm.invoke(f"Schreibe Artikel: {research}")
final = llm.invoke(f"Verbessere: {draft}")

Это работает для простых случаев. Но что, если:

  • Рецензент отклоняет статью и писатель должен переработать? → нужен цикл
  • Ты хочешь делить state между агентами (например все предыдущие сообщения)? → нужно управление состоянием
  • Ты хочешь маршрутизировать на основе содержимого? → нужны условные переходы
  • Ты хочешь приостановить workflow и дождаться подтверждения человека? → нужен human-in-the-loop
  • Ты хочешь запустить несколько агентов параллельно? → нужно параллельное исполнение

Ровно это решает LangGraph.

Архитектура LangGraph

LangGraph моделирует agent workflow’и как State Graphs — направленные графы, в которых:

  • Nodes (узлы) = функции агентов, которые получают состояние, обрабатывают его и возвращают
  • Edges (рёбра) = переходы между узлами, могут быть последовательными или условными
  • State = общий словарь (или Pydantic-модель), который передаётся между всеми узлами
  • Conditional Edges = функции, которые решают на основе состояния, какой узел выполнить следующим
  • Cycles = циклы, которые могут проходить через один узел несколько раз (например Writer → Reviewer → Writer)

Концепция вдохновлена engine’ами workflow’ов типа Apache Airflow или Temporal, но оптимизирована именно для LLM-based агентов.

Почему графы, а не просто цепочки?

ВозможностьПростая цепочкаLangGraph
Линейная последовательность
Условные ветвления
Циклы
Общее состояниеТрудно✅ Natively
Параллельное исполнение
Human-in-the-Loop
Checkpointing / Memory
Визуализация графа

Установка и настройка

pip install langgraph langchain-openai langchain-core

Дополнительно рекомендуется для этого руководства:

pip install langchain-anthropic  # Для Claude
pip install langgraph-checkpoint-sqlite  # Для SQLite-хранилища
pip install grandalf  # Для визуализации графа в терминале

Установка API-ключей:

export OPENAI_API_KEY="sk-..."
# Опционально для Claude:
export ANTHROPIC_API_KEY="sk-ant-..."

Важно: Никогда не hardcode API-ключи в коде. Используй переменные окружения или .env-файлы с python-dotenv.

4 ключевые концепции — Подробно

1. State — Сердце системы

State это общий объект данных, который передаётся между всеми узлами в графе. Каждый узел может читать и обновлять state. Это фундаментальное отличие от простых цепочек, где каждый шаг видит только output предыдущего.

State как TypedDict (рекомендуется для простых случаев):

from typing import TypedDict, List, Optional

class ContentState(TypedDict):
    topic: str                    # Тема для обработки
    research: str                 # Результаты исследования
    draft: str                    # Текущий черновик
    final_article: str            # Финальная статья
    messages: List[str]           # Полный вывод сообщений
    revision_count: int           # Количество переработок
    approved: bool                # Статья одобрена?
    feedback: Optional[str]       # Отзыв рецензента

Почему TypedDict? Это даёт type-safety и автодополнение IDE, сохраняя гибкость словаря. Для более сложных приложений можешь использовать Pydantic-модели:

from pydantic import BaseModel, Field

class ContentState(BaseModel):
    topic: str = Field(description="Тема для обработки")
    research: str = Field(default="", description="Результаты исследования")
    draft: str = Field(default="", description="Текущий черновик")
    revision_count: int = Field(default=0)
    approved: bool = Field(default=False)

State-Reduction — как работают обновления:

По умолчанию возврат из узла перезаписывает state. Но ты можешь определить reducer-функции для накопления полей состояния:

from typing import Annotated
from operator import add

class ContentState(TypedDict):
    # messages добавляется, не перезаписывается
    messages: Annotated[List[str], add]
    # research перезаписывается (по умолчанию)
    research: str

Если теперь два узла вернут messages, списки будут объединены вместо перезаписи. Это крайне полезно для истории сообщений.

2. Nodes — Агенты

Nodes это Python-функции, которые получают State, обрабатывают его и возвращают обновлённый State (или частичное обновление). Каждый Node представляет агента или модуль обработки.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def researcher(state: ContentState) -> dict:
    """Агент исследования: собирает информацию по теме."""
    topic = state["topic"]
    
    prompt = f"""Ты эксперт по исследованиям. Собери ключевую информацию 
    по теме: {topic}.
    
    Структурируй ответ следующим образом:
    1. Определение и основы
    2. Современные разработки
    3. Практические примеры
    4. Распространённые заблуждения
    
    Верни только исследование, без введения."""
    
    response = llm.invoke(prompt)
    
    # Частичное обновление: возвращаем только изменённые поля
    return {
        "research": response.content,
        "messages": [f"Researcher: {response.content[:200]}..."]
    }

Ключевые моменты при работе с Nodes:

  1. Частичные обновления: не нужно возвращать весь State, только изменяемые поля. LangGraph сам объединит их с существующим State.
  2. Отсутствие побочных эффектов: Nodes должны быть чистыми функциями (вход → выход). Побочные эффекты (запись в файлы, вызовы API) нужно обрабатывать явно.
  3. Обработка ошибок: если Node выбросит исключение, весь граф остановится. Используй try/except для операций, подверженных ошибкам.
def researcher(state: ContentState) -> dict:
    try:
        response = llm.invoke(prompt)
        return {"research": response.content}
    except Exception as e:
        # Записываем ошибку в State вместо остановки графа
        return {
            "research": f"Ошибка при исследовании: {str(e)}",
            "messages": [f"Researcher ERROR: {str(e)}"]
        }

3. Edges — Переходы

Edges определяют, какой Node выполнится после текущего. Существует три типа:

Простые edges (фиксированные переходы):

# После researcher всегда идёт writer
workflow.add_edge("researcher", "writer")

Условные edges (зависимые переходы): Это самая мощная возможность LangGraph. Функция маршрутизации решает на основе State, какой Node выполнится следующим:

def route_after_review(state: ContentState) -> str:
    """Определяет, что произойдёт после проверки."""
    if state.get("approved"):
        return "publish"           # Одобрено → публикуем
    elif state.get("revision_count", 0) >= 3:
        return "human_review"      # Слишком много переделок → человек
    else:
        return "writer"            # Не одобрено → автор переделает

# Добавляем условный edge
workflow.add_conditional_edges(
    "reviewer",           # Исходный Node
    route_after_review,   # Функция маршрутизации
    {
        "publish": "publisher",
        "human_review": "human_node",
        "writer": "writer"
    }
)

Начальный Node:

workflow.set_entry_point("researcher")

4. Cycles — Циклы для итеративного совершенствования

Cycles это главное отличие LangGraph от простых цепочек. Цикл позволяет команде агентов работать итеративно:

researcher → writer → reviewer → (не подходит?) → writer → reviewer → (хорошо!) → publish

В LangGraph это реализуется через условные edges, которые ведут обратно к более ранним Nodes:

# Reviewer может отправить Writer обратно (цикл!)
workflow.add_conditional_edges(
    "reviewer",
    route_after_review,  # Может вернуть "writer"
)

# Writer всегда идёт к Reviewer
workflow.add_edge("writer", "reviewer")

Внимание — избегаем бесконечных циклов: Всегда добавляй счётчик или максимальное количество итераций:

def route_after_review(state: ContentState) -> str:
    revision_count = state.get("revision_count", 0)
    
    if state.get("approved"):
        return "publish"
    
    if revision_count >= 5:
        return "human_review"  # Аварийный выход
    
    return "writer"  # Пробуем ещё раз

Полный пример — пошагово

Сейчас создадим полноценную многоагентную систему: Researcher + Writer + Reviewer + Publisher с ручной проверкой, ограничением на переделки и памятью.

Шаг 1: Определяем State

from typing import TypedDict, List, Optional, Annotated
from operator import add

class ContentState(TypedDict):
    topic: str
    research: str
    draft: str
    final_article: str
    messages: Annotated[List[str], add]  # Добавляется, не перезаписывается
    revision_count: int
    approved: bool
    feedback: Optional[str]

Шаг 2: Определяем Nodes (агентов)

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)  # Немного креативности для автора

def researcher(state: ContentState) -> dict:
    """Агент исследования: собирает структурированную информацию."""
    topic = state["topic"]
    
    prompt = f"""Ты эксперт по исследованиям. Собери ключевую информацию 
    по теме: {topic}.
    
    Учитывай:
    - Определение и основы
    - Современные разработки (2025-2026)
    - Практические примеры и варианты использования
    - Распространённые заблуждения и подводные камни
    
    Верни только структурированное исследование."""
    
    response = llm.invoke(prompt)
    return {
        "research": response.content,
        "messages": [f"Researcher: Исследование завершено ({len(response.content)} символов)"]
    }

def writer(state: ContentState) -> dict:
    """Агент автора: пишет статью на основе исследования и отзывов."""
    research = state["research"]
    feedback = state.get("feedback", "")
    revision_count = state.get("revision_count", 0)
    
    if feedback and revision_count > 0:
        prompt = f"""Ты профессиональный автор. Переработай следующую 
        статью на основе отзыва рецензента.
        
        Исследование: {research}
        
        Текущий черновик: {state.get('draft', '')}
        
        Отзыв рецензента: {feedback}
        
        Напиши улучшенную статью. Учти все замечания из отзыва."""
    else:
        prompt = f"""Ты профессиональный автор. Напиши подробную, 
        хорошо структурированную статью по теме: {state['topic']}
        
        Опираясь на это исследование: {research}
        
        Статья должна:
        - Иметь чёткое введение
        - Использовать подзаголовки
        - Содержать практические примеры
        - Завершаться выводами"""
    
    response = llm.invoke(prompt)
    return {
        "draft": response.content,
        "revision_count": revision_count + 1,
        "messages": [f"Writer: Черновик v{revision_count + 1} готов"]
    }

def reviewer(state: ContentState) -> dict:
    """Агент рецензента: проверяет качество и даёт отзывы."""
    draft = state["draft"]
    revision_count = state.get("revision_count", 0)
    
    prompt = f"""Ты требовательный рецензент. Оцени следующую статью 
    по этим критериям:
    
    1. Структура и читаемость (1-10)
    2. Фактическая точность (1-10)
    3. Практическая ценность (1-10)
    4. Полнота (1-10)
    
    Статья: {draft}
    
    Если все критерии набрали минимум 7/10, ответь "APPROVED".
    Иначе ответь "REJECTED" и дай конкретные рекомендации для улучшения."""
    
    response = llm.invoke(prompt)
    content = response.content
    
    approved = "APPROVED" in content.upper()
    
    return {
        "approved": approved,
        "feedback": content if not approved else None,
        "final_article": draft if approved else "",
        "messages": [f"Reviewer: {'Одобрено' if approved else 'Отклонено (Переделка ' + str(revision_count) + ')'}"]
    }

def publisher(state: ContentState) -> dict:
    """Агент издателя: форматирует окончательную статью."""
    article = state["final_article"]
    
    prompt = f"""Отформатируй эту статью как Markdown с:
    - Заголовком как H1
    - Метаданными (автор, дата, время чтения)
    - Чистыми заголовками разделов
    
    Статья: {article}"""
    
    response = llm.invoke(prompt)
    return {
        "final_article": response.content,
        "messages": [f"Publisher: Статья опубликована"]
    }

def human_review(state: ContentState) -> dict:
    """Ручная проверка: человек решает при превышении лимита переделок."""
    print(f"\n=== РУЧНАЯ ПРОВЕРКА ===")
    print(f"Тема: {state['topic']}")
    print(f"Переделок: {state.get('revision_count', 0)}")
    print(f"Отзыв рецензента: {state.get('feedback', 'Нет')}")
    print(f"\nПоследний черновик:\n{state.get('draft', '')[:500]}...")
    
    approval = input("\nОдобрить статью? (y/n): ")
    return {
        "approved": approval.lower() == "y",
        "messages": [f"Human: {'Одобрено' if approval.lower() == 'y' else 'Отклонено'}"]
    }

Шаг 3: определение функций маршрутизации

from langgraph.graph import END

def route_after_review(state: ContentState) -> str:
    """Маршрутизация после проверки с аварийной остановкой."""
    if state.get("approved"):
        return "publisher"
    
    if state.get("revision_count", 0) >= 3:
        return "human_review"
    
    return "writer"  # Вернуться к писателю

def route_after_human(state: ContentState) -> str:
    """Маршрутизация после проверки человеком."""
    if state.get("approved"):
        return "publisher"
    return END  # Прервать, если человек отклонит

Шаг 4: сборка графа

from langgraph.graph import StateGraph

# Создаём граф
workflow = StateGraph(ContentState)

# Добавляем все узлы
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
workflow.add_node("publisher", publisher)
workflow.add_node("human_review", human_review)

# Определяем рёбра
workflow.set_entry_point("researcher")

# researcher → writer (всегда)
workflow.add_edge("researcher", "writer")

# writer → reviewer (всегда)
workflow.add_edge("writer", "reviewer")

# reviewer → условное решение (writer, publisher или human_review)
workflow.add_conditional_edges(
    "reviewer",
    route_after_review,
    {
        "writer": "writer",
        "publisher": "publisher",
        "human_review": "human_review"
    }
)

# human_review → условное решение (publisher или END)
workflow.add_conditional_edges(
    "human_review",
    route_after_human,
    {
        "publisher": "publisher",
        END: END
    }
)

# publisher → END (всегда)
workflow.add_edge("publisher", END)

# Компилируем граф
app = workflow.compile()

Шаг 5: выполнение

# Начальное состояние
initial_state = {
    "topic": "КИ-программирование: Best Practices 2026",
    "research": "",
    "draft": "",
    "final_article": "",
    "messages": [],
    "revision_count": 0,
    "approved": False,
    "feedback": None
}

# Запускаем граф
result = app.invoke(initial_state)

# Выводим результат
print("\n" + "=" * 60)
print("FINAL ARTICLE:")
print("=" * 60)
print(result["final_article"])
print("\n" + "=" * 60)
print(f"Редакторских циклов: {result['revision_count']}")
print(f"История сообщений:")
for msg in result["messages"]:
    print(f"  - {msg}")

Что происходит?

  1. Researcher собирает информацию по теме
  2. Writer пишет первый вариант на основе исследования
  3. Reviewer оценивает вариант по четырём критериям
  4. Если не одобрено: возврат к Writer с обратной связью (цикл!)
  5. После трёх неудачных попыток: Human Review (аварийная остановка)
  6. Если одобрено: Publisher оформляет финальную статью
  7. END — статья готова

Весь workflow является графом с циклом (writer ↔ reviewer) и двумя условными рёбрами (reviewer и human_review).

Продвинутые паттерны

Memory и сохранение состояния с checkpointing

LangGraph может сохранять состояние после выполнения каждого узла (checkpointing). Это даёт возможность:

  • восстанавливаться после сбоев
  • продолжать приостановленные workflows
  • отслеживать диалоги между несколькими сессиями
from langgraph.checkpoint.memory import MemorySaver
# Для сохранения данных между перезагрузками:
# from langgraph.checkpoint.sqlite import SqliteSaver

# In-Memory checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Выполняем с thread_id (важно для memory!)
result = app.invoke(
    initial_state,
    config={"configurable": {"thread_id": "article-ki-programmierung"}}
)

# Позже продолжаем (например, после одобрения человеком)
result2 = app.invoke(
    {"approved": True},  # Отправляем только обновление
    config={"configurable": {"thread_id": "article-ki-programmierung"}}
)

Как это работает: checkpointer сохраняет состояние после выполнения каждого узла. С помощью thread_id можно позже продолжить с того же места. Это особенно полезно для Human-in-the-Loop, где workflow ждёт, пока человек примет решение.

Параллельное выполнение

Можно запускать несколько узлов параллельно и затем объединять результаты:

def researcher_general(state: ContentState) -> dict:
    """Исследует общую информацию."""
    response = llm.invoke(f"Общая информация о: {state['topic']}")
    return {"research_general": response.content}

def researcher_technical(state: ContentState) -> dict:
    """Исследует технические детали."""
    response = llm.invoke(f"Технические детали о: {state['topic']}")
    return {"research_technical": response.content}

def merge_research(state: ContentState) -> dict:
    """Объединяет обе исследования."""
    combined = f"Общее:\n{state['research_general']}\n\nТехническое:\n{state['research_technical']}"
    return {"research": combined}

# В графе:
workflow.add_node("researcher_general", researcher_general)
workflow.add_node("researcher_technical", researcher_technical)
workflow.add_node("merge", merge_research)

# Оба исследователя параллельно со старта
workflow.set_entry_point("researcher_general")
workflow.add_edge("researcher_general", "researcher_technical")
# На самом деле параллельно: оба со старта, оба → merge
# LangGraph автоматически объединяет параллельные узлы

Интеграция инструментов, агенты с tool’ами

Агенты становятся мощными, когда они могут использовать инструменты. LangGraph легко интегрируется с LangChain-инструментами:

from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun

# Web-Search-Tool
search = DuckDuckGoSearchRun()

def researcher_with_tools(state: ContentState) -> dict:
    """Агент исследования с web-поиском."""
    topic = state["topic"]
    
    # Сначала ищем
    search_results = search.run(f"{topic} 2026 latest developments")
    
    # Потом передаём результаты LLM
    prompt = f"""На основе этих результатов поиска создай 
    структурированное исследование по теме {topic}:
    
    Результаты поиска: {search_results}"""
    
    response = llm.invoke(prompt)
    return {"research": response.content}

Sub-Graphs, графы внутри графов

Один LangGraph можно использовать как узел внутри другого LangGraph. Это полезно для модульной архитектуры:

# Sub-граф для исследования
research_graph = StateGraph(ContentState)
research_graph.add_node("web_search", web_searcher)
research_graph.add_node("summarize", summarizer)
research_graph.set_entry_point("web_search")
research_graph.add_edge("web_search", "summarize")
research_graph.add_edge("summarize", END)
research_app = research_graph.compile()

# Основной граф использует sub-граф как узел
main_workflow = StateGraph(ContentState)
main_workflow.add_node("research", research_app)  # Sub-граф как узел!
main_workflow.add_node("write", writer)
main_workflow.set_entry_point("research")
main_workflow.add_edge("research", "write")
main_workflow.add_edge("write", END)

Streaming, результаты в реальном времени

LangGraph поддерживает streaming, поэтому можно видеть промежуточные результаты по мере выполнения графа:

# Транслируем все обновления узлов
for event in app.stream(initial_state):
    for node_name, node_output in event.items():
        print(f"[{node_name}] → {list(node_output.keys())} updated")

# Транслируем только сообщения
for event in app.stream(
    initial_state,
    stream_mode="values"
):
    messages = event.get("messages", [])
    if messages:
        print(f"Latest: {messages[-1]}")

Визуализация графа

LangGraph позволяет визуализировать граф, что чрезвычайно полезно для отладки:

# В Jupyter Notebook:
from IPython.display import Image, display

# Рендеринг графа как изображение
display(Image(app.get_graph().draw_mermaid_png()))

# В терминале:
app.get_graph().print_ascii()

Ты получишь визуальное представление всех узлов, рёбер и условных маршрутов. При работе со сложными графами это становится незаменимым.

LangGraph vs. CrewAI — какой фреймворк для какого случая?

АспектLangGraphCrewAI
ПарадигмаState Graph (теория графов)Role-Based Agents (метафора команды)
ГибкостьОчень высокая — каждый шаг контролируетсяСредняя — абстракция скрывает детали
Кривая обученияКрутая — нужно понимать концепции графовПологая — определи агентов и запусти задачи
КонтрольЗернистый — условные рёбра, управление состояниемАбстрагирован — фреймворк решает многое сам
ОтладкаВизуализация графа, Streaming, Tracing с LangSmithЛоги, простые print-выводы
Циклы и петлиПоддерживаются нативноОграничены (max_iterations)
Memory и StateНативно (Checkpointing, State-Reduction)Ограничены (параметры Memory)
Human-in-the-LoopНативно (interrupt_before, interrupt_after)Реализуется вручную
ПрименениеСложные workflow’ы, production-системыБыстрые прототипы, простые агент-команды
СообществоЭкосистема LangChain (очень большая)Растёт, но меньше

Мои рекомендации:

  • CrewAI для прототипов и простых агент-команд (2-3 агента, линейные процессы)
  • LangGraph для production-систем, сложных workflow’ов с циклами, условиями и Human-in-the-Loop
  • Оба для разных проектов — они не исключают друг друга

Best Practices из практики

1. Держи State максимально компактным

Не всё нужно хранить в State. Локальных переменных в узлах часто достаточно. State должен содержать только данные, которые реально нужно передавать между узлами.

2. Всегда устанавливай лимиты на количество итераций

Без лимита цикл может выполняться бесконечно. Добавь счётчик и аварийный выключатель:

MAX_REVISIONS = 5

def route_after_review(state: ContentState) -> str:
    if state.get("revision_count", 0) >= MAX_REVISIONS:
        return "human_review"  # или END
    # ...

3. Включи Tracing с LangSmith

LangSmith от LangChain даёт детальное представление о каждом шаге:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

В dashboard LangSmith ты увидишь каждый LLM-call, каждый переход состояния и расчёт токенов.

4. Не упускай из виду расходы

Каждый узел делает минимум один LLM-call. Если у тебя 3 агента и 3 ревизии, это 9+ вызовов LLM. С GPT-4o это быстро может стоить 5-10$ за один проход. Используй более дешёвые модели для простых задач, например GPT-4o-mini для ревьюера.

5. Локальные модели для разработки

На этапе разработки и тестирования используй локальные модели (Ollama), чтобы сэкономить на API:

from langchain_community.chat_models import ChatOllama

llm = ChatOllama(model="llama3.1:8b", temperature=0)

Частые ошибки и решения

Проблема: “Graph doesn’t have an entry point”

Решение: Перед компиляцией вызови workflow.set_entry_point("node_name").

Проблема: “Node ‘xyz’ not found”

Решение: Каждый узел, на который ссылаются в рёбрах, должен быть добавлен через add_node до определения рёбер.

Проблема: Бесконечный цикл

Решение: Условное ребро всегда возвращает одинаковый узел. Проверь функцию маршрутизации, она должна возвращать разные значения в зависимости от State.

Проблема: State не обновляется

Решение: Узел возвращает частичное обновление, но поле использует Annotated-Reducer. Убедись, что Reducer определён корректно. При Annotated[List[str], add] узел должен возвращать список, который будет добавлен.

Ключевые концепции

  • LangGraph: фреймворк для stateful multi-agent приложений, основанный на теории графов
  • 4 ключевые концепции: State (общее состояние), Nodes (функции агентов), Edges (переходы), Conditional Edges (условные маршруты)
  • Cycles: петли позволяют итеративное улучшение (Writer ↔ Reviewer)
  • State-Reduction: Annotated-Reducer (например, add) для накопительных полей State
  • Human-in-the-Loop: пауза workflow’а для человеческих решений
  • Checkpointing: сохранение State после каждого узла, восстановление после сбоя
  • Параллельное выполнение: несколько узлов одновременно с автоматическим слиянием
  • Sub-Graphs: граф как узел в другом графе, модульная архитектура
  • Streaming: обновления в реальном времени во время выполнения графа
  • LangSmith: Tracing и отладка для LangGraph workflow’ов
  • Сравнение: LangGraph (гибкий, сложный, production-ready) vs. CrewAI (простой, быстрый, для прототипов)

FAQ

Нужны ли мне знания LangChain для LangGraph? Да, базовое понимание LangChain полезно, так как LangGraph на нём построен (одинаковая интеграция LLM, одинаковые Tools, одинаковые Runnables). Если ты только начинаешь с LangChain, сначала изучи основы.

Стоит ли LangGraph денег? Сам LangGraph — это Open Source (MIT-лицензия). Расходы на LLM (OpenAI, Anthropic) добавляются отдельно. LangSmith (Tracing) имеет бесплатный уровень, но для интенсивного использования платит.

Может ли LangGraph работать с локальными моделями? Да, через Ollama, llama.cpp или LM Studio. Это особенно рекомендуется для разработки и тестирования, чтобы сэкономить на API. Для production облачные модели обычно качественнее.

Как отладить LangGraph workflow? Три инструмента: 1) app.get_graph().draw_mermaid_png() для визуального представления графа. 2) Streaming с app.stream() для обновлений в реальном времени. 3) LangSmith для детального Tracing каждого LLM-call.

В чём разница между LangGraph и LangChain? LangChain — это общий фреймворк для LLM-приложений (Chains, Agents, Tools). LangGraph специализирован на stateful многоагентных workflow’ах с графовой структурой. LangGraph построен на LangChain, но не заменяет его.

Могу ли я использовать LangGraph в production? Да, LangGraph подходит для production. С Checkpointing (Persistence), Error Handling и LangSmith Tracing у тебя есть всё, что нужно для надёжных production-систем. LangGraph Cloud предоставляет Managed Hosting для LangGraph приложений.

Рекомендуемая литература

Keine Bücher für Kategorie "ki-agenten" gefunden.

Назад к блогу
Share:

Nächster Artikel in AI-программирование

Weiterlesen
Vector Database для AI-агентов: хранение и поиск

Похожие статьи