graphgo

README

🕸️ GraphGo

Durable AI agent workflows for Go

Build durable AI agents in Go — a LangGraph-style workflow engine with checkpoints, tools, memory, and local-LLM support. One static binary. Typed state. Production-first.

---

GraphGo lets you model an AI agent as a graph of nodes and edges — with cycles, conditional routing, retries, tools, and human approvals — and runs it on a durable engine that checkpoints every step. Define workflows in YAML or the typed Go SDK, ship them as a single static binary, and resume, replay or inspect any run.

▷ workflow "support-triage"  •  provider=mock  •  store=~/.graphgo/checkpoints.db

▶ run support-triage (run_369d2a11e978a6fb)
  ● classify
    ✓ done 0s
    ⛃ checkpoint ckpt_c1057f65
    → classify ─▶ technical  [technical]
  ● technical
    ✓ done 0s
    ⛃ checkpoint ckpt_38ae22a7
    → technical ─▶ end
✔ run completed

● status=completed steps=2 run=run_369d2a11e978a6fb

Why GraphGo?

• 🧩 LangGraph-style, native Go — nodes, edges, conditional routing, and cycles with compile-time typed state (Go generics), not stringly-typed dicts.
• 💾 Durable by default — every super-step is checkpointed to in-memory or SQLite (pure-Go, no CGO). Crash, resume, and continue.
• ⏸️ Human-in-the-loop — pause a run for approval, persist it, and resume — even in a different process.
• 🛠️ Tools & agents — a ReAct-style agent loop with a tool registry whose schema is MCP- and OpenAI-compatible.
• 🔌 Any model — OpenAI, Anthropic, Gemini, Ollama / local models, any OpenAI-compatible endpoint, plus a deterministic mock for tests.
• 📜 YAML or Go — prototype in YAML, or drop into the typed SDK. Same engine underneath.
• 📦 One binary — graphgo run workflow.yaml. No Python, no runtime, no services. Cross-compiles everywhere.
• 🔍 Observability — streaming execution logs, time-travel replay, run history, and Mermaid/SVG diagram export.

Install

# Go 1.22+
go install github.com/arbazkhan971/graphgo/cmd/graphgo@latest

# Homebrew
brew install arbazkhan971/tap/graphgo

# Docker
docker run --rm ghcr.io/arbazkhan971/graphgo:latest version

# From source
git clone https://github.com/arbazkhan971/graphgo && cd graphgo && make install

60-second quickstart

graphgo init my-agent          # scaffold a starter workflow.yaml
cd my-agent
graphgo run workflow.yaml      # runs offline with the mock provider — no API key needed
graphgo visualize workflow.yaml # print a Mermaid diagram

Point it at a real model by exporting a key and selecting a provider:

export OPENAI_API_KEY=sk-...
graphgo run workflow.yaml --provider openai --model gpt-4o-mini

# or fully local, no keys:
graphgo run workflow.yaml --provider ollama --model llama3.2

Define a workflow in YAML

# research-agent.yaml
workflow:
  name: research-agent
  state:
    topic: string
    notes: array
    answer: string
  nodes:
    research:
      type: llm
      system: "You are a meticulous research assistant."
      prompt: "Research {{topic}}. Produce 3-5 key facts as short notes."
      output: notes
      append: true
    review:
      type: human_approval          # pauses for human approval
      message: "Approve the notes for '{{topic}}'?"
    final:
      type: llm
      prompt: "Write a clear answer about {{topic}} using: {{notes}}"
      output: answer
  edges:
    - { from: start,    to: research }
    - { from: research, to: review }
    - { from: review,   to: final }
    - { from: final,    to: end }

graphgo run research-agent.yaml --set topic="graph databases"

…or in the typed Go SDK

package main

import (
	"context"
	"fmt"

	graphgo "github.com/arbazkhan971/graphgo"
)

type State struct {
	Topic  string `json:"topic"`
	Answer string `json:"answer"`
}

func main() {
	g := graphgo.NewGraph[State]()

	g.AddNode("write", graphgo.LLMNode(graphgo.LLMConfig[State]{
		Provider: graphgo.NewMock("Graphs model relationships as nodes and edges."),
		System:   "You are concise.",
		Prompt:   func(s State) string { return "Explain " + s.Topic },
		Apply:    func(s State, r *graphgo.Response) (State, error) { s.Answer = r.Content; return s, nil },
	}))
	g.SetEntryPoint("write")
	g.SetFinishPoint("write")

	res, _ := g.Run(context.Background(), State{Topic: "graph theory"})
	fmt.Println(res.State.Answer)
}

Conditional edges and cycles are first-class:

g.AddConditionalEdge("research", func(_ context.Context, s State) string {
	if len(s.Notes) >= 3 {
		return "done"
	}
	return "more"
}, map[string]string{"more": "research", "done": "write"}) // loops back to research

Visualize

graphgo visualize code-review.yaml emits a GitHub-renderable Mermaid diagram (or JSON / SVG with -f):

flowchart TD
    analyze[["analyze"]]
    triage{"triage"}
    escalate{{"escalate"}}
    report[["report"]]
    start(("start"))
    end(("end"))

    start --> analyze
    analyze --> triage
    triage -.->|high| escalate
    triage -.->|low| report
    escalate --> report
    report --> end

    classDef llm fill:#ede7f6,stroke:#4527a0,color:#311b92;
    classDef human fill:#fff3e0,stroke:#e65100,color:#bf360c;
    classDef router fill:#fce4ec,stroke:#ad1457,color:#880e4f;
    classDef terminal fill:#eceff1,stroke:#455a64,color:#263238;
    class analyze llm;
    class triage router;
    class escalate human;
    class report llm;
    class start terminal;
    class end terminal;

Durability: checkpoint, resume, replay

Every step is checkpointed, so runs survive crashes and can be paused/resumed:

# A workflow with a human_approval node pauses and persists:
graphgo run research-agent.yaml --set topic="rust"
#   ⏸ paused for approval: Approve the notes for 'rust'?
#   resume with: graphgo run research-agent.yaml --resume run_ab12 --approve

graphgo run research-agent.yaml --resume run_ab12 --approve   # continues to completion

graphgo inspect                 # list every run and its status
graphgo replay run_ab12 --state # time-travel through each step's state

In Go, the same durability is one option away:

store, _ := graphgo.OpenSQLite("runs.db")
res, _ := g.Run(ctx, initial, graphgo.WithStore(store), graphgo.WithRunID("job-42"))
if res.Interrupted() {
	// ...later, even in another process:
	res, _ = g.Resume(ctx, store, "job-42", graphgo.Decision{Approved: true})
}

Node types

Type	YAML type	SDK builder	Purpose
LLM	llm	LLMNode	Prompt a model and write the result to state
Agent	(SDK)	AgentNode	ReAct-style loop that calls tools until done
Tool	tool	ToolNode	Deterministically invoke one tool
Router	router	LLMRouter / AddConditionalEdge	Branch on state or an LLM decision
Human approval	human_approval	HumanApprovalNode	Pause for human input, then resume
Parallel	parallel	ParallelNode	Fan out concurrent branches and merge
Retryable task	task (+retry)	WithRetry	Any node with an automatic retry policy

Providers

Provider	Constructor	Env var	Notes
OpenAI	graphgo.OpenAI(...)	OPENAI_API_KEY	Also any OpenAI-compatible server via BaseURL (vLLM, LM Studio, Together, Groq…)
Anthropic	graphgo.Anthropic(...)	ANTHROPIC_API_KEY	Claude Messages API, tool use
Gemini	graphgo.Gemini(...)	GEMINI_API_KEY	Google Generative Language API
Ollama	graphgo.Ollama(...)	–	Local models, no key, streaming
Mock	graphgo.NewMock(...)	–	Deterministic, offline — powers the test suite

CLI

Command	Description
graphgo init [dir]	Scaffold a starter workflow.yaml
graphgo run <file.yaml>	Run a workflow, streaming and checkpointing each step
graphgo visualize <file.yaml>	Export Mermaid (-f mermaid|json|svg)
graphgo replay <run-id>	Time-travel through a past run's checkpoints
graphgo inspect [run-id]	List runs, or show one run's history and final state
graphgo new <name>	Scaffold a Go SDK agent program
graphgo doctor	Check environment and provider configuration
graphgo version	Print the version

Global flags: --provider, --model, --base-url, --store (memory or a path), --no-color.

How does it compare?

	GraphGo	LangGraph (Python)	Temporal	Hand-rolled Go
Language	Go	Python	Go/Java/…	Go
Deployment	single binary	Python runtime	server + workers	your call
Typed state	✅ generics	⚠️ dict/schema	✅	✅
Graph + cycles	✅	✅	⚠️ code	⚠️ DIY
Checkpoints / resume	✅	✅	✅	❌
Human-in-the-loop	✅	✅	⚠️ signals	❌
Built-in LLM providers	✅	✅	❌	❌
YAML workflows	✅	❌	❌	❌
Local-model friendly	✅	⚠️	⚠️	⚠️

GraphGo borrows LangGraph's excellent mental model (stateful graphs for long-running agents) and Temporal's durability mindset (checkpoint, replay, recover), and delivers them as one dependency-light Go module.

Benchmarks

Single core, Intel Xeon @ 2.20GHz, Go 1.22:

Benchmark	Time/op	Allocs/op
3-node linear graph	4.3 µs	11
3-node linear + checkpoint each step	14.8 µs	29
10-iteration conditional cycle	8.6 µs	25

That's ~230,000 graph runs/sec with zero overhead beyond your node logic. Run them yourself with make bench.

Architecture

cmd/graphgo        the CLI
graphgo.go         the ergonomic SDK facade (NewGraph, LLMNode, HumanApprovalNode, …)
pkg/graph          the execution engine: nodes, edges, cycles, events, retries, interrupts
pkg/state          dynamic Dict state + templating for YAML workflows
pkg/checkpoint     Store interface + in-memory and SQLite backends
pkg/llm            provider interface + OpenAI / Anthropic / Gemini / Ollama / Mock
pkg/tools          tool registry + built-ins (calculator, http_get, now)
pkg/runtime        streaming console printer + time-travel replay
pkg/visualize      Mermaid / JSON / SVG exporters
pkg/workflow       YAML workflow parser and graph builder
examples           runnable YAML workflows and Go SDK programs

Examples

• examples/go/chatbot — a minimal single-node bot
• examples/go/research — cycles + conditional edges
• examples/go/code-review — a tool-using agent loop
• examples/go/human-approval — durable pause & resume with SQLite
• examples/go/ollama — a fully-local workflow
• examples/*.yaml — YAML versions of the above

Roadmap

• Streaming token output surfaced through node events
• First-class MCP client (connect to any MCP tool server)
• Postgres checkpoint store
• Subgraphs / nested graphs as nodes
• Long-term memory store (vector-backed)
• Web UI for run inspection and replay
• State reducers / channels for merge-heavy parallelism

See an issue or open one to shape it.

Contributing

Contributions are very welcome — see CONTRIBUTING.md. In short:

go test ./...     # everything runs offline with the mock provider
make lint         # golangci-lint
make bench        # benchmarks

License

MIT © 2026 Arbaz Khan.

_{If GraphGo helps you build something, consider giving it a ⭐ — it genuinely helps.}

Documentation

Overview

Package graphgo is the ergonomic entry point for building durable, stateful, multi-agent LLM workflows in Go. It re-exports the most common types and builders from the pkg/* subpackages so that a typical application only needs to import "github.com/arbazkhan971/graphgo".

A minimal workflow:

type State struct{ Topic, Answer string }

g := graphgo.NewGraph[State]()
g.AddNode("write", graphgo.LLMNode(graphgo.LLMConfig[State]{
	Provider: graphgo.NewMock("42"),
	Prompt:   func(s State) string { return "Answer about " + s.Topic },
	Apply:    func(s State, r *graphgo.Response) (State, error) { s.Answer = r.Content; return s, nil },
}))
g.SetEntryPoint("write")
g.SetFinishPoint("write")

res, err := g.Run(context.Background(), State{Topic: "Go"})

See the examples directory for complete programs.

Index

• Constants
• func AgentNode[S any](cfg AgentConfig[S]) graph.NodeFunc[S]
• func Anthropic(opts anthropic.Options) llm.Provider
• func Gemini(opts gemini.Options) llm.Provider
• func HumanApprovalNode[S any](cfg HumanApprovalConfig[S]) graph.NodeFunc[S]
• func LLMNode[S any](cfg LLMConfig[S]) graph.NodeFunc[S]
• func LLMRouter[S any](cfg RouterConfig[S]) func(context.Context, S) string
• func NewGraph[S any]() *graph.Graph[S]
• func NewMemoryStore() *checkpoint.MemoryStore
• func NewMock(responses ...string) *llm.Mock
• func NewProvider(cfg ProviderConfig) (llm.Provider, error)
• func Ollama(opts ollama.Options) llm.Provider
• func OpenAI(opts openai.Options) llm.Provider
• func OpenSQLite(path string) (*sqlite.Store, error)
• func ParallelNode[S any](combine func(base S, results []S) S, branches ...graph.NodeFunc[S]) graph.NodeFunc[S]
• func ProviderNames() []string
• func Render(tmpl string, d Dict) string
• func ToolNode[S any](cfg ToolNodeConfig[S]) graph.NodeFunc[S]
• func WithKind[S any](kind string) graph.NodeOption[S]
• func WithListener(l Listener) graph.RunOption
• func WithRetry[S any](p RetryPolicy) graph.NodeOption[S]
• func WithRunID(id string) graph.RunOption
• func WithStore(s Store) graph.RunOption
• type AgentConfig
• type ApprovalRequest
• type Checkpoint
• type Decision
• type Dict
• type Event
• type EventType
• type HumanApprovalConfig
• type LLMConfig
• type Listener
• type Message
• type Provider
• type ProviderConfig
• type Registry
  • func NewRegistry(ts ...Tool) *Registry
• type Request
• type Response
• type RetryPolicy
• type RouterConfig
• type RunInfo
• type Spec
• type Store
• type Tool
  • func Calculator() Tool
  • func HTTPGet(maxBytes int) Tool
• type ToolNodeConfig
• type ToolSpec

Constants

const (
	// Start is the virtual entry node.
	Start = graph.START
	// End is the virtual terminal node; route to it to finish a run.
	End = graph.END
)

Virtual endpoints used when wiring edges.

const Version = "0.1.0"

Version is the GraphGo release version.

Functions

func AgentNode

func AgentNode[S any](cfg AgentConfig[S]) graph.NodeFunc[S]

AgentNode returns a node that runs a ReAct-style loop: the model may request tool calls, which are executed and fed back until the model returns a final answer (a response with no tool calls) or MaxIterations is reached.

func Anthropic

func Anthropic(opts anthropic.Options) llm.Provider

Anthropic returns a Claude provider.

func Gemini

func Gemini(opts gemini.Options) llm.Provider

Gemini returns a Google Gemini provider.

func HumanApprovalNode

func HumanApprovalNode[S any](cfg HumanApprovalConfig[S]) graph.NodeFunc[S]

HumanApprovalNode returns a node that pauses the run for human review.

On first visit it interrupts, checkpointing the run and surfacing an ApprovalRequest. The caller resumes with a Decision (via g.Resume or the CLI). On resume the node applies the decision and continues. When called outside the engine (no Runtime in context) or with AutoApprove set, it approves automatically so unit tests never deadlock.

func LLMNode

func LLMNode[S any](cfg LLMConfig[S]) graph.NodeFunc[S]

LLMNode returns a node that calls a language model and applies the response to state.

func LLMRouter

func LLMRouter[S any](cfg RouterConfig[S]) func(context.Context, S) string

LLMRouter returns a routing function suitable for AddConditionalEdge. It asks the model to pick exactly one of the allowed routes and matches its answer against them (case-insensitive). On any ambiguity it returns Default.

func NewGraph

func NewGraph[S any]() *graph.Graph[S]

NewGraph creates a new typed workflow graph parameterized by the state type S.

func NewMemoryStore

func NewMemoryStore() *checkpoint.MemoryStore

NewMemoryStore returns an in-memory checkpoint store.

func NewMock

func NewMock(responses ...string) *llm.Mock

NewMock returns a deterministic, offline provider used by tests, examples and keyless demos.

func NewProvider

func NewProvider(cfg ProviderConfig) (llm.Provider, error)

NewProvider constructs an llm.Provider from a configuration. It is the entry point used by the CLI to turn --provider/--model flags into a live provider.

func Ollama

func Ollama(opts ollama.Options) llm.Provider

Ollama returns a local Ollama provider (default endpoint http://localhost:11434).

func OpenAI

func OpenAI(opts openai.Options) llm.Provider

OpenAI returns an OpenAI (or OpenAI-compatible) provider. Pass a BaseURL to target local/self-hosted servers such as vLLM, llama.cpp or LM Studio.

func OpenSQLite

func OpenSQLite(path string) (*sqlite.Store, error)

OpenSQLite opens (creating if needed) a durable SQLite-backed checkpoint store at path. The returned store persists run history across process restarts, enabling resume, replay and inspection. It uses a pure-Go SQLite driver, so the binary stays statically linkable with no CGO.

func ParallelNode

func ParallelNode[S any](combine func(base S, results []S) S, branches ...graph.NodeFunc[S]) graph.NodeFunc[S]

ParallelNode runs branches concurrently, each on an independent deep copy of the input state, then merges their results with combine. It models concurrency as a single node so the engine keeps a deterministic, one-active-node execution model. combine receives the original input state and the per-branch results in branch order. If any branch errors, the node fails with that error.

func ProviderNames

func ProviderNames() []string

ProviderNames returns the set of provider identifiers accepted by NewProvider.

func Render

func Render(tmpl string, d Dict) string

Render substitutes {{key}} placeholders in tmpl using values from d.

func ToolNode

func ToolNode[S any](cfg ToolNodeConfig[S]) graph.NodeFunc[S]

ToolNode returns a node that invokes exactly one tool with arguments derived from state and writes the result back into state. Unlike AgentNode there is no model in the loop, making it fully deterministic.

func WithKind

func WithKind[S any](kind string) graph.NodeOption[S]

WithKind sets a node's visualization kind ("llm", "tool", "human", etc.).

func WithListener

func WithListener(l Listener) graph.RunOption

WithListener streams execution events to l.

func WithRetry

func WithRetry[S any](p RetryPolicy) graph.NodeOption[S]

WithRetry attaches a retry policy to a node.

func WithRunID

func WithRunID(id string) graph.RunOption

WithRunID assigns a caller-chosen run identifier.

func WithStore

func WithStore(s Store) graph.RunOption

WithStore persists a checkpoint after every super-step (enables resume, replay, inspection).

Types

type AgentConfig

type AgentConfig[S any] struct {
	Provider    llm.Provider
	Model       string
	System      string
	Prompt      func(S) string
	Tools       *tools.Registry
	Apply       func(S, *llm.Response) (S, error)
	Temperature float64
	MaxTokens   int
	// MaxIterations bounds the tool-use loop (default 6).
	MaxIterations int
	// OnToolCall is an optional observability hook fired for each tool call.
	OnToolCall func(name, args, result string)
}

AgentConfig configures an AgentNode: an LLM that can call tools in a loop until it produces a final answer.

type ApprovalRequest

type ApprovalRequest struct {
	Kind   string `json:"kind"`   // always "approval"
	Prompt string `json:"prompt"` // human-readable description of what to approve
	Node   string `json:"node"`   // the node awaiting approval
}

ApprovalRequest is the interrupt payload produced by a HumanApprovalNode. It is delivered in Result.Interrupt.Payload and stored in the interrupt checkpoint.

type Checkpoint

type Checkpoint = checkpoint.Checkpoint

Checkpoint is a durable run snapshot.

type Decision

type Decision struct {
	Approved bool   `json:"approved"`
	Value    any    `json:"value,omitempty"`
	Feedback string `json:"feedback,omitempty"`
}

Decision is a human-in-the-loop verdict delivered to a HumanApprovalNode via Resume. Value carries any edited payload the reviewer supplied.

type Dict

type Dict = state.Dict

Dict is the dynamic state type used by YAML and untyped workflows.

type Event

type Event = graph.Event

Event is a streaming execution event.

type EventType

type EventType = graph.EventType

EventType enumerates event kinds.

type HumanApprovalConfig

type HumanApprovalConfig[S any] struct {
	// Prompt builds the message shown to the reviewer from the current state.
	Prompt func(S) string
	// Apply updates the state given the reviewer's Decision. If nil, the state
	// is returned unchanged and approval merely gates progression.
	Apply func(state S, d Decision) (S, error)
	// AutoApprove skips the pause and approves immediately (handy for CI).
	AutoApprove bool
	// NodeName labels the approval request; defaults to "human".
	NodeName string
}

HumanApprovalConfig configures a HumanApprovalNode.

type LLMConfig

type LLMConfig[S any] struct {
	// Provider is the language model backend (required).
	Provider llm.Provider
	// Model overrides the provider's default model.
	Model string
	// System is an optional system prompt.
	System string
	// Prompt builds the user message from state. Required unless Messages is set.
	Prompt func(S) string
	// Messages, when set, builds the entire message list and overrides
	// System+Prompt.
	Messages func(S) []llm.Message
	// Apply writes the model's response back into state (required to persist it).
	Apply func(S, *llm.Response) (S, error)

	Temperature float64
	MaxTokens   int
}

LLMConfig configures an LLMNode.

type Listener

type Listener = graph.Listener

Listener receives execution events.

type Message

type Message = llm.Message

Message is a chat message.

type Provider

type Provider = llm.Provider

Provider is a language-model backend.

type ProviderConfig

type ProviderConfig struct {
	Name    string // "mock", "openai", "anthropic", "gemini", "ollama"
	Model   string
	BaseURL string
	APIKey  string
}

ProviderConfig selects and configures an LLM provider by name. Empty fields fall back to each provider's defaults, including reading API keys from the conventional environment variables (OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY/GOOGLE_API_KEY). The "mock" provider needs no configuration and runs fully offline.

type Registry

type Registry = tools.Registry

Registry is a collection of tools.

func NewRegistry

func NewRegistry(ts ...Tool) *Registry

NewRegistry returns a tool registry preloaded with ts.

type Request

type Request = llm.Request

Request is a chat completion request.

type Response

type Response = llm.Response

Response is a chat completion result.

type RetryPolicy

type RetryPolicy = graph.RetryPolicy

RetryPolicy configures automatic retries for a node.

type RouterConfig

type RouterConfig[S any] struct {
	Provider llm.Provider
	Model    string
	Prompt   func(S) string
	// Routes are the allowed route keys the model must choose among.
	Routes []string
	// Default is returned when the model's answer matches no route.
	Default string
}

RouterConfig configures an LLM-based router used with g.AddConditionalEdge.

type RunInfo

type RunInfo = checkpoint.RunInfo

RunInfo summarizes a run.

type Spec

type Spec = graph.Spec

Spec is a serializable description of a graph, used for visualization.

type Store

type Store = checkpoint.Store

Store is the checkpoint persistence interface.

type Tool

type Tool = tools.Tool

Tool is a callable capability exposed to a model.

func Calculator

func Calculator() Tool

Calculator returns the built-in arithmetic tool.

func HTTPGet

func HTTPGet(maxBytes int) Tool

HTTPGet returns the built-in HTTP GET tool (body truncated to maxBytes).

type ToolNodeConfig

type ToolNodeConfig[S any] struct {
	Tool  tools.Tool
	Args  func(S) (json.RawMessage, error)
	Apply func(S, string) (S, error)
}

ToolNodeConfig configures a deterministic single-tool node.

type ToolSpec

type ToolSpec = llm.ToolSpec

ToolSpec advertises a tool to a model.