API 선택하기 (Choosing between Graph and Functional APIs)

API 선택하기 (Choosing between Graph and Functional APIs)

LangGraph는 에이전트 워크플로를 만들기 위한 두 가지 API를 제공해요. 하나는 그래프 API(Graph API), 다른 하나는 **함수형 API(Functional API)**예요. 두 API는 같은 런타임을 공유하고 같은 애플리케이션 안에서 함께 쓸 수도 있지만, 서로 다른 용도와 개발 스타일에 맞게 설계되었어요. 이 문서는 구체적인 요구사항에 따라 어떤 API를 써야 하는지 판단하는 데 도움을 주는 가이드예요.

빠른 판단 가이드 (Quick decision guide)

다음이 필요하다면 그래프 API를 쓰세요.

  • 복잡한 워크플로 시각화 — 디버깅과 문서화에 유용해요
  • 명시적인 상태 관리 — 여러 노드가 공유하는 데이터가 있을 때
  • 조건부 분기(conditional branching) — 판단 지점이 여러 개일 때
  • 병렬 실행 경로 — 나중에 다시 합쳐야 할 때
  • 팀 협업 — 시각적 표현이 이해를 돕는 경우

이런 걸 원한다면 함수형 API를 쓰세요.

  • 기존 절차적 코드에 대한 최소한의 변경 — 이미 있는 코드를 거의 건드리지 않고
  • 표준 제어 흐름 — if/else, 반복문, 함수 호출 같은
  • 함수 범위의 상태(function-scoped state) — 명시적인 상태 관리 없이
  • 빠른 프로토타이핑 — 보일러플레이트(code)가 적어서
  • 단순 분기 로직이 있는 선형 워크플로

상세 비교 (Detailed comparison)

그래프 API는 언제 쓰나요? (When to use the Graph API)

그래프 API는 노드, 엣지, 공유 상태를 정의해 시각적인 그래프 구조를 만드는 선언적(declarative) 접근 방식을 써요.

1. 복잡한 결정 트리와 분기 로직 — 워크플로가 여러 조건에 따라 나뉘는 판단 지점을 많이 갖고 있다면, 그래프 API가 그 분기를 명시적이고 시각적으로 보여줘요.

# Graph API: 결정 경로를 명확하게 시각화
from langgraph.graph import StateGraph
from typing import TypedDict

class AgentState(TypedDict):
    messages: list
    current_tool: str
    retry_count: int

def should_continue(state):
    if state["retry_count"] > 3:
        return "end"
    elif state["current_tool"] == "search":
        return "process_search"
    else:
        return "call_llm"

workflow = StateGraph(AgentState)
workflow.add_node("call_llm", call_llm_node)
workflow.add_node("process_search", search_node)
workflow.add_conditional_edges("call_llm", should_continue)

2. 여러 컴포넌트에 걸친 상태 관리 — 워크플로의 서로 다른 부분 사이에서 상태를 공유하고 조정해야 할 때, 그래프 API의 명시적 상태 관리는 큰 장점이 돼요.

# 여러 노드가 공유 상태에 접근하고 수정
class WorkflowState(TypedDict):
    user_input: str
    search_results: list
    generated_response: str
    validation_status: str

def search_node(state):
    # 공유 상태에 접근
    results = search(state["user_input"])
    return {"search_results": results}

def validation_node(state):
    # 이전 노드의 결과에 접근
    is_valid = validate(state["generated_response"])
    return {"validation_status": "valid" if is_valid else "invalid"}

3. 동기화가 필요한 병렬 처리 — 여러 작업을 병렬로 실행한 뒤 결과를 합쳐야 할 때, 그래프 API가 이 흐름을 자연스럽게 처리해요.

# 여러 데이터 소스의 병렬 처리
workflow.add_node("fetch_news", fetch_news)
workflow.add_node("fetch_weather", fetch_weather)
workflow.add_node("fetch_stocks", fetch_stocks)
workflow.add_node("combine_data", combine_all_data)

# 모든 fetch 작업이 병렬로 실행
workflow.add_edge(START, "fetch_news")
workflow.add_edge(START, "fetch_weather")
workflow.add_edge(START, "fetch_stocks")

# combine은 모든 병렬 작업이 끝난 뒤 실행
workflow.add_edge("fetch_news", "combine_data")
workflow.add_edge("fetch_weather", "combine_data")
workflow.add_edge("fetch_stocks", "combine_data")

4. 팀 개발과 문서화 — 그래프 API의 시각적 특성 덕분에 팀이 복잡한 워크플로를 이해하고, 문서화하고, 유지보수하기 쉬워져요.

# 관심사의 명확한 분리 - 각 팀원이 서로 다른 노드를 작업
workflow.add_node("data_ingestion", data_team_function)
workflow.add_node("ml_processing", ml_team_function)
workflow.add_node("business_logic", product_team_function)
workflow.add_node("output_formatting", frontend_team_function)

함수형 API는 언제 쓰나요? (When to use the Functional API)

함수형 API는 표준 절차적 코드에 LangGraph 기능을 통합하는 명령형(imperative) 접근 방식을 써요.

1. 기존 절차적 코드 — 표준 제어 흐름을 쓰는 코드가 이미 있고, 리팩터링을 최소화하면서 LangGraph 기능을 더하고 싶을 때 적합해요.

# Functional API: 기존 코드에 대한 최소한의 변경
from langgraph.func import entrypoint, task

@task
def process_user_input(user_input: str) -> dict:
    # 최소 변경만 가한 기존 함수
    return {"processed": user_input.lower().strip()}

@entrypoint(checkpointer=checkpointer)
def workflow(user_input: str) -> str:
    # 표준 파이썬 제어 흐름
    processed = process_user_input(user_input).result()

    if "urgent" in processed["processed"]:
        response = handle_urgent_request(processed).result()
    else:
        response = handle_normal_request(processed).result()

    return response

2. 단순 로직을 가진 선형 워크플로 — 워크플로가 주로 순차적이고 조건 로직이 직관적일 때 유용해요. 여기서는 interrupt()를 활용한 인간 검토 체크포인트(human review checkpoint)도 함께 보여줘요.

@entrypoint(checkpointer=checkpointer)
def essay_workflow(topic: str) -> dict:
    # 단순 분기가 있는 선형 흐름
    outline = create_outline(topic).result()

    if len(outline["points"]) < 3:
        outline = expand_outline(outline).result()

    draft = write_draft(outline).result()

    # 인간 검토 체크포인트
    feedback = interrupt({"draft": draft, "action": "Please review"})

    if feedback == "approve":
        final_essay = draft
    else:
        final_essay = revise_essay(draft, feedback).result()

    return {"essay": final_essay}

3. 빠른 프로토타이핑 — 상태 스키마와 그래프 구조를 정의하는 부담 없이 아이디어를 빨리 시험해 보고 싶을 때 좋아요.

@entrypoint(checkpointer=checkpointer)
def quick_prototype(data: dict) -> dict:
    # 빠른 반복 - 상태 스키마가 필요 없음
    step1_result = process_step1(data).result()
    step2_result = process_step2(step1_result).result()

    return {"final_result": step2_result}

4. 함수 범위의 상태 관리 — 상태가 개별 함수에 자연스럽게 국한되고 넓게 공유될 필요가 없을 때 적합해요.

@task
def analyze_document(document: str) -> dict:
    # 함수 내부의 지역 상태 관리
    sections = extract_sections(document)
    summaries = [summarize(section) for section in sections]
    key_points = extract_key_points(summaries)

    return {
        "sections": len(sections),
        "summaries": summaries,
        "key_points": key_points
    }

@entrypoint(checkpointer=checkpointer)
def document_processor(document: str) -> dict:
    analysis = analyze_document(document).result()
    # 상태는 필요에 따라 함수 사이에서 전달
    return generate_report(analysis).result()

두 API를 함께 쓰기 (Combining both APIs)

두 API를 같은 애플리케이션에서 함께 쓸 수 있어요. 시스템의 서로 다른 부분이 서로 다른 요구사항을 가질 때 이 방식이 유용하죠. 예를 들어 복잡한 멀티에이전트 조정(coordination)은 그래프 API로, 단순한 데이터 처리는 함수형 API로 구성하고, 그래프의 노드 안에서 함수형 API 결과를 호출하는 식이에요.

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

# Graph API를 사용한 복잡한 멀티에이전트 조정
coordination_graph = StateGraph(CoordinationState)
coordination_graph.add_node("orchestrator", orchestrator_node)
coordination_graph.add_node("agent_a", agent_a_node)
coordination_graph.add_node("agent_b", agent_b_node)

# Functional API를 사용한 단순 데이터 처리
@entrypoint()
def data_processor(raw_data: dict) -> dict:
    cleaned = clean_data(raw_data).result()
    transformed = transform_data(cleaned).result()
    return transformed

# 그래프 안에서 함수형 API 결과 사용
def orchestrator_node(state):
    processed_data = data_processor.invoke(state["raw_data"])
    return {"processed_data": processed_data}

API 간 마이그레이션 (Migration between APIs)

함수형에서 그래프 API로 (From Functional to Graph API)

함수형 워크플로가 복잡해지면 그래프 API로 이전할 수 있어요. 아래 예시처럼 같은 로직을 함수형에서는 중첩된 if/else로, 그래프에서는 조건부 엣지(conditional edges)로 표현해요.

# Before: Functional API
@entrypoint(checkpointer=checkpointer)
def complex_workflow(input_data: dict) -> dict:
    step1 = process_step1(input_data).result()

    if step1["needs_analysis"]:
        analysis = analyze_data(step1).result()
        if analysis["confidence"] > 0.8:
            result = high_confidence_path(analysis).result()
        else:
            result = low_confidence_path(analysis).result()
    else:
        result = simple_path(step1).result()

    return result

# After: Graph API
class WorkflowState(TypedDict):
    input_data: dict
    step1_result: dict
    analysis: dict
    final_result: dict

def should_analyze(state):
    return "analyze" if state["step1_result"]["needs_analysis"] else "simple_path"

def confidence_check(state):
    return "high_confidence" if state["analysis"]["confidence"] > 0.8 else "low_confidence"

workflow = StateGraph(WorkflowState)
workflow.add_node("step1", process_step1_node)
workflow.add_conditional_edges("step1", should_analyze)
workflow.add_node("analyze", analyze_data_node)
workflow.add_conditional_edges("analyze", confidence_check)
# ... 나머지 노드와 엣지 추가

그래프에서 함수형 API로 (From Graph to Functional API)

반대로, 단순한 선형 처리에는 그래프가 지나치게 복잡해 보일 수 있어요. 그럴 땐 함수형 API로 단순화하면 돼요.

# Before: 과하게 설계된 Graph API
class SimpleState(TypedDict):
    input: str
    step1: str
    step2: str
    result: str

# After: 단순화된 Functional API
@entrypoint(checkpointer=checkpointer)
def simple_workflow(input_data: str) -> str:
    step1 = process_step1(input_data).result()
    step2 = process_step2(step1).result()
    return finalize_result(step2).result()

요약 (Summary)

그래프 API는 워크플로 구조에 대한 명시적 제어, 복잡한 분기, 병렬 처리, 팀 협업의 이점이 필요할 때 선택하세요. 함수형 API는 기존 코드에 최소한의 변경으로 LangGraph 기능을 더하고 싶을 때, 단순한 선형 워크플로를 가졌을 때, 빠른 프로토타이핑이 필요할 때 선택하세요.

두 API는 동일한 핵심 LangGraph 기능(영속화(persistence), 스트리밍(streaming), 인간 개입(human-in-the-loop), 메모리(memory))을 제공하지만, 각기 다른 개발 스타일과 사용 사례에 맞게 다른 패러다임으로 감싸고 있어요. 둘 중 무엇이 더 낫다기보다, 여러분의 상황에 더 잘 맞는 패러다임을 고르는 문제예요.