Functional API 개요

Functional API 개요 (Functional API overview)

Functional API를 사용하면 기존 코드를 최소한으로 바꾸면서 LangGraph의 핵심 기능(persistence, 메모리, human-in-the-loop, 스트리밍)을 애플리케이션에 추가할 수 있어요.

이 API는 분기와 제어 흐름에 표준 언어 원시형(if 문, for 루프, 함수 호출)을 사용할 수 있는 기존 코드에 이런 기능을 통합하도록 설계됐습니다. 코드를 명시적 파이프라인이나 DAG로 재구성해야 하는 많은 데이터 오케스트레이션 프레임워크와 달리, Functional API는 엄격한 실행 모델을 강제하지 않고 이런 기능을 포함시킬 수 있어요.

출처: 문서

본문

Functional API는 두 가지 핵심 빌딩 블록을 사용합니다:

@entrypoint: 함수를 워크플로우의 시작점으로 표시. 로직을 캡슐화하고 실행 흐름을 관리하며, 장기 실행 태스크와 인터럽트 처리를 포함합니다.
@task: API 호출이나 데이터 처리 단계 같은 개별 작업 단위를 나타냅니다. 엔트리포인트 안에서 비동기로 실행될 수 있어요. 태스크는 대기(await)하거나 동기적으로 해결할 수 있는 future 같은 객체를 반환합니다.

이는 상태 관리와 스트리밍이 있는 워크플로우를 만드는 최소한의 추상화를 제공합니다.

Functional API 사용법은 [Use Functional API](/oss/python/langgraph/use-functional-api)를 참고하세요.

Functional API vs. Graph API

더 선언적인(declarative) 접근을 선호하는 사용자를 위해 LangGraph의 Graph API는 그래프 패러다임으로 워크플로우를 정의하게 해줍니다. 두 API는 같은 런타임을 공유하므로 같은 애플리케이션에서 함께 쓸 수 있습니다.

주요 차이점:

제어 흐름: Functional API는 그래프 구조를 고민할 필요가 없어요. 표준 Python 구문으로 워크플로우를 정의할 수 있습니다. 이는 보통 작성해야 할 코드 양을 줄여줍니다.
단기 메모리: GraphAPI는 State를 선언해야 하고, 그래프 상태 갱신을 관리하기 위한 reducers 정의가 필요할 수 있어요. @entrypoint와 @tasks는 상태가 함수 범위에 한정되고 함수 간 공유되지 않으므로 명시적 상태 관리가 필요 없습니다.
체크포인팅: 두 API 모두 체크포인트를 생성하고 사용합니다. Graph API에서는 매 슈퍼스텝마다 새 체크포인트가 생성됩니다. Functional API에서는 태스크가 실행될 때 그 결과가 새 체크포인트를 만들지 않고 주어진 엔트리포인트와 연관된 기존 체크포인트에 저장됩니다.
시각화: Graph API는 워크플로우를 그래프로 시각화하기 쉬워서 디버깅·이해·공유에 유용합니다. Functional API는 그래프가 런타임에 동적으로 생성되므로 시각화를 지원하지 않습니다.

예시 (Example)

에세이를 쓰고 인간 검토를 요청하기 위해 인터럽트하는 간단한 애플리케이션을 보여드릴게요.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt

@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1) # A placeholder for a long-running task.
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt({
        # Any json-serializable payload provided to interrupt as argument.
        # It will be surfaced on the client side as an Interrupt when streaming data
        # from the workflow.
        "essay": essay, # The essay we want reviewed.
        # We can add any additional information that we need.
        # For example, introduce a key called "action" with some instructions.
        "action": "Please approve/reject the essay",
    })

    return {
        "essay": essay, # The essay that was generated
        "is_approved": is_approved, # Response from HIL
    }

이 워크플로우는 "cat" 주제에 대한 에세이를 쓴 뒤 인간의 리뷰를 받기 위해 멈춥니다. 리뷰가 제공될 때까지 워크플로우는 무기한 인터럽트될 수 있어요.

워크플로우가 재개되면 처음부터 다시 실행되지만, writeEssay 태스크의 결과는 이미 저장되어 있어서 태스크 결과를 다시 계산하지 않고 체크포인트에서 로드합니다.

import time

from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt


@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1)  # This is a placeholder for a long-running task.
    return f"An essay about topic: {topic}"


@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt(
        {
            # Any json-serializable payload provided to interrupt as argument.
            # It will be surfaced on the client side as an Interrupt when streaming data
            # from the workflow.
            "essay": essay,  # The essay we want reviewed.
            # We can add any additional information that we need.
            # For example, introduce a key called "action" with some instructions.
            "action": "Please approve/reject the essay",
        }
    )
    return {
        "essay": essay,  # The essay that was generated
        "is_approved": is_approved,  # Response from HIL
    }


thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}
stream = workflow.stream_events("cat", config, version="v3")
_ = stream.output
print({"write_essay": stream.interrupts[0].value["essay"]})
print({"__interrupt__": stream.interrupts})
# {'write_essay': 'An essay about topic: cat'}
# {
#   '__interrupt__': [
#     Interrupt(
#       value={
#           'essay': 'An essay about topic: cat',
#           'action': 'Please approve/reject the essay'
#       },
#       id='369d44b3d93d4a631ae583367ac6b5cc'
#     )
#   ]
# }

에세이가 작성돼 검토 준비가 됐습니다. 리뷰가 제공되면 워크플로우를 재개할 수 있어요:

# Get review from a user (e.g., via a UI)
# In this case, we're using a bool, but this can be any json-serializable value.
human_review = True

resumed_stream = workflow.stream_events(Command(resume=human_review), config, version="v3")
print(resumed_stream.output)
# {'essay': 'An essay about topic: cat', 'is_approved': True}

워크플로우가 완료됐고 리뷰가 에세이에 추가되었습니다.

엔트리포인트 (Entrypoint)

@entrypoint 데코레이터로 함수에서 워크플로우를 만들 수 있어요. 워크플로우 로직을 캡슐화하고 장기 실행 태스크와 인터럽트 처리를 포함한 실행 흐름을 관리합니다.

정의 (Definition)

엔트리포인트는 함수를 @entrypoint 데코레이터로 꾸며서 정의합니다.

함수는 단일 위치 인자를 반드시 받아야 하며, 이 인자가 워크플로우 입력 역할을 합니다. 여러 데이터를 전달해야 한다면 첫 번째 인자의 입력 타입으로 사전(dictionary)을 사용하세요.

entrypoint로 함수를 꾸미면 워크플로우 실행을 관리하는(스트리밍, 재개, 체크포인팅 처리) Pregel 인스턴스가 생성됩니다.

보통 @entrypoint 데코레이터에 체크포인터를 전달해 영속성을 활성화하고 human-in-the-loop 같은 기능을 사용하게 됩니다.

동기(Sync):

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop.
    ...
    return result

비동기(Async):

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop
    ...
    return result

**직렬화** — 엔트리포인트의 **입력과 출력**은 체크포인팅을 지원하기 위해 JSON 직렬화 가능해야 합니다. 자세한 내용은 [serialization](#serialization) 섹션을 참고하세요.

주입 가능한 파라미터 (Injectable parameters)

entrypoint를 선언할 때 런타임에 자동 주입되는 추가 파라미터를 요청할 수 있어요. 파라미터는 다음과 같습니다:

파라미터	설명
previous	주어진 스레드의 이전 `checkpoint`와 연관된 상태에 접근. short-term-memory 참고.
store	[BaseStore][langgraph.store.base.BaseStore]의 인스턴스. 장기 메모리에 유용.
writer	Async Python < 3.11 작업 시 StreamWriter 접근. functional API 스트리밍 참고.
config	런타임 구성 접근. RunnableConfig 참고.

파라미터는 적절한 이름과 타입 어노테이션으로 선언하세요.

from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import StreamWriter

in_memory_checkpointer = InMemorySaver(...)
in_memory_store = InMemoryStore(...)  # An instance of InMemoryStore for long-term memory

@entrypoint(
    checkpointer=in_memory_checkpointer,  # Specify the checkpointer
    store=in_memory_store  # Specify the store
)
def my_workflow(
    some_input: dict,  # The input (e.g., passed via `invoke`)
    *,
    previous: Any = None, # For short-term memory
    store: BaseStore,  # For long-term memory
    writer: StreamWriter,  # For streaming custom data
    config: RunnableConfig  # For accessing the configuration passed to the entrypoint
) -> ...:

실행 (Executing)

@entrypoint를 사용하면 invoke, ainvoke, stream, astream 메서드로 실행할 수 있는 Pregel 객체가 나옵니다.

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
my_workflow.invoke(some_input, config)  # Wait for the result synchronously

비동기 실행:

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
await my_workflow.ainvoke(some_input, config)  # Await result asynchronously

스트리밍:

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = my_workflow.stream_events(some_input, config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

비동기 스트리밍:

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = await my_workflow.astream_events(some_input, config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

재개 (Resuming)

interrupt 후 실행을 재개하려면 Command 원시형에 resume 값을 전달하면 됩니다.

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(Command(resume=some_resume_value), config)

비동기:

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

await my_workflow.ainvoke(Command(resume=some_resume_value), config)

오류 후 재개하기

오류 후 재개하려면 entrypoint를 None과 같은 thread id(config)로 실행하세요. 이는 기반 오류가 해결됐고 실행이 성공적으로 진행될 수 있다고 가정합니다.

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(None, config)

단기 메모리 (Short-term memory)

entrypoint를 checkpointer로 정의하면 같은 thread id의 연속 호출 사이에 정보를 체크포인트에 저장합니다.

이를 통해 previous 파라미터로 이전 호출의 상태에 접근할 수 있어요. 기본적으로 previous 파라미터는 이전 호출의 반환 값입니다.

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> int:
    previous = previous or 0
    return number + previous

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(1, config)  # 1 (previous was None)
my_workflow.invoke(2, config)  # 3 (previous was 1 from the previous invocation)

`entrypoint.final`

entrypoint.final은 엔트리포인트에서 반환할 수 있는 특수 원시형으로, 체크포인트에 저장되는 값을 엔트리포인트의 반환 값에서 분리(decouple) 할 수 있게 해줍니다.

첫 번째 값은 엔트리포인트의 반환 값이고, 두 번째 값은 체크포인트에 저장될 값입니다. 타입 어노테이션은 entrypoint.final[return_type, save_type]입니다.

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    # This will return the previous value to the caller, saving
    # 2 * number to the checkpoint, which will be used in the next invocation
    # for the `previous` parameter.
    return entrypoint.final(value=previous, save=2 * number)

config = {
    "configurable": {
        "thread_id": "1"
    }
}

my_workflow.invoke(3, config)  # 0 (previous was None)
my_workflow.invoke(1, config)  # 6 (previous was 3 * 2 from the previous invocation)

태스크 (Task)

**태스크(task)**는 API 호출이나 데이터 처리 단계 같은 개별 작업 단위를 나타냅니다. 두 가지 핵심 특징이 있어요:

비동기 실행: 태스크는 비동기로 실행되도록 설계되어, 차단 없이 여러 작업을 동시에 실행할 수 있습니다.
체크포인팅: 태스크 결과는 체크포인트에 저장되어 마지막 저장 상태에서 워크플로우를 재개할 수 있게 합니다. (persistence 참고)

정의 (Definition)

태스크는 일반 Python 함수를 감싸는 @task 데코레이터로 정의합니다.

from langgraph.func import task

@task()
def slow_computation(input_value):
    # Simulate a long-running operation
    ...
    return result

**직렬화** — 태스크의 **출력**은 체크포인팅을 지원하기 위해 JSON 직렬화 가능해야 해요.

실행 (Execution)

태스크는 엔트리포인트, 다른 태스크, 또는 상태 그래프 노드 안에서만 호출할 수 있어요.

태스크는 메인 애플리케이션 코드에서 직접 호출할 수 없습니다.

태스크를 호출하면 future 객체와 함께 즉시 반환됩니다. future는 나중에 사용할 수 있는 결과의 자리 표시자입니다.

태스크의 결과를 얻으려면 동기적으로(result() 사용) 기다리거나 비동기적으로(await 사용) 대기할 수 있어요.

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: int) -> int:
    future = slow_computation(some_input)
    return future.result()  # Wait for the result synchronously

비동기:

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: int) -> int:
    return await slow_computation(some_input)  # Await result asynchronously

태스크를 언제 쓸까 (When to use a task)

태스크는 다음 시나리오에서 유용합니다:

체크포인팅: 장기 실행 작업의 결과를 체크포인트에 저장해 워크플로우 재개 시 다시 계산하지 않도록.
Human-in-the-loop: 인간 개입이 필요한 워크플로우를 만들 때는 워크플로우가 올바르게 재개되도록 모든 무작위성(예: API 호출)을 tasks로 캡슐화해야 합니다. 자세한 내용은 determinism 섹션 참고.
병렬 실행: I/O 바운드 태스크에서 tasks는 병렬 실행을 가능하게 해 차단 없이 여러 작업을 동시에 실행합니다(예: 여러 API 호출).
관찰 가능성: 작업을 tasks로 감싸면 LangSmith로 워크플로우 진행과 개별 작업 실행을 추적할 수 있어요.
재시도 가능 작업: 실패나 불일치를 처리하기 위해 재시도가 필요한 작업에서 tasks는 재시도 로직을 캡슐화하고 관리하는 방법을 제공합니다.

직렬화 (Serialization)

LangGraph에는 직렬화에 관한 두 가지 핵심 측면이 있어요:

entrypoint 입력과 출력은 JSON 직렬화 가능해야 합니다.
task 출력은 JSON 직렬화 가능해야 합니다.

이 요구사항은 체크포인팅과 워크플로우 재개를 위해 필요합니다. 사전, 리스트, 문자열, 숫자, 불리언 같은 Python 원시형을 사용해 입력과 출력이 직렬화 가능하도록 하세요.

직렬화는 태스크 결과와 중간 값 같은 워크플로우 상태를 안정적으로 저장·복원할 수 있게 해줍니다. 이는 human-in-the-loop 상호작용, 장애 허용, 병렬 실행에 중요합니다.

직렬화 불가능한 입력이나 출력을 제공하면 워크플로우가 체크포인터로 구성됐을 때 런타임 오류가 발생합니다.

결정성 (Determinism)

워크플로우 런을 재개할 때 코드는 실행이 멈춘 같은 코드 줄에서 재개되지 않아요. 실행은 체크포인트 경계로 돌아가고, 워크플로우는 다시 일시 정지 지점에 도달할 때까지 앞으로 재생(replay) 됩니다.

Functional API에서 재생은 엔트리포인트의 시작에서 시작되며, LangGraph는 완료된 task와 subgraph 결과를 다시 계산하지 않고 체크포인터에서 복원합니다. 이는 장기 실행이거나 비결정적인 task 출력을 포함해 일시 정지 간 기록된 단계 순서를 보존합니다.

human-in-the-loop 같은 기능을 사용하려면 비결정적 작업(예: 무작위 값)과 부작용(예: 파일 쓰기, API 호출)을 tasks에 넣어야 합니다.

워크플로우의 서로 다른 실행은 다른 결과를 만들 수 있지만, 특정 스레드를 재개하면 같은 영속화된 task·subgraph 결과가 재생되어야 합니다.

워크플로우가 결정적이고 일관되게 재생될 수 있도록 다음 지침을 따르세요:

작업 반복 피하기: 엔트리포인트에서 여러 부작용(예: 로깅, 파일 쓰기, 네트워크 호출)을 연결한다면 각각에 자체 task를 줘서 재개 시 출력을 다시 실행하는 대신 체크포인터에서 복원하게 하세요.
비결정적 연산 캡슐화: 시도마다 바뀔 수 있는 값(예: 난수, 벽시계 읽기)을 tasks 안에 두어 재생이 체크포인트된 것과 맞도록 하세요.
멱등 연산 사용: 부분 태스크 실패와 재시도는 Idempotency를 참고하세요.

멱등성 (Idempotency)

멱등성은 같은 연산을 여러 번 실행해도 같은 결과가 나오도록 보장합니다. 이는 단계가 실패로 인해 다시 실행돼도 중복 API 호출과 중복 처리를 방지하는 데 도움이 돼요. API 호출은 항상 tasks 함수 안에 넣어 체크포인팅을 활성화하고, 재실행에 대비해 멱등하도록 설계하세요. 이는 데이터 쓰기를 만드는 연산에서 특히 중요합니다.

워크플로우가 재개되면 LangGraph는 완료된 task 결과를 체크포인트에서 재생합니다. 시작했지만 끝나지 않은 task는 그 재개에서 다시 실행될 수 있으므로 부작용을 멱등하게 설계하세요. 멱등성 키를 사용하거나 기존 결과를 검증해 의도하지 않은 중복을 피하세요.

흔한 함정 (Common pitfalls)

부작용 처리 (Handling side effects)

부작용(예: 파일 쓰기, 이메일 보내기)을 태스크에 캡슐화해 워크플로우 재개 시 여러 번 실행되지 않도록 하세요.

잘못된 예시 — 부작용(파일 쓰기)이 워크플로우에 직접 포함되어, 재개 시 두 번째로 실행됩니다:

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # This code will be executed a second time when resuming the workflow.
    # Which is likely not what you want.
    with open("output.txt", "w") as f:  # [!code highlight]
        f.write("Side effect executed")  # [!code highlight]
    value = interrupt("question")
    return value

올바른 예시 — 부작용이 태스크에 캡슐화되어 재개 시 일관된 실행을 보장합니다:

from langgraph.func import task

@task  # [!code highlight]
def write_to_file():  # [!code highlight]
    with open("output.txt", "w") as f:
        f.write("Side effect executed")

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # The side effect is now encapsulated in a task.
    write_to_file().result()
    value = interrupt("question")
    return value

비결정적 제어 흐름 (Non-deterministic control flow)

매번 다른 결과를 줄 수 있는 작업(현재 시간, 난수 얻기 등)은 태스크에 캡슐화해 재개 시 같은 결과가 반환되도록 해야 합니다.

태스크 안: 난수 얻기 (5) → interrupt → 재개 → (다시 5 반환) → ...
태스크 밖: 난수 얻기 (5) → interrupt → 재개 → 새 난수 얻기 (7) → ...

이는 특히 여러 interrupt 호출이 있는 human-in-the-loop 워크플로우에서 중요합니다. LangGraph는 각 task/entrypoint에 대해 resume 값의 목록을 유지합니다. 인터럽트가 발생하면 해당 resume 값과 매칭됩니다. 이 매칭은 엄격히 인덱스 기반이므로, resume 값의 순서는 인터럽트의 순서와 일치해야 합니다.

재개 시 실행 순서가 유지되지 않으면 하나의 interrupt 호출이 잘못된 resume 값과 매칭되어 잘못된 결과가 나올 수 있습니다.

자세한 내용은 determinism 섹션을 읽어보세요.

잘못된 예시 — 워크플로우가 현재 시간으로 어떤 태스크를 실행할지 결정합니다. 실행 시점에 따라 결과가 달라지므로 비결정적입니다:

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = time.time()  # [!code highlight]

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

올바른 예시 — 워크플로우가 입력 t0으로 어떤 태스크를 실행할지 결정합니다. 입력에만 의존하므로 결정적입니다:

import time

from langgraph.func import task

@task  # [!code highlight]
def get_time() -> float:  # [!code highlight]
    return time.time()

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = get_time().result()  # [!code highlight]

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

Functional API 개요