graph

package
v1.5.3 Latest Latest
Warning

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

 Go to latest Published: Aug 8, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Overview

Package graph implements a LangGraph-style state graph engine for Go.

A StateGraph is a directed graph of nodes that operate on a shared mutable state (a map). Nodes are executed in topological order, with conditional edges routing execution between nodes. The graph supports checkpointing, interrupts, and streaming of node updates.

Index

Constants

This section is empty.

Variables

View Source 
var CommandOut = "__command"

CommandOut is returned by a node when it wants to emit commands.

View Source 
var ErrInterrupted = errors.New("graph: execution interrupted")

ErrInterrupted is returned when the graph pauses at an interrupt point.

Functions

func ApplyTransformers 

func ApplyTransformers(ctx context.Context, in <-chan StreamUpdate, ts ...StreamTransformer) (<-chan StreamUpdate, error)

ApplyTransformers chains the given transformers onto an input stream.

Types

type Checkpoint 

type Checkpoint struct {
	State State
	Next  NodeID
}

Checkpoint captures the state at an interrupt point.

type Checkpointer 

type Checkpointer interface {
	// Get returns the checkpoint for a thread id, or nil if none exists.
	Get(ctx context.Context, threadID string) (*Checkpoint, error)
	// Put stores a checkpoint for a thread id.
	Put(ctx context.Context, threadID string, cp *Checkpoint) error
}

Checkpointer persists and resumes checkpoints.

type Command 

type Command struct {
	// Update is the state update to apply.
	Update State
	// Goto is a node to route to next. If empty, normal edges apply.
	Goto NodeID
	// Resume is a value to resume an interrupted run with.
	Resume any
	// ResumeSet marks whether Resume was explicitly set.
	ResumeSet bool
	// Send contains parallel branches to execute.
	Send []Send
}

Command carries control-flow directives out of a node, mirroring langgraph's Command type. Goto routes to named nodes; Send spawns parallel sub-updates.

type CompiledStateGraph 

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

CompiledStateGraph is a frozen graph ready to run.

func (*CompiledStateGraph) GViz 

func (c *CompiledStateGraph) GViz() string

GViz is a placeholder for future graph visualization.

func (*CompiledStateGraph) Invoke 

func (c *CompiledStateGraph) Invoke(ctx context.Context, input State, opts ...RunOption) (State, error)

Invoke runs the graph to completion (or until an interrupt).

func (*CompiledStateGraph) Stream 

func (c *CompiledStateGraph) Stream(ctx context.Context, input State, opts ...RunOption) (<-chan StreamUpdate, error)

Stream runs the graph, yielding a StreamUpdate after each node executes. Transformers are applied to the emitted stream.

func (*CompiledStateGraph) WithTransformers 

func (c *CompiledStateGraph) WithTransformers(ts ...StreamTransformer) *CompiledStateGraph

WithTransformers registers stream transformers applied to graph.Stream output.

type CondFunc 

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

CondFunc decides the next node given the current state.

type ConditionalEdges 

type ConditionalEdges struct {
	From NodeID
	Cond CondFunc
}

ConditionalEdges attach a CondFunc to a source node.

type Edge 

type Edge struct {
	From NodeID
	To   NodeID
}

Edge is a directed connection between two nodes.

type FilterTransformer 

type FilterTransformer struct {
	Keep map[NodeID]bool
}

FilterTransformer drops updates for nodes whose id is not in keep.

func (FilterTransformer) Transform 

func (f FilterTransformer) Transform(_ context.Context, in <-chan StreamUpdate) (<-chan StreamUpdate, error)

Transform implements StreamTransformer.

type InMemoryCheckpointer 

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

InMemoryCheckpointer is a simple thread-safe in-memory checkpointer.

func NewInMemoryCheckpointer 

func NewInMemoryCheckpointer() *InMemoryCheckpointer

NewInMemoryCheckpointer builds an empty in-memory checkpointer.

func (*InMemoryCheckpointer) Get 

func (c *InMemoryCheckpointer) Get(_ context.Context, threadID string) (*Checkpoint, error)

Get implements Checkpointer.

func (*InMemoryCheckpointer) Put 

func (c *InMemoryCheckpointer) Put(_ context.Context, threadID string, cp *Checkpoint) error

Put implements Checkpointer.

type InterruptedError 

type InterruptedError struct {
	Checkpoint *Checkpoint
}

InterruptedError carries the checkpoint for resumption.

func (*InterruptedError) Error 

func (e *InterruptedError) Error() string

func (*InterruptedError) Unwrap 

func (e *InterruptedError) Unwrap() error

type Message 

type Message struct {
	Role       string
	Content    string
	ToolCalls  []ToolCall
	ToolCallID string
	Name       string
}

Message is a minimal message abstraction used by the graph state.

type Node 

type Node struct {
	ID   NodeID
	Run  NodeFunc
	Name string
}

Node is a node in the graph.

type NodeFunc 

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

NodeFunc is the signature of a graph node. It receives the current state and returns an update to apply to the state, plus an optional set of commands.

type NodeID 

type NodeID string

NodeID identifies a node in the graph. 

const End NodeID = "__end__"

End is the sentinel node id signifying the end of execution. 

const Start NodeID = "__start__"

Start is the sentinel node id for the entry point.

type PassthroughTransformer 

type PassthroughTransformer struct{}

PassthroughTransformer is a no-op transformer that copies its input to its output unchanged.

func (PassthroughTransformer) Transform 

func (p PassthroughTransformer) Transform(_ context.Context, in <-chan StreamUpdate) (<-chan StreamUpdate, error)

Transform implements StreamTransformer.

type RunConfig 

type RunConfig struct {
	// ThreadID identifies the execution thread for checkpointing.
	ThreadID string
	// InterruptBefore/After pause execution at matching nodes.
	InterruptBefore []NodeID
	InterruptAfter  []NodeID
	// Resume supplies a value to resume an interrupted run.
	Resume any
}

RunConfig configures a single graph run.

type RunOption 

type RunOption func(*RunConfig)

RunOption sets a RunConfig field.

func WithInterruptAfter 

func WithInterruptAfter(ids ...NodeID) RunOption

WithInterruptAfter sets nodes to pause after.

func WithInterruptBefore 

func WithInterruptBefore(ids ...NodeID) RunOption

WithInterruptBefore sets nodes to pause before.

func WithResume 

func WithResume(v any) RunOption

WithResume sets a resume value.

func WithThreadID 

func WithThreadID(id string) RunOption

WithThreadID sets the thread id.

type Send 

type Send struct {
	Node  NodeID
	Value any
}

Send describes a parallel branch: a node to run with a partial state update.

type State 

type State = map[string]any

State is the shared mutable state threaded through the graph. It is a map so nodes can add or update arbitrary keys.

type StateGraph 

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

StateGraph is a mutable graph of nodes and edges.

func NewStateGraph 

func NewStateGraph() *StateGraph

NewStateGraph builds an empty state graph.

func (*StateGraph) AddConditionalEdges 

func (g *StateGraph) AddConditionalEdges(from NodeID, cond CondFunc) *StateGraph

AddConditionalEdges attaches a condition function to a source node.

func (*StateGraph) AddEdge 

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

AddEdge connects two nodes. A node may have at most one outgoing edge unless conditional edges are used.

func (*StateGraph) AddInterrupt 

func (g *StateGraph) AddInterrupt(id NodeID) *StateGraph

AddInterrupt marks a node as an interrupt point. Execution pauses before the node runs and can be resumed later.

func (*StateGraph) AddNode 

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

AddNode registers a node in the graph.

func (*StateGraph) Compile 

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

Compile validates and freezes the graph, returning a runnable graph.

func (*StateGraph) SetCheckpointer 

func (g *StateGraph) SetCheckpointer(c Checkpointer) *StateGraph

SetCheckpointer sets the checkpointer used for interrupts and persistence.

func (*StateGraph) SetEntryPoint 

func (g *StateGraph) SetEntryPoint(id NodeID) *StateGraph

SetEntryPoint sets the node execution starts at.

type StreamTransformer 

type StreamTransformer interface {
	// Transform receives the input stream and returns a transformed stream.
	Transform(ctx context.Context, in <-chan StreamUpdate) (<-chan StreamUpdate, error)
}

StreamTransformer transforms the stream of a node's output. It is applied to the update stream produced by graph.Stream. It mirrors langgraph's StreamTransformer factories.

type StreamTransformerFunc 

type StreamTransformerFunc func(ctx context.Context, in <-chan StreamUpdate) (<-chan StreamUpdate, error)

StreamTransformerFunc adapts a function into a StreamTransformer.

func (StreamTransformerFunc) Transform 

func (f StreamTransformerFunc) Transform(ctx context.Context, in <-chan StreamUpdate) (<-chan StreamUpdate, error)

Transform implements StreamTransformer.

type StreamUpdate 

type StreamUpdate struct {
	Node  NodeID
	State State
}

StreamUpdate is a single emission from a graph run.

type Tool 

type Tool struct {
	// Name is the tool's unique name.
	Name string
	// Description describes when to use the tool.
	Description string
	// ArgsSchema is the JSON schema for the tool arguments.
	ArgsSchema map[string]any
	// Func executes the tool.
	Func ToolFn
}

Tool describes a tool available to the graph's ToolNode.

func (Tool) Schema 

func (t Tool) Schema() map[string]any

Schema returns the tool's JSON schema as a map.

type ToolCall 

type ToolCall struct {
	ID    string
	Name  string
	Args  map[string]any
	Index int
}

ToolCall is a structured tool invocation emitted by the model.

type ToolFn 

type ToolFn func(ctx context.Context, input string) (string, error)

ToolFn is the signature of a tool callable by a ToolNode.

type ToolNode 

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

ToolNode executes tool calls found in the state. It reads "messages" from the state, finds assistant tool_calls, executes each via the registered tool, and appends tool messages to the message list.

func NewToolNode 

func NewToolNode(tools ...Tool) *ToolNode

NewToolNode builds a ToolNode from the given tools.

func (*ToolNode) Run 

func (t *ToolNode) Run(ctx context.Context, state State) (State, error)

Run implements NodeFunc. It executes the tool calls in the latest message and returns an update appending tool messages.

func (*ToolNode) ToolsByKey 

func (t *ToolNode) ToolsByKey() map[string]Tool

ToolsByKey returns the tools keyed by name.

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.