Documentation
¶
Index ¶
- Constants
- Variables
- func IsAllowedAwaitBindingTransition(from, to AwaitBindingStatus) bool
- func IsAllowedTaskTransition(from, to TaskStatus) bool
- type AwaitBinding
- type AwaitBindingStatus
- type AwaitSource
- type AwaitType
- type EventGrade
- type NodeRuntime
- type NodeState
- type PatchOp
- type PatchTarget
- type ResumeSpec
- type ReuseKind
- type RuntimePatch
- type SuspendReason
- type Task
- type TaskCostTrace
- type TaskEvent
- type TaskOutput
- type TaskStatus
- type Workflow
- type WorkflowSuspendedError
- type WorkflowVersion
Constants ¶
View Source
const ( TaskEventStarted = "task_started" TaskEventSucceeded = "task_succeeded" TaskEventFailed = "task_failed" TaskEventSuspended = "task_suspended" TaskEventFinalFailed = "task_final_failed" TaskEventNodeCompleteAsync = "node_complete_async" )
View Source
const MaxAutoRetryCount = 5
MaxAutoRetryCount 自动重试上限,防止任务/子任务陷入无限重试循环消耗 API 费用。
Variables ¶
View Source
var AllowedTransitionsAwaitBinding = map[AwaitBindingStatus][]AwaitBindingStatus{ AwaitBindingPending: {AwaitBindingWaiting, AwaitBindingCanceled}, AwaitBindingWaiting: {AwaitBindingCompleting, AwaitBindingTimedOut, AwaitBindingCanceled}, AwaitBindingCompleting: {AwaitBindingCompleted, AwaitBindingFailed}, }
View Source
var AllowedTransitionsNodes = map[NodeState][]NodeState{ NodePending: {NodeReady, NodeSkipped}, NodeReady: {NodeRunning, NodeFailed, NodeSkipped}, NodeRunning: {NodeAwaiting, NodeSuccess, NodeFailed, NodeRetrying, NodeSkipped}, NodeAwaiting: {NodeSuccessPendingEdges, NodeFailedPendingEdges, NodeCanceled}, NodeRetrying: {NodeRunning, NodeFailed, NodeSuccess, NodeSkipped}, NodeSuccessPendingEdges: {NodeSuccess}, NodeFailedPendingEdges: {NodeFailed}, }
AllowedTransitionsNodes 允许的状态迁移规则
View Source
var AllowedTransitionsTasks = map[TaskStatus][]TaskStatus{ TaskPending: {TaskRunning, TaskCanceled}, TaskRunning: {TaskSuspended, TaskSuccess, TaskFailed, TaskCanceled}, TaskSuspended: {TaskRunning, TaskFailed, TaskCanceled}, }
Functions ¶
func IsAllowedAwaitBindingTransition ¶
func IsAllowedAwaitBindingTransition(from, to AwaitBindingStatus) bool
func IsAllowedTaskTransition ¶
func IsAllowedTaskTransition(from, to TaskStatus) bool
Types ¶
type AwaitBinding ¶
type AwaitBinding struct {
ID int64 `json:"id"`
TaskID int64 `json:"task_id"`
RootTaskID int64 `json:"root_task_id"`
NodeName string `json:"node_name"`
WorkflowVersionID int64 `json:"workflow_version_id"`
AwaitType AwaitType `json:"await_type"`
Source AwaitSource `json:"source"`
Status AwaitBindingStatus `json:"status"`
Provider *string `json:"provider,omitempty"`
ProviderTaskID *string `json:"provider_task_id,omitempty"`
APITaskID *string `json:"api_task_id,omitempty"`
ExternalTaskID *string `json:"external_task_id,omitempty"`
SignalName *string `json:"signal_name,omitempty"`
MessageName *string `json:"message_name,omitempty"`
CallbackToken *string `json:"callback_token,omitempty"`
Correlation map[string]any `json:"correlation,omitempty"`
Config map[string]any `json:"config,omitempty"`
LastEventID *string `json:"last_event_id,omitempty"`
LastEventSource *string `json:"last_event_source,omitempty"`
LastEventPayload map[string]any `json:"last_event_payload,omitempty"`
ResultPayload map[string]any `json:"result_payload,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
FallbackPollEnabled bool `json:"fallback_poll_enabled"`
FallbackPollTool *string `json:"fallback_poll_tool,omitempty"`
PollAttempts int `json:"poll_attempts"`
MaxPollAttempts int `json:"max_poll_attempts"`
LastPolledAt *time.Time `json:"last_polled_at,omitempty"`
NextPollAt *time.Time `json:"next_poll_at,omitempty"`
WaitingStartedAt *time.Time `json:"waiting_started_at,omitempty"`
TimeoutAt *time.Time `json:"timeout_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
FailedAt *time.Time `json:"failed_at,omitempty"`
CanceledAt *time.Time `json:"canceled_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
func (*AwaitBinding) CanTransitionTo ¶
func (b *AwaitBinding) CanTransitionTo(to AwaitBindingStatus) bool
type AwaitBindingStatus ¶
type AwaitBindingStatus string
const ( AwaitBindingPending AwaitBindingStatus = "pending" AwaitBindingWaiting AwaitBindingStatus = "waiting" AwaitBindingCompleting AwaitBindingStatus = "completing" AwaitBindingCompleted AwaitBindingStatus = "completed" AwaitBindingFailed AwaitBindingStatus = "failed" AwaitBindingTimedOut AwaitBindingStatus = "timed_out" AwaitBindingCanceled AwaitBindingStatus = "canceled" )
type AwaitSource ¶
type AwaitSource string
const ( AwaitSourceWebhook AwaitSource = "webhook" AwaitSourceSignal AwaitSource = "signal" AwaitSourcePoll AwaitSource = "poll" AwaitSourceWebhookOrPoll AwaitSource = "webhook_or_poll" AwaitSourceMessage AwaitSource = "message" AwaitSourceSubWorkflow AwaitSource = "subworkflow" )
type AwaitType ¶
type AwaitType string
const ( AwaitTypeExternalTask AwaitType = "external_task" AwaitTypeUserInput AwaitType = "user_input" AwaitTypeMessage AwaitType = "message" AwaitTypeApproval AwaitType = "approval" // AwaitTypeSubWorkflow 把"父节点等待 subworkflow 子任务完成"建模成 await binding, // 复用 await 的状态机 / poll / timeout 基建,统一父子任务的"等待—恢复"机制。 AwaitTypeSubWorkflow AwaitType = "subworkflow" )
type EventGrade ¶
type EventGrade string
EventGrade 事件等级,决定事件的持久化和推送策略
const ( GradeTransient EventGrade = "transient" // WS only,不入库不replay GradePersistent EventGrade = "persistent" // DB + Sequence + WS GradeAudit EventGrade = "audit" // DB only,不推送WS )
type NodeRuntime ¶
type NodeRuntime struct {
ID int64 `json:"id"`
TaskID int64 `json:"task_id"`
Name string `json:"name"`
State NodeState `json:"state"`
StartedAt *time.Time `json:"started_at"`
FinishedAt *time.Time `json:"finished_at"`
InputHash string `json:"input_hash"` // 存储上次执行成功时的“指纹”
ResolvedInput map[string]any `json:"resolved_input"` // 节点最终解析后的输入快照,执行期临时字段,不写入数据库
// public contract,给下游/表达式/UI 使用
Output map[string]any `json:"output"`
// internal durable state,给恢复/fork/replay/map fan-in/fan-out 使用
Checkpoint map[string]any `json:"checkpoint"`
Error string `json:"error"`
LastHeartbeat *time.Time `json:"last_heartbeat"`
Progress float64 `json:"progress"` // 当前节点进度 Progress = 0~1
Weight float64 `json:"weight"` // 节点权重 Weight = 0~1
Index int `json:"index"` // 在节点中的位置
BizIndex int `json:"biz_index"` // 在节点中属于业务的位置,去除start和end系统节点得来的
// 每个 Node 只存“自己发出的边”
// {
// "A->B": true,
// "A->C": false
// }
ActivatedEdges map[string]bool `json:"activated_edges"` // 存储已经激活的边
// ===== 新增 =====
OutputHash string `json:"output_hash"` // 输出快照hash
ReusedFromTaskID *int64 `json:"reused_from_task_id"` // 复用自哪个任务
ReusedFromNode *string `json:"reused_from_node"` // 复用自哪个节点
IsInjected bool `json:"is_injected"` // 是否注入输出
IsDirty bool `json:"is_dirty"` // 是否脏节点
DirtyReason string `json:"dirty_reason"` // input_changed / upstream_dirty / manual_override
CheckpointedAt *time.Time `json:"checkpointed_at"` // 快照时间
ReuseKind ReuseKind `json:"reuse_kind"`
ExecutionReason string `json:"execution_reason"`
PlanAction string `json:"plan_action"`
PatchedAt *time.Time `json:"patched_at"`
MaterializedAt *time.Time `json:"materialized_at"`
LastPatchLabel *string `json:"last_patch_label"`
}
func (*NodeRuntime) MergeRuntimePreserveMeta ¶
func (dst *NodeRuntime) MergeRuntimePreserveMeta(src *NodeRuntime)
func (*NodeRuntime) WriteMapItemResult ¶
type NodeState ¶
type NodeState string
const ( NodePending NodeState = "pending" // 调度态 NodeReady NodeState = "ready" // DAG 调度完成后的验证态,在ready期间才可以构造、校验输入参数 NodeRunning NodeState = "running" // 是执行态 NodeAwaiting NodeState = "awaiting" NodeRetrying NodeState = "retrying" NodeSuccess NodeState = "success" NodeFailed NodeState = "failed" // 失败的任务需要需要人为干预在表中设置为retrying才可以在服务重启时重新被执行, NodeSkipped NodeState = "skipped" NodeCanceled NodeState = "canceled" NodeSuccessPendingEdges NodeState = "success_pending_edges" // 任务成功待计算边 NodeFailedPendingEdges = "failed_pending_edges" // 失败待关闭边 )
状态迁移模型
type PatchTarget ¶
type PatchTarget string
const ( PatchTargetNodeOutput PatchTarget = "node_output" PatchTargetNodeCheckpoint PatchTarget = "node_checkpoint" )
type ResumeSpec ¶
type ResumeSpec struct {
ResumeFrom string `json:"resume_from"` // 从哪个节点开始重跑
Patches []RuntimePatch `json:"patches"` // 本次 patch 集合
}
type RuntimePatch ¶
type RuntimePatch struct {
Target PatchTarget `json:"target"` // node_output / node_checkpoint
Node string `json:"node"` // 节点名
Path string `json:"path"` // 点路径,如 intent.scene / results.0.caption
Op PatchOp `json:"op"` // set / delete / merge
Value any `json:"value"` // set/merge 时使用
Label string `json:"label"` // 给 UI 展示,可选
}
type SuspendReason ¶
type SuspendReason string
const ( // SuspendAsyncNode 用来调度异步节点抛出的挂起错误 SuspendAsyncNode SuspendReason = "async_node" // SuspendSubWorkflow 用来调度子工作流节点抛出的挂起错误 SuspendSubWorkflow SuspendReason = "sub_workflow" // SuspendWaitingGate 用来调度时间门控节点的挂起错误 SuspendWaitingGate SuspendReason = "waiting_gate" )
type Task ¶
type Task struct {
ID int64 `json:"id"`
UserID int64 `json:"user_id"`
ParentID *int64 `json:"parent_id"` // 父任务,子任务支持
RootID int64 `json:"root_id"` // 根任务,创建主任务时必须保证RootID=自己的ID,子任务的RootID=父任务的RootID
Type string `json:"type"`
Status TaskStatus `json:"status"`
InputJSON []byte `json:"input_json"`
OutputJSON []byte `json:"output_json"`
RetryCount int `json:"retry_count"`
ErrorMessage string `json:"error_message"`
WorkflowVersionID int64 `json:"workflow_version_id"`
WorkflowDefinitionID int64 `json:"workflow_definition_id"`
SubKey *string `json:"-"` // 恢复子工作流的key
// 任务抢占字段
WorkerID string `json:"-"`
StartedAt time.Time `json:"-"`
ParentNode *string `json:"-"` // 当前任务所属的父节点
MapIndex *int `json:"-"`
Progress float64 `json:"progress"` // 任务进度
// ===== 实现DAG 引擎从执行器升级为可回放、可分叉、可局部重做的运行时 =====
BaseRunID int64 `json:"base_run_id"` // 最初原始任务ID
ForkedFrom *int64 `json:"forked_from"` // 本次fork来源任务ID
RunDepth int `json:"run_depth"` // fork深度,每fork一次就会+1
EditAction *string `json:"edit_action"` // replace_start_image / replace_end_image / edit_user_prompt
EditLabel *string `json:"edit_label"` // 给UI看,比如“替换起始图”
// ===== patch / resume =====
ResumeFrom *string `json:"resume_from"` // 本次 fork 从哪个节点开始恢复
PatchJSON []byte `json:"patch_json"` // []RuntimePatch 的 JSON
// ===== 业务归属字段 =====
EntryType string `json:"entry_type"` // tool / template / workflow
ToolDefinitionID *int64 `json:"tool_definition_id"`
ToolModeID *int64 `json:"tool_mode_id"`
ToolModeVersionID *int64 `json:"tool_mode_version_id"`
TemplateID *int64 `json:"template_id"`
TemplateVersionID *int64 `json:"template_version_id"`
// 展示冗余字段(方便列表页直出)
EntryTitle *string `json:"entry_title"` // 比如“视频生成”
EntrySubtitle *string `json:"entry_subtitle"` // 比如“图生视频”
RouteKey *string `json:"route_key"` // video_generation
ModeKey *string `json:"mode_key"` // image_to_video
EstimatedCostTotal float64 `json:"estimated_cost_total"`
ActualCostTotal float64 `json:"actual_cost_total"`
CostStatus string `json:"cost_status"`
CreatedAt time.Time
UpdatedAt time.Time
}
func (*Task) IsTerminal ¶
type TaskCostTrace ¶
type TaskCostTrace struct {
ID int64 `json:"id"`
TaskID int64 `json:"task_id"`
RootTaskID int64 `json:"root_task_id"`
NodeRuntimeID *int64 `json:"node_runtime_id,omitempty"`
WorkflowName string `json:"workflow_name,omitempty"`
NodeName string `json:"node_name,omitempty"`
StepName string `json:"step_name,omitempty"`
ResourceType string `json:"resource_type"`
Provider string `json:"provider"`
Model string `json:"model,omitempty"`
ProviderRequestID string `json:"provider_request_id,omitempty"`
UsageQuantity float64 `json:"usage_quantity"`
UsageUnit string `json:"usage_unit"`
UnitPrice float64 `json:"unit_price"`
EstimatedCost float64 `json:"estimated_cost"`
ActualCost float64 `json:"actual_cost"`
Currency string `json:"currency"`
Status string `json:"status"`
IdempotencyKey string `json:"idempotency_key"`
TracePayload map[string]any `json:"trace_payload,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type TaskEvent ¶
type TaskEvent struct {
ID int64 `json:"-"`
TaskID int64 `json:"task_id"` // 任务ID
RootTaskID int64 `json:"root_task_id"`
Step string `json:"step"` // 步骤就是节点
Message string `json:"message"` // UI 文案
Error string `json:"error"` // 错误信息
Meta map[string]any `json:"meta"` // 结构化数据
CreatedAt time.Time `json:"created_at"` // 创建时间
Type string `json:"type"` // 事件类型
Progress float64 `json:"progress"` // 节点运行的进度,当前Step节点内的进度,非任务进度
Level string `json:"level"` // info system debug
// 任务进度计算: overall_progress = (node_index + node_progress) / node_total
NodeIndex int `json:"node_index"`
NodeTotal int `json:"node_total"`
// 事件分层
Grade EventGrade `json:"grade"` // transient / persistent / audit
Sequence int64 `json:"sequence"` // 全局递增序号,仅 Persistent 事件有值
}
TaskEvent 用于日志输出或外部监控(Step、Message、时间)
type TaskOutput ¶
type TaskOutput struct {
ResultType string `json:"result_type"`
PrimaryFileUrl string `json:"primary_file_url"`
CoverUrl *string `json:"cover_url,omitempty"`
PreviewUrl *string `json:"preview_url,omitempty"`
Width *int64 `json:"width,omitempty"` // 像素用整数
Height *int64 `json:"height,omitempty"` // 像素用整数
Duration *float64 `json:"duration,omitempty"` // 时长用浮点数(秒)
Extras map[string]any `json:"extras,omitempty"` // 扩展
}
TaskOutput 最终结果层(给客户端的 final Response)。
该结构结构性地对应工作流 DSL 的 definition.OutputDefinition: 引擎按 OutputDefinition 的字段把节点输出物化到这里,两者字段一一对应。 不再内嵌任何业务展示类型——「创意详情/时间轴」等派生视图已在 v1.0.4 起 收敛为 definition.OutputSlices,由业务 Service 从 task_nodes 按需再生, 不属于核心输出契约。
func ParseFinal ¶
func ParseFinal(outputJSON []byte) (*TaskOutput, error)
type TaskStatus ¶
type TaskStatus string
const ( TaskPending TaskStatus = "pending" TaskRunning TaskStatus = "running" TaskSuccess TaskStatus = "success" TaskFailed TaskStatus = "failed" TaskSuspended TaskStatus = "suspended" TaskCanceled TaskStatus = "canceled" )
type Workflow ¶
type Workflow struct {
ID int64 `gorm:"primaryKey" json:"id"`
Name string `json:"name"`
UserID *int64 `json:"user_id"`
Description string `json:"description"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
Workflow 表示一个逻辑工作流
type WorkflowSuspendedError ¶
type WorkflowSuspendedError struct {
Reason SuspendReason
}
func (*WorkflowSuspendedError) Error ¶
func (e *WorkflowSuspendedError) Error() string
type WorkflowVersion ¶
type WorkflowVersion struct {
ID int64 `json:"id"`
WorkflowID int64 `json:"workflow_id"`
Version int64 `json:"version"`
// DefinitionJSON (DAG) 包含每一个 节点的定义NodeDefinition
// 表示节点结构(DAG), 当前版本的工作流每个节点的定义都存储在这里面
DefinitionJSON datatypes.JSON `json:"definition_json"`
Hash string `json:"hash"`
CreatedAt time.Time `json:"created_at"`
}
WorkflowVersion 工作流执行的版本
Source Files
¶
Click to show internal directories.
Click to hide internal directories.