nodes

package
v1.0.3 Latest Latest
Warning

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

 Go to latest Published: Jul 7, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Index

Constants

View Source 
const TaskEventFanoutProgress = "fanout_progress"

Variables

View Source 
var AwaitNodeSchema = &NodeTypeSchema{
	Type:        definition.NodeAwait,
	Description: "Await external event/input node",
	ConfigSchema: tool.DataSchema{
		Fields: map[string]tool.FieldSchema{
			"await_type": {
				Type:     "string",
				Required: true,
			},
			"source": {
				Type:     "string",
				Required: true,
			},
			"provider": {
				Type:     "string",
				Required: false,
			},
			"signal_name": {
				Type:     "string",
				Required: false,
			},
			"callback_token_expr": {
				Type:     "string",
				Required: false,
			},
			"correlation": {
				Type:     "object",
				Required: false,
			},
			"completion": {
				Type:     "object",
				Required: false,
			},
			"fallback_poll": {
				Type:     "object",
				Required: false,
			},
			"timeout_seconds": {
				Type:     "number",
				Required: false,
			},
		},
	},
}
View Source 
var EndNodeSchema = &NodeTypeSchema{
	Type:        definition.NodeEnd,
	Description: "End Node",
	ConfigSchema: tool.DataSchema{
		Fields: map[string]tool.FieldSchema{},
	},
}
View Source 
var LoopNodeSchema = &NodeTypeSchema{

	Type: definition.NodeLoop,

	Description: "Loop execution node",

	ConfigSchema: tool.DataSchema{

		Fields: map[string]tool.FieldSchema{

			"items": {
				Type:     "string",
				Required: true,
			},
			"iterator": {
				Type:     "string",
				Required: true,
			},
			"workflow": {
				Type: "string",
				Desc: "Workflow name",
			},

			"carry": {
				Type:     "object",
				Required: false,
				Desc:     "每一轮循环携带的参数",
			},

			"initial": {
				Type:     "object",
				Required: false,
				Desc:     "初始时的参数",
			},
		},
	},
	ExprConfigFields: map[string]bool{
		"items": true,
	},
}
View Source 
var MapNodeSchema = &NodeTypeSchema{

	Type: definition.NodeMap,

	Description: "Map execution node",

	ConfigSchema: tool.DataSchema{

		Fields: map[string]tool.FieldSchema{

			"items": {
				Type:     "string",
				Required: true,
			},

			"iterator": {
				Type:     "string",
				Required: true,
			},
			"workflow": {
				Type: "string",
				Desc: "Workflow name",
			},
			"parallel": {
				Type:     "number",
				Required: false,
				Desc:     "最大并发数",
			},
			"failure_policy": {
				Type:     "string",
				Required: false,
				Desc:     "子任务失败策略: fail_fast（默认，任一失败则Map失败）| partial（允许部分成功，失败项使用fallback兜底）",
			},
			"max_child_retries": {
				Type:     "number",
				Required: false,
				Desc:     "每个子任务最大自动重试次数，-1使用全局默认",
			},
			"fallback_source": {
				Type:     "string",
				Required: false,
				Desc:     "fallback数据来源: item（使用原始迭代项）",
			},
			"max_fallback_ratio": {
				Type:     "number",
				Required: false,
				Desc:     "允许的最大 fallback 比率（0.0-1.0），超出时 Map 失败（Phase 2 强制执行）",
			},
		},
	},
	ExprConfigFields: map[string]bool{
		"items": true,
	},
}
View Source 
var StartNodeSchema = &NodeTypeSchema{
	Type:        definition.NodeStart,
	Description: "Start Node",
	ConfigSchema: tool.DataSchema{
		Fields: map[string]tool.FieldSchema{},
	},
}
View Source 
var SubWorkflowNodeSchema = &NodeTypeSchema{

	Type: definition.NodeSubWorkflow,

	Description: "SubWorkflow node",

	ConfigSchema: tool.DataSchema{

		Fields: map[string]tool.FieldSchema{

			"workflow": {
				Type:     "string",
				Required: true,
			},
		},
	},
}
View Source 
var ToolNodeSchema = &NodeTypeSchema{
	Type:        definition.NodeTool,
	Description: "Tool execution node",
	ConfigSchema: tool.DataSchema{
		Fields: map[string]tool.FieldSchema{
			"tool": {
				Type:     "string",
				Required: true,
				Desc:     "Tool name",
			},
		},
	},
}

Functions

func CPFanoutKind 

func CPFanoutKind() string

func CalculateMapItemHash 

func CalculateMapItemHash(item any) string

CalculateMapItemHash 计算 item hash

func EmitFanoutProgress 

func EmitFanoutProgress(execCtx *NodeExecContext, p FanoutProgress)

func IsSystemNode 

func IsSystemNode(nodeName string) bool

IsSystemNode 是否是系统节点，只有start和end是系统节点

func RegisterBuiltinNodes 

func RegisterBuiltinNodes(reg *NodeRegistry)

RegisterBuiltinNodes registers the 6 built-in node types that every workflow engine needs: start, end, subworkflow, map, loop, await.

The caller must also register tool nodes via: 

reg.Register(NewToolFactory(toolReg), ToolNodeSchema)

with their own tool.Registry.

func ValidateFieldTypeStrict 

func ValidateFieldTypeStrict(fs tool.FieldSchema, val any) error

ValidateFieldTypeStrict 类型校验

Types

type AwaitNodeFactory 

type AwaitNodeFactory struct{}

func NewAwaitNodeFactory 

func NewAwaitNodeFactory() *AwaitNodeFactory

func (*AwaitNodeFactory) Create 

func (f *AwaitNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*AwaitNodeFactory) Type 

func (f *AwaitNodeFactory) Type() string

type AwaitStep 

type AwaitStep struct {
	AwaitType string
	Source    string
}

AwaitStep 本身不在 Step.Run 中执行真正的等待逻辑。 与 subworkflow 类似，Engine.executeNode 会接管该节点： 1. 创建 AwaitBinding 2. 将节点置为 awaiting 3. 挂起任务

func NewAwaitStep 

func NewAwaitStep(awaitType, source string) *AwaitStep

func (*AwaitStep) InputSchema 

func (s *AwaitStep) InputSchema() tool.DataSchema

func (*AwaitStep) Mode 

func (s *AwaitStep) Mode() tool.ExecutionMode

func (*AwaitStep) Name 

func (s *AwaitStep) Name() string

func (*AwaitStep) OutputSchema 

func (s *AwaitStep) OutputSchema() tool.DataSchema

func (*AwaitStep) RetryPolicy 

func (s *AwaitStep) RetryPolicy() RetryPolicy

func (*AwaitStep) Run 

func (s *AwaitStep) Run(ctx *NodeExecContext) error

type Context 

type Context struct {
	Ctx  context.Context
	Task *domain.Task

	Workflow *definition.WorkflowDefinition

	// Input 任务的输入结构
	Input map[string]any

	// Output 是最终输出的持久化输出结构（final data contract） 最终会写入数据库，所以Output = 持久化结构
	// 主要用于：API、DB、DAG UI、Debug、Replay
	// 强制统一结构：
	// {
	// "input": {
	//   "topic": "AI Workflow"
	// },
	// "nodes": {
	//
	//   "generate_script": {
	//       "status": "success",
	//       "started_at": "2026-03-06T10:00",
	//       "finished_at": "2026-03-06T10:01",
	//       "output": {
	//           "script": "AI is transforming..."
	//       }
	//   },
	//
	//   "generate_image": {
	//       "status": "success",
	//       "started_at": "...",
	//       "finished_at": "...",
	//       "output": {
	//           "image_url": "..."
	//       }
	//   }
	//
	// }
	//}
	Output map[string]any
	// Runtime 运行时节点
	Runtime map[string]*domain.NodeRuntime
	// ActivatedEdges 激活边（active edge）
	// 已激活的边，运行时真实发生的路径（Runtime Path）
	ActivatedEdges map[string]bool // from->to

	EventBus *eventbus.EventBus
	// 递归限制，防止无限创建 Task
	Depth int

	// ===== 新增 =====
	ParentSnapshot *ReuseSnapshot
	// InjectedOutputs 不是调度必需字段，而是“本次 run 中哪些节点是从父快照注入复用的”这一层运行时观测数据
	InjectedOutputs map[string]map[string]any // nodeName -> output
	DirtyNodes      map[string]bool
	MapItemReuse    map[string]map[int]bool

	// ✅ 降级为 debug / UI 辅助信息
	// ❌ 不再参与执行逻辑
	Patches      []domain.RuntimePatch
	ResumeFrom   string
	PatchedNodes map[string]bool
	// contains filtered or unexported fields
}

func (*Context) ActivatedEdgesMerge 

func (c *Context) ActivatedEdgesMerge(m map[string]bool)

ActivatedEdgesMerge 合并激活边

func (*Context) CalculateInputHash 

func (c *Context) CalculateInputHash(
	nodeVersion string,
	input map[string]any,
) string

CalculateInputHash 计算节点的输入哈希（带 nodeVersion 盐值）

func (*Context) CalculateOutputHash 

func (c *Context) CalculateOutputHash(
	output map[string]any,
) string

CalculateOutputHash 计算节点的输出哈希（不带盐值，纯粹计算数据本身）

func (*Context) CalculateTaskProgress 

func (c *Context) CalculateTaskProgress() float64

CalculateTaskProgress 计算任务进度

func (*Context) ClearNodeOutput 

func (c *Context) ClearNodeOutput(nodeName string)

func (*Context) EmitNodeEvent 

func (c *Context) EmitNodeEvent(nodeName string, event NodeEvent)

func (*Context) EmitToolEvent 

func (c *Context) EmitToolEvent(nodeName string, event tool.ToolEvent)

func (*Context) EnsureOutputInitialized 

func (c *Context) EnsureOutputInitialized()

EnsureOutputInitialized Output初始化函数

func (*Context) EvalAny 

func (c *Context) EvalAny(exprStr string) (any, error)

EvalAny 通用表达式

func (*Context) EvalBool 

func (c *Context) EvalBool(exprStr string) (bool, error)

EvalBool 专门用于 Condition 的表达式

func (*Context) GetFinal 

func (c *Context) GetFinal() (*domain.TaskOutput, error)

GetFinal 按照规范输出最终结果

func (*Context) GetNodeIndex 

func (c *Context) GetNodeIndex(node string) int

func (*Context) GetNodeOutput 

func (c *Context) GetNodeOutput(node string) map[string]any

func (*Context) GetNodeTotal 

func (c *Context) GetNodeTotal() int

func (*Context) Progress 

func (c *Context) Progress() float64

Progress 计算任务执行的估算进度

func (*Context) SetNodeOutput 

func (c *Context) SetNodeOutput(
	nodeName string,
	data map[string]any,
	schema tool.DataSchema,
) error

SetNodeOutput 将单个节点的成功结果存入全局上下文

func (*Context) UpdateNodeStatus 

func (c *Context) UpdateNodeStatus(
	nodeName string,
	status string,
)

UpdateNodeStatus 更新节点状态到Output

type DefaultFactory 

type DefaultFactory struct {
	ToolRegistry *registry2.Registry
}

func (*DefaultFactory) Create 

func (f *DefaultFactory) Create(nd definition.NodeDefinition) (Step, error)

type EndNodeFactory 

type EndNodeFactory struct{}

func NewEndNodeFactory 

func NewEndNodeFactory() *EndNodeFactory

func (*EndNodeFactory) Create 

func (f *EndNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*EndNodeFactory) Type 

func (f *EndNodeFactory) Type() string

type EndStep 

type EndStep struct{}

func (*EndStep) InputSchema 

func (e *EndStep) InputSchema() tool.DataSchema

func (*EndStep) Mode 

func (v *EndStep) Mode() tool.ExecutionMode

func (*EndStep) Name 

func (e *EndStep) Name() string

func (*EndStep) OutputSchema 

func (e *EndStep) OutputSchema() tool.DataSchema

func (*EndStep) RetryPolicy 

func (e *EndStep) RetryPolicy() RetryPolicy

func (*EndStep) Run 

func (e *EndStep) Run(ctx *NodeExecContext) error

type FanoutKind 

type FanoutKind string
const (
	FanoutKindNone        FanoutKind = ""
	FanoutKindMap         FanoutKind = "map"
	FanoutKindLoop        FanoutKind = "loop"
	FanoutKindSubWorkflow FanoutKind = "subworkflow"
)

type FanoutProgress 

type FanoutProgress struct {
	Kind         FanoutKind
	ParentNode   string
	ParentLabel  string
	Total        int
	Done         int
	Running      int
	Failed       int
	Reused       int
	CurrentIndex int
	Progress     float64
}

type LoopNodeFactory 

type LoopNodeFactory struct{}

func NewLoopNodeFactory 

func NewLoopNodeFactory() *LoopNodeFactory

func (*LoopNodeFactory) Create 

func (l *LoopNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*LoopNodeFactory) Type 

func (l *LoopNodeFactory) Type() string

type LoopNodeStep 

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

func NewLoopStep 

func NewLoopStep(
	items string,
	iterator string,
	workflow string,
	carry map[string]string,
	initial map[string]any,
) *LoopNodeStep

func (*LoopNodeStep) InputSchema 

func (l *LoopNodeStep) InputSchema() tool.DataSchema

func (*LoopNodeStep) Mode 

func (l *LoopNodeStep) Mode() tool.ExecutionMode

func (*LoopNodeStep) Name 

func (l *LoopNodeStep) Name() string

func (*LoopNodeStep) OutputSchema 

func (l *LoopNodeStep) OutputSchema() tool.DataSchema

func (*LoopNodeStep) RetryPolicy 

func (l *LoopNodeStep) RetryPolicy() RetryPolicy

func (*LoopNodeStep) Run 

func (l *LoopNodeStep) Run(execCtx *NodeExecContext) error

type MapNodeFactory 

type MapNodeFactory struct{}

func NewMapNodeFactory 

func NewMapNodeFactory() *MapNodeFactory

func (*MapNodeFactory) Create 

func (f *MapNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*MapNodeFactory) Type 

func (f *MapNodeFactory) Type() string

type MapNodeStep 

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

func NewMapStep 

func NewMapStep(
	items string,
	iterator string,
	workflow string,
	parallel int,
) *MapNodeStep

func (*MapNodeStep) InputSchema 

func (m *MapNodeStep) InputSchema() tool.DataSchema

func (*MapNodeStep) Mode 

func (m *MapNodeStep) Mode() tool.ExecutionMode

func (*MapNodeStep) Name 

func (m *MapNodeStep) Name() string

func (*MapNodeStep) OutputSchema 

func (m *MapNodeStep) OutputSchema() tool.DataSchema

func (*MapNodeStep) RetryPolicy 

func (m *MapNodeStep) RetryPolicy() RetryPolicy

func (*MapNodeStep) Run 

func (m *MapNodeStep) Run(execCtx *NodeExecContext) error

func (*MapNodeStep) WithFailurePolicy 

func (m *MapNodeStep) WithFailurePolicy(policy string) *MapNodeStep

WithFailurePolicy sets the failure policy for this map step.

func (*MapNodeStep) WithFallbackSource 

func (m *MapNodeStep) WithFallbackSource(source string) *MapNodeStep

WithFallbackSource sets the fallback data source.

func (*MapNodeStep) WithMaxChildRetries 

func (m *MapNodeStep) WithMaxChildRetries(retries int) *MapNodeStep

WithMaxChildRetries sets the max child-level retries.

func (*MapNodeStep) WithMaxFallbackRatio 

func (m *MapNodeStep) WithMaxFallbackRatio(ratio float64) *MapNodeStep

WithMaxFallbackRatio sets the max allowed fallback ratio (Phase 2 enforcement).

type Node 

type Node struct {
	Name    string
	Label   string // 面向客户端的展示名
	Step    Step   // Step 是顶部步骤的代码 不存数据库
	Version string // 代码逻辑版本号。当你改了 Tool 代码，手动改这个值（如 "v1.1"）

	Config       map[string]any    // 节点配置，静态参数
	InputMapping map[string]string // 数据流映射，运行时输
	Type         definition.NodeType
	Weight       float64 // 节点权重
}

Node DAG 节点，带状态、依赖、重试策略

type NodeEmitter 

type NodeEmitter interface {
	EmitNodeEvent(event NodeEvent)
}

type NodeEvent 

type NodeEvent struct {
	Type     string
	Message  string
	Progress float64 // 本节点的执行进度
	Data     map[string]any
}

NodeEvent 节点事件

type NodeExecContext 

type NodeExecContext struct {
	TaskContext *Context // 全局只读
	// Input (当前节点的“入参表”)：这是执行前，从全局 Output 搬运过来的、专门给当前这个 Step 用的数据。比如：{"prompt": "画一只猫"}
	Input map[string]any
	// Output 当前节点的“成绩单”)：这是 Step.Run 执行完后，产出的结果。比如：{"image_url": "http://..."}。
	Output map[string]any

	NodeDef  *definition.NodeDefinition
	Executor WorkflowExecutor
	// contains filtered or unexported fields
}

NodeExecContext 每一个节点的执行上下文，包含一个只读的TaskContext全局上下文

func (*NodeExecContext) EmitNodeEvent 

func (n *NodeExecContext) EmitNodeEvent(event NodeEvent)

EmitNodeEvent 发送节点事件

func (*NodeExecContext) EmitToolEvent 

func (n *NodeExecContext) EmitToolEvent(event tool.ToolEvent)

EmitToolEvent 发送工具事件

func (*NodeExecContext) SetOutput 

func (n *NodeExecContext) SetOutput(key string, val any)

type NodeFactory 

type NodeFactory interface {
	// Type 节点类型
	Type() string
	// Create 创建Step
	//例如：LLMFactory
	// ToolFactory
	//HTTPFactory
	//ScriptFactory
	Create(def definition.NodeDefinition) (Step, error)
}

NodeFactory 用于节点Step的构建

type NodeRegistry 

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

NodeRegistry 节点注册中心 注意：NodeRegistry不是 ToolRegistry，它负责NodeType -> NodeFactory 例如：start tool llm condition code

func InitNodeRegistry deprecated 

func InitNodeRegistry(toolReg *tool.Registry) *NodeRegistry

InitNodeRegistry creates a new NodeRegistry, registers built-in nodes and all tools from the given tool.Registry, then returns the registry.

Deprecated: use NewNodeRegistry + RegisterBuiltinNodes + explicit tool node registration instead. This function uses a package-level global and prevents multiple independent registries in the same process.

func NewNodeRegistry 

func NewNodeRegistry() *NodeRegistry

func (*NodeRegistry) Get 

func (r *NodeRegistry) Get(nodeType string) (NodeFactory, error)

func (*NodeRegistry) GetSchema 

func (r *NodeRegistry) GetSchema(nodeType definition.NodeType) (*NodeTypeSchema, error)

func (*NodeRegistry) IsExprConfigField 

func (r *NodeRegistry) IsExprConfigField(nodeType definition.NodeType, field string) bool

func (*NodeRegistry) List 

func (r *NodeRegistry) List() []string

func (*NodeRegistry) Register 

func (r *NodeRegistry) Register(factory NodeFactory, schema *NodeTypeSchema)

func (*NodeRegistry) ValidateNode 

func (r *NodeRegistry) ValidateNode(def *definition.NodeDefinition) error

type NodeTypeSchema 

type NodeTypeSchema struct {

	// 节点类型
	Type definition.NodeType

	// 描述
	Description string

	// 配置Schema
	ConfigSchema tool.DataSchema

	// 表达式字段，定义哪些字段是表达式字段，运行时输入构造时，expression config 要 Eval
	ExprConfigFields map[string]bool
}

NodeTypeSchema 描述节点类型

func (*NodeTypeSchema) Validate 

func (s *NodeTypeSchema) Validate(
	def *definition.NodeDefinition,
) error

Validate 校验 NodeDefinition Config

type RetryPolicy 

type RetryPolicy struct {
	MaxRetries int
	Interval   time.Duration
}

RetryPolicy 重试机制

type ReuseSnapshot 

type ReuseSnapshot struct {
	TaskID int64
	Nodes  map[string]*domain.NodeRuntime
	Output map[string]any
}

ReuseSnapshot 恢复快照

type StartNodeFactory 

type StartNodeFactory struct{}

StartNodeFactory 是 StartStep的工程结构

func NewStartNodeFactory 

func NewStartNodeFactory() *StartNodeFactory

func (*StartNodeFactory) Create 

func (f *StartNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*StartNodeFactory) Type 

func (f *StartNodeFactory) Type() string

type StartStep 

type StartStep struct{}

StartStep 是Start 节点，职责只有一个： 把 TaskContext.Input 传递给 DAG，例如：用户输入： 

	{
	 "topic": "AI创业"
	}

 Start 输出：

	{
	 "topic": "AI创业"
	}

func (*StartStep) InputSchema 

func (s *StartStep) InputSchema() tool.DataSchema

func (*StartStep) Mode 

func (v *StartStep) Mode() tool.ExecutionMode

func (*StartStep) Name 

func (s *StartStep) Name() string

func (*StartStep) OutputSchema 

func (s *StartStep) OutputSchema() tool.DataSchema

func (*StartStep) RetryPolicy 

func (s *StartStep) RetryPolicy() RetryPolicy

func (*StartStep) Run 

func (s *StartStep) Run(ctx *NodeExecContext) error

type Step 

type Step interface {
	Name() string
	Run(ctx *NodeExecContext) error
	RetryPolicy() RetryPolicy

	InputSchema() tool.DataSchema
	OutputSchema() tool.DataSchema

	Mode() tool.ExecutionMode
}

Step 节点执行

func NewToolStepAdapter 

func NewToolStepAdapter(
	t tool.Tool,
	config map[string]any,
	version string,
) Step

type StepFactory 

type StepFactory interface {
	Create(node definition.NodeDefinition) (Step, error)
}

type SubWorkflowNodeFactory 

type SubWorkflowNodeFactory struct{}

func NewSubWorkflowNodeFactory 

func NewSubWorkflowNodeFactory() *SubWorkflowNodeFactory

func (*SubWorkflowNodeFactory) Create 

func (f *SubWorkflowNodeFactory) Create(def definition.NodeDefinition) (Step, error)

func (*SubWorkflowNodeFactory) Type 

func (f *SubWorkflowNodeFactory) Type() string

type SubWorkflowStep 

type SubWorkflowStep struct {
	WorkflowName string
}

func NewSubWorkflowStep 

func NewSubWorkflowStep(workflowName string) *SubWorkflowStep

func (*SubWorkflowStep) InputSchema 

func (s *SubWorkflowStep) InputSchema() tool.DataSchema

func (*SubWorkflowStep) Mode 

func (s *SubWorkflowStep) Mode() tool.ExecutionMode

func (SubWorkflowStep) Name 

func (s SubWorkflowStep) Name() string

func (*SubWorkflowStep) OutputSchema 

func (s *SubWorkflowStep) OutputSchema() tool.DataSchema

func (*SubWorkflowStep) RetryPolicy 

func (s *SubWorkflowStep) RetryPolicy() RetryPolicy

func (*SubWorkflowStep) Run 

func (s *SubWorkflowStep) Run(ctx *NodeExecContext) error

type ToolFactory 

type ToolFactory struct {
	ToolRegistry *tool.Registry
}

func NewToolFactory 

func NewToolFactory(toolRegistry *tool.Registry) *ToolFactory

func (*ToolFactory) Create 

func (f *ToolFactory) Create(def definition.NodeDefinition) (Step, error)

func (*ToolFactory) Type 

func (f *ToolFactory) Type() string

type ToolStepAdapter 

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

ToolStepAdapter 桥接 Tool → Step 的适配器

func (*ToolStepAdapter) BuildUsageFacts 

func (t *ToolStepAdapter) BuildUsageFacts(input map[string]any, output map[string]any) ([]map[string]any, error)

func (*ToolStepAdapter) InputSchema 

func (t *ToolStepAdapter) InputSchema() tool.DataSchema

func (*ToolStepAdapter) Mode 

func (t *ToolStepAdapter) Mode() tool.ExecutionMode

func (*ToolStepAdapter) Name 

func (t *ToolStepAdapter) Name() string

func (*ToolStepAdapter) OutputSchema 

func (t *ToolStepAdapter) OutputSchema() tool.DataSchema

func (*ToolStepAdapter) RetryPolicy 

func (t *ToolStepAdapter) RetryPolicy() RetryPolicy

func (*ToolStepAdapter) Run 

func (t *ToolStepAdapter) Run(ctx *NodeExecContext) error

func (*ToolStepAdapter) UsageSchema 

func (t *ToolStepAdapter) UsageSchema() tool.DataSchema

type UsageAwareStep 

type UsageAwareStep interface {
	UsageSchema() tool.DataSchema
	BuildUsageFacts(input map[string]any, output map[string]any) ([]map[string]any, error)
}

UsageAwareStep 是节点 step 的可选能力接口。 只有有成本语义的 step/tool 才需要实现它。

type WorkflowExecutor 

type WorkflowExecutor interface {
	RunSubWorkflow(
		execCtx *NodeExecContext,
		workflowName string,
		input map[string]any,
	) (map[string]any, error)

	TaskRepo() repository.TaskRepository
	NodeRepo() repository.NodeRuntimeRepository
}

WorkflowExecutor 工作流执行协议

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.