Migration von LangGraph

Code-Vergleich und Konzept-Mapping für Entwickler, die von LangGraph zu JamJet wechseln.

Konzept-Mapping

LangGraph	JamJet
`TypedDict` State	`pydantic.BaseModel` State — validiert bei jedem Schritt
`StateGraph`	`Workflow`
`graph.add_node("name", fn)`	`@workflow.step` Dekorator
`graph.add_conditional_edges(node, router_fn)`	`@workflow.step(next={"target": predicate})`
`graph.add_edge(A, B)`	Sequenziell standardmäßig; `next=` für Verzweigungen
`graph.compile()`	`workflow.compile()` → IR für die Rust-Runtime
`app.invoke(state)`	`workflow.run_sync(state)` (lokal)
`app.astream(state)`	`workflow.run(state)` (async, lokal)
`MemorySaver` / `PostgresSaver`	In die Rust-Runtime integriert — automatisch
`interrupt_before` (human-in-the-loop)	`type: wait` Node oder `human_approval=True` am Step

Vergleichsbeispiel

Ein mehrstufiger Agent mit bedingtem Routing — entscheide, ob gesucht werden soll, und erzeuge dann eine Antwort.

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

# Lokale Ausführung — kein Server erforderlich

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

# Produktion: wf.compile() + jamjet dev

LangGraph verwendet TypedDict — Dictionary-Zugriff ohne Validierung. JamJet nutzt Pydantic — Felder werden bei jedem Schritt-Übergang validiert. Wenn ein Schritt die falsche Struktur zurückgibt, erhältst du sofort einen Fehler, statt stiller Datenkorruption im weiteren Verlauf.

Routing-Syntax

LangGraph erfordert eine separate Routing-Funktion, die an add_conditional_edges übergeben wird. Bei JamJet ist das Routing inline im Step:

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

Für einfache lineare Workflows schreibst du nichts — Steps werden in Deklarationsreihenfolge ausgeführt.

Durability

LangGraphs Checkpointing ist opt-in und in-process (SQLite, Redis, Postgres Adapter, die du konfigurierst und verwaltest). JamJets Rust-Runtime ist standardmäßig durable — jeder Schritt-Übergang ist ein event-sourced Write. Absturz bei Schritt 7 von 12? Fortsetzung ab Schritt 7, nicht Schritt 1.

Lokal vs. Produktion

Beide Modi verwenden dieselbe API:


# Development (in-process, kein Server)

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

# Produktion (durable Rust-Runtime)

ir = wf.compile()

# jamjet dev  ← Runtime in separatem Terminal starten

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

Schnelleinstieg Migration

pip install jamjet

Ersetze TypedDict durch pydantic.BaseModel
Ersetze StateGraph + add_node + add_edge durch @wf.step
Für bedingtes Routing: @wf.step(next={"target": lambda s: s.flag})
Ersetze app.invoke(state) durch wf.run_sync(State(...))
Wenn produktionsbereit: wf.compile() → jamjet dev

Tipp: Vollständige funktionierende Beispiele in jamjet-labs/jamjet-benchmarks.