core

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Aug 27, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package core provides the fundamental building blocks for creating and executing graph-based workflows in GoLangGraph.

The core package implements a graph execution engine that allows you to define workflows as directed graphs where nodes represent computational units and edges define the flow of execution. This package is the foundation of the GoLangGraph framework and provides the essential abstractions for building AI agent workflows.

Graph Execution Model

The core execution model revolves around two main concepts:

  • Graph: A directed graph structure containing nodes and edges that defines the workflow topology
  • BaseState: A thread-safe state container that carries data between nodes during execution

Basic Usage

Creating and executing a simple graph: 

graph := core.NewGraph("my-workflow")

// Add nodes with processing functions
graph.AddNode("start", "Start Node", func(ctx context.Context, state *core.BaseState) (*core.BaseState, error) {
	state.Set("message", "Hello, World!")
	return state, nil
})

graph.AddNode("end", "End Node", func(ctx context.Context, state *core.BaseState) (*core.BaseState, error) {
	message, _ := state.Get("message")
	fmt.Println(message)
	return state, nil
})

// Connect nodes
graph.AddEdge("start", "end", nil)
graph.SetStartNode("start")
graph.AddEndNode("end")

// Execute the graph
initialState := core.NewBaseState()
ctx := context.Background()
finalState, err := graph.Execute(ctx, initialState)

Conditional Execution

Graphs support conditional edges that determine the next node based on the current state: 

graph.AddEdge("decision", "path_a", func(ctx context.Context, state *core.BaseState) (string, error) {
	if condition, _ := state.Get("condition"); condition == "A" {
		return "path_a", nil
	}
	return "path_b", nil
})

State Management

The BaseState provides thread-safe access to workflow data: 

state := core.NewBaseState()
state.Set("key", "value")
value, exists := state.Get("key")
state.SetMetadata("execution_id", "12345")

// Clone state for parallel processing
clonedState := state.Clone()

// Merge states from parallel branches
state.Merge(otherState)

Streaming Execution

For long-running workflows, use streaming execution to receive intermediate results: 

resultChan := make(chan *core.BaseState, 10)
go func() {
	err := graph.Stream(ctx, initialState, resultChan)
	close(resultChan)
}()

for state := range resultChan {
	// Process intermediate state
}

Error Handling

The package provides comprehensive error handling with automatic retries and graceful degradation:

  • Node execution errors are wrapped with context information
  • Validation errors prevent invalid graph configurations
  • Timeout handling for long-running operations
  • Interrupt support for graceful cancellation

Thread Safety

All core types are designed to be thread-safe:

  • BaseState uses read-write mutexes for concurrent access
  • Graph execution supports parallel node processing
  • State cloning enables safe parallel branches

Performance Considerations

The core package is optimized for performance:

  • Minimal memory allocation during execution
  • Efficient state management with copy-on-write semantics
  • Lazy evaluation of conditional edges
  • Configurable retry policies and timeouts

For more advanced usage patterns and integration with other GoLangGraph packages, see the examples in the examples/ directory and the comprehensive documentation in the docs/ directory.

Index

Constants

View Source 
const (
	START = "__start__"
	END   = "__end__"
)

START and END constants for graph flow control

Variables

View Source 
var (
	// ErrGraphInvalid indicates the graph structure failed validation.
	ErrGraphInvalid = errors.New("graph validation failed")
	// ErrRecursionLimit indicates execution exceeded GraphConfig.MaxIterations.
	// This mirrors LangGraph's GraphRecursionError.
	ErrRecursionLimit = errors.New("recursion limit exceeded")
	// ErrInterrupted indicates execution was stopped via Interrupt.
	ErrInterrupted = errors.New("execution interrupted")
	// ErrNodePanic indicates a node function panicked. The panic is recovered
	// and converted to this error; the engine never leaves locks held.
	ErrNodePanic = errors.New("node panicked")
	// ErrNoRoute indicates no outgoing edge matched from a node.
	ErrNoRoute = errors.New("no valid next node")
	// ErrGraphClosed indicates the graph has been closed via Close.
	ErrGraphClosed = errors.New("graph is closed")
)

Sentinel errors returned by graph execution. Callers should use errors.Is to classify failures rather than matching on message text.

View Source 
var ErrSubgraphInterrupted = errors.New("subgraph interrupted")

ErrSubgraphInterrupted wraps an interrupt raised inside a nested graph.

Functions

func RouteByMessageType 

func RouteByMessageType(ctx context.Context, state *BaseState) (string, error)

RouteByMessageType routes based on the type of the last message

func RouteByToolCalls 

func RouteByToolCalls(ctx context.Context, state *BaseState) (string, error)

RouteByToolCalls routes based on whether tool calls are present

Types

type BaseState 

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

BaseState represents the base state structure

func NewBaseState 

func NewBaseState() *BaseState

NewBaseState creates a new base state

func (*BaseState) Clone 

func (bs *BaseState) Clone() *BaseState

Clone creates a deep copy of the state. Cloning a nil state yields a new empty state so that a node returning nil can never crash the engine.

func (*BaseState) CreateSnapshot 

func (bs *BaseState) CreateSnapshot() StateSnapshot

CreateSnapshot creates a snapshot of the current state

func (*BaseState) Delete 

func (bs *BaseState) Delete(key string)

Delete removes a key from the state

func (*BaseState) FromJSON 

func (bs *BaseState) FromJSON(data []byte) error

FromJSON loads the state from JSON

func (*BaseState) Get 

func (bs *BaseState) Get(key string) (StateValue, bool)

Get retrieves a value from the state

func (*BaseState) GetAll 

func (bs *BaseState) GetAll() map[string]StateValue

GetAll returns a copy of all data in the state

func (*BaseState) GetHistory 

func (bs *BaseState) GetHistory() *StateHistory

GetHistory returns the state history

func (*BaseState) GetMetadata 

func (bs *BaseState) GetMetadata(key string) (interface{}, bool)

GetMetadata retrieves metadata from the state

func (*BaseState) Keys 

func (bs *BaseState) Keys() []string

Keys returns all keys in the state

func (*BaseState) MarshalJSON 

func (bs *BaseState) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

BaseState keeps its data in unexported fields, so without this method encoding/json serializes it as "{}" and every persisted checkpoint, API response and WebSocket frame silently loses the entire state.

func (*BaseState) Merge 

func (bs *BaseState) Merge(other *BaseState)

Merge merges another state into this state using last-write-wins semantics for every key. Use MergeWithSchema to apply reducers.

func (*BaseState) MergeWithSchema 

func (bs *BaseState) MergeWithSchema(other *BaseState, schema *StateSchema)

MergeWithSchema merges another state into this one, applying the schema's reducer for each key. Keys without a reducer use last-write-wins, matching LangGraph's default channel behavior.

func (*BaseState) RestoreFromSnapshot 

func (bs *BaseState) RestoreFromSnapshot(snapshot StateSnapshot)

RestoreFromSnapshot restores the state from a snapshot

func (*BaseState) Set 

func (bs *BaseState) Set(key string, value StateValue)

Set sets a value in the state

func (*BaseState) SetMetadata 

func (bs *BaseState) SetMetadata(key string, value interface{})

SetMetadata sets metadata for the state

func (*BaseState) ToJSON 

func (bs *BaseState) ToJSON() ([]byte, error)

ToJSON converts the state to JSON

func (*BaseState) UnmarshalJSON 

func (bs *BaseState) UnmarshalJSON(raw []byte) error

UnmarshalJSON implements json.Unmarshaler and accepts both the canonical {"data":...,"metadata":...} envelope and a bare object of state values, so older payloads and hand-written requests both load.

func (*BaseState) Update 

func (bs *BaseState) Update(schema *StateSchema, key string, value StateValue)

Update applies a single key update through the schema reducer, if any.

type Channel 

type Channel struct {
	Key     string
	Reducer Reducer
	Default func() StateValue
}

Channel describes one key of the graph state: how updates are combined and what the value is before anything has been written.

type ConditionalEdge 

type ConditionalEdge struct {
	ID        string                 `json:"id"`
	From      string                 `json:"from"`
	Condition EdgeCondition          `json:"-"`
	Routes    map[string]string      `json:"routes"` // condition result -> target node
	Metadata  map[string]interface{} `json:"metadata"`
}

ConditionalEdge represents a conditional edge that can route to different nodes

type ConditionalRouter 

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

ConditionalRouter manages conditional routing logic

func NewConditionalRouter 

func NewConditionalRouter(fallback string) *ConditionalRouter

NewConditionalRouter creates a new conditional router

func (*ConditionalRouter) AddRoute 

func (cr *ConditionalRouter) AddRoute(condition string, router RouterFunction)

AddRoute adds a route with a condition. Routes are evaluated in the order they were added, which keeps routing deterministic.

func (*ConditionalRouter) Route 

func (cr *ConditionalRouter) Route(ctx context.Context, state *BaseState) (string, error)

Route determines the next node based on state. Conditions are evaluated in insertion order; the first non-empty result wins. A condition that returns an error is skipped, and the fallback is used when nothing matches.

type Edge 

type Edge struct {
	ID        string                 `json:"id"`
	From      string                 `json:"from"`
	To        string                 `json:"to"`
	Condition EdgeCondition          `json:"-"`
	Metadata  map[string]interface{} `json:"metadata"`
}

Edge represents an edge in the graph

type EdgeCondition 

type EdgeCondition func(ctx context.Context, state *BaseState) (string, error)

EdgeCondition represents a condition function for conditional edges.

For per-edge conditions (AddEdge), the function returns the target node ID to take that edge, or "" to decline it. For routed conditional edges (AddConditionalEdges), the function returns a routing key that is mapped through the route table.

type ExecuteOptions 

type ExecuteOptions struct {
	// ThreadID scopes checkpoints for this run. Empty uses the graph default.
	ThreadID string
	// StartNode overrides the entry point (used by Resume).
	StartNode string
	// Stream, when non-nil, receives per-step results for this run only.
	// It is closed when the run finishes.
	Stream chan<- *ExecutionResult
	// ResumeStep sets the starting step counter (used by Resume).
	ResumeStep int
}

ExecuteOptions customizes a single run.

type ExecutionResult 

type ExecutionResult struct {
	NodeID       string        `json:"node_id"`
	Success      bool          `json:"success"`
	Error        error         `json:"-"`
	ErrorMessage string        `json:"error,omitempty"`
	Duration     time.Duration `json:"duration"`
	Timestamp    time.Time     `json:"timestamp"`
	State        *BaseState    `json:"state,omitempty"`
	// Step is the 0-based index of this node execution within the run.
	Step int `json:"step"`
	// Attempts is the number of attempts made (1 when the node succeeded first try).
	Attempts int `json:"attempts"`
}

ExecutionResult represents the result of node execution.

Error holds the Go error and is not serialisable; ErrorMessage carries the same information over JSON/WebSocket so clients such as GoLangGraph Studio can display failures.

type Graph 

type Graph struct {
	ID        string                 `json:"id"`
	Name      string                 `json:"name"`
	Nodes     map[string]*Node       `json:"nodes"`
	Edges     map[string]*Edge       `json:"edges"`
	StartNode string                 `json:"start_node"`
	EndNodes  []string               `json:"end_nodes"`
	Config    *GraphConfig           `json:"config"`
	Metadata  map[string]interface{} `json:"metadata"`
	// contains filtered or unexported fields
}

Graph represents the execution graph.

A Graph is safe for concurrent use: Execute keeps all mutable run state in a per-invocation structure, so multiple goroutines may execute the same graph simultaneously without interfering with each other.

func NewGraph 

func NewGraph(name string) *Graph

NewGraph creates a new graph

func (*Graph) AddConditionalEdges 

func (g *Graph) AddConditionalEdges(from string, condition EdgeCondition, routes map[string]string) error

AddConditionalEdges adds conditional edges to the graph, mirroring LangGraph's add_conditional_edges: a single path function is evaluated once per visit and its result is mapped through the route table.

An empty routes map means the condition returns the destination node ID directly. END is accepted as a destination.

func (*Graph) AddEdge 

func (g *Graph) AddEdge(from, to string, condition EdgeCondition) *Edge

AddEdge adds an edge to the graph. Edges are followed in insertion order, which makes routing deterministic.

func (*Graph) AddEndNode 

func (g *Graph) AddEndNode(nodeID string) error

AddEndNode adds an end node to the graph

func (*Graph) AddNode 

func (g *Graph) AddNode(id, name string, fn NodeFunc) *Node

AddNode adds a node to the graph. Adding a node with an existing ID or an empty ID records a build error surfaced by Validate.

func (*Graph) AddSubgraph 

func (g *Graph) AddSubgraph(id, name string, sub *Graph, opts *SubgraphOptions) (*Node, error)

AddSubgraph registers a compiled graph as a node of this graph, mirroring LangGraph's ability to use a compiled graph as a node.

The subgraph runs to completion with its own recursion limit and routing; its resulting state is merged back into the parent according to opts.

func (*Graph) AddUpdateNode 

func (g *Graph) AddUpdateNode(id, name string, fn UpdateFunc) *Node

AddUpdateNode registers a node that returns partial channel updates. The updates are merged into the running state using the graph's state schema, so reducers such as Append and AddMessages apply exactly as they do in LangGraph.

func (*Graph) Close 

func (g *Graph) Close()

Close closes the graph and cleans up resources. It is idempotent and safe to call concurrently with execution: in-flight runs are interrupted first.

func (*Graph) Execute 

func (g *Graph) Execute(ctx context.Context, initialState *BaseState) (*BaseState, error)

Execute executes the graph with the given initial state.

Execute is safe for concurrent use; each call carries its own state and history. On failure it returns the last known good state alongside the error so callers can inspect partial progress.

func (*Graph) ExecuteParallel 

func (g *Graph) ExecuteParallel(ctx context.Context, nodeIDs []string, state *BaseState) (map[string]*ExecutionResult, error)

ExecuteParallel executes multiple nodes concurrently against a shared input state (a LangGraph super-step). Each node receives its own copy of the state; merge the results with MergeResults or a StateSchema reducer.

func (*Graph) ExecuteParallelUpdates 

func (g *Graph) ExecuteParallelUpdates(ctx context.Context, nodeIDs []string, state *BaseState) (*BaseState, error)

ExecuteParallelUpdates runs several update-style nodes concurrently against the same input state and merges their updates through the schema's reducers, implementing a LangGraph super-step over parallel branches.

Branch updates are applied in the order the node IDs were supplied, so the merged result is deterministic regardless of completion order.

func (*Graph) ExecuteWithOptions 

func (g *Graph) ExecuteWithOptions(ctx context.Context, initialState *BaseState, opts *ExecuteOptions) (*BaseState, error)

ExecuteWithOptions runs the graph with per-run options.

func (*Graph) GetConditionalEdge 

func (g *Graph) GetConditionalEdge(nodeID string) (*ConditionalEdge, bool)

GetConditionalEdge retrieves a conditional edge for a node

func (*Graph) GetCurrentState 

func (g *Graph) GetCurrentState() *BaseState

GetCurrentState returns the state of the most recent run.

func (*Graph) GetExecutionHistory 

func (g *Graph) GetExecutionHistory() []*ExecutionResult

GetExecutionHistory returns the execution history of the most recent run.

func (*Graph) GetNextNodes 

func (g *Graph) GetNextNodes(ctx context.Context, currentNodeID string, state *BaseState) ([]string, error)

GetNextNodes determines the next nodes to execute based on current node and state. It delegates to the same routing logic the engine uses, so callers and the executor can never disagree about where a node leads.

func (*Graph) GetNodesByType 

func (g *Graph) GetNodesByType(nodeType string) []*Node

GetNodesByType returns nodes filtered by metadata type

func (*Graph) GetTopology 

func (g *Graph) GetTopology() map[string][]string

GetTopology returns the graph topology as adjacency list, including conditional routes so visualisers and Studio see the full reachable graph.

func (*Graph) Interrupt 

func (g *Graph) Interrupt()

Interrupt interrupts all in-flight executions. It is safe to call at any time, including after Close and when nothing is running.

func (*Graph) IsEndNode 

func (g *Graph) IsEndNode(nodeID string) bool

IsEndNode checks if a node is an end node

func (*Graph) IsRunning 

func (g *Graph) IsRunning() bool

IsRunning returns whether the graph is currently executing

func (*Graph) IsStartNode 

func (g *Graph) IsStartNode(nodeID string) bool

IsStartNode checks if a node is the start node

func (*Graph) Reset 

func (g *Graph) Reset()

Reset resets the graph observability state.

func (*Graph) Resume 

func (g *Graph) Resume(ctx context.Context, ie *InterruptError) (*BaseState, error)

Resume continues a run from a previously interrupted point.

func (*Graph) SetLogger 

func (g *Graph) SetLogger(l *logrus.Logger)

SetLogger replaces the graph logger. A nil logger is ignored.

func (*Graph) SetStartNode 

func (g *Graph) SetStartNode(nodeID string) error

SetStartNode sets the starting node for execution

func (*Graph) StateSchema 

func (g *Graph) StateSchema() *StateSchema

StateSchema returns the graph's state schema, if any.

func (*Graph) Stream 

func (g *Graph) Stream() <-chan *ExecutionResult

Stream returns a channel for streaming execution results from any run. Results are dropped rather than blocking execution if the consumer is slow; use ExecuteWithOptions with a per-run Stream for lossless streaming.

func (*Graph) Subgraph 

func (g *Graph) Subgraph(nodeID string) (*Graph, bool)

Subgraph returns the nested graph registered under a node ID.

func (*Graph) Subgraphs 

func (g *Graph) Subgraphs() map[string]*Graph

Subgraphs returns nested graphs by node ID.

func (*Graph) Validate 

func (g *Graph) Validate() error

Validate validates the graph structure

func (*Graph) WithCheckpointer 

func (g *Graph) WithCheckpointer(saver StateSaver, threadID string) *Graph

WithCheckpointer attaches a state saver and thread ID used to persist state after every node execution, enabling durable execution and resume.

func (*Graph) WithStateSchema 

func (g *Graph) WithStateSchema(schema *StateSchema) *Graph

WithStateSchema attaches a state schema whose reducers are applied to the updates returned by nodes registered with AddUpdateNode.

type GraphConfig 

type GraphConfig struct {
	// MaxIterations bounds the number of node executions in a single run,
	// mirroring LangGraph's recursion_limit. Exceeding it returns ErrRecursionLimit.
	MaxIterations int `json:"max_iterations"`
	// Timeout bounds total run duration. Zero means no timeout.
	Timeout           time.Duration `json:"timeout"`
	EnableStreaming   bool          `json:"enable_streaming"`
	EnableCheckpoints bool          `json:"enable_checkpoints"`
	ParallelExecution bool          `json:"parallel_execution"`
	// RetryAttempts is the number of additional attempts after the first for
	// every node. It defaults to 0: node functions frequently perform
	// non-idempotent work (LLM calls, tool side effects), so silent retries are
	// opt-in rather than the default. Set a per-node RetryPolicy for finer control.
	RetryAttempts int           `json:"retry_attempts"`
	RetryDelay    time.Duration `json:"retry_delay"`
	// InterruptBefore pauses execution before the listed nodes run.
	InterruptBefore []string `json:"interrupt_before,omitempty"`
	// InterruptAfter pauses execution after the listed nodes run.
	InterruptAfter []string `json:"interrupt_after,omitempty"`
}

GraphConfig represents configuration for graph execution

func DefaultGraphConfig 

func DefaultGraphConfig() *GraphConfig

DefaultGraphConfig returns default configuration

func (*GraphConfig) Clone 

func (c *GraphConfig) Clone() *GraphConfig

Clone returns a deep copy of the configuration.

type InterruptError 

type InterruptError struct {
	NodeID   string
	Before   bool
	State    *BaseState
	Step     int
	ThreadID string
}

InterruptError is returned when execution pauses at an interrupt point. It carries the state at the pause and the node that would run next, so the run can be resumed with Resume.

func (*InterruptError) Error 

func (e *InterruptError) Error() string

func (*InterruptError) Is 

func (e *InterruptError) Is(target error) bool

Is lets errors.Is(err, ErrInterrupted) match interrupt pauses.

type Node 

type Node struct {
	ID       string                 `json:"id"`
	Name     string                 `json:"name"`
	Function NodeFunc               `json:"-" yaml:"-"`
	Metadata map[string]interface{} `json:"metadata"`
	// Retry, when non-nil, overrides GraphConfig retry settings for this node.
	Retry *RetryPolicy `json:"retry,omitempty"`
	// contains filtered or unexported fields
}

Node represents a node in the graph

type NodeFunc 

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

NodeFunc represents a function that can be executed as a node.

Returning (nil, nil) means "no state update" and mirrors LangGraph's behavior when a node returns None: the incoming state is carried forward unchanged.

type PanicError 

type PanicError struct {
	// Where identifies the node or condition that panicked.
	Where string
	// Value is the recovered panic value.
	Value interface{}
	// Stack is the goroutine stack captured at recovery time, for logs only.
	Stack []byte
}

PanicError reports a recovered panic from user code. The stack is kept in a separate field rather than in the message so that error strings surfaced to API clients do not leak internal paths and goroutine dumps.

func (*PanicError) Error 

func (e *PanicError) Error() string

func (*PanicError) Unwrap 

func (e *PanicError) Unwrap() error

Unwrap lets errors.Is(err, ErrNodePanic) succeed.

type Reducer 

type Reducer func(existing, update StateValue) StateValue

Reducer combines an existing channel value with an update, mirroring LangGraph's channel reducers (for example operator.add or add_messages).

A reducer must not mutate its arguments; it returns the new value.

type RetryPolicy 

type RetryPolicy struct {
	// MaxAttempts is the number of *additional* attempts after the first.
	MaxAttempts int `json:"max_attempts"`
	// Delay is the wait between attempts.
	Delay time.Duration `json:"delay"`
	// Backoff multiplies Delay after each failed attempt. Values <= 1 mean a
	// constant delay.
	Backoff float64 `json:"backoff"`
	// RetryIf decides whether an error is retryable. Nil means "retry all".
	RetryIf func(error) bool `json:"-" yaml:"-"`
}

RetryPolicy controls per-node retry behavior.

type RouterFunction 

type RouterFunction func(ctx context.Context, state *BaseState) (string, error)

RouterFunction represents a function that determines the next node based on state

func RouteByCondition 

func RouteByCondition(conditionKey string, trueRoute string, falseRoute string) RouterFunction

RouteByCondition routes based on a boolean condition in state

func RouteByCounter 

func RouteByCounter(counterKey string, maxCount int, continueRoute string, exitRoute string) RouterFunction

RouteByCounter routes based on a counter value

func RouteByStateValue 

func RouteByStateValue(key string, routes map[interface{}]string, defaultRoute string) RouterFunction

RouteByStateValue routes based on a specific state value

type StateHistory 

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

StateHistory manages the history of state changes

func NewStateHistory 

func NewStateHistory(maxSize int) *StateHistory

NewStateHistory creates a new state history with a maximum size

func (*StateHistory) AddSnapshot 

func (sh *StateHistory) AddSnapshot(snapshot StateSnapshot)

AddSnapshot adds a new snapshot to the history

func (*StateHistory) GetSnapshot 

func (sh *StateHistory) GetSnapshot(id string) (*StateSnapshot, error)

GetSnapshot returns a specific snapshot by ID

func (*StateHistory) GetSnapshots 

func (sh *StateHistory) GetSnapshots() []StateSnapshot

GetSnapshots returns all snapshots in the history

type StateManager 

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

StateManager manages multiple states and provides advanced operations

func NewStateManager 

func NewStateManager() *StateManager

NewStateManager creates a new state manager

func (*StateManager) CreateState 

func (sm *StateManager) CreateState(id string) *BaseState

CreateState creates a new state with the given ID

func (*StateManager) DeleteState 

func (sm *StateManager) DeleteState(id string)

DeleteState removes a state by ID

func (*StateManager) GetState 

func (sm *StateManager) GetState(id string) (*BaseState, bool)

GetState retrieves a state by ID

func (*StateManager) ListStates 

func (sm *StateManager) ListStates() []string

ListStates returns all state IDs

type StateSaver 

type StateSaver interface {
	SaveState(ctx context.Context, threadID, nodeID string, step int, state *BaseState) error
}

StateSaver is the minimal checkpointing hook the engine needs. The persistence package provides an adapter implementing it, which keeps core free of a dependency on any storage backend.

type StateSchema 

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

StateSchema declares the channels of a graph state. It is the GoLangGraph equivalent of a LangGraph TypedDict state annotated with reducers.

A nil schema, or a key with no declared channel, uses last-write-wins.

func NewStateSchema 

func NewStateSchema() *StateSchema

NewStateSchema creates an empty schema.

func (*StateSchema) AddChannel 

func (s *StateSchema) AddChannel(key string, reducer Reducer, def func() StateValue) *StateSchema

AddChannel declares a channel with a reducer and optional default factory. Re-declaring a key replaces the previous channel.

func (*StateSchema) ApplyUpdates 

func (s *StateSchema) ApplyUpdates(state *BaseState, updates map[string]StateValue)

ApplyUpdates merges a map of channel updates into a state through reducers. Keys are applied in sorted order so the result is deterministic.

func (*StateSchema) Default 

func (s *StateSchema) Default(key string) StateValue

Default returns the zero value for a key before any write.

func (*StateSchema) Keys 

func (s *StateSchema) Keys() []string

Keys returns declared channel keys in declaration order.

func (*StateSchema) NewState 

func (s *StateSchema) NewState() *BaseState

NewState builds a state pre-populated with each channel's default value.

func (*StateSchema) Reducer 

func (s *StateSchema) Reducer(key string) Reducer

Reducer returns the reducer for a key, or nil when the key uses last-write-wins.

type StateSnapshot 

type StateSnapshot struct {
	ID        string                 `json:"id"`
	Timestamp time.Time              `json:"timestamp"`
	Data      map[string]StateValue  `json:"data"`
	Metadata  map[string]interface{} `json:"metadata"`
}

StateSnapshot represents a snapshot of the state at a specific point in time

type StateValue 

type StateValue interface{}

StateValue represents any value that can be stored in state

func AddMessages 

func AddMessages(existing, update StateValue) StateValue

AddMessages appends messages and replaces any existing message that shares an "id" with an incoming one, matching LangGraph's add_messages reducer.

func Append 

func Append(existing, update StateValue) StateValue

Append concatenates slice updates onto the existing slice, mirroring LangGraph's operator.add on list channels. Non-slice updates are appended as a single element.

func LastValue 

func LastValue(existing, update StateValue) StateValue

LastValue overwrites the existing value. This is the default behavior for channels without a reducer.

func MergeMap 

func MergeMap(existing, update StateValue) StateValue

MergeMap merges map updates key-by-key into the existing map.

func SumFloat 

func SumFloat(existing, update StateValue) StateValue

SumFloat adds float updates to the existing value.

func SumInt 

func SumInt(existing, update StateValue) StateValue

SumInt adds integer updates to the existing value.

type SubgraphOptions 

type SubgraphOptions struct {
	// InputKeys restricts what the subgraph sees. Empty means the whole state.
	InputKeys []string
	// OutputKeys restricts what is written back to the parent. Empty means the
	// whole subgraph output state.
	OutputKeys []string
	// Namespace, when set, writes the subgraph output under this single key as
	// a map[string]interface{} instead of merging keys into the parent state.
	Namespace string
	// Schema applies reducers when merging subgraph output into parent state.
	// When nil, the parent graph's schema is used.
	Schema *StateSchema
	// PropagateInterrupts surfaces a subgraph interrupt to the parent instead of
	// treating it as an error.
	PropagateInterrupts bool
}

SubgraphOptions controls how a nested graph exchanges state with its parent.

type UpdateFunc 

type UpdateFunc func(ctx context.Context, state *BaseState) (map[string]StateValue, error)

UpdateFunc is a node that returns only the channels it changed, the way a LangGraph node returns a partial state dict. Returning nil means no update.

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.