graph

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 25, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package graph provides the StateGraph durable execution runtime.

Index

Constants

View Source 
const (
	StartNode = "__start__"
	EndNode   = "__end__"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type AppendReducer added in v0.3.0 

type AppendReducer struct{}

AppendReducer appends update elements to an existing slice. If the existing value is not a slice, it is replaced. If the update is a single value, it is wrapped in a slice and appended.

func (AppendReducer) Reduce added in v0.3.0 

func (AppendReducer) Reduce(existing, update any) any

type CompiledGraph 

type CompiledGraph struct {
	ID      string
	Nodes   map[string]*Node
	AdjList map[string][]*Edge // from -> edges
	Entry   string
}

CompiledGraph is the immutable, validated graph ready for execution.

func RegisterEntrypoint added in v0.3.0 

func RegisterEntrypoint(name string, fn EntrypointFunc) (*CompiledGraph, error)

RegisterEntrypoint wraps a Go function as a graph entrypoint. The returned CompiledGraph has a single node that runs the function, integrating with checkpointing and durable execution.

func TaskGraph added in v0.3.0 

func TaskGraph(id string, tasks map[string]TaskFunc) (*CompiledGraph, error)

TaskGraph creates a CompiledGraph from a sequence of named tasks. Tasks are chained linearly: the output of each becomes the input of the next.

func (*CompiledGraph) ToDOT added in v0.3.0 

func (cg *CompiledGraph) ToDOT() string

ToDOT generates a Graphviz DOT format representation of the compiled graph.

func (*CompiledGraph) ToMermaid added in v0.3.0 

func (cg *CompiledGraph) ToMermaid() string

ToMermaid generates a Mermaid flowchart definition of the compiled graph.

type Edge 

type Edge struct {
	From      string
	To        string        // static target (mutually exclusive with Condition)
	Condition EdgeCondition // dynamic routing (mutually exclusive with To)
}

Edge represents a transition between nodes.

type EdgeCondition 

type EdgeCondition func(state State) string

EdgeCondition decides which node to transition to based on state. Returns the target node ID.

type EntrypointFunc added in v0.3.0 

type EntrypointFunc func(ctx context.Context, input any) (any, error)

EntrypointFunc is a function that serves as a graph entrypoint.

type MergeFunc added in v0.3.0 

type MergeFunc func(original State, results []State) (State, error)

MergeFunc combines the results of parallel branches into a single state. It receives the original pre-fork state and the results from each branch.

type MergeMapReducer added in v0.3.0 

type MergeMapReducer struct{}

MergeMapReducer performs a shallow merge of two maps. Update keys overwrite existing keys. Non-map values are replaced.

func (MergeMapReducer) Reduce added in v0.3.0 

func (MergeMapReducer) Reduce(existing, update any) any

type Message 

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

Message records a message exchanged during the run.

type Node 

type Node struct {
	ID string
	Fn NodeFunc
	// Interrupt, if true, causes the runner to checkpoint and pause before executing this node.
	Interrupt bool
}

Node represents a node in the state graph.

type NodeFunc 

type NodeFunc func(ctx context.Context, state State) (State, error)

NodeFunc is the function signature for a graph node. It receives the current state and returns updated state.

func FanOut added in v0.3.0 

func FanOut(branches []NodeFunc, merge MergeFunc) NodeFunc

FanOut creates a NodeFunc that executes multiple node functions in parallel. Each function receives a copy of the input state. Results are merged back using a MergeFunc.

func RegisterTask added in v0.3.0 

func RegisterTask(name string, fn TaskFunc) NodeFunc

RegisterTask marks a function as a checkpoint-able task. The result is automatically cached by task name and input hash. If the task was already completed in a previous run (via checkpoint), the cached result is returned without re-executing the function.

func SubgraphNode added in v0.3.0 

func SubgraphNode(sub *CompiledGraph) NodeFunc

SubgraphNode wraps a CompiledGraph as a NodeFunc so it can be used as a node within another graph. The subgraph executes inline, sharing the parent's state.

type Reducer added in v0.3.0 

type Reducer interface {
	Reduce(existing, update any) any
}

Reducer defines how a state value is combined when a node returns an update. The default behavior (nil reducer) is full replacement, but specific keys can be configured to append to slices, merge maps, etc.

type ReducerFunc added in v0.3.0 

type ReducerFunc func(existing, update any) any

ReducerFunc is a convenience adapter for functions.

func (ReducerFunc) Reduce added in v0.3.0 

func (f ReducerFunc) Reduce(existing, update any) any

type ReplaceReducer added in v0.3.0 

type ReplaceReducer struct{}

ReplaceReducer always replaces the existing value with the update. This is the default behavior when no reducer is configured.

func (ReplaceReducer) Reduce added in v0.3.0 

func (ReplaceReducer) Reduce(_, update any) any

type RunState 

type RunState struct {
	RunID       string    `json:"run_id"`
	SessionID   string    `json:"session_id"`
	GraphID     string    `json:"graph_id"`
	CurrentNode string    `json:"current_node"`
	Status      RunStatus `json:"status"`
	State       State     `json:"state"`
	SeqNum      int64     `json:"seq_num"`
	StartedAt   time.Time `json:"started_at"`
	UpdatedAt   time.Time `json:"updated_at"`

	// Rich output (Agno-inspired)
	Messages   []Message        `json:"messages,omitempty"`
	ToolCalls  []ToolCallRecord `json:"tool_calls,omitempty"`
	TotalUsage UsageStats       `json:"total_usage"`
}

RunState captures the full runtime state for a single graph execution.

type RunStatus 

type RunStatus string

RunStatus represents the status of a graph run. 

const (
	RunStatusRunning   RunStatus = "running"
	RunStatusPaused    RunStatus = "paused"
	RunStatusCompleted RunStatus = "completed"
	RunStatusFailed    RunStatus = "failed"
)

type Runner 

type Runner struct {
	// contains filtered or unexported fields
}

Runner executes a CompiledGraph with durable checkpointing.

func NewRunner 

func NewRunner(g *CompiledGraph, store storage.Storage) *Runner

NewRunner creates a runner for the given compiled graph.

func (*Runner) ForkFrom added in v0.3.0 

func (r *Runner) ForkFrom(ctx context.Context, checkpointID string, stateUpdate map[string]any) (*RunState, error)

ForkFrom creates a new execution branch from a specific checkpoint with modified state. The original checkpoint history is preserved; the fork continues independently.

func (*Runner) ReplayFrom added in v0.3.0 

func (r *Runner) ReplayFrom(ctx context.Context, checkpointID string) (*RunState, error)

ReplayFrom loads a checkpoint and re-executes the graph from the checkpoint's node. All nodes before the checkpoint node are skipped (treated as cached). Re-execution starts at the checkpoint's node and continues to END.

func (*Runner) Resume 

func (r *Runner) Resume(ctx context.Context, sessionID string) (*RunState, error)

Resume continues execution from the latest checkpoint for the given session.

func (*Runner) ResumeFromCheckpoint 

func (r *Runner) ResumeFromCheckpoint(ctx context.Context, checkpointID string) (*RunState, error)

ResumeFromCheckpoint resumes from a specific checkpoint (time-travel).

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context, sessionID string, initial State) (*RunState, error)

Run starts a new execution of the graph with the given initial state.

func (*Runner) Stream 

func (r *Runner) Stream() <-chan StreamEvent

Stream returns a channel of execution events for real-time observability.

func (*Runner) WithBroker 

func (r *Runner) WithBroker(b *stream.Broker) *Runner

WithBroker attaches an SSE Broker so the runner publishes events to SSE subscribers.

func (*Runner) WithTracer 

func (r *Runner) WithTracer(t *trace.Collector) *Runner

WithTracer attaches a trace.Collector for span-based execution tracing.

type State 

type State map[string]any

State is the mutable state bag passed through graph execution.

func MergeAll added in v0.3.0 

func MergeAll(original State, results []State) (State, error)

MergeAll is a simple merge strategy that applies all branch results on top of the original state. Later branches overwrite earlier ones for conflicting keys.

type StateGraph 

type StateGraph struct {
	// contains filtered or unexported fields
}

StateGraph is a builder for defining directed graphs with durable execution.

func New 

func New(id string) *StateGraph

New creates a new StateGraph with the given ID.

func (*StateGraph) AddConditionalEdge 

func (g *StateGraph) AddConditionalEdge(from string, condition EdgeCondition) *StateGraph

AddConditionalEdge adds a dynamic edge that routes based on state.

func (*StateGraph) AddEdge 

func (g *StateGraph) AddEdge(from, to string) *StateGraph

AddEdge adds a static edge from one node to another.

func (*StateGraph) AddInterruptNode 

func (g *StateGraph) AddInterruptNode(id string, fn NodeFunc) *StateGraph

AddInterruptNode registers a node that will pause execution for human approval.

func (*StateGraph) AddNode 

func (g *StateGraph) AddNode(id string, fn NodeFunc) *StateGraph

AddNode registers a node with the given ID and handler function.

func (*StateGraph) AddSubgraph added in v0.3.0 

func (g *StateGraph) AddSubgraph(id string, sub *CompiledGraph) *StateGraph

AddSubgraph registers a compiled graph as a node within this graph.

func (*StateGraph) Compile 

func (g *StateGraph) Compile() (*CompiledGraph, error)

Compile validates the graph and returns a CompiledGraph.

func (*StateGraph) SetEntryPoint 

func (g *StateGraph) SetEntryPoint(nodeID string) *StateGraph

SetEntryPoint sets the starting node of the graph.

func (*StateGraph) SetFinishPoint 

func (g *StateGraph) SetFinishPoint(nodeID string) *StateGraph

SetFinishPoint sets the ending node of the graph.

type StateSchema added in v0.3.0 

type StateSchema map[string]Reducer

StateSchema maps state keys to their reducers. Keys not in the schema use the default ReplaceReducer behavior.

func (StateSchema) ApplyUpdate added in v0.3.0 

func (s StateSchema) ApplyUpdate(existing, update State) State

ApplyUpdate merges an update into the existing state using the schema's reducers.

type StreamEvent 

type StreamEvent struct {
	Type      string    `json:"type"` // node_start, node_end, edge_transition, checkpoint, interrupt, error
	NodeID    string    `json:"node_id,omitempty"`
	State     State     `json:"state,omitempty"`
	Error     string    `json:"error,omitempty"`
	Timestamp time.Time `json:"timestamp"`
}

StreamEvent is emitted during graph execution for real-time observability.

type TaskFunc added in v0.3.0 

type TaskFunc func(ctx context.Context, input any) (any, error)

TaskFunc is a function that serves as a checkpoint-able task.

type ToolCallRecord 

type ToolCallRecord struct {
	Name   string         `json:"name"`
	Args   map[string]any `json:"args"`
	Result any            `json:"result"`
	Error  string         `json:"error,omitempty"`
}

ToolCallRecord records a tool invocation during the run.

type UsageStats 

type UsageStats struct {
	PromptTokens     int `json:"prompt_tokens"`
	CompletionTokens int `json:"completion_tokens"`
	TotalTokens      int `json:"total_tokens"`
}

UsageStats tracks cumulative token usage across a run.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.