Migrating from LangGraph

This guide is for LangGraph users moving an app to loomflow. It translates the patterns you use most often, side by side. The table below is the summary; the sections after it show concrete code.

LangGraph concept	Loomflow equivalent
`StateGraph` + nodes + edges	`Workflow` (`add_node` / `add_edge` / `add_router`, `START` / `END`) when you define the graph; `Agent` + `Architecture` (twelve shipped) when the model drives
`add_messages` reducer	Automatic. Message history rehydrates from `session_id`
`config={"configurable": {"thread_id": "..."}}`	`agent.run(prompt, session_id="...")`
`config={"configurable": {"user_id": "..."}}` (convention)	`agent.run(prompt, user_id="...")` (built-in)
Checkpointer (`MemorySaver`, `SqliteSaver`)	`SqliteRuntime` + `Tuning(checkpoint=True)` for resume, `Memory` for recall, joined by `session_id`
Store API (`store.put(namespace, key, value)`)	`Memory.recall_facts` + `Fact` (typed, bi-temporal)
`tool_node` + tool routing	`tools=[...]` on `Agent`; the framework dispatches
`RunnableConfig` propagation	`RunContext` via `get_run_context()`
`stream_mode="values" / "updates" / "messages" / "debug"`	One `Event` stream with backpressure
`interrupt` / human-in-the-loop	Permission policies + an `approval_handler`; park a run on a signal and wake it with `agent.signal(...)`
Subgraphs	A `Workflow` or `Agent` as a node in another `Workflow`; multi-agent architectures (`Supervisor`, `Debate`, `Swarm`, …) compose `Agent` instances directly

Hello world


# LangGraph
from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-4.1-mini")
 
def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}
 
graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot)
graph.set_entry_point("chatbot")
graph.set_finish_point("chatbot")
app = graph.compile()
 
result = app.invoke({"messages": [{"role": "user", "content": "Hi."}]})
print(result["messages"][-1].content)


# Loomflow
import asyncio
from loomflow import Agent
 
async def main():
    agent = Agent("Be helpful.", model="gpt-4.1-mini")
    result = await agent.run("Hi.")
    print(result.output)
 
asyncio.run(main())

Tool calling


# LangGraph
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"sunny in {city}"
 
graph = create_react_agent(model="gpt-4.1-mini", tools=[get_weather])
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Weather in Tokyo?"}]}
)


# Loomflow
from loomflow import Agent, tool
 
@tool
async def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"sunny in {city}"
 
agent = Agent(
    "Use the weather tool when asked about weather.",
    model="gpt-4.1-mini",
    tools=[get_weather],
)
result = await agent.run("Weather in Tokyo?")

Explicit graphs

When you wire the nodes yourself, Workflow is the direct equivalent of StateGraph. Nodes receive the previous node’s return value instead of a shared state dict.


# LangGraph
from langgraph.graph import StateGraph, START, END
 
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("billing", billing)
graph.add_node("tech", tech)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", pick_route, {"billing": "billing", "tech": "tech"})
graph.add_edge("billing", END)
graph.add_edge("tech", END)
app = graph.compile()


# Loomflow
from loomflow import Workflow, START, END
 
wf = Workflow("support")
wf.add_node("classify", classify)
wf.add_node("billing", billing)
wf.add_node("tech", tech)
wf.add_edge(START, "classify")
wf.add_router("classify", pick_route, {"billing": "billing", "tech": "tech"})
wf.add_edge("billing", END)
wf.add_edge("tech", END)
result = await wf.run("My invoice is wrong", user_id="alice")

A node can be a function, an Agent or another Workflow. See the Workflow reference.

Multi-tenant memory

The biggest correctness gap in LangGraph: user_id is a string in config["configurable"]. Typo it once and you silently leak data across tenants. Loomflow makes user_id a typed argument that the framework honours, and every memory backend partitions by it automatically.


# LangGraph: user_id is a CONVENTION you have to honour by hand
from langgraph.store.memory import InMemoryStore
 
store = InMemoryStore()
 
def my_node(state, config):
    user_id = config["configurable"]["user_id"]   # ← typo here = leak
    namespace = (user_id, "memories")
    memories = store.search(namespace, query=state["messages"][-1].content)
    # ...


# Loomflow: user_id is a typed argument; partitioning is automatic
from loomflow import Agent, get_run_context, tool
 
# Inside any tool: no need to pass user_id through signatures.
@tool
async def fetch_orders() -> str:
    ctx = get_run_context()
    return await db.query("orders", user_id=ctx.user_id)
 
agent = Agent("...", model="gpt-4.1-mini", tools=[fetch_orders])
 
# user_id is the call kwarg; framework partitions memory recall.
await agent.run("show my orders", user_id="alice")

Conversation continuity


# LangGraph: the checkpointer wires up state replay
from langgraph.checkpoint.memory import MemorySaver
 
graph = ... .compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "conv-42"}}
 
graph.invoke({"messages": ["Hi, I'm Alice."]}, config)
graph.invoke({"messages": ["What's my name?"]}, config)
# → "Alice", because the checkpointer rehydrated the thread state.


# Loomflow: reuse the same session_id and the conversation continues
agent = Agent("...", model="gpt-4.1-mini")
await agent.run("Hi, I'm Alice.", session_id="conv-42", user_id="alice")
await agent.run("What's my name?", session_id="conv-42", user_id="alice")
# → "Alice", because session_messages rehydrated the prior turns.

Structured output

LangGraph offloads this to whatever model adapter you wire up (ChatOpenAI(model="...").with_structured_output(MySchema)). Loomflow ships it as a built-in kwarg with retry-on-validation-failure.


# LangGraph
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-4.1-mini").with_structured_output(MyOutput)
result = llm.invoke([{"role": "user", "content": "..."}])
# result is a MyOutput instance, but errors aren't fed back as a
# retry; you handle ValidationError yourself.


# Loomflow
from loomflow import Agent
 
agent = Agent("...", model="gpt-4.1-mini")
result = await agent.run("...", output_schema=MyOutput)
output: MyOutput = result.parsed   # ← validated; retry-on-fail built in

Streaming


# LangGraph: pick a stream_mode (4 incompatible flavours)
async for chunk in graph.astream(input_, config, stream_mode="messages"):
    ...


# Loomflow: one Event stream, backpressure-aware
async for event in agent.stream(prompt, session_id="conv-42"):
    if event.kind == "model_chunk":
        chunk = event.payload["chunk"]
        if chunk["kind"] == "text":
            print(chunk["text"], end="", flush=True)

Multi-agent

LangGraph’s subgraphs are compiled graphs inside compiled graphs, which has known issues around config propagation and checkpointing. Loomflow’s multi-agent architectures compose Agent instances directly, and sub-agents inherit the parent’s RunContext automatically.


# Loomflow: the Team builder covers the common shapes
from loomflow import Agent
from loomflow.team import Team
 
researcher = Agent("Research the topic.", model="gpt-4.1-mini")
writer = Agent("Draft the answer.", model="gpt-4.1-mini")
 
team = Team.supervisor(
    workers={"researcher": researcher, "writer": writer},
    model="gpt-4.1-mini",
)
result = await team.run("Write a brief about Acme Corp.", user_id="alice")

Things Loomflow does NOT have

No shared-state reducers. In a Workflow, each node’s return value is passed along its edges to the next node (a join receives a list). Inside an Agent, the loop is a strategy; twelve are shipped. For something custom, implement the Architecture protocol (a name, an async-generator run, and declared_workers).
No RunnableConfig / configurable dict. Use kwargs (user_id, session_id, metadata) and the typed RunContext.
No 4-mode streaming. One Event stream covers all cases; filter by event.kind if you only want a subset.