node

package
v1.1.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrLoopCancelled = errors.New("loop cancelled")
	ErrLoopTimeout   = errors.New("loop timeout")
)

Loop 相关的错误

Functions

func BuildChain added in v1.1.0 

func BuildChain(aspects []Aspect, core func() (map[string]any, error)) func(ac *AspectContext) (map[string]any, error)

BuildChain 将切面列表构建为洋葱调用链 最外层的切面最先执行（index 0 在最外层） core 是最内层的实际执行逻辑

func BuildNodeChain added in v1.1.0 

func BuildNodeChain(aspects []Aspect, node Node, runFunc func(ac *AspectContext) (map[string]any, error)) func(ac *AspectContext) (map[string]any, error)

BuildNodeChain 为节点构建切面调用链 aspects: 全局切面 + 节点切面 node: 当前节点 runFunc: 节点执行逻辑，接收 AspectContext 以使用切面级 context 等待数据

Types

type Aspect 

type Aspect interface {
	Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)
}

Aspect 统一切面接口 所有切面（Before/After/Around/Retry/Timeout/CircuitBreaker）都实现此接口

func AfterFunc added in v1.1.0 

func AfterFunc(fn func(ctx *AspectContext, node Node, err error)) Aspect

AfterFunc 创建后置切面：调用 next 后执行 fn

func AroundFunc added in v1.1.0 

func AroundFunc(fn func(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)) Aspect

AroundFunc 创建环绕切面

func BeforeFunc added in v1.1.0 

func BeforeFunc(fn func(ctx *AspectContext, node Node)) Aspect

BeforeFunc 创建前置切面：执行 fn 后调用 next

type AspectContext added in v1.1.0 

type AspectContext struct {
	Flow *flow.FlowContext // 工作流共享数据
	// contains filtered or unexported fields
}

AspectContext 切面执行上下文 每层拦截器拥有独立的可取消 context，实现超时、熔断等节点级控制

func NewAspectContext added in v1.1.0 

func NewAspectContext(flowCtx *flow.FlowContext, parent context.Context) *AspectContext

NewAspectContext 创建切面上下文

func WithTimeout added in v1.1.0 

func WithTimeout(parent *AspectContext, timeout time.Duration) (*AspectContext, context.CancelFunc)

WithTimeout 创建带超时的 AspectContext 子节点

func (*AspectContext) Cancel added in v1.1.0 

func (ac *AspectContext) Cancel(err error)

Cancel 取消本层 context

func (*AspectContext) Context added in v1.1.0 

func (ac *AspectContext) Context() context.Context

Context 获取本层的可取消 context（用于超时、取消控制）

func (*AspectContext) Done added in v1.1.0 

func (ac *AspectContext) Done() <-chan struct{}

Done 返回本层 context 的 Done channel

type CircuitBreakerAspect added in v1.1.0 

type CircuitBreakerAspect struct {
	HalfOpenMaxCalls int

	FallbackFunc func(ctx *AspectContext, node Node) (map[string]any, error)
	// contains filtered or unexported fields
}

CircuitBreakerAspect 熔断降级切面 当连续失败次数达到阈值时，进入熔断状态，直接返回 Fallback，不再执行真实逻辑

func NewCircuitBreakerAspect added in v1.1.0 

func NewCircuitBreakerAspect(threshold int, timeout time.Duration) *CircuitBreakerAspect

func (*CircuitBreakerAspect) Around added in v1.1.0 

func (cb *CircuitBreakerAspect) Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

type CircuitState 

type CircuitState int

CircuitState 熔断器状态 

const (
	StateClosed   CircuitState = iota // 关闭（正常通行）
	StateOpen                         // 打开（熔断）
	StateHalfOpen                     // 半开（试探）
)

type ErrorSwallowAspect added in v1.1.0 

type ErrorSwallowAspect struct {
	FallbackFunc func(ctx *AspectContext, node Node, err error) (map[string]any, error)
}

ErrorSwallowAspect 拦截 error，执行降级逻辑 作为最外层切面使用，确保节点级别的错误不会传播到工作流层

func NewErrorSwallowAspect added in v1.1.0 

func NewErrorSwallowAspect(fallback func(ctx *AspectContext, node Node, err error) (map[string]any, error)) *ErrorSwallowAspect

func (*ErrorSwallowAspect) Around added in v1.1.0 

func (e *ErrorSwallowAspect) Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

type LoopConfig 

type LoopConfig struct {
	MaxIterations int             // 最大循环次数（0表示无限制）
	Timeout       time.Duration   // 超时时间（0表示无超时）
	Context       context.Context // 外部context，用于取消（nil则使用background）
}

LoopConfig 循环节点配置

type LoopResult added in v1.1.0 

type LoopResult struct {
	Status     LoopStatus `json:"status"`     // 结束状态
	Iterations int        `json:"iterations"` // 实际执行次数
	Message    string     `json:"message"`    // 详细描述
}

LoopResult 循环执行结果

func NewLoopCancelledResult added in v1.1.0 

func NewLoopCancelledResult(iterations int, err error) *LoopResult

NewLoopCancelledResult 创建被取消的循环结果

func NewLoopCompletedResult added in v1.1.0 

func NewLoopCompletedResult(iterations int) *LoopResult

NewLoopCompletedResult 创建正常完成的循环结果

func NewLoopMaxIterationsResult added in v1.1.0 

func NewLoopMaxIterationsResult(maxIterations, actualIterations int) *LoopResult

NewLoopMaxIterationsResult 创建达到最大次数的循环结果

func NewLoopResult added in v1.1.0 

func NewLoopResult(status LoopStatus, iterations int, message string) *LoopResult

NewLoopResult 创建循环结果

func NewLoopTimeoutResult added in v1.1.0 

func NewLoopTimeoutResult(iterations int) *LoopResult

NewLoopTimeoutResult 创建超时的循环结果

func (*LoopResult) IsError added in v1.1.0 

func (r *LoopResult) IsError() bool

IsError 判断循环是否因错误而结束

func (*LoopResult) IsSuccess added in v1.1.0 

func (r *LoopResult) IsSuccess() bool

IsSuccess 判断循环是否成功完成

func (*LoopResult) ToMap added in v1.1.0 

func (r *LoopResult) ToMap() map[string]any

ToMap 转换为 map，方便放入 FlowContext

type LoopStatus added in v1.1.0 

type LoopStatus string

LoopStatus 循环执行状态 

const (
	LoopStatusCompleted       LoopStatus = "completed"        // 正常完成
	LoopStatusMaxIterations   LoopStatus = "max_iterations"   // 达到最大循环次数
	LoopStatusTimeout         LoopStatus = "timeout"          // 超时
	LoopStatusCancelled       LoopStatus = "cancelled"        // 被取消
	LoopStatusConditionFailed LoopStatus = "condition_failed" // 条件不满足
)

func (LoopStatus) String added in v1.1.0 

func (s LoopStatus) String() string

String 实现 Stringer 接口

type Node 

type Node interface {
	ID() string
	Inputs() []string  // 依赖的输入 keys
	Outputs() []string // 输出 keys

	// Run 执行节点业务逻辑
	Run(ctx *flow.FlowContext, inputs map[string]any) (outputs map[string]any, err error)

	// Aspects 节点自己的切面（AOP）
	Aspects() []Aspect
}

Node 工作流节点

type RecoveryAspect added in v1.1.0 

type RecoveryAspect struct {
	FallbackFunc func(ctx *AspectContext, node Node, recoverVal any) (map[string]any, error)
}

RecoveryAspect 捕获 panic 并执行兜底逻辑

func NewRecoveryAspect added in v1.1.0 

func NewRecoveryAspect(fallback func(ctx *AspectContext, node Node, recoverVal any) (map[string]any, error)) *RecoveryAspect

func (*RecoveryAspect) Around added in v1.1.0 

func (r *RecoveryAspect) Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (outputs map[string]any, err error)

type RetryAspect added in v1.1.0 

type RetryAspect struct {
	MaxAttempts int           // 最大尝试次数（至少为1）
	Delay       time.Duration // 每次重试间隔
	ShouldRetry func(err error) bool
}

RetryAspect 节点失败时自动重试

func NewRetryAspect added in v1.1.0 

func NewRetryAspect(maxAttempts int, delay time.Duration) *RetryAspect

func (*RetryAspect) Around added in v1.1.0 

func (r *RetryAspect) Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

type SimpleNode 

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

SimpleNode 通用节点

func NewConditionNode 

func NewConditionNode(
	id string,
	inputKey string,
	condition func(value any) bool,
	trueKey string,
	falseKey string,
) *SimpleNode

NewConditionNode 创建【条件判断节点】 id: 节点ID inputKey: 要判断的输入key condition: 条件函数（返回true/false） trueKey: 条件成立时输出的key falseKey: 条件不成立时输出的key

func NewLLMStreamNode 

func NewLLMStreamNode(
	id string,
	promptKey string,
	outputKey string,
	model chatmodel.BaseModel,
	copies uint,
) *SimpleNode

NewLLMStreamNode 创建【流式LLM节点】 id: 节点ID promptKey: 从上下文获取提示词的key OutputKey: 返回一个 []*StreamReader model: *chatmodel.BaseModel 实例 copies: StreamReader的数量

func NewLoopNode 

func NewLoopNode(
	id string,
	controlKey string,
	condition func(ctx *flow.FlowContext) bool,
	loopBody func(ctx *flow.FlowContext),
	outputKey string,
	config *LoopConfig,
) *SimpleNode

NewLoopNode 创建【循环节点】（while 模式：条件为真就一直执行） id: 节点ID controlKey: 循环控制key（节点会等待这个key来启动循环） condition: 循环条件函数，返回true=继续循环，false=退出循环 loopBody: 循环体内执行的逻辑 outputKey: 循环结束后输出的结果key config: 循环配置（最大次数、超时、context）

func NewNode 

func NewNode(
	id string,
	inputs []string,
	outputs []string,
	runFunc func(ctx *flow.FlowContext, inputs map[string]any) (map[string]any, error),
) *SimpleNode

NewNode 🌟 最友好的节点初始化函数 只需要传：ID、输入列表、输出列表、执行逻辑

func NewParallelNode 

func NewParallelNode(
	id string,
	waitKeys []string,
	outputKey string,
) *SimpleNode

NewParallelNode 创建【并行汇聚节点】 作用：等待所有输入全部就绪 → 然后输出完成信号 id: 节点ID waitKeys: 要等待的所有输入key（数组） outputKey: 全部完成后输出的key

func (*SimpleNode) AddAspect 

func (n *SimpleNode) AddAspect(aspect Aspect)

AddAspect 给节点追加切面

func (*SimpleNode) Aspects 

func (n *SimpleNode) Aspects() []Aspect

func (*SimpleNode) ID 

func (n *SimpleNode) ID() string

func (*SimpleNode) Inputs 

func (n *SimpleNode) Inputs() []string

func (*SimpleNode) Outputs 

func (n *SimpleNode) Outputs() []string

func (*SimpleNode) Run 

func (n *SimpleNode) Run(ctx *flow.FlowContext, inputs map[string]any) (map[string]any, error)

type TimeoutAspect added in v1.1.0 

type TimeoutAspect struct {
	Timeout time.Duration
}

TimeoutAspect 限制节点执行时间，超时时取消本层 context 并返回错误

func NewTimeoutAspect added in v1.1.0 

func NewTimeoutAspect(timeout time.Duration) *TimeoutAspect

func (*TimeoutAspect) Around added in v1.1.0 

func (t *TimeoutAspect) Around(ctx *AspectContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

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.