Coming from LangGraph or LangChain

If you know LangGraph or LangChain's agents, most of CriogenAI will look familiar: a graph of steps over a shared state, run in steps, saved by a checkpointer, paused for a person. This page maps the ideas one to one, shows the same program in both, and lists the differences that change how you write code. The LangGraph snippets are short sketches to show the idea, not complete programs.

The map#

LangGraph / LangChainCriogenAINotes
StateGraph(State) with a TypedDictWorkflow()The state is plain JSON: no schema class
add_node(name, fn)@wf.node, wf.add_node(name, fn), .node(name, fn)Same contract: return the keys you change
add_edge(START, "a")wf.start("a")
add_edge("a", "b")wf.edge("a", "b")
add_conditional_edges("a", fn, path_map)wf.route("a", fn, {label: node})Labels must be declared; an unknown label is an error (unknown_route)
Annotated[list, operator.add]Workflow(reducers={"key": "append"})Two writes to a key in one step without a reducer fail in both (here: update_conflict)
add_messages"upsert_by_id"Agent nodes write the whole transcript back, so keep the default for messages
Command(goto=..., update=...)goto(targets, update)The node's possible targets are declared with wf.goto(node, [...])
Waiting for several brancheswf.join([a, b], c)
Send (a dynamic fan-out)noneFixed fan-out: several edges and a join. Dynamic: asyncio.gather / Promise.all inside one node
compile(checkpointer=...)wf.compile(storage=...), Agent(storage=...)
InMemorySaver, SqliteSaver, PostgresSaverStorage(), Storage("sqlite://..."), Storage("postgres://...")One storage also holds memory and the audit trail
config={"configurable": {"thread_id": ...}}agent.thread(id), app.thread(id)A handle instead of a config dict
interrupt(value) and Command(resume=...)suspend(payload) and thread.resume({node: value})The paused node runs again with ctx.resume, as with interrupt
Human approval of tool callsAgent(approve=[...]), then approve, reject, decideNo node to write: the agent pauses before the tools run
get_state_history, time travelthread.history(), thread.rewind(id)No update_state: rewind, then send
stream_mode="messages" / "updates"thread.stream(...) eventstext, tool_call_started, approval_needed, ...
recursion_limitmax_steps (workflows), limits (agents)Agents always have limits (25 model calls by default)
RetryPolicy on a nodemodel_retry, tool_retry on agentsPer-node retry policies are in Rust only
create_react_agent / create_agentAgent(model, tools=[...])The loop runs in the engine; no graph to build
@tool (langchain_core.tools)@tool (Python), tool({...}) (TypeScript)Arguments are checked against the schema by the engine before your function runs
ToolNodebuilt into the agentTool calls run concurrently; results in call order
BaseStore: put, search with tuple namespacesstorage.remember, storage.search with path namespaces"users/1" instead of ("users", "1")
Memory tools for the agentAgent(memory=Memory.by_state(...))Recall before each model call, plus remember / forget / search_memory
Middleware / callbacksHooksBefore and after every model and tool call
LangSmithTracing: OTLP, Langfuse, a local viewer
init_chat_model("openai:...")"openai:gpt-5-mini"The same provider:model string style

A graph#

LangGraph, in outline:

# LangGraph (sketch)
class State(TypedDict):
    steps: list[str]
    answer: str
    log: Annotated[list[str], operator.add]

graph = StateGraph(State)
graph.add_node("plan", plan)
graph.add_node("write", write)
graph.add_edge(START, "plan")
graph.add_conditional_edges("plan", lambda s: "write" if s["steps"] else END)
graph.add_edge("write", END)
app = graph.compile()
app.invoke({"question": "..."})

CriogenAI:

from criogenai import END, Workflow

wf = Workflow(reducers={"log": "append"})  # like an Annotated reducer on a state key

@wf.node
def plan(state: dict) -> dict:
    return {"steps": ["search", "write"], "log": ["planned"]}

@wf.node
def write(state: dict) -> dict:
    return {"answer": f"Done in {len(state['steps'])} steps", "log": ["written"]}

wf.start("plan")  # START -> plan
wf.route("plan", lambda s: "write" if s["steps"] else "stop", {"write": "write", "stop": END})
wf.edge("write", END)
app = wf.compile()  # validates the graph; no state class needed

result = await app.run({"question": "..."})
print(result.state["answer"], result.state["log"])
import { END, Workflow } from "criogenai";

const app = new Workflow({ reducers: { log: "append" } }) // like a reducer on a state channel
  .node("plan", () => ({ steps: ["search", "write"], log: ["planned"] }))
  .node("write", (state) => ({ answer: `Done in ${(state.steps as string[]).length} steps`, log: ["written"] }))
  .start("plan") // START -> plan
  .route("plan", (state) => ((state.steps as string[]).length ? "write" : "stop"), { write: "write", stop: END })
  .edge("write", END)
  .compile(); // validates the graph; no state schema needed

const result = await app.run({ question: "..." });
console.log(result.state.answer, result.state.log);

What changes:

  • No state class. Keys are whatever nodes write; give a reducer only to keys that several nodes write in the same step.
  • compile() checks the whole graph and reports every problem at once (unknown nodes, unreachable nodes, nodes with no way out), as a ValidationError with details["issues"].
  • A router sees the state after the step's updates are merged, including what parallel nodes of the same step wrote.

An agent with a checkpointer#

LangGraph, in outline:

# LangGraph (sketch)
agent = create_react_agent(model, tools=[get_weather], checkpointer=SqliteSaver(conn))
agent.invoke(
    {"messages": [{"role": "user", "content": "Weather in Paris?"}]},
    config={"configurable": {"thread_id": "user-1"}},
)

CriogenAI:

from criogenai import Agent, Storage

agent = Agent(
    "openai:gpt-5-mini",
    tools=[get_weather],
    instructions="Answer briefly.",
    storage=Storage("sqlite://chat.db"),  # the checkpointer
)
# thread_id in the config becomes a thread handle
reply = await agent.thread("user-1").send("Weather in Paris?")
print(reply.text)
import { Agent, Storage } from "criogenai";

const agent = new Agent({
  model: "openai:gpt-5-mini",
  tools: [getWeather],
  instructions: "Answer briefly.",
  storage: new Storage("sqlite://chat.db"), // the checkpointer
});
// thread_id in the config becomes a thread handle
const reply = await agent.thread("user-1").send("Weather in Paris?");
console.log(reply.text);

The thread handle replaces the config dict, and the reply carries the answer, the transcript, usage and the checkpoint it saved. One run uses a thread at a time: a second send waits (or is rejected) instead of racing.

Human in the loop#

In LangGraph you call interrupt() inside a node (often a node in front of the tools) and resume with Command(resume=...). In CriogenAI, approval of tool calls is an agent option; the four answers (approve, edit, reject, respond) are built in:

from criogenai import Agent, Storage

agent = Agent(
    "openai:gpt-5-mini",
    tools=[get_weather],
    approve=["get_weather"],  # pause before this tool runs: no interrupt() call to write
    storage=Storage("sqlite://chat.db"),
)
chat = agent.thread("user-1")
reply = await chat.send("Weather in Paris?")
if reply.needs_approval:  # like an __interrupt__ in the result
    reply = await chat.approve()  # like resuming with a Command
const careful = new Agent({
  model: "openai:gpt-5-mini",
  tools: [getWeather],
  approve: ["get_weather"], // pause before this tool runs: no interrupt() call to write
  storage: new Storage("sqlite://chat.db"),
});
const chat = careful.thread("user-2");
let answer = await chat.send("Weather in Paris?");
if (answer.needsApproval) {
  // like an __interrupt__ in the result
  answer = await chat.approve(); // like resuming with a Command
}

For pauses that are not about tool calls, a workflow node returns suspend(payload); thread.resume({node: value}) continues it, and the node runs again with ctx.resume == value, much like interrupt returning its resume value. See Human approval and Workflows.

A store for long-term memory#

LangGraph's store keeps JSON documents under tuple namespaces:

# LangGraph (sketch)
store.put(("users", "1"), "language", {"value": "French"})
store.search(("users", "1"), query="language")

CriogenAI's memory lives in the same storage as threads, under path namespaces, and an agent can use it directly:

from criogenai import Agent, Memory, Storage

storage = Storage("sqlite://chat.db")
await storage.remember("users/1", "language", "French")  # namespace as a path, not a tuple
facts = await storage.search("users/1", "language")

# Or let the agent recall and save facts itself, per user:
agent = Agent("openai:gpt-5-mini", storage=storage, memory=Memory.by_state("user_id", "users"))
await agent.thread("user-1").send("Bonjour", user_id="1")
import { Memory } from "criogenai";

const storage = new Storage("sqlite://chat.db");
await storage.remember("users/1", "language", "French"); // namespace as a path, not an array
const facts = await storage.search("users/1", "language");

// Or let the agent recall and save facts itself, per user:
const remembering = new Agent({ model: "openai:gpt-5-mini", storage, memory: Memory.byState("user_id", "users") });
await remembering.thread("user-3").send("Bonjour", { state: { user_id: "1" } });

Differences that change how you write code#

  • Strict by default. Tool arguments, edited approvals, structured output and graphs are validated, and problems are errors with stable codes, not warnings. A typo in a route label fails at once.
  • The engine is not in your language. Scheduling, merging, validation, retries and saving run in Rust; your Python or TypeScript functions are the nodes and tools. The state and messages are JSON, so a thread saved from Python can be continued from TypeScript.
  • Agents are not graphs you assemble. The model/tools loop, approvals, structured output, memory recall and retries are options of Agent. You build a graph only for your own steps.
  • Saved per tool call. Finished tool calls are saved one by one, so retry() after a crash never repeats them.
  • One run per thread. Concurrent calls on one thread wait, are rejected, or interrupt the running one (on_busy).

What CriogenAI does not have (yet)#

Be sure these are not load-bearing for you before you move:

  • No Send: a dynamic number of parallel branches needs one node that runs them itself.
  • No functional API (@entrypoint / @task): use a Workflow.
  • No update_state: you cannot edit a thread's state directly; rewind or send a new message.
  • No subgraph type: put an agent in a node, or call another compiled workflow from a node.
  • No large catalog of integrations (document loaders, many vector stores): CriogenAI has its own collections and MCP for outside tools.

The status page lists what is built per language.

Moving a project#

  1. Install CriogenAI next to LangGraph; nothing conflicts.
  2. Move tools first: a LangChain @tool function usually needs only the import changed in Python. Check the schema with tool.schema.
  3. Replace create_react_agent / create_agent with Agent, and the checkpointer with a Storage. Approvals become approve=[...].
  4. Move custom graphs node by node: the node functions often stay as they are (they take the state and return a dict).
  5. Existing LangGraph checkpoints cannot be read by CriogenAI: start new threads, or copy the messages you need into the first message of a new thread.
  6. Test with FakeModel (Testing) before switching traffic.

Next#