Migrating from LangGraph

Side-by-side translations of the patterns LangGraph users hit most often. The table below is the executive summary; the sections after it show concrete code.

LangGraph concept	Loom equivalent
`StateGraph` + nodes + edges + reducers	`Agent` + `Architecture` (twelve shipped)
`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` + `Memory` together. Split into journaling and 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`. 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	Hooks (`@agent.before_tool` returns a denial) + permission policies
Subgraphs	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)


# Loom
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}"
 
agent = create_react_agent(model="gpt-4.1-mini", tools=[get_weather])
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Weather in Tokyo?"}]}
)


# Loom
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?")

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. Loom makes user_id a typed primitive that the framework honours. 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)
    # ...


# Loom — user_id is a typed primitive; partition is automatic
from loomflow import Agent, get_run_context
 
# Inside any tool — never plumb 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.


# Loom — same session_id reused = 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)). Loom 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.


# Loom
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"):
    ...


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

Multi-agent

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


# Loom — Team facade for 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 Loom does NOT have

No graph editor / state-graph DSL. The agent loop is a strategy; twelve are shipped. If you need something custom, implement the Architecture protocol (one async generator method).
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.