Точки входа (@entrypoint), задачи (@task), сравнение графового и функционального API, параллельная обработка.
Функциональный API позволяет добавить возможности LangGraph к существующему коду с минимальными изменениями — без построения явного графа.
@entrypoint отмечает функцию как точку входа рабочего процесса. @task — отдельная единица работы, результат которой сохраняется в контрольной точке.
from langgraph.func import entrypoint, task
from langgraph.types import interrupt
@task
def write_essay(topic: str) -> str:
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}Точка входа принимает ровно один позиционный аргумент. Входные и выходные данные должны быть JSON-сериализуемыми. Задачи можно вызывать только из точки входа, другой задачи или узла графа. Задачи возвращают future — вызовите .result() для получения значения.
При объявлении точки входа можно запросить previous (состояние предыдущей контрольной точки), store (хранилище для долговременной памяти), writer (для стриминга) и config (конфигурация выполнения).
Графовый API — декларативный подход с явными узлами, рёбрами и общим состоянием. Функциональный API — императивный подход со стандартными Python-конструкциями. Оба используют один runtime, их можно комбинировать.
| Характеристика | Графовый API | Функциональный API |
|---|---|---|
| Управление потоком | Явные рёбра и условия | Стандартный Python |
| Состояние | Общее для всех узлов | Инкапсулировано в функции |
| Контрольные точки | После каждого супершага | После выполнения задач |
| Визуализация | Поддерживается | Не поддерживается |
Задачи можно запускать параллельно — они возвращают future немедленно:
@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}Через entrypoint.final можно разделить возвращаемое и сохраняемое значения — возвратить пользователю одно, а в контрольную точку сохранить другое:
@entrypoint(checkpointer=checkpointer)
def workflow(data: str, *, previous: str = None) -> entrypoint.final[str, str]:
if previous:
return entrypoint.final(value=previous, save=previous)
result = process(data).result()
return entrypoint.final(value=result, save=result)Далее: Мультиагентные системы