함수형 API 사용하기

함수형 API 사용하기 (Use the functional API)

기존 코드를 최소한으로 바꾸면서 LangGraph의 핵심 기능을 앱에 더하고 싶다면 **함수형 API(Functional API)**가 정답이에요. 지속성(persistence), 메모리, human-in-the-loop, 스트리밍을 지원하니까요.

출처: 공식문서

간단한 워크플로 만들기

entrypoint를 정의할 때 입력은 함수의 첫 번째 인자로 제한돼요. 여러 입력을 넘기려면 딕셔너리를 쓰면 돼요.

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    value = inputs["value"]
    another_value = inputs["another_value"]
    ...

my_workflow.invoke({"value": 1, "another_value": 2})

@task와 @entrypoint 장식자(decorator)를 쓰는 전형적인 예시예요. 체크포인터를 제공했으므로 워크플로 결과가 체크포인터에 영속화돼요.

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

# Task that checks if a number is even
@task
def is_even(number: int) -> bool:
    return number % 2 == 0

# Task that formats a message
@task
def format_message(is_even: bool) -> str:
    return "The number is even." if is_even else "The number is odd."

# Create a checkpointer for persistence
checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: dict) -> str:
    """Simple workflow to classify a number."""
    even = is_even(inputs["number"]).result()
    return format_message(even).result()

# Run the workflow with a unique thread ID
config = {"configurable": {"thread_id": str(uuid7())}}
result = workflow.invoke({"number": 7}, config=config)
print(result)

LLM을 쓰는 에세이 생성 예시도 있어요.

import uuid
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model('gpt-3.5-turbo')

# Task: generate essay using an LLM
@task
def compose_essay(topic: str) -> str:
    """Generate an essay about the given topic."""
    return model.invoke([
        {"role": "system", "content": "You are a helpful assistant that writes essays."},
        {"role": "user", "content": f"Write an essay about {topic}."}
    ]).content

# Create a checkpointer for persistence
checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(topic: str) -> str:
    """Simple workflow that generates an essay with an LLM."""
    return compose_essay(topic).result()

# Execute the workflow
config = {"configurable": {"thread_id": str(uuid7())}}
result = workflow.invoke("the history of flight", config=config)
print(result)

병렬 실행

테스크를 동시에 호출하고 결과를 기다리면 병렬로 실행할 수 있어요. LLM용 API 호출처럼 I/O 바운드 작업에서 성능을 올리는 데 유용해요.

@task
def add_one(number: int) -> int:
    return number + 1

@entrypoint(checkpointer=checkpointer)
def graph(numbers: list[int]) -> list[str]:
    futures = [add_one(i) for i in numbers]
    return [f.result() for f in futures]

여러 LLM 호출을 @task로 병렬 실행해 각기 다른 주제의 문단을 생성하고 결과를 하나의 텍스트로 합치는 예시예요. LangGraph의 동시성 모델이 LLM 완성 같은 I/O가 섞인 작업의 실행 시간을 줄여줘요.

import uuid
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model("gpt-3.5-turbo")

# Task that generates a paragraph about a given topic
@task
def generate_paragraph(topic: str) -> str:
    response = model.invoke([
        {"role": "system", "content": "You are a helpful assistant that writes educational paragraphs."},
        {"role": "user", "content": f"Write a paragraph about {topic}."}
    ])
    return response.content

# Create a checkpointer for persistence
checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(topics: list[str]) -> str:
    """Generates multiple paragraphs in parallel and combines them."""
    futures = [generate_paragraph(topic) for topic in topics]
    paragraphs = [f.result() for f in futures]
    return "\n\n".join(paragraphs)

config = {"configurable": {"thread_id": str(uuid7())}}
result = workflow.invoke(["quantum computing", "climate change", "history of aviation"], config=config)
print(result)

그래프 호출하기

함수형 API와 그래프 API는 같은 런타임을 공유하므로 한 앱 안에서 함께 쓸 수 있어요.

from langgraph.func import entrypoint
from langgraph.graph import StateGraph

builder = StateGraph()
...
some_graph = builder.compile()

@entrypoint()
def some_workflow(some_input: dict) -> int:
    # Call a graph defined using the graph API
    result_1 = some_graph.invoke(...)
    # Call another graph defined using the graph API
    result_2 = another_graph.invoke(...)
    return {
        "result_1": result_1,
        "result_2": result_2
    }

그래프 API로 double 노드를 정의하고, 함수형 API 워크플로에서 그 그래프를 호출하는 전체 예시예요.

import uuid
from typing import TypedDict
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

class State(TypedDict):
    foo: int

def double(state: State) -> State:
    return {"foo": state["foo"] * 2}

builder = StateGraph(State)
builder.add_node("double", double)
builder.set_entry_point("double")
graph = builder.compile()

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(x: int) -> dict:
    result = graph.invoke({"foo": x})
    return {"bar": result["foo"]}

config = {"configurable": {"thread_id": str(uuid7())}}
print(workflow.invoke(5, config=config))  # Output: {'bar': 10}

다른 entrypoint 호출하기

entrypoint나 task 안에서 다른 entrypoint를 호출할 수 있어요. 부모 entrypoint의 체크포인터가 자동으로 사용돼요.

@entrypoint() # Will automatically use the checkpointer from the parent entrypoint
def some_other_workflow(inputs: dict) -> int:
    return inputs["value"]

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    value = some_other_workflow.invoke({"value": 1})
    return value

스트리밍

함수형 API는 그래프 API와 같은 스트리밍 메커니즘을 사용해요. 자세한 내용은 스트리밍 가이드에서 다루고, 여기서는 워크플로 실행에서 value 청크를 스트리밍하는 예시를 볼게요.

config = {"configurable": {"thread_id": str(uuid7())}}

stream = main.stream_events({"x": 5}, config=config, version="v3")
for mode, chunk in stream.interleave("values"):
    print(f"{mode}: {chunk}")
# values: 10

1. langgraph.config에서 get_stream_writer를 임포트한다.
2. entrypoint 안에서 스트림 라이터 인스턴스를 얻는다.
3. 계산이 시작되기 전에 커스텀 데이터를 발행한다.
4. 결과를 계산한 뒤 다시 커스텀 메시지를 발행한다.
5. stream_events()로 스트리밍 출력을 처리한다.
6. interleave("values")가 주는 (mode, chunk) 쌍을 순회한다.

Python < 3.11에서의 async — Python < 3.11에서 async 코드를 쓸 때는 get_stream_writer가 동작하지 않아요. 대신 StreamWriter 클래스를 직접 사용하세요. 자세한 내용은 Async with Python < 3.11 문서를 봐요.

from langgraph.types import StreamWriter

@entrypoint(checkpointer=checkpointer)
async def main(inputs: dict, writer: StreamWriter) -> int:
    ...

재시도 정책 (Retry policy)

RetryPolicy로 특정 오류에 대한 재시도를 설정할 수 있어요. 기본 RetryPolicy는 특정 네트워크 오류에 최적화되어 있어요.

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

# This variable is just used for demonstration purposes to simulate a network failure.
# It's not something you will have in your actual code.
attempts = 0

# Let's configure the RetryPolicy to retry on ValueError.
retry_policy = RetryPolicy(retry_on=ValueError)

@task(retry_policy=retry_policy)
def get_info():
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError('Failure')
    return "OK"

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def main(inputs, writer):
    return get_info().result()

config = {"configurable": {"thread_id": "1"}}
main.invoke({'any_input': 'foobar'}, config=config)

실행 결과는 'OK'예요.

task·entrypoint 타임아웃 설정

@task나 @entrypoint의 timeout 파라미터로 단일 async 시도의 실행 시간을 제한할 수 있어요. 시간은 초 단위 또는 datetime.timedelta로 전달해요.

import asyncio

from langgraph.errors import NodeTimeoutError
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy


@task(
    timeout=1.0,
    retry_policy=RetryPolicy(retry_on=NodeTimeoutError),
)
async def call_api(url: str) -> str:
    await asyncio.sleep(2)
    return f"result from {url}"


@entrypoint(timeout=5.0)
async def workflow(inputs: dict) -> str:
    return await call_api(inputs["url"])


try:
    await workflow.ainvoke({"url": "https://example.com"})
except NodeTimeoutError:
    print("Task timed out")

타임아웃은 async task·entrypoint에서만 지원돼요. 동기 함수에 timeout을 설정하면 LangGraph가 task·entrypoint가 선언될 때 오류를 일으켜요. task·entrypoint가 타임아웃을 넘기면 NodeTimeoutError가 발생하는데, 이는 Python 내장 TimeoutError의 서브클래스예요. 재시도 정책이 TimeoutError나 NodeTimeoutError를 재시도하도록 설정돼 있으면, 타임아웃된 시도가 재시도돼요. 타임아웃은 각 시도마다 독립적으로 적용되므로 재시도할 때마다 타이머가 초기화돼요.

태스크 캐싱

CachePolicy와 엔트리포인트의 cache= 인자로 태스크 결과를 캐시할 수 있어요. ttl은 초 단위로 지정하며, 이 시간이 지나면 캐시가 무효화돼요.

import time
from langgraph.cache.memory import InMemoryCache
from langgraph.func import entrypoint, task
from langgraph.types import CachePolicy

@task(cache_policy=CachePolicy(ttl=120))
def slow_add(x: int) -> int:
    time.sleep(1)
    return x * 2

@entrypoint(cache=InMemoryCache())
def main(inputs: dict) -> dict[str, int]:
    result1 = slow_add(inputs["x"]).result()
    result2 = slow_add(inputs["x"]).result()
    return {"result1": result1, "result2": result2}

오류 후 재개하기

체크포인터가 있으면 오류 후 재개할 때 이미 저장된 결과를 다시 실행하지 않아요. 아래 예시에서 slow_task는 첫 호출에 1초가 걸리고 get_info는 첫 시도에서 실패해요.

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

attempts = 0

@task()
def get_info():
    global attempts
    attempts += 1
    if attempts < 2:
        raise ValueError("Failure")
    return "OK"

@task
def slow_task():
    time.sleep(1)
    return "Ran slow task."

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def main(inputs, writer: StreamWriter):
    slow_task_result = slow_task().result()
    get_info().result()  # Exception will be raised here on the first attempt
    return slow_task_result

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

try:
    main.invoke({'any_input': 'foobar'}, config=config)
except ValueError:
    pass

재개하면 slow_task의 결과는 이미 체크포인트에 저장돼 있으므로 다시 실행하지 않아요.

main.invoke(None, config=config)

결과는 'Ran slow task.'예요.

Human-in-the-loop

함수형 API는 interrupt 함수와 Command 프리미티브로 human-in-the-loop 워크플로를 지원해요.

간단한 예시로 세 개의 task를 만들게요. 1) "bar"를 붙인다. 2) 인간 입력을 위해 잠시 멈추고, 재개되면 인간 입력을 붙인다. 3) "qux"를 붙인다.

from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt


@task
def step_1(input_query):
    """Append bar."""
    return f"{input_query} bar"


@task
def human_feedback(input_query):
    """Append user input."""
    feedback = interrupt(f"Please provide feedback: {input_query}")
    return f"{input_query} {feedback}"


@task
def step_3(input_query):
    """Append qux."""
    return f"{input_query} qux"

이 task들을 entrypoint에서 조합해요. interrupt()가 task 안에서 호출되면 인간이 이전 task의 출력을 검토·편집할 수 있고, 이전 task(step_1)의 결과는 영속화되어 재실행되지 않아요.

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()


@entrypoint(checkpointer=checkpointer)
def graph(input_query):
    result_1 = step_1(input_query).result()
    result_2 = human_feedback(result_1).result()
    result_3 = step_3(result_2).result()
    return result_3

반환값과 저장값 분리하기

entrypoint.final을 쓰면 호출자에게 반환하는 값과 체크포인트에 저장하는 값을 분리할 수 있어요. 이때 유용한 상황이 두 가지예요.

• 계산된 결과(요약이나 상태)를 반환하되, 다음 호출에 쓸 다른 내부 값을 저장하고 싶을 때.
• 다음 실행에서 previous 파라미터로 넘길 값을 제어해야 할 때.

from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def accumulate(n: int, *, previous: int | None) -> entrypoint.final[int, int]:
    previous = previous or 0
    total = previous + n
    # Return the *previous* value to the caller but save the *new* total to the checkpoint.
    return entrypoint.final(value=previous, save=total)

config = {"configurable": {"thread_id": "my-thread"}}

print(accumulate.invoke(1, config=config))  # 0
print(accumulate.invoke(2, config=config))  # 1
print(accumulate.invoke(3, config=config))  # 3

챗봇 예시

함수형 API와 InMemorySaver 체크포인터를 쓴 간단한 챗봇이에요. 이전 대화를 기억하고 이어서 진행할 수 있어요.

from langchain.messages import BaseMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-6")

@task
def call_model(messages: list[BaseMessage]):
    response = model.invoke(messages)
    return response

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))

config = {"configurable": {"thread_id": "1"}}
input_message = {"role": "user", "content": "hi! I'm bob"}
stream = workflow.stream_events([input_message], config, version="v3")
for snapshot in stream.values:
    print(snapshot)

장기 메모리

장기 메모리는 서로 다른 thread id에 걸쳐 정보를 저장해요. 한 대화에서 특정 사용자에 대해 학습한 정보를 다른 대화에서도 쓰는 데 유용해요.

더 알아보기 (Learn more)

• Workflows and agents — 함수형 API로 워크플로를 만드는 추가 예시
• Functional API — 함수형 API 개념 문서
• 다른 프레임워크에 LangGraph 기능 추가하기 — 지속성·메모리·스트리밍을 기본 제공하지 않는 다른 에이전트 프레임워크에 함수형 API로 추가