core

Documentation

Index

• func ParseNodeInputs(inputs []Inputs, parentOutputs *ExecutionContext) (map[string]any, error)
• func ParseOutput(data any, output Output) (any, error)
• func ProcessNodeOutput(data map[string]any, outputs []Output) (map[string]any, error)
• func ReleaseExecutionContext(ctx *ExecutionContext)
• func ValidateOutput(data any, output Output) error
• type Connection
• type Content
• type ContextOption
  • func WithTracer(tracer opentracing.Tracer) ContextOption
• type ExecutionContext
  • func NewExecutionContext(ctx context.Context, serialID string, totalNodes int64, params map[string]any) *ExecutionContext
  • func (ctx *ExecutionContext) Cancel()
  • func (ctx *ExecutionContext) CheckRoute(key []string) bool
  • func (c *ExecutionContext) Deadline() (deadline time.Time, ok bool)
  • func (c *ExecutionContext) Done() <-chan struct{}
  • func (c *ExecutionContext) Err() error
  • func (ctx *ExecutionContext) GetAllResults() map[string]*NodeResult
  • func (ctx *ExecutionContext) GetNodeResult(nodeID string) (*NodeResult, bool)
  • func (ctx *ExecutionContext) GetVariable(key string) (any, bool)
  • func (ctx *ExecutionContext) SetError(nodeID string, err error)
  • func (ctx *ExecutionContext) SetNodeResult(nodeID string, result *NodeResult)
  • func (ctx *ExecutionContext) SetTracer(tracer opentracing.Tracer)
  • func (ctx *ExecutionContext) SetVariable(key string, value any)
  • func (c *ExecutionContext) Value(key interface{}) interface{}
• type ExecutionStatus
• type Inputs
• type NodeDefinition
• type NodeResult
• type Output
• type Result
• type RetryPolicy
• type ValidationError
• type Value
• type WorkflowConfig
• type WorkflowDef
• type WorkflowState
  • func NewWorkflowState() *WorkflowState

Functions

func ParseNodeInputs

func ParseNodeInputs(inputs []Inputs, parentOutputs *ExecutionContext) (map[string]any, error)

ParseNodeInputs 解析节点输入

func ParseOutput

func ParseOutput(data any, output Output) (any, error)

ParseOutput 解析单个输出

func ProcessNodeOutput

func ProcessNodeOutput(data map[string]any, outputs []Output) (map[string]any, error)

ProcessNodeOutput 处理节点的所有输出

func ReleaseExecutionContext

func ReleaseExecutionContext(ctx *ExecutionContext)

ReleaseExecutionContext 释放执行上下文

func ValidateOutput

func ValidateOutput(data any, output Output) error

ValidateOutput 验证输出数据是否符合Schema定义

Types

type Connection

type Connection struct {
	From      string `json:"from"`
	To        string `json:"to"`
	Condition string `json:"condition,omitempty"`
}

Connection 节点连接

type Content

type Content struct {
	BlockID string `json:"blockId"` // 区块 ID
	Name    string `json:"name"`    // 名称
	Value   string `json:"value"`   // 值
}

Content 表示值的内容

type ContextOption

type ContextOption func(*ExecutionContext)

ContextOption 上下文配置选项

func WithTracer

func WithTracer(tracer opentracing.Tracer) ContextOption

WithTracer 设置追踪器

type ExecutionContext

type ExecutionContext struct {
	context.Context
	Id string

	State *WorkflowState
	Route map[string]struct{}

	TotalNodes int64
	Expiration time.Time // 添加过期时间
	// contains filtered or unexported fields
}

ExecutionContext 工作流执行上下文

func NewExecutionContext

func NewExecutionContext(ctx context.Context, serialID string, totalNodes int64, params map[string]any) *ExecutionContext

NewExecutionContext 创建新的执行上下文

func (*ExecutionContext) Cancel

func (ctx *ExecutionContext) Cancel()

Cancel 取消执行

func (*ExecutionContext) CheckRoute

func (ctx *ExecutionContext) CheckRoute(key []string) bool

CheckRoute 检查路由

func (*ExecutionContext) Deadline

func (c *ExecutionContext) Deadline() (deadline time.Time, ok bool)

Deadline 实现 context.Context 接口

func (*ExecutionContext) Done

func (c *ExecutionContext) Done() <-chan struct{}

Done 实现 context.Context 接口

func (*ExecutionContext) Err

func (c *ExecutionContext) Err() error

Err 实现 context.Context 接口

func (*ExecutionContext) GetAllResults

func (ctx *ExecutionContext) GetAllResults() map[string]*NodeResult

GetAllResults 获取所有节点的执行结果

func (*ExecutionContext) GetNodeResult

func (ctx *ExecutionContext) GetNodeResult(nodeID string) (*NodeResult, bool)

GetNodeResult 获取指定节点的执行结果

func (*ExecutionContext) GetVariable

func (ctx *ExecutionContext) GetVariable(key string) (any, bool)

GetVariable 获取变量值

func (*ExecutionContext) SetError

func (ctx *ExecutionContext) SetError(nodeID string, err error)

func (*ExecutionContext) SetNodeResult

func (ctx *ExecutionContext) SetNodeResult(nodeID string, result *NodeResult)

func (*ExecutionContext) SetTracer

func (ctx *ExecutionContext) SetTracer(tracer opentracing.Tracer)

func (*ExecutionContext) SetVariable

func (ctx *ExecutionContext) SetVariable(key string, value any)

SetVariable 设置变量值

func (*ExecutionContext) Value

func (c *ExecutionContext) Value(key interface{}) interface{}

Value 实现 context.Context 接口

type ExecutionStatus

type ExecutionStatus string

ExecutionStatus 执行状态

const (
	StatusPending   ExecutionStatus = "pending"
	StatusRunning   ExecutionStatus = "running"
	StatusPaused    ExecutionStatus = "paused"
	StatusCompleted ExecutionStatus = "completed"
	StatusFailed    ExecutionStatus = "failed"
	StatusCanceled  ExecutionStatus = "canceled"
)

type Inputs

type Inputs struct {
	Name  string `json:"name"`  // 输入的 name 定义
	Type  string `json:"type"`  // 输入的类型
	Value Value  `json:"value"` // 输入的值
}

type NodeDefinition

type NodeDefinition struct {
	ID      string   `json:"id"`
	Name    string   `json:"name"`
	Type    string   `json:"type"`
	Inputs  []Inputs `json:"inputs"`
	Outputs []Output `json:"outputs"`
	// 组件配置
	Config map[string]any `json:"config"`
}

NodeDefinition 节点定义

type NodeResult

type NodeResult struct {
	Input    any
	Output   any
	Route    []string
	NodeID   string
	Duration int64 // ms
	Error    string
	Type     string
}

type Output

type Output struct {
	Name      string   `json:"name"`      // 输出的名称
	Type      string   `json:"type"`      // 输出的类型
	ItemType  string   `json:"itemType"`  // 输出的 itemType 定义
	DeftValue any      `json:"deftValue"` // 输出的默认值
	Desc      string   `json:"desc"`      // 输出的描述
	Schema    []Output `json:"schema"`    // 输出的 schema 定义
	Required  bool     `json:"required"`  // 输出的 required 定义
}

Output 表示每个输出的结构

type Result

type Result struct {
	Output any
	Route  []string
}

Result 组件执行结果

type RetryPolicy

type RetryPolicy struct {
	MaxAttempts int           `json:"maxAttempts"`
	Interval    time.Duration `json:"interval"`
	MaxInterval time.Duration `json:"maxInterval"`
	Multiplier  float64       `json:"multiplier"`
}

RetryPolicy 重试策略

type ValidationError

type ValidationError struct {
	Field      string `json:"field"`
	Message    string `json:"message"`
	Code       string `json:"code"`
	Severity   string `json:"severity"`
	Suggestion string `json:"suggestion"`
}

ValidationError 验证错误

type Value

type Value struct {
	Content Content `json:"content"` // 输入的 content 定义
	Type    string  `json:"type"`    // 输入的类型 ref 引用  fix 固定值
}

type WorkflowConfig

type WorkflowConfig struct {
	Timeout     time.Duration `json:"timeout"`
	RetryPolicy *RetryPolicy  `json:"retryPolicy,omitempty"`
}

WorkflowConfig 工作流配置

type WorkflowDef

type WorkflowDef struct {
	// 基础信息
	ID          string    `json:"id"`
	Name        string    `json:"name"`
	Description string    `json:"description"`
	Version     string    `json:"version"`
	CreateTime  time.Time `json:"createTime"`
	UpdateTime  time.Time `json:"updateTime"`

	// 节点定义
	Nodes       map[string]*NodeDefinition `json:"nodes"`
	Connections []Connection               `json:"connections"`

	// 配置信息
	Config WorkflowConfig `json:"config"`
}

WorkflowDef 工作流定义

type WorkflowState

type WorkflowState struct {
	Status   ExecutionStatus
	Progress float64
	Result   map[string]*NodeResult
	Errors   []error
}

func NewWorkflowState

func NewWorkflowState() *WorkflowState