runnable

type Assign ¶

type Assign[I any] struct {
	// contains filtered or unexported fields
}

Assign creates a parallel runnable that passes the input through unchanged while also computing additional keys. The result is a map[string]any containing the original input plus the computed values.

Usage:

chain := Assign(
    map[string]core.Runnable[map[string]any, any]{
        "context": retriever,
    },
)

This adds a "context" key to the input map while preserving existing keys.

func NewAssign[I any](additions map[string]func(ctx context.Context, input I, opts ...core.Option) (any, error)) *Assign[I]

func (a *Assign[I]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]map[string]any, error)

func (a *Assign[I]) GetName() string

func (a *Assign[I]) Invoke(ctx context.Context, input I, opts ...core.Option) (map[string]any, error)

func (a *Assign[I]) Stream(ctx context.Context, input I, opts ...core.Option) (*core.StreamIterator[map[string]any], error)

type Branch ¶

type Branch[I, O any] struct {
	// contains filtered or unexported fields
}

Branch selects which runnable to execute based on conditions. It evaluates conditions in order and runs the first one that returns true. If no conditions match, the default runnable is used. It implements Runnable[I, O].

func NewBranch[I, O any](
	conditions []BranchCondition[I, O],
	defaultBranch core.Runnable[I, O],
) *Branch[I, O]

func (b *Branch[I, O]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]O, error)

func (b *Branch[I, O]) GetName() string

func (b *Branch[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

func (b *Branch[I, O]) Stream(ctx context.Context, input I, opts ...core.Option) (*core.StreamIterator[O], error)

func (b *Branch[I, O]) WithName(name string) *Branch[I, O]

type BranchCondition ¶

type BranchCondition[I, O any] struct {
	Condition func(input I) bool
	Runnable  core.Runnable[I, O]
}

BranchCondition pairs a condition function with a runnable.

type Lambda ¶

type Lambda[I, O any] struct {
	// contains filtered or unexported fields
}

Lambda wraps a Go function as a Runnable. It implements Runnable[I, O].

func NewLambda[I, O any](fn func(ctx context.Context, input I) (O, error)) *Lambda[I, O]

func (l *Lambda[I, O]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]O, error)

func (l *Lambda[I, O]) GetName() string

func (l *Lambda[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

func (l *Lambda[I, O]) Stream(ctx context.Context, input I, opts ...core.Option) (*core.StreamIterator[O], error)

func (l *Lambda[I, O]) WithName(name string) *Lambda[I, O]

type Parallel ¶

type Parallel[I any] struct {
	// contains filtered or unexported fields
}

Parallel runs multiple runnables in parallel with the same input, collecting their outputs into a map[string]any. It implements Runnable[I, map[string]any].

func NewParallel[I, O any](branches map[string]core.Runnable[I, O]) *Parallel[I]

func NewParallelAny[I any](branches map[string]func(ctx context.Context, input I, opts ...core.Option) (any, error)) *Parallel[I]

func (p *Parallel[I]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]map[string]any, error)

func (p *Parallel[I]) GetName() string

func (p *Parallel[I]) Invoke(ctx context.Context, input I, opts ...core.Option) (map[string]any, error)

func (p *Parallel[I]) Stream(ctx context.Context, input I, opts ...core.Option) (*core.StreamIterator[map[string]any], error)

func (p *Parallel[I]) WithName(name string) *Parallel[I]

type Passthrough ¶

type Passthrough[T any] struct {
	// contains filtered or unexported fields
}

Passthrough passes its input through unchanged. It implements Runnable[T, T].

func NewPassthrough[T any]() *Passthrough[T]

func (p *Passthrough[T]) Batch(ctx context.Context, inputs []T, opts ...core.Option) ([]T, error)

func (p *Passthrough[T]) GetName() string

func (p *Passthrough[T]) Invoke(ctx context.Context, input T, opts ...core.Option) (T, error)

func (p *Passthrough[T]) Stream(ctx context.Context, input T, opts ...core.Option) (*core.StreamIterator[T], error)

func (p *Passthrough[T]) WithName(name string) *Passthrough[T]

type Sequence ¶

type Sequence[I, O any] struct {
	// contains filtered or unexported fields
}

Sequence chains multiple runnables together: the output of each becomes the input of the next. Because Go generics don't support heterogeneous type lists, the intermediate types are erased to `any`. The overall Sequence is typed on the first input and last output.

func Pipe(runnables ...core.Runnable[any, any]) *Sequence[any, any]

func Pipe2[A, B, C any](
	first core.Runnable[A, B],
	second core.Runnable[B, C],
) *Sequence[A, C]

func Pipe3[A, B, C, D any](
	first core.Runnable[A, B],
	second core.Runnable[B, C],
	third core.Runnable[C, D],
) *Sequence[A, D]

func Pipe4[A, B, C, D, E any](
	first core.Runnable[A, B],
	second core.Runnable[B, C],
	third core.Runnable[C, D],
	fourth core.Runnable[D, E],
) *Sequence[A, E]

func (s *Sequence[I, O]) Batch(ctx context.Context, inputs []I, opts ...core.Option) ([]O, error)

func (s *Sequence[I, O]) GetName() string

func (s *Sequence[I, O]) Invoke(ctx context.Context, input I, opts ...core.Option) (O, error)

func (s *Sequence[I, O]) Stream(ctx context.Context, input I, opts ...core.Option) (*core.StreamIterator[O], error)

func (s *Sequence[I, O]) WithName(name string) *Sequence[I, O]