nodes

package
v0.6.0 Latest Latest
Warning

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

Go to latest
Published: Sep 15, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Log nodes. Input is a map so a log node can be inserted anywhere in a pipeline without a projection: the previous node's output is logged as structured attributes and passed through unchanged.

Convention:

"message"  : string   — the log line (optional)
other keys : any      — structured attributes

Zero allocation when the level is disabled: slog.Default().Enabled is checked before any work is done.

Index

Constants

View Source
const (
	LogInfoName  = "log.info"
	LogWarnName  = "log.warn"
	LogErrorName = "log.error"
)

Variables

This section is empty.

Functions

func All added in v0.6.0

func All() []action.AnyAction

All returns every domain-neutral action this package provides.

Used by flow.Library() so a runtime that only needs the flow DSL can get a working action set without pulling in AI, sandbox, or provider dependencies.

func DistributeMaxItemsForTest added in v0.5.0

func DistributeMaxItemsForTest() int

func NewBenchCompareAction added in v0.5.0

func NewBenchCompareAction() action.AnyAction

func NewBenchRunAction added in v0.5.0

func NewBenchRunAction() action.AnyAction

NewBenchRunAction benchmarks another action. The target action is resolved against the registry that the flow compiler placed in the execution context, so this node works the same whether it is called directly or from inside a pipeline.

func NewBenchSaveAction added in v0.5.0

func NewBenchSaveAction() action.AnyAction

func NewDistributeMapAction added in v0.5.0

func NewDistributeMapAction() action.AnyAction

NewDistributeMapAction invokes another action once per item, bounded by concurrency. The target action is resolved against the registry placed in the execution context by the flow compiler.

func NewDistributeReduceAction added in v0.5.0

func NewDistributeReduceAction() action.AnyAction

func NewDynamicSaga

func NewDynamicSaga(name string, steps []SagaStep) *action.Builder[any, any]

NewDynamicSaga chains steps: each step's output feeds the next step's input. If a step fails, completed steps are compensated in LIFO order.

This is deliberately distinct from kernel/action.NewSaga, which runs every step against the same input. Here we need transformation chaining with rollback, which the flow DSL authors expect from `A -> B -> C` syntax.

func NewLogErrorAction added in v0.5.0

func NewLogErrorAction() action.AnyAction

func NewLogInfoAction added in v0.5.0

func NewLogInfoAction() action.AnyAction

func NewLogWarnAction added in v0.5.0

func NewLogWarnAction() action.AnyAction

func NewLoopAction

func NewLoopAction(
	bodyAction action.AnyAction,
	untilCondition string,
	maxTurns int,
) (*action.BuiltAction[any, any], error)

func NewProjectionAction

func NewProjectionAction(code string) (*action.BuiltAction[any, any], error)

NewProjectionAction compiles inline data shaping { key: expr } using precompiled expr bytecode. Supports standard expressions, JQ-style field prefixes ({ to: .user }), and root access ({ out: . }).

func NewPromptNode

func NewPromptNode(cfg PromptConfig) action.AnyAction

NewPromptNode creates a reusable, pre-configured prompt template action.

func NewSupervisorNode

func NewSupervisorNode(name string) action.AnyAction

NewSupervisorNode builds a supervisor that compiles and runs child pipelines on the fly. The compiler is obtained from the execution context, so the node works inside any flow without constructor args.

Types

type BenchCompareReq added in v0.5.0

type BenchCompareReq struct {
	Baseline     string  `json:"baseline" validate:"required" usage:"Baseline JSON path (validated by xfs.Rel)"`
	TolerancePct float64 `json:"tolerance_pct,omitempty"      usage:"Allowed regression percent (default 5)"`
	BenchRunRes
}

type BenchCompareRes added in v0.5.0

type BenchCompareRes struct {
	Baseline    string                 `json:"baseline"`
	Pass        bool                   `json:"pass"`
	Regressions []string               `json:"regressions,omitempty"`
	Metrics     map[string]BenchMetric `json:"metrics"`
}

type BenchMetric added in v0.5.0

type BenchMetric struct {
	Baseline  float64 `json:"baseline"`
	Current   float64 `json:"current"`
	DeltaPct  float64 `json:"delta_pct"`
	Regressed bool    `json:"regressed"`
}

type BenchRunReq added in v0.5.0

type BenchRunReq struct {
	Action     string         `json:"action"               validate:"required" usage:"Node name to benchmark"`
	Iterations int            `json:"iterations,omitempty"                    usage:"Timed iterations (default 50)"`
	Warmup     int            `json:"warmup,omitempty"                        usage:"Discarded pre-runs (default 3)"`
	Payload    map[string]any `json:"payload,omitempty"                       usage:"Request passed to each invocation"`
}

type BenchRunRes added in v0.5.0

type BenchRunRes struct {
	Action     string  `json:"action"`
	Iterations int     `json:"iterations"`
	Warmup     int     `json:"warmup"`
	Errors     int     `json:"errors"`
	MinMs      float64 `json:"min_ms"`
	MaxMs      float64 `json:"max_ms"`
	MeanMs     float64 `json:"mean_ms"`
	P50Ms      float64 `json:"p50_ms"`
	P95Ms      float64 `json:"p95_ms"`
	P99Ms      float64 `json:"p99_ms"`
	RPS        float64 `json:"rps"`
	ElapsedMs  int64   `json:"elapsed_ms"`
}

type BenchSaveReq added in v0.5.0

type BenchSaveReq struct {
	File string `json:"file" validate:"required" usage:"Relative path to write (validated by xfs.Rel)"`
	BenchRunRes
}

type BenchSaveRes added in v0.5.0

type BenchSaveRes struct {
	File  string `json:"file"`
	Bytes int    `json:"bytes"`
	BenchRunRes
}

type ChildResult

type ChildResult struct {
	TaskID   string `json:"task_id"`
	Output   any    `json:"output,omitempty"`
	Error    string `json:"error,omitempty"`
	Duration int64  `json:"duration_ms"`
}

type ChildTask

type ChildTask struct {
	ID        string `json:"id"`
	DSL       string `json:"dsl"`
	Payload   any    `json:"payload"`
	TimeoutMS int64  `json:"timeout_ms,omitempty"`
}

type DistributeItem added in v0.5.0

type DistributeItem struct {
	OK     bool   `json:"ok"`
	Result any    `json:"result,omitempty"`
	Error  string `json:"error,omitempty"`
}

type DistributeMapReq added in v0.5.0

type DistributeMapReq struct {
	Action      string `json:"action"      validate:"required" usage:"Node name to invoke for each item"`
	Concurrency int    `json:"concurrency,omitempty"           usage:"Max in-flight invocations (default 4, max 256)"`
	Items       []any  `json:"items"       validate:"required" usage:"Items to fan out. Passed unchanged to Action."`
}

type DistributeMapRes added in v0.5.0

type DistributeMapRes struct {
	Items     []DistributeItem `json:"items"`
	Succeeded int              `json:"succeeded"`
	Failed    int              `json:"failed"`
}

type DistributeReduceReq added in v0.5.0

type DistributeReduceReq struct {
	Strategy string           `json:"strategy" validate:"required,oneof=collect all_pass any_pass first_success count"`
	Items    []DistributeItem `json:"items"    validate:"required"`
}

type DistributeReduceRes added in v0.5.0

type DistributeReduceRes struct {
	Strategy  string `json:"strategy"`
	AllPass   bool   `json:"all_pass,omitempty"`
	AnyPass   bool   `json:"any_pass,omitempty"`
	Succeeded int    `json:"succeeded,omitempty"`
	Failed    int    `json:"failed,omitempty"`
	Result    any    `json:"result,omitempty"`
	Results   []any  `json:"results,omitempty"`
}

type PromptConfig

type PromptConfig struct {
	Name          string         `json:"name"`
	Description   string         `json:"description"`
	SystemPrompt  string         `json:"system_prompt"`
	UserTemplate  string         `json:"user_template"`
	DefaultParams map[string]any `json:"default_params,omitempty"`
	Timeout       time.Duration  `json:"timeout,omitempty"`
}

type PromptReq

type PromptReq struct {
	Input  any            `json:"input,omitempty"`
	Params map[string]any `json:"params,omitempty"`
	Prompt string         `json:"prompt,omitempty"`
}

type PromptRes

type PromptRes struct {
	SystemPrompt string         `json:"system_prompt"`
	RenderedUser string         `json:"rendered_user"`
	Params       map[string]any `json:"params"`
}

type SagaStep

type SagaStep struct {
	NodeID     string
	Forward    action.AnyAction
	Compensate action.AnyAction
}

type SupervisorReq

type SupervisorReq struct {
	Tasks []ChildTask `json:"tasks" validate:"required"`
}

type SupervisorRes

type SupervisorRes struct {
	Total     int           `json:"total"`
	Succeeded int           `json:"succeeded"`
	Failed    int           `json:"failed"`
	Results   []ChildResult `json:"results"`
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL