기능 API 개요 (Functional API Overview)

Functional API를 쓰면 LangGraph의 핵심 기능( persistence, memory, human-in-the-loop, streaming)을 기존 코드를 크게 바꾸지 않고도 애플리케이션에 붙일 수 있어요.

이 API는 분기와 제어 흐름에 표준 언어 기본 요소를 사용하는 기존 코드, 예를 들어 if 문, for 루프, 함수 호출 같은 코드에 이 기능들을 통합하도록 설계됐어요. 많은 데이터 오케스트레이션 프레임워크가 코드를 명시적인 파이프라인이나 DAG로 재구성하도록 요구하는 것과 달리, Functional API는 그런 경직된 실행 모델을 강요하지 않으면서도 이런 기능을 담을 수 있게 해줘요.

Functional API는 두 가지 핵심 구성 요소를 사용해요.

@entrypoint: 함수를 워크플로의 시작점으로 표시해요. 로직을 캡슐화하고 실행 흐름을 관리하는데, 오래 걸리는 작업과 인터럽트 처리까지 포함해요.
@task: API 호출이나 데이터 처리 단계 같은 하나의 원자적인 작업 단위를 표현해요. 엔트리포인트 안에서 비동기로 실행될 수 있고, task는 future와 비슷한 객체를 반환해서 await하거나 동기적으로 resolve할 수 있어요.

이렇게 해서 상태 관리와 스트리밍을 갖춘 워크플로를 만들기 위한 최소한의 추상화를 제공해요.

Functional API를 사용하는 방법은 Use Functional API 문서를 참고해 주세요.

Functional API vs. Graph API

좀 더 선언적인 방식을 선호하는 분을 위해 LangGraph의 Graph API는 Graph 패러다임으로 워크플로를 정의하게 해줘요. 두 API는 같은 기반 런타임을 공유해서, 같은 애플리케이션 안에서 함께 쓸 수 있어요.

핵심 차이점 몇 가지를 정리하면 이래요.

제어 흐름(Control flow): Functional API는 그래프 구조를 생각할 필요가 없어요. 표준 파이썬 구문으로 워크플로를 정의하면 돼요. 보통 이렇게 하면 작성해야 할 코드 양이 줄어들어요.
단기 기억(Short-term memory): GraphAPI는 State를 선언해야 하고, 그래프 상태 업데이트를 관리하려고 reducers를 정의해야 할 수도 있어요. 반면 @entrypoint와 @task는 상태가 함수 안으로 한정되고 함수 간에 공유되지 않기 때문에 명시적 상태 관리가 필요 없어요.
체크포인팅(Checkpointing): 두 API 모두 체크포인트를 만들고 사용해요. Graph API는 superstep이 끝날 때마다 새 체크포인트가 생겨요. Functional API에서는 task가 실행되면 그 결과가 주어진 엔트리포인트와 연결된 기존 체크포인트에 저장되어, 새 체크포인트를 만들지 않아요.
시각화(Visualization): Graph API는 워크플로를 그래프로 시각화하기 쉬워서 디버깅, 이해, 공유에 유용해요. Functional API는 그래프가 런타임에 동적으로 생성되기 때문에 시각화를 지원하지 않아요.

예제

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

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"이라는 주제에 대한 에세이를 작성한 뒤, 인간의 검토를 받기 위해 멈춰요. 검토가 들어올 때까지 워크플로는 무기한으로 인터럽트된 상태로 있을 수 있어요.

워크플로가 다시 이어질 때는 처음부터 실행되지만, write_essay task의 결과가 이미 저장돼 있기 때문에 task 결과는 다시 계산하는 대신 체크포인트에서 불러와요.

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 데코레이터는 함수로부터 워크플로를 만들 때 사용해요. 이 데코레이터가 워크플로 로직을 캡슐화하고 실행 흐름을 관리하는데, 오래 걸리는 작업 과 인터럽트 처리까지 포함해요.

정의

엔트리포인트는 함수에 @entrypoint 데코레이터를 붙여서 정의해요.

이 함수는 위치 인자를 하나만 받아야 해요. 그 인자가 워크플로 입력 역할을 해요. 여러 데이터를 넘겨야 한다면 첫 번째 인자의 입력 타입으로 딕셔너리를 쓰면 돼요.

함수에 entrypoint를 붙이면 Pregel 인스턴스가 생성되는데, 이 인스턴스가 워크플로 실행을 관리해 줘요. 예를 들어 스트리밍, 재개(resumption), 체크포인팅을 처리해요.

보통 지속성(persistence)을 활성화하고 human-in-the-loop 같은 기능을 쓰려면 @entrypoint 데코레이터에 checkpointer를 넘기게 돼요.

동기식(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

직렬화(Serialization)

체크포인팅을 지원하려면 엔트리포인트의 입력과 출력이 JSON으로 직렬화 가능해야 해요. 자세한 내용은 직렬화 섹션을 참고해 주세요.

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

entrypoint를 선언할 때 런타임에 자동으로 주입되는 추가 파라미터를 요청할 수 있어요. 이 파라미터들은 다음과 같아요.

파라미터	설명
previous	주어진 thread에 대한 이전 `checkpoint`와 연결된 상태에 접근해요. 단기 기억 참고.
store	BaseStore의 인스턴스예요. 장기 기억에 유용해요.
writer	3.11 미만의 Async Python에서 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를 쓰면 Pregel 객체가 생기는데, 이 객체는 invoke, ainvoke, stream, astream 메서드로 실행할 수 있어요.

Invoke

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

Async Invoke

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

Stream

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)

Async Stream

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)

인터럽트 이후 실행을 재개하려면 Command 기본 요소에 resume 값을 넘기면 돼요.

Invoke

from langgraph.types import Command

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

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

Async Invoke

from langgraph.types import Command

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

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

Stream

from langgraph.types import Command

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

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

Async Stream

from langgraph.types import Command

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

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

오류 후 재개하기

오류 이후에 재개하려면 entrypoint를 None과 같은 thread id(config)로 실행해요. 이때 기본적인 오류가 해결됐다고 가정하고, 그 뒤 실행이 성공적으로 진행될 수 있다는 전제가 깔려 있어요.

Invoke

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

my_workflow.invoke(None, config)

Async Invoke

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

await my_workflow.ainvoke(None, config)

Stream

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

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

Async Stream

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

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

단기 기억(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 호출이나 데이터 처리 단계 같은 하나의 원자적인 작업 단위를 나타내요. 두 가지 핵심 특징이 있어요.

비동기 실행: task는 비동기로 실행되도록 설계되어, 여러 작업을 블로킹 없이 동시에 실행할 수 있어요.
체크포인팅: task 결과는 체크포인트에 저장되어, 마지막으로 저장된 상태에서 워크플로를 재개할 수 있게 해줘요. (자세한 내용은 persistence 참고.)

정의

task는 일반 파이썬 함수를 감싸는 @task 데코레이터로 정의해요.

from langgraph.func import task

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

직렬화(Serialization)

체크포인팅을 지원하려면 task의 출력이 JSON으로 직렬화 가능해야 해요.

실행(Execution)

task는 entrypoint 안, 다른 task 안, 또는 state graph node 안에서만 호출할 수 있어요.

task는 메인 애플리케이션 코드에서 직접 호출할 수 없어요.

task를 호출하면 future 객체와 함께 즉시 반환돼요. future는 나중에 사용할 수 있는 결과의 자리 표시자예요. task의 결과를 얻으려면 동기적으로 기다리거나(result()), 비동기로 await하면 돼요.

동기식 호출(Synchronous Invocation)

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

비동기식 호출(Asynchronous Invocation)

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

언제 task를 써야 하나요(When to use a task)

task는 이런 상황에서 유용해요.

체크포인팅: 오래 걸리는 작업의 결과를 체크포인트에 저장해야 할 때. 워크플로를 재개할 때 다시 계산하지 않아도 되도록요.
Human-in-the-loop: 인간의 개입이 필요한 워크플로를 만들 때는 반드시 task를 써서 무작위성(예: API 호출)을 캡슐화해야 해요. 그래야 워크플로를 올바르게 재개할 수 있어요. 자세한 내용은 결정론 섹션을 참고해 주세요.
병렬 실행(Parallel Execution): I/O 바운드 작업의 경우 task는 병렬 실행을 가능하게 해서, 여러 작업(예: 여러 API 호출)을 블로킹 없이 동시에 실행할 수 있게 해줘요.
관측 가능성(Observability): 작업을 task로 감싸면 LangSmith로 워크플로 진행 상황을 추적하고 개별 작업의 실행을 모니터링할 수 있어요.
재시도 가능한 작업(Retryable Work): 실패나 불일치를 처리하기 위해 작업을 재시도해야 할 때, task는 재시도 로직을 캡슐화하고 관리하는 방법을 제공해요.

직렬화(Serialization)

LangGraph에서 직렬화에는 두 가지 핵심 측면이 있어요.

entrypoint의 입력과 출력은 JSON으로 직렬화 가능해야 해요.
task의 출력은 JSON으로 직렬화 가능해야 해요.

이 요구 사항은 체크포인팅과 워크플로 재개를 가능하게 하기 위해 필요해요. 입력과 출력을 직렬화 가능하게 만들려면 딕셔너리, 리스트, 문자열, 숫자, 불리언 같은 파이썬 기본 요소를 사용해 주세요.

직렬화 덕분에 task 결과와 중간 값 같은 워크플로 상태를 안정적으로 저장하고 복원할 수 있어요. 이것은 human-in-the-loop 상호작용, 내결함성, 병렬 실행을 가능하게 하는 데 중요해요. 워크플로가 체크포인터로 설정되어 있는데 직렬화할 수 없는 입력이나 출력을 제공하면 런타임 오류가 발생해요.

결정론(Determinism)

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

Functional API에서 리플레이는 entrypoint의 시작부터 다시 시작되는데, LangGraph는 완료된 task와 subgraph 결과를 다시 계산하는 대신 체크포인터에서 복원해요. 덕분에 오래 걸리거나 비결정적인 task 출력을 포함해서도, 멈춤을 넘나드는 기록된 단계 순서가 보존돼요.

human-in-the-loop 같은 기능을 쓰려면 비결정적 작업(예: 임의 값)과 부수 효과(예: 파일 쓰기, API 호출)를 task에 넣어야 해요.

워크플로의 서로 다른 실행은 서로 다른 결과를 낼 수 있지만, 특정 thread를 재개하면 같은 지속된 task·subgraph 결과가 리플레이되어야 해요.

워크플로가 결정적이고 일관되게 리플레이될 수 있게 하려면 이 지침을 따르세요.

반복 작업 피하기: entrypoint 안에서 여러 부수 효과(예: 로깅, 파일 쓰기, 네트워크 호출)를 연결한다면 각각에 자기 task를 주세요. 그래야 재개할 때 그 결과물을 다시 실행하는 대신 체크포인터에서 복원해요.
비결정적 작업 캡슐화하기: 시도 사이에 바뀔 수 있는 값(예: 난수나 벽시계 시간 읽기)은 task 안에 넣어서 리플레이가 체크포인트된 내용과 일치하게 해주세요.
멱등 연산 사용하기: 부분 task 실패와 재시도에 대해서는 멱등성 섹션을 참고해 주세요.

멱등성(Idempotency)

멱등성은 같은 연산을 여러 번 실행해도 같은 결과가 나오도록 보장해요. 이렇게 하면 단계가 실패 때문에 다시 실행될 때 중복 API 호출이나 불필요한 처리를 막아줘요. 항상 API 호출을 task 함수 안에 두어 체크포인팅하고, 재실행될 경우를 대비해 멱등이 되도록 설계하세요.

특히 데이터 쓰기를 유발하는 연산에서 이게 중요해요.

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

흔한 함정(Common pitfalls)

부수 효과 처리하기(Handling side effects)

파일 쓰기나 이메일 보내기 같은 부수 효과를 task에 캡슐화해서, 워크플로 재개 시 여러 번 실행되지 않도록 하세요.

잘못된 예(Incorrect)

이 예에서는 부수 효과(파일 쓰기)가 워크플로에 직접 포함되어 있어서, 워크플로를 재개할 때 두 번째로 실행돼요.

@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:
        f.write("Side effect executed")
    value = interrupt("question")
    return value

올바른 예(Correct)

이 예에서는 부수 효과가 task에 캡슐화되어, 재개 시 일관되게 실행되도록 보장해요.

from langgraph.func import task

@task
def write_to_file():
    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)

현재 시간 구하기나 난수처럼 실행할 때마다 다른 결과를 줄 수 있는 연산은 task에 캡슐화해서, 재개 시 같은 결과가 반환되도록 해야 해요.

task 안에 있을 때: 난수 5 얻기 → interrupt → resume → (5를 다시 얻음) → …
task 안에 없을 때: 난수 5 얻기 → interrupt → resume → 새 난수 7 얻기 → …

특히 human-in-the-loop 워크플로에서 interrupt 호출이 여러 개일 때 이게 중요해요. LangGraph는 각 task/entrypoint에 대해 resume 값 목록을 유지해요. interrupt가 발생하면 해당하는 resume 값과 매칭되는데, 이 매칭은 엄격하게 인덱스 기반이라서 resume 값의 순서가 interrupt의 순서와 일치해야 해요.

재개 시 실행 순서가 유지되지 않으면 하나의 interrupt 호출이 잘못된 resume 값과 매칭되어 잘못된 결과를 초래할 수 있어요.

자세한 내용은 결정론 섹션을 읽어보세요.

잘못된 예(Incorrect)

이 예에서 워크플로는 현재 시간으로 어느 task를 실행할지 결정해요. 이는 워크플로의 결과가 실행되는 시점에 달려 있기 때문에 비결정적이에요.

from langgraph.func import entrypoint

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

    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
    }

올바른 예(Correct)

이 예에서 워크플로는 입력 t0으로 어느 task를 실행할지 결정해요. 이는 워크플로의 결과가 입력에만 의존하기 때문에 결정적이에요.

import time

from langgraph.func import task

@task
def get_time() -> float:
    return time.time()

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

    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
    }

기능 API 개요 (Functional API Overview)