runnables

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 runnables implements a LangChain-v1 style runnable (LCEL) core for Go.

A Runnable is a composable unit of computation. Everything in the framework (chains, graphs, agents, tools, models) is a Runnable, so they can be combined with operators such as Pipe, Branch, Fallbacks, and Retry.

Index

Constants

This section is empty.

Variables

View Source 
var ErrNotStreamable = errors.New("runnable: not streamable")

ErrNotStreamable is returned when a runnable cannot stream incrementally.

Functions

func AStream 

func AStream(ctx context.Context, r Runnable, input Input, opts ...Option) (<-chan Output, error)

AStream streams r asynchronously, falling back to the sync Stream.

func Invoke 

func Invoke[T any](ctx context.Context, r Runnable, input T, opts ...Option) (T, error)

Invoke is a convenience helper that types the input/output.

Types

type AsyncRunnable 

type AsyncRunnable interface {
	// AInvoke runs the runnable asynchronously.
	AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)
	// AStream yields incremental outputs asynchronously.
	AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)
	// ABatch runs the runnable over multiple inputs asynchronously.
	ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)
}

AsyncRunnable is implemented by runnables that support async execution. The default implementations in the basic operators fall back to the sync methods when this interface is not implemented.

type CallConfig 

type CallConfig struct {
	// RunID identifies a single run, used for tracing and callbacks.
	RunID string
	// Callbacks receives lifecycle events for the run.
	// Callbacks []Callback
	// Metadata is arbitrary key/value data attached to the run.
	Metadata map[string]any
	// Tags are labels attached to the run.
	Tags []string
	// MaxRetries overrides the default retry count for RunnableRetry.
	MaxRetries int
	// Timeout optionally bounds the whole invocation.
	Timeout any
}

CallConfig carries per-invocation configuration.

type ErrRetryable 

type ErrRetryable interface {
	Retryable() bool
}

ErrRetryable reports whether an error is safe to retry.

type Input 

type Input = any

Input and Output are type aliases used throughout the runnable definitions. They alias any so that the core stays untyped, mirroring LangChain's duck-typed runnables that are validated at runtime.

type Option 

type Option func(*CallConfig)

Option is a functional option applied when invoking a runnable.

func WithMaxRetries 

func WithMaxRetries(n int) Option

WithMaxRetries overrides the retry count for a RunnableRetry.

func WithMetadata 

func WithMetadata(m map[string]any) Option

WithMetadata attaches metadata to the run.

func WithRunID 

func WithRunID(id string) Option

WithRunID sets the run id.

func WithTags 

func WithTags(tags ...string) Option

WithTags attaches tags to the run.

type Output 

type Output = any

Input and Output are type aliases used throughout the runnable definitions. They alias any so that the core stays untyped, mirroring LangChain's duck-typed runnables that are validated at runtime.

func ABatch 

func ABatch(ctx context.Context, r Runnable, inputs []Input, opts ...Option) ([]Output, error)

ABatch batches r asynchronously, falling back to the sync Batch.

func AInvoke 

func AInvoke(ctx context.Context, r Runnable, input Input, opts ...Option) (Output, error)

AInvoke invokes r asynchronously, falling back to the sync Invoke if r does not implement AsyncRunnable.

func Batch 

func Batch(ctx context.Context, r Runnable, inputs []Input, opts ...Option) ([]Output, error)

Batch runs a runnable over each input sequentially, preserving order. If a step fails, the error is returned immediately.

func ConcurrentBatch 

func ConcurrentBatch(ctx context.Context, r Runnable, inputs []Input, workers int, opts ...Option) ([]Output, error)

ConcurrentBatch runs the runnable over the inputs concurrently with up to workers goroutines, preserving input order in the output slice. It is the fundamental async primitive used by async runnable implementations.

type RunErr 

type RunErr struct {
	RunID string
	Cause error
}

RunErr is a typed error that carries the run id and the underlying cause.

func (*RunErr) Error 

func (e *RunErr) Error() string

func (*RunErr) Unwrap 

func (e *RunErr) Unwrap() error

type Runnable 

type Runnable interface {
	// Invoke runs the runnable once with the given input and returns its output.
	Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)
	// Stream yields incremental outputs. If the runnable is not natively
	// streaming, it may emit a single final chunk.
	Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)
	// Batch runs the runnable over a slice of inputs, returning outputs in order.
	Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)
}

Runnable is the core composable interface.

type RunnableBranch 

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

RunnableBranch routes the input to one of several branches based on predicates.

func NewRunnableBranch 

func NewRunnableBranch(defaultR any, branches ...any) (*RunnableBranch, error)

NewRunnableBranch builds a runnable that evaluates each condition in order and invokes the first matching branch. If none match, defaultR is used. Conditions are funcs; runnables may be Runnable or plain func.

func (*RunnableBranch) ABatch 

func (r *RunnableBranch) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableBranch) AInvoke 

func (r *RunnableBranch) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnableBranch) AStream 

func (r *RunnableBranch) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableBranch) Batch 

func (r *RunnableBranch) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableBranch) Invoke 

func (r *RunnableBranch) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableBranch) Stream 

func (r *RunnableBranch) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableConfigurable 

type RunnableConfigurable struct {

	// Key is the config field used to select an alternative. It is read from
	// the input map under this key.
	Key string
	// Alternatives maps a config value to the runnable to use.
	Alternatives map[any]Runnable
	// contains filtered or unexported fields
}

RunnableConfigurable wraps a runnable and swaps its underlying implementation based on a resolved configuration key. It mirrors langchain's RunnableConfigurableAlternatives.

func NewRunnableConfigurable 

func NewRunnableConfigurable(defaultR any, key string, alternatives map[any]any) (*RunnableConfigurable, error)

NewRunnableConfigurable builds a configurable runnable from a default runnable, a config key, and a map of value -> runnable alternatives.

func (*RunnableConfigurable) ABatch 

func (r *RunnableConfigurable) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableConfigurable) AInvoke 

func (r *RunnableConfigurable) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnableConfigurable) AStream 

func (r *RunnableConfigurable) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableConfigurable) Batch 

func (r *RunnableConfigurable) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableConfigurable) Invoke 

func (r *RunnableConfigurable) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableConfigurable) Stream 

func (r *RunnableConfigurable) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableFallback 

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

RunnableFallback tries the primary runnable and falls back to the alternatives in order if it fails.

func NewRunnableFallback 

func NewRunnableFallback(primary any, fallbacks ...any) (*RunnableFallback, error)

NewRunnableFallback builds a fallback from the primary runnable and any number of alternative runnables/funcs.

func (*RunnableFallback) ABatch 

func (r *RunnableFallback) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableFallback) AInvoke 

func (r *RunnableFallback) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnableFallback) AStream 

func (r *RunnableFallback) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableFallback) Batch 

func (r *RunnableFallback) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableFallback) Invoke 

func (r *RunnableFallback) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableFallback) Stream 

func (r *RunnableFallback) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableFunc 

type RunnableFunc func(ctx context.Context, input Input, opts ...Option) (Output, error)

RunnableFunc adapts a plain function into a Runnable.

func (RunnableFunc) Batch 

func (f RunnableFunc) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (RunnableFunc) Invoke 

func (f RunnableFunc) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (RunnableFunc) Stream 

func (f RunnableFunc) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable by emitting a single final chunk.

type RunnableLambda 

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

RunnableLambda wraps an arbitrary function as a Runnable.

func NewRunnableLambda 

func NewRunnableLambda(fn func(ctx context.Context, input Input) (Output, error)) *RunnableLambda

NewRunnableLambda wraps fn into a RunnableLambda.

func (*RunnableLambda) ABatch 

func (r *RunnableLambda) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableLambda) AInvoke 

func (r *RunnableLambda) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnableLambda) AStream 

func (r *RunnableLambda) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableLambda) Batch 

func (r *RunnableLambda) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableLambda) Invoke 

func (r *RunnableLambda) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableLambda) Stream 

func (r *RunnableLambda) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableParallel 

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

RunnableParallel invokes several runnables concurrently on the same input and merges their outputs into a map keyed by name.

func NewRunnableParallel 

func NewRunnableParallel(branches map[string]any) (*RunnableParallel, error)

NewRunnableParallel builds a parallel runnable from a map of name -> runnable.

func (*RunnableParallel) ABatch 

func (r *RunnableParallel) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableParallel) AInvoke 

func (r *RunnableParallel) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable (parallel is already concurrent).

func (*RunnableParallel) AStream 

func (r *RunnableParallel) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableParallel) Batch 

func (r *RunnableParallel) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableParallel) Invoke 

func (r *RunnableParallel) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableParallel) Stream 

func (r *RunnableParallel) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnablePassthrough 

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

RunnablePassthrough passes its input through unchanged. It can also be configured with an assignment function that merges new keys into a map input.

func NewRunnablePassthrough 

func NewRunnablePassthrough() *RunnablePassthrough

NewRunnablePassthrough returns a passthrough that returns its input unchanged.

func NewRunnablePassthroughAssign 

func NewRunnablePassthroughAssign(assign func(ctx context.Context, m map[string]any) (map[string]any, error)) *RunnablePassthrough

NewRunnablePassthroughAssign returns a passthrough that calls assign to add keys to a map input.

func (*RunnablePassthrough) ABatch 

func (r *RunnablePassthrough) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnablePassthrough) AInvoke 

func (r *RunnablePassthrough) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnablePassthrough) AStream 

func (r *RunnablePassthrough) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnablePassthrough) Assign 

func (r *RunnablePassthrough) Assign(assign func(ctx context.Context, m map[string]any) (map[string]any, error)) *RunnablePassthrough

Assign sets the assign function on the passthrough.

func (*RunnablePassthrough) Batch 

func (r *RunnablePassthrough) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnablePassthrough) Invoke 

func (r *RunnablePassthrough) Invoke(ctx context.Context, input Input, _ ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnablePassthrough) Stream 

func (r *RunnablePassthrough) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableRetry 

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

RunnableRetry retries the wrapped runnable on failure up to maxRetries times.

func NewRunnableRetry 

func NewRunnableRetry(r any, maxRetries int) (*RunnableRetry, error)

NewRunnableRetry wraps r with a default of 3 retries.

func (*RunnableRetry) ABatch 

func (r *RunnableRetry) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableRetry) AInvoke 

func (r *RunnableRetry) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable.

func (*RunnableRetry) AStream 

func (r *RunnableRetry) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableRetry) Batch 

func (r *RunnableRetry) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableRetry) Invoke 

func (r *RunnableRetry) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableRetry) Stream 

func (r *RunnableRetry) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable.

type RunnableSequence 

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

RunnableSequence pipes the output of each runnable into the input of the next.

func NewRunnableSequence 

func NewRunnableSequence(steps ...any) (*RunnableSequence, error)

NewRunnableSequence builds a sequence from the given steps. A single plain function is wrapped as RunnableLambda.

func Pipe 

func Pipe(r Runnable, next ...any) (*RunnableSequence, error)

Pipe returns a sequence of r followed by the given runnables/funcs.

func (*RunnableSequence) ABatch 

func (r *RunnableSequence) ABatch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

ABatch implements AsyncRunnable.

func (*RunnableSequence) AInvoke 

func (r *RunnableSequence) AInvoke(ctx context.Context, input Input, opts ...Option) (Output, error)

AInvoke implements AsyncRunnable by piping each step asynchronously.

func (*RunnableSequence) AStream 

func (r *RunnableSequence) AStream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

AStream implements AsyncRunnable.

func (*RunnableSequence) Batch 

func (r *RunnableSequence) Batch(ctx context.Context, inputs []Input, opts ...Option) ([]Output, error)

Batch implements Runnable.

func (*RunnableSequence) Invoke 

func (r *RunnableSequence) Invoke(ctx context.Context, input Input, opts ...Option) (Output, error)

Invoke implements Runnable.

func (*RunnableSequence) Steps 

func (r *RunnableSequence) Steps() []Runnable

Steps returns the underlying steps.

func (*RunnableSequence) Stream 

func (r *RunnableSequence) Stream(ctx context.Context, input Input, opts ...Option) (<-chan Output, error)

Stream implements Runnable by piping the stream of each step into the next. Only the final step streams incrementally; intermediate steps are invoked to completion because their partial output is not generally streamable.

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.