Migrando desde LangGraph

Comparación de código lado a lado y mapeo de conceptos para desarrolladores que migran de LangGraph a JamJet.

Mapeo de conceptos

LangGraph	JamJet
Estado `TypedDict`	Estado `pydantic.BaseModel` — validado en cada paso
`StateGraph`	`Workflow`
`graph.add_node("name", fn)`	Decorador `@workflow.step`
`graph.add_conditional_edges(node, router_fn)`	`@workflow.step(next={"target": predicate})`
`graph.add_edge(A, B)`	Secuencial por defecto; `next=` para ramas
`graph.compile()`	`workflow.compile()` → IR para el runtime de Rust
`app.invoke(state)`	`workflow.run_sync(state)` (local)
`app.astream(state)`	`workflow.run(state)` (asíncrono, local)
`MemorySaver` / `PostgresSaver`	Integrado en el runtime de Rust — automático
`interrupt_before` (humano en el bucle)	Nodo `type: wait` o `human_approval=True` en el paso

Ejemplo comparativo

Un agente multi-paso con enrutamiento condicional — decide si buscar, luego sintetiza una respuesta.

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

# Ejecución local — no necesitas servidor

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

# Producción: wf.compile() + jamjet dev

LangGraph usa TypedDict — acceso a diccionarios sin validación. JamJet usa Pydantic — los campos se validan en cada transición de paso. Si un paso devuelve la forma incorrecta, obtienes un error inmediatamente en lugar de corrupción silenciosa de datos más adelante.

Sintaxis de enrutamiento

LangGraph requiere una función de enrutamiento separada pasada a add_conditional_edges. El enrutamiento de JamJet está integrado en el paso:

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

Para flujos de trabajo lineales simples, no escribes nada — los pasos se ejecutan en orden de declaración.

Durabilidad

El checkpointing de LangGraph es opcional y en proceso (adaptadores SQLite, Redis, Postgres que configuras y gestionas tú). El runtime de Rust de JamJet es durable por defecto — cada transición de paso es una escritura basada en eventos. ¿Falla en el paso 7 de 12? Se reanuda desde el paso 7, no desde el paso 1.

Local vs producción

Ambos modos usan la misma API:


# Desarrollo (en proceso, sin servidor)

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

# Producción (runtime durable de Rust)

ir = wf.compile()

# jamjet dev  ← iniciar runtime en otra terminal

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

Migración rápida

pip install jamjet

Reemplaza TypedDict con pydantic.BaseModel
Reemplaza StateGraph + add_node + add_edge con @wf.step
Para enrutamiento condicional: @wf.step(next={"target": lambda s: s.flag})
Reemplaza app.invoke(state) con wf.run_sync(State(...))
Cuando estés listo para producción: wf.compile() → jamjet dev

consejo: Ejemplos completos en jamjet-labs/jamjet-benchmarks.