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 / LangChain | CriogenAI | Notes |
|---|---|---|
StateGraph(State) with a TypedDict | Workflow() | 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 branches | wf.join([a, b], c) | |
Send (a dynamic fan-out) | none | Fixed fan-out: several edges and a join. Dynamic: asyncio.gather / Promise.all inside one node |
compile(checkpointer=...) | wf.compile(storage=...), Agent(storage=...) | |
InMemorySaver, SqliteSaver, PostgresSaver | Storage(), 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 calls | Agent(approve=[...]), then approve, reject, decide | No node to write: the agent pauses before the tools run |
get_state_history, time travel | thread.history(), thread.rewind(id) | No update_state: rewind, then send |
stream_mode="messages" / "updates" | thread.stream(...) events | text, tool_call_started, approval_needed, ... |
recursion_limit | max_steps (workflows), limits (agents) | Agents always have limits (25 model calls by default) |
RetryPolicy on a node | model_retry, tool_retry on agents | Per-node retry policies are in Rust only |
create_react_agent / create_agent | Agent(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 |
ToolNode | built into the agent | Tool calls run concurrently; results in call order |
BaseStore: put, search with tuple namespaces | storage.remember, storage.search with path namespaces | "users/1" instead of ("users", "1") |
| Memory tools for the agent | Agent(memory=Memory.by_state(...)) | Recall before each model call, plus remember / forget / search_memory |
| Middleware / callbacks | Hooks | Before and after every model and tool call |
| LangSmith | Tracing: 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 aValidationErrorwithdetails["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 Commandconst 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 aWorkflow. - 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#
- Install CriogenAI next to LangGraph; nothing conflicts.
- Move tools first: a LangChain
@toolfunction usually needs only the import changed in Python. Check the schema withtool.schema. - Replace
create_react_agent/create_agentwithAgent, and the checkpointer with aStorage. Approvals becomeapprove=[...]. - Move custom graphs node by node: the node functions often stay as they are (they take the state and return a dict).
- 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.
- Test with
FakeModel(Testing) before switching traffic.
Next#
- Tutorial: build a support agent.
- Agents and the loop and How a run works.
- Benchmarks: measured comparisons, with how they were measured.