minigraph

README

minigraph

A stateful, graph-based agent runtime for Go — in ~420 lines.

LangGraph's core ideas (typed state, cyclic graphs, streaming, checkpointing, human-in-the-loop, parallel fan-out) distilled into one dependency-free package you can read in a sitting.

---

Build agents and workflows as a graph of nodes that transform a shared, typed state, wired with edges — static or conditional. Cycles are first-class: a node can route back to an earlier one, which is exactly what turns a pipeline into an agent that loops until it's done.

app, _ := minigraph.New[State]().
    AddNode("agent", callModel).
    AddNode("tools", runTools).
    AddEdge(minigraph.Start, "agent").
    AddRouter("agent", func(ctx context.Context, s State) (string, error) {
        if s.Done {
            return minigraph.End, nil
        }
        return "tools", nil // otherwise call tools, then loop back to the agent
    }).
    AddEdge("tools", "agent").
    Compile()

final, err := app.Invoke(ctx, State{Question: "..."})

flowchart LR
    Start([Start]) --> agent
    agent -- done --> End([End])
    agent -- needs tools --> tools
    tools --> agent

Why

LangGraph is excellent — and large. minigraph keeps the ~80% of the model that matters for most agents and drops the machinery you rarely touch, trading a heap of features for something you can actually hold in your head.

	Lines of code (no tests)
LangGraph core + checkpoint libraries (Python)	~33,700
minigraph (Go)	419

Same shape — StateGraph, conditional edges, cycles, invoke/stream, checkpointers, interrupt(), parallel fan-out — at ~1.2% of the code, with zero dependencies beyond the Go standard library, and typed state checked by the compiler instead of at runtime.

Features

• 🧩 Typed state via generics — no map[string]any, no schema DSL. Your struct is the schema; the compiler enforces it.
• 🔁 Cycles are first-class — the difference between a DAG pipeline and a real agent loop.
• 🌊 Streaming built in — Stream is a Go 1.23 iter.Seq2, so observing every step is just for step, err := range.
• 💾 Checkpointing & durable threads — every step is a resume point; crash, restart, continue where it stopped.
• 🙋 Human-in-the-loop — a node returns an Interrupt to pause for approval, then resumes with the edited state.
• 🍴 Parallel fan-out/join — run branches concurrently as a single graph step, with a merge you control.
• 🪆 Free subgraphs — a compiled App.Invoke is a Node, so graphs nest with no special support.
• ✅ Fails at Compile, not mid-run — dangling edges, dead ends, and duplicate nodes are all reported up front.
• 🧵 Concurrency-safe — a compiled App is immutable; run it from as many goroutines as you like.

Install

go get github.com/tomerfooks/minigraph

Requires Go 1.24+ (uses iter, maps, and slices from the standard library).

Examples

Four runnable programs, each a self-contained pattern:

Command	Pattern
go run ./examples/agent	Minimal agent ⇄ tools loop
go run ./examples/react	Full ReAct loop: Thought → Action → Observation, with a trace and a swappable mock LLM
go run ./examples/approval	Human-in-the-loop: interrupt, reject with feedback, redraft, approve — on a durable thread
go run ./examples/fanout	Parallel researchers merged into one report

A tour of the API

Streaming

Stream returns a standard iterator, so watching a run is an ordinary loop — break stops it, and the Step you keep is a checkpoint you can resume from:

for step, err := range app.Stream(ctx, initial) {
    if err != nil { /* handle */ }
    fmt.Println(step.Node, step.State)
}

Checkpoints & durable threads

Every Step a run yields is a checkpoint; InvokeFrom(ctx, step) continues from exactly that point. The Checkpointer interface (with an in-memory MemorySaver included) persists the latest step per thread, and the *Thread methods make a run durable — die anywhere, run again, continue:

saver := &minigraph.MemorySaver[State]{}
final, err := app.InvokeThread(ctx, saver, "thread-42", initial)

Human-in-the-loop

A node pauses by returning an Interrupt. Unlike an error, the state it returns is kept — the outside world answers by editing that state and resuming from the interrupted node:

final, err := app.Invoke(ctx, initial)
var intr *minigraph.Interrupt
if errors.As(err, &intr) {
    fmt.Println("agent asks:", intr.Payload)
    final.Approved = true // the human's answer, written into the state
    final, err = app.InvokeFrom(ctx, minigraph.Step[State]{Node: intr.Node, State: final})
}

Parallel fan-out

Parallel composes nodes into one node: branches run concurrently on copies of the state, then a merge you write folds the results. Because App.Invoke has a node's signature, whole compiled subgraphs can be branches:

research := minigraph.Parallel(
    func(ctx context.Context, base State, results []State) (State, error) {
        for _, r := range results {
            base.Findings = append(base.Findings, r.Finding)
        }
        return base, nil
    },
    searchWeb, searchDocs, subgraph.Invoke,
)
g.AddNode("research", research)

Coming from LangGraph

LangGraph	minigraph
StateGraph(State)	New[State]() — state is any Go type, checked at compile time
node function	func(ctx, S) (S, error) — takes state, returns the next state
add_edge	AddEdge(from, to)
add_conditional_edges	AddRouter(from, router) — the router returns the next node's name
START / END	minigraph.Start / minigraph.End
compile()	Compile() — validates the whole graph, returns an immutable App
invoke	Invoke(ctx, state)
stream	Stream(ctx, state) — an iter.Seq2[Step[S], error]
recursion limit	App.MaxSteps (default 25)
checkpointer + thread_id	Checkpointer / MemorySaver, InvokeThread / StreamThread
interrupt()	return &Interrupt{Payload: ...} from a node, InvokeFrom to continue
Send / parallel supersteps	Parallel(merge, branches...) — fan-out/join as a node combinator
subgraphs	free: App.Invoke is a valid Node, so g.AddNode("sub", sub.Invoke)

Design choices

• Typed state, no reducers. A node returns the whole next state; branching logic reads it directly. The Go compiler is the schema validator. This one decision is what lets minigraph skip LangGraph's largest subsystem (channels and per-key reducers).
• Exactly one outgoing edge per node. A static edge is just a router that ignores the state, so control flow always has one place to look.
• Every Step is a checkpoint. There's no separate checkpoint type or replay engine: a run is a (node, state) pair advancing, so any yielded pair is a resume point. Interrupts, durability, and retries all fall out of that one fact.
• Fork/join instead of supersteps. Parallelism lives inside Parallel, where you write the merge — the graph stays deterministic and the state stays a plain typed value. (Inside a branch, treat shared reference fields as read-only and let the merge reconcile.)
• Fail at Compile, not mid-run. Wiring mistakes accumulate and come back joined, all at once.

Deliberately out of scope: per-key state reducers, multiple streaming modes, token streaming, retry policies, and the platform layer (studio, tracing, deployment). Nodes are plain functions — LLM calls, retries, and telemetry compose from the outside.

Testing

go test -race ./...

The suite covers linear and cyclic runs, every compile-time validation, streaming and early break, context cancellation, node/router errors, interrupt-and-resume, durable threads, and concurrent invocation — with more test code than library code.

License

MIT © Tomer Fooks

Documentation

Overview

Package minigraph is a minimal LangGraph for Go: a state machine of nodes that transform a shared, typed state, wired with static or conditional edges. Build a Graph, Compile it, then Invoke or Stream.

Index

• Constants
• Variables
• type App
  • func (r *App[S]) Invoke(ctx context.Context, state S) (S, error)
  • func (r *App[S]) InvokeFrom(ctx context.Context, from Step[S]) (S, error)
  • func (r *App[S]) InvokeThread(ctx context.Context, saver Checkpointer[S], thread string, initial S) (S, error)
  • func (r *App[S]) Stream(ctx context.Context, state S) iter.Seq2[Step[S], error]
  • func (r *App[S]) StreamFrom(ctx context.Context, from Step[S]) iter.Seq2[Step[S], error]
  • func (r *App[S]) StreamThread(ctx context.Context, saver Checkpointer[S], thread string, initial S) iter.Seq2[Step[S], error]
• type Checkpointer
• type Graph
  • func New[S any]() *Graph[S]
  • func (g *Graph[S]) AddEdge(from, to string) *Graph[S]
  • func (g *Graph[S]) AddNode(name string, fn Node[S]) *Graph[S]
  • func (g *Graph[S]) AddRouter(from string, r Router[S]) *Graph[S]
  • func (g *Graph[S]) Compile() (*App[S], error)
• type Interrupt
  • func (i *Interrupt) Error() string
• type MemorySaver
  • func (m *MemorySaver[S]) Load(_ context.Context, thread string) (Step[S], bool, error)
  • func (m *MemorySaver[S]) Save(_ context.Context, thread string, step Step[S]) error
• type Node
  • func Parallel[S any](merge func(ctx context.Context, base S, results []S) (S, error), ...) Node[S]
• type Router
• type Step

Constants

const (
	Start = "__start__"
	End   = "__end__"
)

Start and End are the reserved endpoints of every run: wire Start to your entry node, and route to End to finish.

Variables

var ErrMaxSteps = errors.New("minigraph: max steps exceeded")

ErrMaxSteps is returned when a run exceeds App.MaxSteps node executions — usually a cycle that never routes to End.

Types

type App

type App[S any] struct {

	// MaxSteps caps node executions per run, catching endless cycles.
	// Compile sets it to 25; override freely.
	MaxSteps int
	// contains filtered or unexported fields
}

App is a compiled, immutable graph. Safe for concurrent use.

func (*App[S]) Invoke

func (r *App[S]) Invoke(ctx context.Context, state S) (S, error)

Invoke runs the graph to completion and returns the final state. On error it returns the last state a node produced successfully.

func (*App[S]) InvokeFrom

func (r *App[S]) InvokeFrom(ctx context.Context, from Step[S]) (S, error)

InvokeFrom runs the graph to completion from a previously yielded Step.

func (*App[S]) InvokeThread

func (r *App[S]) InvokeThread(ctx context.Context, saver Checkpointer[S], thread string, initial S) (S, error)

InvokeThread runs a thread to completion (or its next Interrupt) and returns the final state. Calling it again on a finished thread is a no-op that returns the same final state. To answer an Interrupt, edit the returned state, Save it under the interrupted node, and call InvokeThread again:

saver.Save(ctx, thread, Step[S]{Node: intr.Node, State: edited})

func (*App[S]) Stream

func (r *App[S]) Stream(ctx context.Context, state S) iter.Seq2[Step[S], error]

Stream runs the graph from Start, yielding after every node. An error arrives as the final pair, with the Step beside it carrying the last completed node and last good state — so resuming an error Step retries the step that failed. Breaking out of the loop stops the run.

func (*App[S]) StreamFrom

func (r *App[S]) StreamFrom(ctx context.Context, from Step[S]) iter.Seq2[Step[S], error]

StreamFrom continues a run from a previously yielded Step: routing restarts from from.Node with from.State, so the checkpointed node is not re-executed. MaxSteps counts from zero again.

func (*App[S]) StreamThread

func (r *App[S]) StreamThread(ctx context.Context, saver Checkpointer[S], thread string, initial S) iter.Seq2[Step[S], error]

StreamThread is Stream with durability. If the thread has a saved Step the run resumes from it and initial is ignored; otherwise the run starts fresh. Every successful Step is saved before it is yielded, and an Interrupt's Step is saved too — so a process can die at any point and the thread continues where it left off. Ordinary node errors are not saved: rerunning the thread retries from the last good Step.

type Checkpointer

type Checkpointer[S any] interface {
	Save(ctx context.Context, thread string, step Step[S]) error
	Load(ctx context.Context, thread string) (Step[S], bool, error)
}

Checkpointer persists the latest Step of each thread, making runs durable: a run that crashed, was interrupted, or was broken out of continues from its last saved Step. Implementations should serialize S (e.g. JSON) when storing outside process memory.

type Graph

type Graph[S any] struct {
	// contains filtered or unexported fields
}

Graph is a mutable builder. Add nodes and edges, then Compile. Building mistakes accumulate silently and are all reported by Compile.

func New

func New[S any]() *Graph[S]

New returns an empty graph over state type S.

func (*Graph[S]) AddEdge

func (g *Graph[S]) AddEdge(from, to string) *Graph[S]

AddEdge wires from → to unconditionally. Every node has exactly one outgoing edge; from may be Start, to may be End.

func (*Graph[S]) AddNode

func (g *Graph[S]) AddNode(name string, fn Node[S]) *Graph[S]

AddNode registers fn under name. Names must be unique and not Start or End.

func (*Graph[S]) AddRouter

func (g *Graph[S]) AddRouter(from string, r Router[S]) *Graph[S]

AddRouter wires from → r(state), letting the state pick the next node at run time. r must return a node name or End.

func (*Graph[S]) Compile

func (g *Graph[S]) Compile() (*App[S], error)

Compile validates the graph and freezes it into a App. The builder can keep changing afterwards without affecting compiled runnables.

type Interrupt

type Interrupt struct {
	Payload any    // what the node wants to tell whoever resumes the run
	Node    string // set by the engine: the node that paused
}

Interrupt pauses a run for outside input — human approval, missing data, anything the graph can't decide alone. Return one from a node:

return s, &Interrupt{Payload: "ok to send this email?"}

Unlike an ordinary error, the state returned alongside an Interrupt is kept: the run yields it as the final Step, which is the checkpoint to continue from once the outside world has answered (usually after editing the state):

var intr *Interrupt
if errors.As(err, &intr) {
    state.Approved = true
    final, err = app.InvokeFrom(ctx, Step[S]{Node: intr.Node, State: state})
}

InvokeFrom routes onward from the interrupted node; the node itself does not re-run. Inside Parallel branches an Interrupt cannot pause the run and is treated as a plain error.

func (*Interrupt) Error

func (i *Interrupt) Error() string

type MemorySaver

type MemorySaver[S any] struct {
	// contains filtered or unexported fields
}

MemorySaver is an in-process Checkpointer, good for tests and single-run durability (interrupt/resume). The zero value is ready to use. Steps are stored by value: reference fields inside S still point at shared data.

func (*MemorySaver[S]) Load

func (m *MemorySaver[S]) Load(_ context.Context, thread string) (Step[S], bool, error)

func (*MemorySaver[S]) Save

func (m *MemorySaver[S]) Save(_ context.Context, thread string, step Step[S]) error

type Node

type Node[S any] func(ctx context.Context, state S) (S, error)

Node transforms the state: it receives the current state and returns the next one.

func Parallel

func Parallel[S any](merge func(ctx context.Context, base S, results []S) (S, error), branches ...Node[S]) Node[S]

Parallel composes branches into a single Node: every branch runs concurrently on the same starting state, then merge folds the results — ordered like the branches — into the next state. It is fan-out/join as a combinator: the graph stays sequential and this counts as one step.

Because a App's Invoke has a Node's signature, branches can be whole compiled subgraphs: Parallel(merge, subA.Invoke, subB.Invoke).

Branches receive shallow copies of the state, so reference fields (slices, maps, pointers) are shared: treat them as read-only inside a branch and put new data in the branch's own copy for merge to reconcile. The first branch error cancels the siblings and fails the node; an Interrupt inside a branch is treated as a plain error.

type Router

type Router[S any] func(ctx context.Context, state S) (string, error)

Router inspects the state after a node ran and names the next node, or End.

type Step

type Step[S any] struct {
	Node  string
	State S
}

Step is one node execution: the node's name and the state it produced. Every Step is also a checkpoint — feed one to InvokeFrom or StreamFrom to continue a run from exactly that point.