Less Sloppy Slop - Tinier Abstractions for LLM

TSAICode

My "Initial Commit" post did not age well. Regardless, in an attempt to revitalize my blog, I would like to discuss something I have been working on lately. That is, reducing slop in a slop world.

Everyone and their grandmother is building GenAI applications it feels like. And, while I recognize the magnitude of the hype does not match the magnitude of the value, I still enjoy poking around with cool-shiny-things especially as an avid /r/localllama peruser.

So, keep that in mind that any workflow, graph, or even basic application I work with in my personal time will not inherit the capabilities of SOTA models that can get away with far longer, more detailed, or even more vague instructions. It needs to be distilled, simple from a prompt perspective, and fail fast.

What do we really need?

If I were to ask someone at random to recommend a framework for building GenAI applications, I would likely hear LangChain, LangGraph, maybe PydanticAI if someone feels rather creative, but I would hazard a guess no, "OpenAI package is all you need" would be said. Doing so might send shivers down an AI-wrapper-start up founder's spine.

Humor aside, at the bare minimum we need a few things:

Some "pipeline" composed of steps.
A meaningful way of maintaining state across the pipeline.
A code-driven approach for transitioning between nodes.

The philosophical framework aligns with KISS (keep-it-simple-stupid). In general, building with AI becomes easier if you assume the model has a default flag for stupid: true. This becomes exponentially more apparent as parameter size decreases.

Before diving into a pseudo-code implementation of the basics, I think it is worth mentioning, this approach is not specific to AI. In fact, graphs like this are useful for data-processing, in-game world simulations, and more. A state graph is nothing more than a fancy mental model for a set of functions, function calls, variables, and conditionals. I do, however, prefer this tiny amount of abstraction as it compartmentalizes each node, and it's I/O, from the entire application reducing the cognitive load and increasing the productivity and reusability.

Common Pattern

Let's start with what I would consider a typical RAG workflow.

stateDiagram-v2
    direction LR
    userprompt : User prompt
    grading: Grader Agent
    redosearch: Rewriter Agent
    qanode: Answer Node

    [*] --> userprompt
    userprompt --> Embedding
    Embedding --> Retrieval
    Retrieval --> grading
    grading --> qanode
    grading --> redosearch: rel docs < 1
    redosearch --> Embedding: n < 3
    redosearch --> qanode
    qanode --> [*]

The idea here is that we have to answer a user's question based off a set of embedded documents. While a large model would likely get less-thrown off by irrelevant documents returned, smaller models tend to get side-tracked far more easily.

So, we add a "Grader Agent" or, essentially, an LLM call with structured output that determines relevance in accordance to what the user asked.

Likewise, there is a non-zero chance the LLM decides to rain on our parade, or that the results are sub-par, we could call a "Rewriter Agent" to reformulate the query grounded on the sources we found and the structure/content of the documents, or to broaden the search.

Pseudocode for Graph

This might seem rudimentary, but bear with me. My high school professor was obsessed with finite-state-machines and game development which shaped my early learning, but I realize a lot of programmers aren't as familiar with this concept. For example, in university, we spent more time on AVL Trees and Discrete mathematics than we did basic graphs. The former two are often less useful in the enterprise software world than the latter.

Sidebar: My high school professor was awesome. Getting exposed to the rabbit hole of cellular automaton, game theory, and more at a younger age was deeply impactful on how I view the world.

1BEGIN NODE
2    SIGNATURE: Execute(CURRENT_STATE, AVAILABLE_TRANSITIONS = None) -> (
3        PATCH, NEXT_NODE
4    )
5
6    [PLACEHOLDER LOGIC]  # Each node will implement this
7
8    RETURN: PATCH        # a state diff to be applied with apply_patch()
9    RETURN: NEXT_NODE    # node identifier or None
10END NODE

1BEGIN GRAPH
2    SIGNATURE: Run(INITIAL_STATE) -> (CURRENT_STATE)
3
4    VAR: NEXT_NODE = START_NODE
5    VAR: CURRENT_STATE = INITIAL_STATE
6
7    WHILE NEXT_NODE != None
8        VAR: PATCH, NEXT_NODE = NEXT_NODE.Execute(
9            CURRENT_STATE, AVAILABLE_TRANSITIONS = None
10        )
11
12        CURRENT_STATE = apply_patch(CURRENT_STATE, PATCH)
13    ENDWHILE
14
15    RETURN: CURRENT_STATE
16END GRAPH

This is really all you need for a directed graph. I like to think of this as a O(n) complexity of implementation (excluding types) just because it is that straight forward. Thank brython for existing so you can run it as-is in the browser.

Python Implementation

1from typing import Callable, Dict, Optional, Tuple, Any
2
3
4def apply_patch(
5    state: Dict[str, Any], patch: Optional[Dict[str, Any]]
6) -> Dict[str, Any]:
7    """
8    Return a new state with the shallow patch applied (None => no-op).
9    """
10    new = dict(state)
11    if patch:
12        new.update(patch)
13    return new
14
15class Node:
16    """
17    Minimal node abstraction. Provide a handler callable with signature:
18      handler(current_state: dict, available_transitions: Optional[dict]) ->
19        (patch: dict|None, next_node_id: Optional[str])
20    """
21
22    def __init__(self, node_id: str, handler: Optional[Callable] = None):
23        self.id = node_id
24        if handler is None:
25            # default handler does nothing and terminates graph
26            handler = lambda state, avail: (None, None)
27        self._handler = handler
28
29    def execute(
30        self,
31        current_state: Dict[str, Any],
32        available_transitions: Optional[Dict] = None,
33    ) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
34        return self._handler(current_state, available_transitions)
35
36
37class Graph:
38    """
39    Graph runs nodes by id. nodes: mapping of id -> Node. start: start node id.
40    """
41
42    def __init__(self, nodes: Dict[str, Node], start: str):
43        if start not in nodes:
44            raise KeyError("start node not in nodes")
45        self.nodes = nodes
46        self.start = start
47
48    def run(self, initial_state: Dict[str, Any]) -> Dict[str, Any]:
49        current_state = dict(initial_state)
50        next_node = self.start
51
52        while next_node is not None:
53            node = self.nodes[next_node]
54            patch, next_node = node.execute(current_state, available_transitions=None)
55            current_state = apply_patch(current_state, patch)
56        return current_state
57
58
59def example_usage():
60    def start_handler(state, _):
61        return ({"count": 0}, "inc")
62
63    def inc_handler(state, _):
64        n = state.get("count", 0) + 1
65        if n >= 3:
66            return ({"count": n}, None)
67        return ({"count": n}, "inc")
68
69    nodes = {
70        "start": Node("start", start_handler),
71        "inc": Node("inc", inc_handler),
72    }
73
74    g = Graph(nodes, start="start")
75    final = g.run({})
76
77    return final

Room for Improvement

All the above is what I would define as the building blocks for more complex applications. However, I acknowledge, the use-case influences the framework. So, in my pursuit of not reinventing the wheel with each POC, I built tinygraph-ts. Like Pocketflow, it can also be ported in a single-shot to any language. However, it does not prioritize LoC, dependencies, and operator overloads. It just seeks to be simple and to encourage good patterns.

⚠️ Note: This is just my rationale and not the critique of any developer. It is my thought process on why this library is not for me.

I really wanted to like Pocketflow, but what I thought was a superpower of being small, was a library that attempts to be minimal, if only for the sake of it. The project cites, "separation of concerns" as the rational for the structure of a node:

prep
exec
post

However, if the majority of your prep is pulling from shared state, why does this need to be a separate function?
If the complexity of your node is already low, or the complexity is hidden by an external function call by a more robust implementation, why separate exec and post?
I will skip over operator overload in Python, that is largely preferential.
Finally, why have a separate async implementation if you could handle async versus sync using inspect.iscoroutinefunction(object)? You could even use concurrent.futures to offload them to a ThreadPoolExecutor and have fine-grained concurrency controls.

That said, I love the philosophy of the project. Keep all the dependencies of the graph outside the graph. You are not tied to any library and which dependencies it wraps. This is such a beautiful concept. If I want to POC using a specific VectorDB locally, but use a different one in production, I don't have to shop around or even think in two places before I have started the project. If one works locally but doesn't work down the line, I can build a tiny adapter.

Libraries should enable, not hinder and for all I critique about Pocketflow, simplicity in 2026 is a welcome change and other libraries should take note.

Fork it, skip it, use it. Whatever you do with or without this library, enjoy being a developer and take time to smell the roses. Cheers.

Example of TinyGraph

1import { Graph, type Node, type Ctx } from "tinygraph-ts";
2
3// fetch.ts
4class FetchNode implements Node<State> {
5  async next(state: Ctx<State>) {
6    console.log("Made to fetch!");
7    const r = await fetch(state.site as string).then((response) =>
8      response.json()
9    );
10
11    return {
12      context: { json: r },
13      transition: "log",
14    };
15  }
16}
17// log.ts
18class LogNode implements Node<State> {
19  async next(state: Ctx<State>) {
20    console.log(`User ID: ${state.json?.userId}`);
21    // Note: No transition returned, ends the graph here.
22  }
23}
24// index.ts
25// Define type-safe graph. `site` is required to run so it is guranteed to exist,
26// but the result is *not* if an error occurs or `FetchNode` has not been run.
27type State = {
28  json?: {
29    userId: number;
30    id: number;
31    title: string;
32    completed: boolean;
33  };
34  site: string;
35};
36
37const graphRes = new Graph<State>()
38  .node("fetch", new FetchNode())
39  .node("log", new LogNode())
40  .edge("fetch", "log")
41  .setStart("fetch")
42  .run({ site: "https://jsonplaceholder.typicode.com/todos/1" });
43
44console.log(await graphRes);