Key facts
- A minimal runtime needs only nodes, action transitions, a flow runner, and shared state to teach the core idea.
- Model clients, vector stores, and web search should begin as application utilities outside the orchestration core.
- Deterministic tests should assert state transitions without real model calls.
The goal is a readable control-flow core
Do not start by building a framework. Start by writing the smallest runtime that can run one graph. It should have a node lifecycle, transitions, a shared state object, and predictable error behavior. Model clients, embedding functions, vector stores, web search, and persistence should be utilities outside the core.
This mirrors Pocket Flow's philosophy without copying its implementation. The point is not to hit exactly 100 lines. The point is to keep the orchestration surface small enough that the team can read it before changing it.
"Zero bloat, zero dependencies"
A tiny Python runtime
This is intentionally plain. It is not production-hardened, but it shows the irreducible pieces: each node can prepare, execute, postprocess, and return an action; each edge maps an action to the next node; each flow starts from one node and walks until no transition matches.
class Node:
def __init__(self, max_retries=1, wait=0):
self.max_retries = max_retries
self.wait = wait
self.transitions = {}
def prep(self, shared):
return None
def exec(self, prep_result):
return None
def post(self, shared, prep_result, exec_result):
return "default"
def fallback(self, prep_result, error):
raise error
def run(self, shared):
prep_result = self.prep(shared)
for attempt in range(self.max_retries):
try:
exec_result = self.exec(prep_result)
break
except Exception as error:
if attempt == self.max_retries - 1:
exec_result = self.fallback(prep_result, error)
action = self.post(shared, prep_result, exec_result)
return action or "default"
def on(self, action, node):
self.transitions[action] = node
return node
class Flow:
def __init__(self, start):
self.start = start
def run(self, shared):
current = self.start
while current:
action = current.run(shared)
current = current.transitions.get(action)A small search-and-answer graph
The graph below uses fake functions because provider calls are not the orchestration framework. A production app can call OpenAI, Anthropic, Gemini, a local model, a database, or a search API inside node methods without changing the runtime.
class Decide(Node):
def prep(self, shared):
return shared["question"], shared.get("notes", [])
def exec(self, inputs):
question, notes = inputs
if len(notes) == 0:
return {"action": "search", "query": question}
return {"action": "answer"}
def post(self, shared, prep_result, result):
if result["action"] == "search":
shared["query"] = result["query"]
return result["action"]
class Search(Node):
def prep(self, shared):
return shared["query"]
def exec(self, query):
return search_web(query)
def post(self, shared, prep_result, result):
shared.setdefault("notes", []).append(result)
return "decide"
class Answer(Node):
def prep(self, shared):
return shared["question"], shared["notes"]
def exec(self, inputs):
question, notes = inputs
return call_llm(question, context=notes)
def post(self, shared, prep_result, answer):
shared["answer"] = answer
return "done"
decide = Decide()
search = decide.on("search", Search())
answer = decide.on("answer", Answer())
search.on("decide", decide)
flow = Flow(decide)
shared = {"question": "How does a node graph agent work?"}
flow.run(shared)Test the graph, not the model
A minimal runtime is easy to test if you keep LLM calls outside the runtime. Write fake nodes that return known actions. Assert that shared state changed as expected. Assert that a repair action loops back. Assert that an unknown action stops or fails according to your policy.
The most important tests are state-transition tests. Model output quality needs evals, fixtures, and human review, but the orchestrator should have deterministic tests. If your graph cannot be tested without real model calls, your nodes own too much.
What not to add too early
Resist the temptation to add plugin loading, provider registries, built-in vector stores, prompt libraries, background queues, tracing dashboards, or deployment descriptors before the graph proves its value. Each feature may be useful later, but each one also changes the core from a readable runtime into an ecosystem.
Add production features at the edge first. Wrap provider calls. Persist shared state. Emit structured logs. Add per-node timing. Once the same concern appears in multiple graphs, then promote it into the runtime. That sequence keeps the core honest.
Sources used on this page
- Pocket Flow documentation The Pocket. Accessed July 6, 2026. Primary source for the 100-line claim, Graph + Shared Store abstraction, core patterns, and utility philosophy.
- The-Pocket/PocketFlow GitHub repository GitHub. Accessed July 6, 2026. Primary repository and README for installation, license, language ports, examples, and project framing.
- Pocket Flow Node documentation The Pocket. Accessed July 6, 2026. Primary source for prep, exec, post, retries, and fallback behavior.
- Pocket Flow Flow documentation The Pocket. Accessed July 6, 2026. Primary source for action-based transitions, branching, loops, and nested flows.
- Pocket Flow Communication documentation The Pocket. Accessed July 6, 2026. Primary source for Shared Store and Params guidance.
Cite this page
Build your own minimal orchestrator. PocketFlow AI Guide. Updated July 6, 2026. https://pocketflowai.com/build-your-own/
PocketFlow AI Guide. "Build your own minimal orchestrator." Accessed July 6, 2026. https://pocketflowai.com/build-your-own/