Help us shape the future of Go! Take the 2026 Go Developer Survey

graph

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 6, 2026 License: MIT Imports: 14 Imported by: 0

README

graph

CI Latest release Go Reference

graph builds and runs typed directed acyclic graphs in Go.

The package provides these features:

  • Typed input and output ports
  • Reusable node types and unique node instances
  • Parallel execution with a fixed limit
  • Fail-fast cancellation and partial results
  • Strict JSON and YAML definitions
  • Ordered lifecycle events
  • Aggregated validation diagnostics
  • Optional ports with explicit absence
  • Concurrent reuse of compiled graphs

How it works

  1. Implement reusable node types.
  2. Register node factories.
  3. Build or decode a graph definition.
  4. Compile the definition.
  5. Run the immutable graph.

Compilation checks the complete graph before execution. Runtime checks protect graph input and node output boundaries.

Install

Run this command:

go get github.com/bpradana/graph

Quick start

Define a node with two inputs and two outputs:

type arithmeticNode struct{}

func (*arithmeticNode) Inputs() []graph.InputPort {
	integer := reflect.TypeFor[int]()
	return []graph.InputPort{
		{Name: "left", TypeID: "integer/v1", Type: integer},
		{Name: "right", TypeID: "integer/v1", Type: integer},
	}
}

func (*arithmeticNode) Outputs() []graph.OutputPort {
	integer := reflect.TypeFor[int]()
	return []graph.OutputPort{
		{Name: "sum", TypeID: "integer/v1", Type: integer},
		{Name: "product", TypeID: "integer/v1", Type: integer},
	}
}

func (*arithmeticNode) Run(
	_ context.Context,
	inputs graph.Inputs,
) (graph.Outputs, error) {
	left, err := graph.ValueAs[int](inputs, "left")
	if err != nil {
		return graph.Outputs{}, err
	}
	right, err := graph.ValueAs[int](inputs, "right")
	if err != nil {
		return graph.Outputs{}, err
	}
	return graph.NewValues(map[string]any{
		"sum":     left + right,
		"product": left * right,
	}), nil
}

Register the reusable node type:

registry := graph.NewRegistry()
err := registry.Register(
	"math/arithmetic",
	func(graph.Config) (graph.Node, error) {
		return &arithmeticNode{}, nil
	},
)
if err != nil {
	return err
}

Create a definition and compile it:

definitionData := []byte(`
apiVersion: graph/v1
name: arithmetic
inputs:
  - name: left
    type: integer/v1
    targets: [{node: math, port: left}]
  - name: right
    type: integer/v1
    targets: [{node: math, port: right}]
outputs:
  - name: sum
    type: integer/v1
    source: {node: math, port: sum}
nodes:
  - id: math
    type: math/arithmetic
    inputs:
      - {name: left, type: integer/v1}
      - {name: right, type: integer/v1}
    outputs:
      - {name: sum, type: integer/v1}
      - {name: product, type: integer/v1}
`)

definition, err := graph.DecodeYAML(definitionData)
if err != nil {
	return err
}

compiled, err := graph.Compile(definition, registry)
if err != nil {
	return err
}

Run the graph:

executor, err := graph.NewExecutor(graph.WithMaxParallel(4))
if err != nil {
	return err
}

ctx := context.Background()
result, err := executor.Run(ctx, compiled, graph.NewValues(map[string]any{
	"left":  6,
	"right": 7,
}))
if err != nil {
	return err
}

sum, err := graph.ValueAs[int](result.Outputs, "sum")

See example_test.go for a Go example test. See the runnable examples for complete programs.

Documentation

Feature map

Feature Documentation
Nodes, ports, and type IDs Implement a node
Registry and configuration Register reusable node types
Programmatic definitions Define a graph with Builder
YAML and JSON Schema reference
Edges and fan-out Port connections
Optional values Optional ports
Compilation diagnostics Compilation validation
Parallel execution Scheduling
Errors and partial results Failures and cancellation
Lifecycle events Observers
Concurrency and ownership Thread safety
Package limitations Limitations
Common errors Troubleshooting

Requirements

  • Go 1.26.5 or a compatible later version

Documentation

Overview

Package graph builds and runs directed acyclic graphs.

A graph connects typed node ports. Independent nodes can run at the same time. Definitions can come from Go values, JSON, or YAML.

Compile resolves reusable node types and validates the complete graph. Executor runs the immutable graph with bounded parallelism and fail-fast cancellation.

Values move through named ports. Each port has a portable type ID and a declared Go type. Optional ports represent values that can be absent.

Example 
package main

import (
	"context"
	"fmt"
	"reflect"

	graph "github.com/bpradana/graph"
)

type arithmeticNode struct{}

func (*arithmeticNode) Inputs() []graph.InputPort {
	integer := reflect.TypeFor[int]()
	return []graph.InputPort{
		{Name: "left", TypeID: "integer/v1", Type: integer},
		{Name: "right", TypeID: "integer/v1", Type: integer},
	}
}

func (*arithmeticNode) Outputs() []graph.OutputPort {
	integer := reflect.TypeFor[int]()
	return []graph.OutputPort{
		{Name: "sum", TypeID: "integer/v1", Type: integer},
		{Name: "product", TypeID: "integer/v1", Type: integer},
	}
}

func (*arithmeticNode) Run(_ context.Context, inputs graph.Inputs) (graph.Outputs, error) {
	left, err := graph.ValueAs[int](inputs, "left")
	if err != nil {
		return graph.Outputs{}, err
	}
	right, err := graph.ValueAs[int](inputs, "right")
	if err != nil {
		return graph.Outputs{}, err
	}
	return graph.NewValues(map[string]any{
		"sum":     left + right,
		"product": left * right,
	}), nil
}

func main() {
	registry := graph.NewRegistry()
	_ = registry.Register("math/arithmetic", func(graph.Config) (graph.Node, error) {
		return &arithmeticNode{}, nil
	})

	integerPorts := []graph.PortDefinition{
		{Name: "left", TypeID: "integer/v1"},
		{Name: "right", TypeID: "integer/v1"},
	}
	definition := graph.Definition{
		APIVersion: graph.APIVersion,
		Name:       "arithmetic",
		Inputs: []graph.GraphInputDefinition{
			{PortDefinition: integerPorts[0], Targets: []graph.Endpoint{{Node: "math", Port: "left"}}},
			{PortDefinition: integerPorts[1], Targets: []graph.Endpoint{{Node: "math", Port: "right"}}},
		},
		Outputs: []graph.GraphOutputDefinition{
			{PortDefinition: graph.PortDefinition{Name: "sum", TypeID: "integer/v1"}, Source: graph.Endpoint{Node: "math", Port: "sum"}},
			{PortDefinition: graph.PortDefinition{Name: "product", TypeID: "integer/v1"}, Source: graph.Endpoint{Node: "math", Port: "product"}},
		},
		Nodes: []graph.NodeDefinition{{
			ID: "math", Type: "math/arithmetic",
			Inputs: integerPorts,
			Outputs: []graph.PortDefinition{
				{Name: "sum", TypeID: "integer/v1"},
				{Name: "product", TypeID: "integer/v1"},
			},
		}},
	}

	compiled, _ := graph.Compile(definition, registry)
	executor, _ := graph.NewExecutor(graph.WithMaxParallel(2))
	result, _ := executor.Run(context.Background(), compiled, graph.NewValues(map[string]any{
		"left":  6,
		"right": 7,
	}))
	sum, _ := graph.ValueAs[int](result.Outputs, "sum")
	product, _ := graph.ValueAs[int](result.Outputs, "product")
	fmt.Println(sum, product)
}

Output:
13 42

Index

Examples

Constants

View Source 
const APIVersion = "graph/v1"

APIVersion identifies the supported serialized definition format.

Variables

This section is empty.

Functions

func EncodeJSON 

func EncodeJSON(definition Definition) ([]byte, error)

EncodeJSON encodes a graph definition as indented JSON.

Encoding does not validate graph semantics.

func EncodeYAML 

func EncodeYAML(definition Definition) ([]byte, error)

EncodeYAML encodes a graph definition as YAML.

Encoding does not preserve source comments or formatting.

func LookupAs 

func LookupAs[T any](values Values, name string) (T, bool, error)

LookupAs returns an optional value with the requested Go type.

The Boolean result is false when the name is absent. It returns an error when a present value has the wrong Go type.

func ValueAs 

func ValueAs[T any](values Values, name string) (T, error)

ValueAs returns a required value with the requested Go type.

It returns an error when the name is absent or the Go type does not match.

Types

type Builder 

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

Builder creates a graph definition through fluent node bindings.

Builder resolves node contracts during Build. Builder collects errors and returns them from Build. Builder does not support concurrent use.

func NewBuilder 

func NewBuilder(name string, resolver Resolver) *Builder

NewBuilder creates a fluent builder for a named graph.

func (*Builder) Build 

func (b *Builder) Build() (Definition, error)

Build resolves node contracts and returns the generated graph definition.

Build returns ValidationError when a builder operation or connection is invalid.

func (*Builder) Input added in v1.1.0 

func (b *Builder) Input(name string) *InputRef

Input declares one graph input and returns a reusable source reference.

func (*Builder) Node added in v1.1.0 

func (b *Builder) Node(id, nodeType string) *NodeRef

Node declares one node instance and returns its fluent reference.

func (*Builder) Output added in v1.1.0 

func (b *Builder) Output(name string, source *OutputRef) *Builder

Output exposes one node output as a named graph output.

type Config 

type Config map[string]any

Config contains portable node configuration values.

JSON and YAML definitions can contain maps, lists, numbers, strings, Boolean values, and null values.

func (Config) Decode 

func (c Config) Decode(target any) error

Decode converts the configuration to a typed value.

Target must be a non-nil pointer. Decode rejects fields that do not exist in the target type.

func (*Config) UnmarshalJSON 

func (c *Config) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes configuration and preserves JSON number values.

Applications normally use DecodeJSON instead of calling this method.

type Definition 

type Definition struct {
	// APIVersion identifies the serialized schema version.
	APIVersion string `json:"apiVersion" yaml:"apiVersion"`
	// Name identifies the graph in diagnostics and events.
	Name string `json:"name" yaml:"name"`
	// Inputs declares values that a caller can supply.
	Inputs []GraphInputDefinition `json:"inputs,omitempty" yaml:"inputs,omitempty"`
	// Outputs declares values that a caller can read.
	Outputs []GraphOutputDefinition `json:"outputs,omitempty" yaml:"outputs,omitempty"`
	// Nodes declares all node instances.
	Nodes []NodeDefinition `json:"nodes" yaml:"nodes"`
	// Edges connects node output ports to node input ports.
	Edges []EdgeDefinition `json:"edges,omitempty" yaml:"edges,omitempty"`
}

Definition describes a serializable graph.

func DecodeJSON 

func DecodeJSON(data []byte) (Definition, error)

DecodeJSON decodes one strict JSON graph definition.

It rejects unknown fields, trailing values, malformed data, and unsupported API versions.

func DecodeYAML 

func DecodeYAML(data []byte) (Definition, error)

DecodeYAML decodes one strict YAML graph definition.

It rejects unknown fields, duplicate keys, multiple documents, malformed data, and unsupported API versions.

Example 
package main

import (
	"fmt"

	graph "github.com/bpradana/graph"
)

func main() {
	data := []byte(`
apiVersion: graph/v1
name: empty
nodes: []
`)
	definition, err := graph.DecodeYAML(data)
	fmt.Println(definition.Name, err)
}

Output:
empty <nil>

type EdgeDefinition 

type EdgeDefinition struct {
	// From identifies the source output port.
	From Endpoint `json:"from" yaml:"from"`
	// To identifies the target input port.
	To Endpoint `json:"to" yaml:"to"`
}

EdgeDefinition connects one node output to one node input.

type Endpoint 

type Endpoint struct {
	// Node identifies one node instance.
	Node string `json:"node" yaml:"node"`
	// Port identifies one port on the node instance.
	Port string `json:"port" yaml:"port"`
}

Endpoint identifies one port on one node instance.

type Event 

type Event struct {
	// Kind identifies the lifecycle transition.
	Kind EventKind
	// GraphName identifies the running graph.
	GraphName string
	// NodeID identifies the applicable node instance.
	NodeID string
	// Err contains the applicable failure.
	Err error
	// Time records when the executor emitted the event.
	Time time.Time
}

Event describes one graph lifecycle transition.

type EventKind 

type EventKind string

EventKind identifies one graph lifecycle transition. 

const (
	// EventGraphStarted reports the start of one graph run.
	EventGraphStarted EventKind = "graph_started"
	// EventNodeStarted reports the start of one node run.
	EventNodeStarted EventKind = "node_started"
	// EventNodeSucceeded reports successful node completion.
	EventNodeSucceeded EventKind = "node_succeeded"
	// EventNodeFailed reports failed node completion.
	EventNodeFailed EventKind = "node_failed"
	// EventGraphSucceeded reports successful graph completion.
	EventGraphSucceeded EventKind = "graph_succeeded"
	// EventGraphFailed reports failed graph completion.
	EventGraphFailed EventKind = "graph_failed"
)

type Executor 

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

Executor runs compiled graphs.

One executor supports concurrent Run calls.

func NewExecutor 

func NewExecutor(options ...ExecutorOption) (*Executor, error)

NewExecutor creates an executor.

The default parallel limit is runtime.GOMAXPROCS(0).

func (*Executor) Run 

func (e *Executor) Run(ctx context.Context, graph *Graph, inputs Values) (RunResult, error)

Run executes one compiled graph.

Run validates all graph inputs before it starts a node. Run returns successful partial results when execution fails. It waits for every active node before it returns.

type ExecutorOption 

type ExecutorOption interface {
	// contains filtered or unexported methods
}

ExecutorOption configures an executor.

func WithMaxParallel 

func WithMaxParallel(limit int) ExecutorOption

WithMaxParallel sets the maximum number of active nodes in each run.

NewExecutor returns an error when limit is not positive.

func WithObserver 

func WithObserver(observer Observer) ExecutorOption

WithObserver adds one lifecycle observer.

The executor calls observers in option order. NewExecutor rejects a nil observer.

type Graph 

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

Graph is an immutable compiled graph.

Graph supports concurrent runs. Each runtime node must support concurrent Run calls.

func Compile 

func Compile(definition Definition, resolver Resolver) (*Graph, error)

Compile resolves and validates a graph definition.

Compile returns one ValidationError with all detected issues. It does not return a graph when validation fails.

func (*Graph) Name 

func (g *Graph) Name() string

Name returns the graph name.

type GraphInputDefinition 

type GraphInputDefinition struct {
	// PortDefinition declares the graph input contract.
	PortDefinition `json:",inline" yaml:",inline"`
	// Targets identifies every node input that receives the value.
	Targets []Endpoint `json:"targets" yaml:"targets"`
}

GraphInputDefinition exposes one graph input.

type GraphOutputDefinition 

type GraphOutputDefinition struct {
	// PortDefinition declares the graph output contract.
	PortDefinition `json:",inline" yaml:",inline"`
	// Source identifies the node output that supplies the value.
	Source Endpoint `json:"source" yaml:"source"`
}

GraphOutputDefinition exposes one graph output.

type InputPort 

type InputPort struct {
	// Name identifies the port inside one node.
	Name string
	// TypeID identifies the portable value contract.
	TypeID string
	// Type identifies the required Go value type.
	Type reflect.Type
	// Optional permits the input value to be absent.
	Optional bool
}

InputPort describes one input that a node accepts.

type InputRef added in v1.1.0 

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

InputRef identifies one graph input created by Builder.Input. Its fields are intentionally hidden.

type Inputs 

type Inputs = Values

Inputs contains the values that a node receives.

type Issue 

type Issue struct {
	// Path identifies the invalid definition location.
	Path string
	// Message describes the validation failure.
	Message string
}

Issue describes one invalid part of a graph definition.

type Node 

type Node interface {
	// Inputs returns all input ports that the node accepts.
	Inputs() []InputPort
	// Outputs returns all output ports that the node can produce.
	Outputs() []OutputPort
	// Run executes the node with one input collection.
	Run(context.Context, Inputs) (Outputs, error)
}

Node runs one operation in a graph.

Inputs and Outputs declare the complete runtime port contract. Compile compares this contract with the node definition.

Run receives values only for present input ports. It must return all required outputs. Run can be called concurrently when a compiled graph has concurrent runs.

type NodeDefinition 

type NodeDefinition struct {
	// ID uniquely identifies this node instance inside the graph.
	ID string `json:"id" yaml:"id"`
	// Type selects a reusable node type through the resolver.
	Type string `json:"type" yaml:"type"`
	// Inputs repeats the complete runtime input contract.
	Inputs []PortDefinition `json:"inputs,omitempty" yaml:"inputs,omitempty"`
	// Outputs repeats the complete runtime output contract.
	Outputs []PortDefinition `json:"outputs,omitempty" yaml:"outputs,omitempty"`
	// Config contains configuration for this node instance.
	Config Config `json:"config,omitempty" yaml:"config,omitempty"`
}

NodeDefinition describes one node instance.

type NodeFactory 

type NodeFactory func(Config) (Node, error)

NodeFactory creates one runtime node from instance configuration.

type NodeRef added in v1.1.0 

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

NodeRef identifies one node instance created by Builder.Node.

func (*NodeRef) Bind added in v1.1.0 

func (n *NodeRef) Bind(input string, source Source) *NodeRef

Bind connects one source to a named node input.

func (*NodeRef) Configure added in v1.1.0 

func (n *NodeRef) Configure(config Config) *NodeRef

Configure replaces the node instance configuration.

func (*NodeRef) Out added in v1.1.0 

func (n *NodeRef) Out(output string) *OutputRef

Out returns a reference to one named node output.

type Observer 

type Observer interface {
	Observe(context.Context, Event)
}

Observer receives ordered graph lifecycle events.

The executor calls an observer synchronously. It never calls one observer concurrently. An observer must not call Run on the same executor.

type ObserverError 

type ObserverError struct {
	// Value contains the recovered panic value.
	Value any
	// Stack contains the recovered goroutine stack.
	Stack []byte
	// Event contains the event that caused the observer panic.
	Event Event
}

ObserverError reports a recovered observer panic.

func (*ObserverError) Error 

func (e *ObserverError) Error() string

Error returns the recovered observer panic value.

type ObserverFunc 

type ObserverFunc func(context.Context, Event)

ObserverFunc adapts a function to the Observer interface.

func (ObserverFunc) Observe 

func (f ObserverFunc) Observe(ctx context.Context, event Event)

Observe sends one event to the function.

type OutputPort 

type OutputPort struct {
	// Name identifies the port inside one node.
	Name string
	// TypeID identifies the portable value contract.
	TypeID string
	// Type identifies the required Go value type.
	Type reflect.Type
	// Optional permits the output value to be absent.
	Optional bool
}

OutputPort describes one output that a node produces.

type OutputRef added in v1.1.0 

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

OutputRef identifies one node output created by NodeRef.Out. Its fields are intentionally hidden.

type Outputs 

type Outputs = Values

Outputs contains the values that a node returns.

type PanicError 

type PanicError struct {
	// Value contains the recovered panic value.
	Value any
	// Stack contains the recovered goroutine stack.
	Stack []byte
}

PanicError reports a recovered node panic.

func (*PanicError) Error 

func (e *PanicError) Error() string

Error returns the recovered panic value.

type PortDefinition 

type PortDefinition struct {
	// Name identifies the port inside one node or graph boundary.
	Name string `json:"name" yaml:"name"`
	// TypeID identifies the portable value contract.
	TypeID string `json:"type" yaml:"type"`
	// Optional permits the value to be absent.
	Optional bool `json:"optional,omitempty" yaml:"optional,omitempty"`
}

PortDefinition describes a portable port contract.

type Registry 

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

Registry resolves node types through registered factories.

Registry supports concurrent registration and resolution.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates an empty node registry.

func (*Registry) Register 

func (r *Registry) Register(nodeType string, factory NodeFactory) error

Register adds a factory for one stable node type.

Register rejects an empty type, a nil factory, or a duplicate type.

func (*Registry) Resolve 

func (r *Registry) Resolve(definition NodeDefinition) (Node, error)

Resolve creates a new runtime node for a node definition.

Resolve returns an error for an unknown type, a factory error, or a nil node.

type Resolver 

type Resolver interface {
	Resolve(NodeDefinition) (Node, error)
}

Resolver creates a runtime node for one node definition.

Compile calls Resolve once for each valid node instance.

type RunError 

type RunError struct {
	// NodeID identifies the failed node instance.
	NodeID string
	// Err contains the original node error.
	Err error
}

RunError reports a failed node.

func (*RunError) Error 

func (e *RunError) Error() string

Error returns the node failure summary.

func (*RunError) Unwrap 

func (e *RunError) Unwrap() error

Unwrap returns the original node error.

type RunResult 

type RunResult struct {
	// Outputs contains present graph boundary outputs.
	Outputs Values
	// NodeResults contains outputs from every successful completed node.
	NodeResults map[string]Outputs
}

RunResult contains graph outputs and successful node results.

type Source added in v1.1.0 

type Source interface {
	// contains filtered or unexported methods
}

Source identifies a value that can supply one node input.

InputRef and OutputRef are the only implementations.

type ValidationError 

type ValidationError struct {
	// Issues contains every detected validation issue.
	Issues []Issue
}

ValidationError contains all detected definition issues.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

Error returns a summary of the validation issues.

type Values 

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

Values contains named graph values.

The zero value is an empty collection. NewValues copies the source map. Values stored inside the map are not copied.

func NewValues 

func NewValues(source map[string]any) Values

NewValues creates a value collection from a map.

A nil source creates an empty collection.

func (Values) Len 

func (v Values) Len() int

Len returns the number of present values.

func (Values) Names 

func (v Values) Names() []string

Names returns the present names in lexical order.

Source Files

View all Source files

Directories

Path Synopsis
docs
examples/concurrency command
examples/json command
examples/observer command
examples/optional command
examples/programmatic command
examples/yaml command

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.