LangGraph에서 마이그레이션하기

LangGraph에서 JamJet으로 전환하는 개발자를 위한 코드 비교 및 개념 매핑.

개념 매핑

LangGraph	JamJet
`TypedDict` state	`pydantic.BaseModel` state — 모든 단계에서 검증됨
`StateGraph`	`Workflow`
`graph.add_node("name", fn)`	`@workflow.step` 데코레이터
`graph.add_conditional_edges(node, router_fn)`	`@workflow.step(next={"target": predicate})`
`graph.add_edge(A, B)`	기본적으로 순차적; 분기는 `next=` 사용
`graph.compile()`	`workflow.compile()` → Rust 런타임용 IR
`app.invoke(state)`	`workflow.run_sync(state)` (로컬)
`app.astream(state)`	`workflow.run(state)` (비동기, 로컬)
`MemorySaver` / `PostgresSaver`	Rust 런타임에 내장 — 자동
`interrupt_before` (human-in-the-loop)	`type: wait` 노드 또는 스텝에 `human_approval=True`

병렬 비교 예제

조건부 라우팅이 있는 멀티 스텝 에이전트 — 검색 여부를 결정한 후 답변을 생성합니다.

LangGraph

from typing import Literal, TypedDict
from langgraph.graph import END, START, StateGraph

class State(TypedDict):
    question: str
    needs_search: bool
    search_results: list[str]
    answer: str

def route(state: State) -> State:
    q = state["question"].lower()
    needs = any(w in q for w in ["latest", "current", "today"])
    return {**state, "needs_search": needs}

def search(state: State) -> State:
    return {**state, "search_results": [f"[result for: {state['question']}]"]}

def answer(state: State) -> State:
    ctx = "\n".join(state.get("search_results", []))
    return {**state, "answer": f"Answer: {state['question']}\n{ctx}"}

def should_search(state: State) -> Literal["search", "answer"]:
    return "search" if state["needs_search"] else "answer"

graph = StateGraph(State)
graph.add_node("route", route)
graph.add_node("search", search)
graph.add_node("answer", answer)
graph.add_edge(START, "route")
graph.add_conditional_edges("route", should_search)
graph.add_edge("search", "answer")
graph.add_edge("answer", END)

app = graph.compile()
result = app.invoke({"question": "...", "needs_search": False, "search_results": [], "answer": ""})
print(result["answer"])

JamJet

from pydantic import BaseModel
from jamjet import Workflow

class State(BaseModel):
    question: str
    needs_search: bool = False
    search_results: list[str] = []
    answer: str = ""

wf = Workflow("research-agent")

@wf.state
class AgentState(State):
    pass

@wf.step
async def route(state: AgentState) -> AgentState:
    q = state.question.lower()
    needs = any(w in q for w in ["latest", "current", "today"])
    return state.model_copy(update={"needs_search": needs})

@wf.step(next={"search": lambda s: s.needs_search})
async def check_route(state: AgentState) -> AgentState:
    return state  # pure routing step

@wf.step
async def search(state: AgentState) -> AgentState:
    results = [f"[result for: {state.question}]"]
    return state.model_copy(update={"search_results": results})

@wf.step
async def answer(state: AgentState) -> AgentState:
    ctx = "\n".join(state.search_results)
    return state.model_copy(update={"answer": f"Answer: {state.question}\n{ctx}"})

# 로컬 실행 — 서버 불필요

result = wf.run_sync(AgentState(question="..."))
print(result.state.answer)

# 프로덕션: wf.compile() + jamjet dev

주요 차이점

상태 검증

LangGraph는 TypedDict를 사용합니다 — 검증 없는 딕셔너리 접근. JamJet은 Pydantic을 사용합니다 — 모든 단계 전환마다 필드가 검증됩니다. 스텝이 잘못된 형태를 반환하면 다운스트림에서 조용히 데이터가 손상되는 대신 즉시 오류가 발생합니다.

라우팅 문법

LangGraph는 add_conditional_edges에 전달되는 별도의 라우팅 함수가 필요합니다. JamJet 라우팅은 스텝에 인라인으로 작성합니다:

@wf.step(next={"branch_a": lambda s: s.flag, "branch_b": lambda s: not s.flag})
async def my_step(state: State) -> State: ...

단순한 선형 워크플로의 경우 아무것도 작성할 필요 없이 스텝이 선언 순서대로 실행됩니다.

내구성

LangGraph의 체크포인팅은 선택 사항이며 인프로세스 방식입니다(직접 설정하고 관리하는 SQLite, Redis, Postgres 어댑터). JamJet의 Rust 런타임은 기본적으로 내구성을 제공합니다 — 모든 단계 전환이 이벤트 소싱 방식의 쓰기입니다. 12개 스텝 중 7번째에서 크래시? 1번이 아닌 7번 스텝부터 재개합니다.

로컬 vs 프로덕션

두 모드 모두 동일한 API를 사용합니다:


# 개발 (인프로세스, 서버 없음)

result = wf.run_sync(State(question="..."))

# 프로덕션 (내구성 있는 Rust 런타임)

ir = wf.compile()

# jamjet dev  ← 다른 터미널에서 런타임 시작

# jamjet run workflow.yaml --input '{"question": "..."}'

빠른 마이그레이션

pip install jamjet

TypedDict를 pydantic.BaseModel로 교체
StateGraph + add_node + add_edge를 @wf.step으로 교체
조건부 라우팅: @wf.step(next={"target": lambda s: s.flag})
app.invoke(state)를 wf.run_sync(State(...))로 교체
프로덕션 준비 완료 시: wf.compile() → jamjet dev

팁: jamjet-labs/jamjet-benchmarks에서 전체 실행 예제를 확인하세요.

LangGraph에서 마이그레이션하기

On this page