Точки входа (@entrypoint), задачи (@task), сравнение графового и функционального API, параллельная обработка.
Функциональный API добавляет возможности LangGraph к обычному коду на Python, не заставляя строить явный граф.
Графовый API требует описать процесс как структуру: состояние, узлы, рёбра. Это окупается, когда нужны параллелизм, ветвление и визуализация. Но если у вас уже есть работающая функция на сто строк с привычными if и for, переписывание её в граф ради контрольных точек выглядит несоразмерной платой.
Функциональный API даёт те же возможности рантайма (сохранение состояния, прерывания, стриминг, восстановление после сбоя) при сохранении обычного потока управления. Оба API работают на одном движке и сочетаются в одном приложении.
@entrypoint помечает функцию как точку входа рабочего процесса. @task помечает отдельную единицу работы, результат которой сохраняется в контрольной точке.
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt
@task
def write_essay(topic: str) -> str:
"""Долгая работа: вызов модели, обращение к API, расчёт."""
return f"Эссе на тему: {topic}"
@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
essay = write_essay(topic).result()
is_approved = interrupt({"essay": essay, "action": "Одобрить?"})
return {"essay": essay, "is_approved": is_approved}Здесь видно главное преимущество. Когда человек одобряет эссе через несколько часов, процесс возобновляется, но write_essay заново не выполняется: её результат взят из контрольной точки. В графовом варианте того же эффекта пришлось бы добиваться разбиением на узлы.
@taskЗадача — это граница восстановления. Всё, что помечено @task, выполняется один раз и запоминается; код между задачами при возобновлении выполняется заново.
Отсюда правило размещения: в задачи выносят дорогое и небезопасное к повтору. Вызов модели, обращение к платёжному шлюзу, отправку письма. Простые вычисления оборачивать незачем, они дешевле, чем запись в контрольную точку.
Задача возвращает future немедленно, а значение забирают вызовом .result():
future = write_essay("LangGraph")
essay = future.result()В асинхронном коде задача ожидается через await:
@entrypoint(checkpointer=checkpointer)
async def workflow(topic: str) -> str:
return await write_essay(topic)Раз задача возвращает future сразу, несколько задач можно запустить одновременно и собрать результаты позже. Это самый простой способ распараллелить работу в функциональном стиле:
@task
def fetch_page(url: str) -> str:
return download(url)
@entrypoint(checkpointer=InMemorySaver())
def crawl_website(urls: list[str]) -> dict:
futures = [fetch_page(url) for url in urls] # запуск всех сразу
pages = [f.result() for f in futures] # ожидание результатов
return {"pages": pages}Порядок здесь важен. Если написать [fetch_page(url).result() for url in urls], каждая задача будет дождана до запуска следующей, и параллелизм исчезнет. Сначала запускаем все, потом собираем.
Точка входа принимает ровно один позиционный аргумент. Если данных несколько, их упаковывают в словарь или модель. Ограничение связано с тем, что вход сохраняется в контрольной точке и должен иметь предсказуемую форму.
Вход и выход должны быть сериализуемыми. Соединение с базой или открытый файл в возвращаемом значении сломают сохранение.
Задачи вызываются только изнутри точки входа, другой задачи или узла графа. Вызов задачи из обычной функции вне контекста LangGraph завершится ошибкой, потому что записывать результат некуда.
Точка входа может запросить дополнительные параметры, и рантайм подставит их сам.
previous содержит состояние, сохранённое предыдущим запуском на этом же потоке. Это способ накапливать данные между вызовами без явного состояния:
from typing import Any
@entrypoint(checkpointer=InMemorySaver())
def accumulate(value: int, *, previous: Any = None) -> int:
total = (previous or 0) + value
return totalКаждый вызов с одним thread_id продолжает счёт: возвращённое значение становится previous для следующего запуска.
store даёт долговременное хранилище, writer позволяет отправлять данные в поток, config открывает доступ к конфигурации запуска. Объявляются они как именованные параметры с аннотациями.
Иногда пользователю нужно вернуть одно, а в контрольную точку записать другое. Скажем, наружу отдать ответ, а внутри сохранить накопленный контекст. Для этого есть entrypoint.final:
@entrypoint(checkpointer=checkpointer)
def workflow(data: str, *, previous: str = None) -> entrypoint.final[str, str]:
result = process(data).result()
return entrypoint.final(value=result, save=f"{previous or ''}\n{result}")Параметр value уходит вызывающему коду, save попадает в контрольную точку и придёт как previous в следующий раз. Без этого механизма пришлось бы возвращать пару и разбирать её на стороне вызова.
Прерывания работают так же, как в графах: interrupt() останавливает выполнение, Command(resume=...) продолжает. Правила тоже те же, включая главное: код от начала точки входа до прерывания выполняется заново, а результаты задач берутся из контрольной точки.
from langgraph.types import Command
config = {"configurable": {"thread_id": "essay-1"}}
workflow.invoke("LangGraph", config=config) # остановится на interrupt
workflow.invoke(Command(resume=True), config=config) # продолжитЭто и объясняет, почему дорогие операции стоит оборачивать в @task: без обёртки они повторятся при каждом возобновлении.
Графовый API — это декларативное описание: узлы, рёбра, общее состояние. Функциональный — это императивный код с обычным потоком управления.
| Характеристика | Графовый API | Функциональный API |
|---|---|---|
| Управление потоком | Явные рёбра и условия | Стандартные конструкции Python |
| Состояние | Общее для всех узлов | Локальные переменные функции |
| Контрольные точки | После каждого супершага | После выполнения задач |
| Визуализация | Поддерживается | Не поддерживается |
| Порог входа | Выше | Ниже |
Потеря визуализации существеннее, чем кажется на первый взгляд. На графе видно, какие ветки существуют и куда ведут, а функциональный процесс приходится читать целиком.
Функциональный API уместен, когда логика линейна или ветвится обычными условиями, когда нужно быстро добавить контрольные точки и прерывания к существующему коду, когда команде важнее привычность, чем схема.
Графовый API оправдан, когда есть настоящий параллелизм с объединением результатов, когда процесс нужно показывать и обсуждать как схему, когда несколько агентов передают управление друг другу, когда важно вмешиваться в состояние снаружи через update_state.
Смешивать их допустимо: задачу можно вызвать из узла графа, а граф целиком запустить внутри точки входа как обычную функцию.
Вызов задачи с немедленным .result() в цикле. Параллелизм пропадает, хотя код выглядит так, будто он есть.
Дорогие операции вне @task. При каждом возобновлении после прерывания они выполняются заново.
Несколько аргументов у точки входа. Разрешён ровно один позиционный, остальное упаковывают в структуру.
Вызов задачи вне контекста LangGraph, например из обычного теста. Результат некуда сохранять, поэтому вызов падает; логику для юнит-тестов выносят в отдельную функцию без декоратора.
Далее: Мультиагентные системы