node

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: May 23, 2026 License: Apache-2.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func IsCompleted 

func IsCompleted(plan *Plan) bool

func TrimJSONWrapper 

func TrimJSONWrapper(raw string) string

Types

type AfterAspect 

type AfterAspect struct {
	Fn func(ctx *flow.FlowContext, node Node, err error)
}

AfterAspect 简易实现：只执行 After

func (*AfterAspect) After 

func (a *AfterAspect) After(ctx *flow.FlowContext, node Node, err error)

func (*AfterAspect) Before 

func (a *AfterAspect) Before(ctx *flow.FlowContext, node Node)

type AroundAspect 

type AroundAspect struct {
	BeforeFn func(ctx *flow.FlowContext, node Node)
	AfterFn  func(ctx *flow.FlowContext, node Node, err error)
}

AroundAspect 简易实现：前后都执行

func (*AroundAspect) After 

func (a *AroundAspect) After(ctx *flow.FlowContext, node Node, err error)

func (*AroundAspect) Before 

func (a *AroundAspect) Before(ctx *flow.FlowContext, node Node)

type Aspect 

type Aspect interface {
	// Before 节点执行前调用
	Before(ctx *flow.FlowContext, node Node)

	// After 节点执行后调用
	After(ctx *flow.FlowContext, node Node, err error)
}

Aspect 切面接口 对应你要的三种类型

type BeforeAspect 

type BeforeAspect struct {
	Fn func(ctx *flow.FlowContext, node Node)
}

BeforeAspect 简易实现：只执行 Before

func (*BeforeAspect) After 

func (a *BeforeAspect) After(ctx *flow.FlowContext, node Node, err error)

func (*BeforeAspect) Before 

func (a *BeforeAspect) Before(ctx *flow.FlowContext, node Node)

type CircuitBreakerInterceptor 

type CircuitBreakerInterceptor struct {
	HalfOpenMaxCalls int // 半开时最多允许的试探次数，达到即关闭

	// FallbackFunc 降级函数，熔断时调用。若 nil，则返回错误。
	FallbackFunc func(ctx *flow.FlowContext, node Node) (map[string]any, error)
	// contains filtered or unexported fields
}

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

func NewCircuitBreakerInterceptor 

func NewCircuitBreakerInterceptor(threshold int, timeout time.Duration) *CircuitBreakerInterceptor

NewCircuitBreakerInterceptor 创建熔断拦截器 threshold: 触发熔断的失败次数阈值 timeout: 熔断后多久尝试恢复（进入半开）

func (*CircuitBreakerInterceptor) After 

func (cb *CircuitBreakerInterceptor) After(ctx *flow.FlowContext, node Node, err error)

func (*CircuitBreakerInterceptor) Around 

func (cb *CircuitBreakerInterceptor) Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

func (*CircuitBreakerInterceptor) Before 

func (cb *CircuitBreakerInterceptor) Before(ctx *flow.FlowContext, node Node)

type CircuitState 

type CircuitState int

CircuitState 熔断器状态 

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

type ErrorSwallowInterceptor 

type ErrorSwallowInterceptor struct {
	FallbackFunc func(ctx *flow.FlowContext, node Node, err error) (map[string]any, error)
}

ErrorSwallowInterceptor 拦截节点（或内层拦截器）返回的 error，执行降级逻辑。 作为最外层拦截器使用，确保节点级别的错误不会传播到工作流层触发 ctx.Cancel。

func NewErrorSwallowInterceptor 

func NewErrorSwallowInterceptor(fallback func(ctx *flow.FlowContext, node Node, err error) (map[string]any, error)) *ErrorSwallowInterceptor

func (*ErrorSwallowInterceptor) After 

func (e *ErrorSwallowInterceptor) After(ctx *flow.FlowContext, node Node, err error)

func (*ErrorSwallowInterceptor) Around 

func (e *ErrorSwallowInterceptor) Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

func (*ErrorSwallowInterceptor) Before 

func (e *ErrorSwallowInterceptor) Before(ctx *flow.FlowContext, node Node)

type Interceptor 

type Interceptor interface {
	Aspect
	// Around 包装节点执行，构建洋葱调用链
	// next: 调用下一个拦截器或实际节点执行
	// 返回: 节点输出和错误
	Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (map[string]any, error)
}

Interceptor 是能够拦截节点执行的切面（AOP 增强） 它包装了实际的节点执行逻辑，可实现重试、超时、熔断、兜底等高级控制。 Interceptor 也是 Aspect，因此可以直接通过 AddAspect 添加到节点或工作流。

type LoopConfig 

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

LoopConfig 循环节点配置

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 Plan 

type Plan struct {
	Goal  string `json:"goal"`
	Tasks []Task `json:"tasks"`
	// contains filtered or unexported fields
}

Plan 执行计划 新增状态变更通知机制：支持 channel 通知和条件变量等待，替代轮询

func NewPlan 

func NewPlan(goal string) *Plan

func Planning 

func Planning(ctx context.Context, goal string, agent agent.AgentInterface) (*Plan, error)

func RePlan 

func RePlan(ctx context.Context, plan *Plan, failedTask *Task, agent agent.AgentInterface) (*Plan, error)

func (*Plan) FindFailedTask 

func (p *Plan) FindFailedTask() *Task

FindFailedTask 查找第一个失败的任务（线程安全）

func (*Plan) GetMu 

func (p *Plan) GetMu() *sync.Mutex

func (*Plan) GetStateChannel 

func (p *Plan) GetStateChannel() <-chan struct{}

GetStateChannel 获取状态变更通知 channel 调用方可通过 select 监听此 channel，实现非轮询的状态监控

func (*Plan) IsAllCompleted 

func (p *Plan) IsAllCompleted() bool

IsAllCompleted 检查是否全部完成（线程安全）

func (*Plan) IsAnyFailed 

func (p *Plan) IsAnyFailed() bool

IsAnyFailed 检查是否有失败任务（线程安全）

func (*Plan) Snapshot 

func (p *Plan) Snapshot() Plan

Snapshot 获取计划当前状态的快照（线程安全）

func (*Plan) WaitForStateChange 

func (p *Plan) WaitForStateChange()

WaitForStateChange 阻塞等待状态变更（替代轮询） 返回当前计划的一个快照，调用方可检查任务状态

type RecoveryInterceptor 

type RecoveryInterceptor struct {
	// FallbackFunc 兜底函数，接收 panic 值，返回兜底结果。
	// 若 nil，则 panic 被转为 error 返回。
	FallbackFunc func(ctx *flow.FlowContext, node Node, recoverVal any) (map[string]any, error)
}

RecoveryInterceptor 捕获 panic 并执行兜底逻辑，防止单个节点拖垮整个工作流。

func NewRecoveryInterceptor 

func NewRecoveryInterceptor(fallback func(ctx *flow.FlowContext, node Node, recoverVal any) (map[string]any, error)) *RecoveryInterceptor

NewRecoveryInterceptor 创建兜底拦截器

func (*RecoveryInterceptor) After 

func (r *RecoveryInterceptor) After(ctx *flow.FlowContext, node Node, err error)

func (*RecoveryInterceptor) Around 

func (r *RecoveryInterceptor) Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (outputs map[string]any, err error)

func (*RecoveryInterceptor) Before 

func (r *RecoveryInterceptor) Before(ctx *flow.FlowContext, node Node)

type RetryInterceptor 

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

RetryInterceptor 节点失败时自动重试

func NewRetryInterceptor 

func NewRetryInterceptor(maxAttempts int, delay time.Duration) *RetryInterceptor

NewRetryInterceptor 创建重试拦截器

func (*RetryInterceptor) After 

func (r *RetryInterceptor) After(ctx *flow.FlowContext, node Node, err error)

func (*RetryInterceptor) Around 

func (r *RetryInterceptor) Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

func (*RetryInterceptor) Before 

func (r *RetryInterceptor) Before(ctx *flow.FlowContext, node Node)

type SimpleNode 

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

SimpleNode 通用节点

func BatchNewTaskNode 

func BatchNewTaskNode(plannerNodeID string, plan *Plan, agent agent.AgentInterface) []*SimpleNode

BatchNewTaskNode 批量创建任务节点（保留兼容） 注意：返回的节点列表可直接用于 Workflow.AddNode，也可传入 NewTopologicalNode 进行拓扑排序执行

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 NewPlannerNode 

func NewPlannerNode(
	id string,
	agent agent.AgentInterface,
) *SimpleNode

NewPlannerNode 创建规划节点 id: 节点ID user_goal: 目标 model: 使用的模型 规划的结果会存到: ID_plan 中

func NewScheduleLoopNode 

func NewScheduleLoopNode(
	plannerNodeID string,
	agent agent.AgentInterface,
) *SimpleNode

NewScheduleLoopNode 调度循环：监听任务完成/失败，触发重规划 修复：使用 Plan 的状态变更通知机制替代忙等待轮询 plannerNodeName: 需要监控的 planner 的节点ID 最终结果会存到: final_answer 中

func NewTaskNode 

func NewTaskNode(plannerNodeID string, task Task, agent agent.AgentInterface) *SimpleNode

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 Task 

type Task struct {
	ID          string         `json:"id"`
	Description string         `json:"description"`
	Inputs      []string       `json:"inputs"`
	Outputs     []string       `json:"outputs"`
	State       TaskState      `json:"state"`
	Result      map[string]any `json:"result"`
	Error       string         `json:"error"`
}

Task 规划任务

type TaskState 

type TaskState string

TaskState 任务状态 

const (
	TaskPending   TaskState = "pending"
	TaskRunning   TaskState = "running"
	TaskSuccess   TaskState = "success"
	TaskFailed    TaskState = "failed"
	TaskCancelled TaskState = "cancelled"
)

type TimeoutInterceptor 

type TimeoutInterceptor struct {
	Timeout time.Duration
}

TimeoutInterceptor 限制节点执行时间，超时时返回错误

func NewTimeoutInterceptor 

func NewTimeoutInterceptor(timeout time.Duration) *TimeoutInterceptor

NewTimeoutInterceptor 创建超时拦截器

func (*TimeoutInterceptor) After 

func (t *TimeoutInterceptor) After(ctx *flow.FlowContext, node Node, err error)

func (*TimeoutInterceptor) Around 

func (t *TimeoutInterceptor) Around(ctx *flow.FlowContext, node Node, next func() (map[string]any, error)) (map[string]any, error)

func (*TimeoutInterceptor) Before 

func (t *TimeoutInterceptor) Before(ctx *flow.FlowContext, node Node)

type TopologicalNode 

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

TopologicalNode 拓扑排序节点包装器 将一组有依赖关系的任务节点按拓扑序组织，确保依赖先执行 核心设计理念保留：依旧通过任务等待机制（ctx.Wait/ctx.WaitAll）控制执行 执行策略：按拓扑分层，同层节点并行执行（通过goroutine），层间串行等待

func NewTopologicalNode 

func NewTopologicalNode(id string, nodes []Node, outputKeys []string) (*TopologicalNode, error)

NewTopologicalNode 创建拓扑排序节点 id: 节点ID nodes: 一组有依赖关系的任务节点 outputKeys: 最终输出的key列表

func (*TopologicalNode) AddAspect 

func (tn *TopologicalNode) AddAspect(aspect Aspect)

AddAspect 添加切面

func (*TopologicalNode) Aspects 

func (tn *TopologicalNode) Aspects() []Aspect

Aspects 实现 Node 接口

func (*TopologicalNode) GetDependencyGraph 

func (tn *TopologicalNode) GetDependencyGraph() map[string][]string

GetDependencyGraph 获取依赖图（用于调试）

func (*TopologicalNode) GetExecutionOrder 

func (tn *TopologicalNode) GetExecutionOrder() []string

GetExecutionOrder 获取拓扑排序后的执行顺序（用于调试）

func (*TopologicalNode) GetLayers 

func (tn *TopologicalNode) GetLayers() [][]string

GetLayers 获取分层结果（用于调试）

func (*TopologicalNode) ID 

func (tn *TopologicalNode) ID() string

ID 实现 Node 接口

func (*TopologicalNode) Inputs 

func (tn *TopologicalNode) Inputs() []string

Inputs 实现 Node 接口：返回所有外部输入（不被任何节点生产的input）

func (*TopologicalNode) Outputs 

func (tn *TopologicalNode) Outputs() []string

Outputs 实现 Node 接口

func (*TopologicalNode) Run 

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

Run 按拓扑分层执行所有子节点 核心设计理念保留：依旧通过 ctx.WaitAll 等待机制控制执行 执行策略：

  • 按拓扑分层，每层内部节点并行执行（通过 goroutine）
  • 层间串行：前一层全部完成后，后一层才能开始
  • 节点通过 ctx.WaitAll 等待自己的输入就绪（来自前一层节点的输出或外部输入）

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.