함수형 API 사용하기

함수형 API를 쓰면 LangGraph의 핵심 기능인 persistence, memory, human-in-the-loop, streaming을 기존 코드를 거의 바꾸지 않고 애플리케이션에 추가할 수 있어요.

함수형 API에 대한 개념 설명은 [함수형 API](/oss/python/langgraph/functional-api)를 보세요.

출처: 함수형 API 사용하기 - 공식 문서

간단한 워크플로 만들기

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})

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

숫자가 짝수인지 확인하는 태스크

@task def is_even(number: int) -> bool: return number % 2 == 0

메시지를 포맷하는 태스크

@task def format_message(is_even: bool) -> str: return "The number is even." if is_even else "The number is odd."

영속화를 위한 체크포인터 생성

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer) def workflow(inputs: dict) -> str: """숫자를 분류하는 간단한 워크플로.""" even = is_even(inputs["number"]).result() return format_message(even).result()

고유한 thread ID로 워크플로 실행

config = {"configurable": {"thread_id": str(uuid7())}} result = workflow.invoke({"number": 7}, config=config) print(result)

</Accordion>

<Accordion title="확장 예제: LLM으로 에세이 작성">
이 예제는 `@task`와 `@entrypoint` 데코레이터를 문법적으로 어떻게 쓰는지 보여줘요. 체크포인터가 제공되므로 워크플로 결과는 체크포인터에 영속됩니다.

```python
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')

# 태스크: LLM으로 에세이 생성
@task
def compose_essay(topic: str) -> str:
    """주어진 주제에 대한 에세이를 생성한다."""
    return model.invoke([
        {"role": "system", "content": "You are a helpful assistant that writes essays."},
        {"role": "user", "content": f"Write an essay about {topic}."}
    ]).content

# 영속화를 위한 체크포인터 생성
checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(topic: str) -> str:
    """LLM으로 에세이를 생성하는 간단한 워크플로."""
    return compose_essay(topic).result()

# 워크플로 실행
config = {"configurable": {"thread_id": str(uuid7())}}
result = workflow.invoke("the history of flight", config=config)
print(result)

병렬 실행

태스크를 동시에 호출하고 결과를 기다려 병렬로 실행할 수 있어요. IO 바운드 태스크(예: LLM용 API 호출)에서 성능을 높이는 데 유용해요.

@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]

이 예제는 `@task`로 여러 LLM 호출을 병렬로 실행하는 법을 보여줘요. 각 호출이 다른 주제의 단락을 만들고, 결과는 단일 텍스트 출력으로 합쳐져요.

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

# LLM 모델 초기화
model = init_chat_model("gpt-3.5-turbo")

# 주어진 주제에 대한 단락을 생성하는 태스크
@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

# 영속화를 위한 체크포인터 생성
checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(topics: list[str]) -> str:
    """여러 단락을 병렬로 생성해 합친다."""
    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)

이 예제는 LangGraph의 동시성 모델을 이용해 실행 시간을 개선해요. 특히 LLM 완성 같은 I/O가 포함된 태스크에서 효과적이죠.

그래프 호출

함수형 API와 Graph 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:
    # 그래프 API로 정의한 그래프 호출
    result_1 = some_graph.invoke(...)
    # 그래프 API로 정의한 다른 그래프 호출
    result_2 = another_graph.invoke(...)
    return {
        "result_1": result_1,
        "result_2": result_2
    }

```python 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}

그래프 API로 그래프 구축

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

함수형 API 워크플로 정의

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}

</Accordion>

## 다른 entrypoint 호출

**entrypoint**나 **task** 안에서 다른 **entrypoint**를 호출할 수 있어요.

```python
@entrypoint() # 부모 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

```python import uuid from langgraph.func import entrypoint from langgraph.checkpoint.memory import InMemorySaver

체크포인터 초기화

checkpointer = InMemorySaver()

숫자를 곱하는 재사용 가능한 서브 워크플로

@entrypoint() def multiply(inputs: dict) -> int: return inputs["a"] * inputs["b"]

서브 워크플로를 호출하는 메인 워크플로

@entrypoint(checkpointer=checkpointer) def main(inputs: dict) -> dict: result = multiply.invoke({"a": inputs["x"], "b": inputs["y"]}) return {"product": result}

메인 워크플로 실행

config = {"configurable": {"thread_id": str(uuid7())}} print(main.invoke({"x": 6, "y": 7}, config=config)) # Output: {'product': 42}

</Accordion>

## 스트리밍

**함수형 API**는 **Graph API**와 같은 스트리밍 메커니즘을 사용해요. 자세한 내용은 [**스트리밍 가이드**](/oss/python/langgraph/streaming)를 보세요.

워크플로 실행에서 값 청크를 스트리밍하는 예제예요.

```python
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

langgraph.config에서 get_stream_writer를 import 해요.
entrypoint 안에서 스트림 라이터 인스턴스를 얻어요.
계산이 시작되기 전에 커스텀 데이터를 내보내요.
결과를 계산한 뒤 다른 커스텀 메시지를 내보내요.
stream_events()로 스트리밍 출력을 처리해요.
interleave("values")에서 (mode, chunk) 쌍을 순회해요.

**Python \< 3.11에서의 비동기** Python \< 3.11에서 비동기 코드를 작성한다면 [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer)는 동작하지 않아요. 대신 `StreamWriter` 클래스를 직접 사용하세요. 자세한 내용은 [Python \< 3.11에서의 비동기](/oss/python/langgraph/streaming#async)를 보세요.

from langgraph.types import StreamWriter

@entrypoint(checkpointer=checkpointer)
async def main(inputs: dict, writer: StreamWriter) -> int:  # [!code highlight]
...

재시도 정책

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

# 이 변수는 네트워크 장애를 시뮬레이션하기 위한 데모용일 뿐이에요.
# 실제 코드에는 없어요.
attempts = 0

# ValueError에 재시도하도록 RetryPolicy를 구성해요.
# 기본 RetryPolicy는 특정 네트워크 오류 재시도에 최적화되어 있어요.
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'

태스크·entrypoint 타임아웃 설정

@task나 @entrypoint에 timeout 파라미터를 쓰면 단일 비동기 시도가 실행될 수 있는 시간을 제한할 수 있어요. 타임아웃을 초 단위나 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")

타임아웃은 비동기 태스크와 entrypoint에서만 지원돼요. 동기 함수에 timeout을 설정하면, 태스크나 entrypoint가 선언될 때 LangGraph가 오류를 발생시켜요.

태스크나 entrypoint가 타임아웃을 넘으면 LangGraph는 NodeTimeoutError를 발생시켜요. 이 오류는 Python 내장 TimeoutError의 서브클래스죠. 재시도 정책이 TimeoutError나 NodeTimeoutError를 재시도한다면 시간 초과된 시도는 재시도돼요. 타임아웃은 각 시도에 독립적으로 적용되므로, 재시도마다 타이머가 리셋돼요.

태스크 캐싱

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))    # [!code highlight]
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}


stream = main.stream_events({"x": 5}, version="v3")
for snapshot in stream.values:
    print(snapshot)

ttl은 초 단위예요. 이 시간이 지나면 캐시가 무효화돼요.

오류 후 재개

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():
    """
    한 번 실패한 후 성공하는 태스크를 시뮬레이션한다.
    첫 번째 시도에서 예외를 발생시키고, 이후 시도에서 "OK"를 반환한다.
    """
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError("Failure")  # 첫 번째 시도에서 실패를 시뮬레이션
    return "OK"

# 영속화를 위한 인메모리 체크포인터 초기화
checkpointer = InMemorySaver()

@task
def slow_task():
    """
    1초 지연을 넣어 느리게 실행되는 태스크를 시뮬레이션한다.
    """
    time.sleep(1)
    return "Ran slow task."

@entrypoint(checkpointer=checkpointer)
def main(inputs, writer: StreamWriter):
    """
    slow_task와 get_info 태스크를 순차적으로 실행하는 메인 워크플로 함수.

    입력값과 스트림 라이터를 받는다. slow_task를 먼저 실행하고,
    첫 번째 호출에서 실패할 get_info를 실행하려 한다.
    """
    slow_task_result = slow_task().result()  # slow_task에 대한 블로킹 호출
    get_info().result()  # 첫 번째 시도에서 여기서 예외 발생
    return slow_task_result

# 고유한 스레드 식별자를 가진 워크플로 실행 설정
config = {
    "configurable": {
        "thread_id": "1"  # 워크플로 실행을 추적하는 고유 식별자
    }
}

# 이 호출은 slow_task 실행 때문에 약 1초 걸림
try:
    # `get_info` 태스크가 실패하므로 첫 번째 호출은 예외 발생
    main.invoke({'any_input': 'foobar'}, config=config)
except ValueError:
    pass  # 실패를 우아하게 처리

실행을 재개할 때는 slow_task를 다시 실행할 필요가 없어요. 그 결과가 이미 체크포인트에 저장되어 있기 때문이에요.

main.invoke(None, config=config)

'Ran slow task.'

휴먼 인 더 루프

함수형 API는 interrupt 함수와 Command 기본 요소로 human-in-the-loop 워크플로를 지원해요.

기본 휴먼 인 더 루프 워크플로

세 개의 태스크를 만들게요:

"bar"를 붙여요.
인간 입력을 위해 멈추고, 재개할 때 인간 입력을 붙여요.
"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"

이 태스크들을 entrypoint로 조합할 수 있어요:

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

interrupt()는 태스크 안에서 호출되어, 인간이 이전 태스크의 출력을 검토·편집할 수 있게 해요. 이전 태스크(step_1)의 결과는 영속되므로, interrupt 이후에 다시 실행되지 않아요.

쿼리 문자열을 보내볼게요:

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

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

step_1 이후 interrupt로 멈췄다는 점에 주목하세요. 이 인터럽트는 실행을 재개하는 방법을 알려줘요. 재개하려면 human_feedback 태스크가 기대하는 데이터를 담은 Command를 발행해요.

# 실행 계속하기
stream = graph.stream_events(Command(resume="baz"), config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

재개 후 실행은 남은 단계를 거쳐 예상대로 종료돼요.

툴 콜 검토

툴 콜을 실행 전에 검토하려면 interrupt를 호출하는 review_tool_call 함수를 추가해요. 이 함수가 호출되면, 재개 명령을 발행할 때까지 실행이 멈춰요.

툴 콜이 주어지면 우리 함수는 인간 검토를 위해 interrupt해요. 이 시점에 다음 중 하나를 할 수 있어요:

툴 콜 수락
툴 콜 수정 후 계속
커스텀 툴 메시지 생성(예: 모델에게 툴 콜 재포맷 지시)

from typing import Union

def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:
    """Review a tool call, returning a validated version."""
    human_review = interrupt(
        {
            "question": "Is this correct?",
            "tool_call": tool_call,
        }
    )
    review_action = human_review["action"]
    review_data = human_review.get("data")
    if review_action == "continue":
        return tool_call
    elif review_action == "update":
        updated_tool_call = {**tool_call, **{"args": review_data}}
        return updated_tool_call
    elif review_action == "feedback":
        return ToolMessage(
            content=review_data, name=tool_call["name"], tool_call_id=tool_call["id"]
        )

이제 entrypoint를 갱신해 생성된 툴 콜을 검토할 수 있어요. 툴 콜이 수락되거나 수정되면 이전과 같은 방식으로 실행돼요. 그렇지 않으면 인간이 제공한 ToolMessage를 추가해요. 이전 태스크(여기서는 초기 모델 호출)의 결과는 영속되므로, interrupt 이후에 다시 실행되지 않아요.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt


checkpointer = InMemorySaver()


@entrypoint(checkpointer=checkpointer)
def agent(messages, previous):
    if previous is not None:
        messages = add_messages(previous, messages)

    model_response = call_model(messages).result()
    while True:
        if not model_response.tool_calls:
            break

        # 툴 콜 검토
        tool_results = []
        tool_calls = []
        for i, tool_call in enumerate(model_response.tool_calls):
            review = review_tool_call(tool_call)
            if isinstance(review, ToolMessage):
                tool_results.append(review)
            else:  # 검증된 툴 콜
                tool_calls.append(review)
                if review != tool_call:
                    model_response.tool_calls[i] = review  # 메시지 갱신

        # 남은 툴 콜 실행
        tool_result_futures = [call_tool(tool_call) for tool_call in tool_calls]
        remaining_tool_results = [fut.result() for fut in tool_result_futures]

        # 메시지 목록에 추가
        messages = add_messages(
            messages,
            [model_response, *tool_results, *remaining_tool_results],
        )

        # 모델을 다시 호출
        model_response = call_model(messages).result()

    # 최종 응답 생성
    messages = add_messages(messages, model_response)
    return entrypoint.final(value=model_response, save=messages)

단기 메모리

단기 메모리는 같은 thread id의 서로 다른 invocation 간에 정보를 저장할 수 있게 해요. 자세한 내용은 short-term memory를 보세요.

체크포인트 관리

체크포인터가 저장한 정보는 조회·삭제할 수 있어요.

스레드 상태 보기

config = {
    "configurable": {
        "thread_id": "1",  # [!code highlight]
        # 선택적으로 특정 체크포인트의 ID를 제공할 수 있어요.
        # 제공하지 않으면 최신 체크포인트가 보여요.
        # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"  # [!code highlight]

    }
}
graph.get_state(config)  # [!code highlight]

StateSnapshot(
    values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
    metadata={
        'source': 'loop',
        'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
        'step': 4,
        'parents': {},
        'thread_id': '1'
    },
    created_at='2025-05-05T16:01:24.680462+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
    tasks=(),
    interrupts=()
)

스레드 히스토리 보기

config = {
    "configurable": {
        "thread_id": "1"  # [!code highlight]
    }
}
list(graph.get_state_history(config))  # [!code highlight]

[
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:24.680462+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")]},
        next=('call_model',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863421+00:00',
        parent_config={...}
        tasks=(PregelTask(id='8ab4155e-6b15-b885-9ce5-bed69a2c305c', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Your name is Bob.')}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},
        next=('__start__',),
        config={...},
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863173+00:00',
        parent_config={...}
        tasks=(PregelTask(id='24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "what's my name?"}]}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},
        next=(),
        config={...},
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.862295+00:00',
        parent_config={...}
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob")]},
        next=('call_model',),
        config={...},
        metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:22.278960+00:00',
        parent_config={...}
        tasks=(PregelTask(id='8cbd75e0-3720-b056-04f7-71ac805140a0', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': []},
        next=('__start__',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:22.277497+00:00',
        parent_config=None,
        tasks=(PregelTask(id='d458367b-8265-812c-18e2-33001d199ce6', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}),),
        interrupts=()
    )
]

반환 값과 저장 값 분리

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 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)

input_message = {"role": "user", "content": "what's my name?"}
stream = workflow.stream_events([input_message], config, version="v3")
for snapshot in stream.values:
    print(snapshot)

장기 메모리

long-term memory은 서로 다른 thread id 간에 정보를 저장할 수 있게 해요. 한 대화에서 특정 사용자에 대해 배운 정보를 다른 대화에 쓰는 데 유용하죠.

워크플로

함수형 API로 워크플로를 만드는 더 많은 예제는 Workflows and agent 가이드에서 볼 수 있어요.

다른 라이브러리와 연동

다른 프레임워크에 함수형 API로 LangGraph 기능 추가: 영속성·메모리·스트리밍 같은 LangGraph 기능을 기본 제공하지 않는 다른 에이전트 프레임워크에 더해요.

더 알아보기 (Learn more)

함수형 API — 개념 설명
Streaming — 스트리밍 메커니즘
Interrupts — 휴먼 인 더 루프