Site icon Efficient Coder

ClearFlow: The Tiny Type-Safe LLM Workflow Engine for Reliable AI Applications

高效码农

Build Reliable LLM Workflows with ClearFlow: A Practical 3,000-Word Guide

“

Reading time: ~12 minutes

Table of Contents

  1. What Exactly Is ClearFlow?
  2. Why Not Just Write Plain Python?
  3. One-Command Installation & Your First 60-Second “Hello LLM”
  4. The Three Core Concepts—Node, NodeResult, Flow
  5. End-to-End Walkthrough: A Multi-Step Data Pipeline
  6. Testing, Debugging & Lessons From the Trenches
  7. ClearFlow vs. PocketFlow: Side-by-Side Facts
  8. Frequently Asked Questions (FAQ)
  9. Where to Go Next

1. What Exactly Is ClearFlow?

ClearFlow is a tiny, type-safe, async-first workflow engine for language-model applications.
Everything you need is contained in a single 166-line file with zero runtime dependencies. You bring your own LLM client—OpenAI, Anthropic, or a local model—and ClearFlow handles routing, state immutability, and graceful termination.

Key Traits

  • Explicit routing – every outcome maps to exactly one next step
  • Immutable state – each node receives a frozen copy, eliminating side effects
  • Single exit rule – one mandatory None route prevents runaway jobs
  • Tiny API surface – only three concepts: Node, NodeResult, Flow
  • 100 % test coverage – every line is exercised in CI
  • Zero runtime dependencies – the install footprint is < 5 kB

2. Why Not Just Write Plain Python?

Imagine a typical script:

# pseudo-code, the naive way
result = llm.chat(messages)
if result.ok:
    processed = result.text * 2
    save(processed)
else:
    log_error(result.error)

Problems you will meet:

Naive Script ClearFlow Approach
State can be mutated anywhere, making bugs hard to trace State is frozen after every node
Business logic and routing are mixed together Routing is declared separately and read like a map
Unit tests need mocks for the entire world Each node is a pure async function—tested like math
Dependency hell on deployment day ClearFlow has no runtime dependencies

3. One-Command Installation & Your First 60-Second “Hello LLM”

3.1 Install

# If you do not have uv yet
pip install --user uv      # or: pipx install uv

# Install ClearFlow from PyPI
pip install clearflow

3.2 The Smallest Working Example

Below you will build a chatbot that replies “Hello!” to any incoming “Hi”.
It demonstrates state definition, node creation, flow assembly, and async execution—all in 25 lines.

import asyncio
from typing import TypedDict
from clearflow import Flow, Node, NodeResult

# 1. Define the shape of your state
class ChatState(TypedDict):
    messages: list[dict[str, str]]

# 2. Create a node that appends an assistant reply
class ChatNode(Node[ChatState]):
    async def exec(self, state: ChatState) -> NodeResult[ChatState]:
        # In real code, call your LLM here
        # reply = await openai_client.chat.completions.create(...)
        reply = {"role": "assistant", "content": "Hello!"}
        new_state: ChatState = {"messages": [*state["messages"], reply]}
        return NodeResult(new_state, outcome="success")

# 3. Wire the nodes into a flow
chat = ChatNode()
flow = (
    Flow[ChatState]("ChatBot")
    .start_with(chat)
    .route(chat, "success", None)  # “success” terminates the flow
    .build()
)

# 4. Run it
async def main():
    result = await flow({"messages": [{"role": "user", "content": "Hi"}]})
    print(result.state["messages"][-1]["content"])  # -> Hello!

if __name__ == "__main__":
    asyncio.run(main())

Run the file and you will see:

Hello!

4. The Three Core Concepts—Node, NodeResult, Flow

4.1 Node: the smallest unit of work

A node is any class that inherits from Node[T] and implements at least one required method.

class MyNode(Node[int]):
    async def prep(self, state: int) -> int:
        # Optional: validation or enrichment
        return state

    async def exec(self, state: int) -> NodeResult[int]:
        # Required: the actual business logic
        return NodeResult(state + 1, outcome="ok")

    async def post(self, result: NodeResult[int]) -> NodeResult[int]:
        # Optional: logging, metrics, cleanup
        return result

Guidelines:

  • Keep each node stateless; it only receives a copy of the data.
  • All three methods are async—feel free to use await inside.

4.2 NodeResult: a simple envelope

NodeResult(new_state, outcome="continue")
  • new_state is the immutable snapshot passed to the next node.
  • outcome is a plain string used by the flow router.

4.3 Flow: the declarative map

Think of a flow as a train timetable: every station (node) and every possible departure (outcome) must be listed.

flow = (
    Flow[int]("MyFlow")
    .start_with(validate)
    .route(validate, "ok", process)
    .route(validate, "bad", error_handler)
    .route(process, "done", None)  # single termination
    .build()
)

Rules:

  • Exactly one route must end with None—this is your single exit gate.
  • A flow is itself a node; you can nest flows inside larger flows.

5. End-to-End Walkthrough: A Multi-Step Data Pipeline

Scenario:

  • Accept an integer from the user.
  • If negative, print an error and stop.
  • If non-negative, double it and print the result.

5.1 Define the state

class State(TypedDict):
    value: int

5.2 Implement the three nodes

class Validate(Node[State]):
    async def exec(self, s: State) -> NodeResult[State]:
        if s["value"] >= 0:
            return NodeResult(s, "valid")
        return NodeResult(s, "invalid")

class Process(Node[State]):
    async def exec(self, s: State) -> NodeResult[State]:
        return NodeResult({"value": s["value"] * 2}, "success")

class Output(Node[State]):
    async def exec(self, s: State) -> NodeResult[State]:
        print("Final:", s["value"])
        return NodeResult(s, "done")

5.3 Assemble and run

flow = (
    Flow[State]("Pipeline")
    .start_with(Validate())
    .route("Validate", "valid", Process())
    .route("Validate", "invalid", Output())
    .route("Process", "success", Output())
    .route("Output", "done", None)
    .build()
)

# First run
await flow({"value": 21})
# Console: Final: 42

# Second run
await flow({"value": -5})
# Console: Final: -5

6. Testing, Debugging & Lessons From the Trenches

6.1 Unit-testing a node

Because nodes are pure async functions, you can test them like algebra:

import pytest
from clearflow import Node, NodeResult

class Increment(Node[int]):
    async def exec(self, x: int) -> NodeResult[int]:
        return NodeResult(x + 1, "ok")

@pytest.mark.asyncio
async def test_increment():
    res = await Increment()(0)
    assert res.state == 1 and res.outcome == "ok"

6.2 Debugging checklist

  • Forgot a route? The .build() call will raise ValueError: exactly one None route required.
  • State looks wrong? Add a print in post; the state is frozen, so you can safely log the entire object.
  • Asyncio warning? Ensure you run the flow inside asyncio.run() or an event loop.

7. ClearFlow vs. PocketFlow: Side-by-Side Facts

Aspect ClearFlow PocketFlow
State handling Immutable copies Shared mutable dict
Routing style Explicit (node, outcome) table Graph with labelled edges
Termination rule Exactly one None route enforced Multiple exit patterns allowed
Type safety Full Python 3.13+ generics Dynamic typing
Lines of source 166 100

In short:

  • Choose ClearFlow when type safety, explicit control, and long-term maintainability matter.
  • Choose PocketFlow when you need the shortest script possible and are comfortable with a shared state.

8. Frequently Asked Questions (FAQ)

Q1: Does ClearFlow support parallel or concurrent nodes?
A: Nodes are async, so you can use asyncio.gather inside any node. The framework itself does not yet provide built-in parallel-split syntax.

Q2: Can I embed a flow inside another flow?
A: Yes. After calling .build(), a flow is a node. Pass it to .route() like any other node.

Q3: How do I plug in OpenAI or Claude?
A: Inside the node’s exec method, call your preferred SDK and place the returned text into the new state. ClearFlow does not ship with any HTTP client.

Q4: Is Python 3.13 mandatory?
A: The examples use 3.13-style generics. On 3.10+ you can switch to from typing import List, Dict.

Q5: How do I log in production?
A: Use the optional post method. Because the state is immutable, you can safely serialize and persist it without worrying about concurrent mutations.

9. Where to Go Next

If you have reached this point, you already know enough to ship a small production service. Recommended next steps:

  1. Read the source: clearflow.py is 166 lines—perfect for a coffee break.
  2. Browse examples: the examples/chat and examples/structured_output folders in the repository.
  3. Clone for development: 
    git clone https://github.com/consent-ai/ClearFlow.git
cd ClearFlow
uv sync --group dev
./quality-check.sh

May your workflows stay predictable and your debugging sessions short.

Exit mobile version