Graph API와 Functional API 중 무엇을 쓸까
Graph API와 Functional API 중 무엇을 쓸까 (Choosing between Graph and Functional APIs)
LangGraph는 에이전트 워크플로우를 만들 수 있는 두 가지 API를 제공해요. Graph API와 Functional API가 그것이죠. 두 API는 같은 런타임을 공유하고 같은 애플리케이션 안에서 함께 쓸 수 있지만, 각각 다른 사용 사례와 개발 선호도를 위한 설계라는 차이가 있습니다. 이 가이드에서는 여러분의 요구에 맞춰 어떤 API를 써야 할지 판단하는 데 도움을 드릴게요.
출처: 문서
본문
빠른 판단 가이드 (Quick decision guide)
다음이 필요하다면 Graph API를 쓰세요:
- 복잡한 워크플로우 시각화 — 디버깅과 문서화에 유용할 때
- 명시적인 상태 관리 — 여러 노드(node)가 공유하는 데이터가 있을 때
- 조건부 분기 — 판단 지점(decision point)이 여러 개일 때
- 병렬 실행 경로 — 나중에 다시 합쳐야 하는 병렬 흐름이 있을 때
- 팀 협업 — 시각적 표현이 이해를 돕는 환경일 때
다음을 원한다면 Functional API를 쓰세요:
- 기존 절차 코드의 수정을 최소화
- 표준 제어 흐름 (if/else, 반복문, 함수 호출)
- 명시적 상태 관리 없이 함수 단위로 국한된 상태
- 보일러플레이트가 적은 빠른 프로토타이핑
- 단순한 분기 로직을 가진 선형 워크플로우
상세 비교 (Detailed comparison)
Graph API를 언제 쓸까
Graph API는 노드, 엣지(edge), 공유 상태를 정의해 시각적인 그래프 구조를 만드는 선언적(declarative) 방식이에요.
1. 복잡한 의사 결정 트리와 분기 로직
워크플로우에 다양한 조건에 의존하는 판단 지점이 여러 개라면, Graph API가 그 분기를 명시적으로 드러내고 쉽게 시각화할 수 있게 해줘요.
# Graph API: Clear visualization of decision paths
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. 여러 컴포넌트에 걸친 상태 관리
워크플로우의 서로 다른 부분 사이에서 상태를 공유하고 조정해야 한다면, Graph API의 명시적 상태 관리가 유용해요.
# Multiple nodes can access and modify shared state
class WorkflowState(TypedDict):
user_input: str
search_results: list
generated_response: str
validation_status: str
def search_node(state):
# Access shared state
results = search(state["user_input"])
return {"search_results": results}
def validation_node(state):
# Access results from previous node
is_valid = validate(state["generated_response"])
return {"validation_status": "valid" if is_valid else "invalid"}
3. 동기화가 필요한 병렬 처리
여러 작업을 병렬로 실행한 뒤 그 결과를 합쳐야 한다면, Graph API가 자연스럽게 처리해 줍니다.
# Parallel processing of multiple data sources
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)
# All fetch operations run in parallel
workflow.add_edge(START, "fetch_news")
workflow.add_edge(START, "fetch_weather")
workflow.add_edge(START, "fetch_stocks")
# Combine waits for all parallel operations to complete
workflow.add_edge("fetch_news", "combine_data")
workflow.add_edge("fetch_weather", "combine_data")
workflow.add_edge("fetch_stocks", "combine_data")
4. 팀 개발과 문서화
Graph API의 시각적 특성 덕분에 팀이 복잡한 워크플로우를 이해·문서화·유지보수하기 쉬워져요.
# Clear separation of concerns - each team member can work on different nodes
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)
Functional API를 언제 쓸까
Functional API는 표준 절차 코드에 LangGraph 기능을 통합하는 명령형(imperative) 방식이에요.
1. 기존 절차 코드가 있을 때
표준 제어 흐름을 쓰는 기존 코드에 최소한의 리팩토링으로 LangGraph 기능을 더하고 싶을 때 유용합니다.
# Functional API: Minimal changes to existing code
from langgraph.func import entrypoint, task
@task
def process_user_input(user_input: str) -> dict:
# Existing function with minimal changes
return {"processed": user_input.lower().strip()}
@entrypoint(checkpointer=checkpointer)
def workflow(user_input: str) -> str:
# Standard Python control flow
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. 단순한 로직의 선형 워크플로우
워크플로우가 주로 순차적이고 조건 로직이 간단할 때 쓰기 좋아요.
@entrypoint(checkpointer=checkpointer)
def essay_workflow(topic: str) -> dict:
# Linear flow with simple branching
outline = create_outline(topic).result()
if len(outline["points"]) < 3:
outline = expand_outline(outline).result()
draft = write_draft(outline).result()
# Human review checkpoint
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:
# Fast iteration - no state schema needed
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:
# Local state management within function
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()
# State is passed between functions as needed
return generate_report(analysis).result()
두 API 함께 쓰기 (Combining both APIs)
같은 애플리케이션에서 두 API를 함께 사용할 수 있어요. 시스템의 서로 다른 부분이 서로 다른 요구를 가질 때 유용합니다.
from langgraph.graph import StateGraph
from langgraph.func import entrypoint
# Complex multi-agent coordination using 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)
# Simple data processing using Functional API
@entrypoint()
def data_processor(raw_data: dict) -> dict:
cleaned = clean_data(raw_data).result()
transformed = transform_data(cleaned).result()
return transformed
# Use the functional API result in the graph
def orchestrator_node(state):
processed_data = data_processor.invoke(state["raw_data"])
return {"processed_data": processed_data}
API 간 마이그레이션 (Migration between APIs)
Functional에서 Graph API로
기능형 워크플로우가 복잡해지면 Graph API로 전환할 수 있어요:
# 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)
# ... add remaining nodes and edges
Graph에서 Functional API로
단순한 선형 처리에는 그래프가 과하게 복잡해질 때가 있어요:
# Before: Over-engineered Graph API
class SimpleState(TypedDict):
input: str
step1: str
step2: str
result: str
# After: Simplified 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)
워크플로우 구조를 명시적으로 제어해야 하거나, 복잡한 분기·병렬 처리가 필요하거나, 팀 협업의 이점이 필요하다면 Graph API를 선택하세요.
기존 코드에 최소한의 변경으로 LangGraph 기능을 더하고 싶거나, 단순한 선형 워크플로우를 만들거나, 빠른 프로토타이핑이 필요하다면 Functional API를 선택하세요.
두 API 모두 LangGraph의 핵심 기능(영속성, 스트리밍, human-in-the-loop, 메모리)을 제공하지만, 개발 스타일과 사용 사례에 맞게 서로 다른 패러다임으로 구성되어 있습니다.
더 알아보기 (Learn more)
- Graph API — 노드·엣지·상태로 그래프를 만드는 선언적 API.
- Functional API — 절차 코드에 LangGraph 기능을 통합하는 명령형 API.