工作流块 YAML 模式
工作流块的 YAML 配置参考
模式定义
type: object
required:
- type
- name
- inputs
properties:
type:
type: string
enum: [workflow]
description: Block type identifier
name:
type: string
description: Display name for this workflow block
inputs:
type: object
required:
- workflowId
properties:
workflowId:
type: string
description: ID of the workflow to execute
inputMapping:
type: object
description: Map current workflow data to sub-workflow inputs
additionalProperties:
type: string
description: Input value or reference to parent workflow data
environmentVariables:
type: object
description: Environment variables to pass to sub-workflow
additionalProperties:
type: string
description: Environment variable value
timeout:
type: number
description: Maximum execution time in milliseconds
default: 300000
minimum: 1000
maximum: 1800000
connections:
type: object
properties:
success:
type: string
description: Target block ID for successful workflow completion
error:
type: string
description: Target block ID for error handling连接配置
连接定义了根据子工作流结果工作流的走向：
connections:
success: <string> # Target block ID for successful completion
error: <string> # Target block ID for error handling (optional)示例
简单工作流执行
data-processor:
type: workflow
name: "Data Processing Workflow"
inputs:
workflowId: "data-processing-v2"
inputMapping:
rawData: <start.input>
userId: <user-validator.userId>
environmentVariables:
PROCESSING_MODE: "production"
LOG_LEVEL: "info"
connections:
success: process-results
error: workflow-error-handler内容生成管道
content-generator:
type: workflow
name: "Content Generation Pipeline"
inputs:
workflowId: "content-generation-v3"
inputMapping:
topic: <start.topic>
style: <style-analyzer.recommendedStyle>
targetAudience: <audience-detector.audience>
brandGuidelines: <brand-config.guidelines>
environmentVariables:
CONTENT_API_KEY: "{{CONTENT_API_KEY}}"
QUALITY_THRESHOLD: "high"
timeout: 120000
connections:
success: review-content
error: content-generation-failed多步骤分析工作流
analysis-workflow:
type: workflow
name: "Analysis Workflow"
inputs:
workflowId: "comprehensive-analysis"
inputMapping:
document: <document-processor.content>
analysisType: "comprehensive"
includeMetrics: true
outputFormat: "structured"
environmentVariables:
ANALYSIS_MODEL: "gpt-4o"
OPENAI_API_KEY: "{{OPENAI_API_KEY}}"
CLAUDE_API_KEY: "{{CLAUDE_API_KEY}}"
connections:
success: compile-analysis-report
error: analysis-workflow-error条件工作流执行
customer-workflow-router:
type: condition
name: "Customer Workflow Router"
inputs:
conditions:
if: <customer-type.type> === "enterprise"
else-if: <customer-type.type> === "premium"
else: true
connections:
conditions:
if: enterprise-workflow
else-if: premium-workflow
else: standard-workflow
enterprise-workflow:
type: workflow
name: "Enterprise Customer Workflow"
inputs:
workflowId: "enterprise-customer-processing"
inputMapping:
customerData: <customer-data.profile>
accountManager: <account-assignment.manager>
tier: "enterprise"
environmentVariables:
PRIORITY_LEVEL: "high"
SLA_REQUIREMENTS: "strict"
connections:
success: enterprise-complete
premium-workflow:
type: workflow
name: "Premium Customer Workflow"
inputs:
workflowId: "premium-customer-processing"
inputMapping:
customerData: <customer-data.profile>
supportLevel: "premium"
environmentVariables:
PRIORITY_LEVEL: "medium"
connections:
success: premium-complete
standard-workflow:
type: workflow
name: "Standard Customer Workflow"
inputs:
workflowId: "standard-customer-processing"
inputMapping:
customerData: <customer-data.profile>
environmentVariables:
PRIORITY_LEVEL: "standard"
connections:
success: standard-complete并行工作流执行
parallel-workflows:
type: parallel
name: "Parallel Workflow Processing"
inputs:
parallelType: collection
collection: |
[
{"workflowId": "sentiment-analysis", "focus": "sentiment"},
{"workflowId": "topic-extraction", "focus": "topics"},
{"workflowId": "entity-recognition", "focus": "entities"}
]
connections:
success: merge-workflow-results
execute-analysis-workflow:
type: workflow
name: "Execute Analysis Workflow"
parentId: parallel-workflows
inputs:
workflowId: <parallel.currentItem.workflowId>
inputMapping:
content: <start.content>
analysisType: <parallel.currentItem.focus>
environmentVariables:
ANALYSIS_API_KEY: "{{ANALYSIS_API_KEY}}"
connections:
success: workflow-complete错误处理工作流
main-workflow:
type: workflow
name: "Main Processing Workflow"
inputs:
workflowId: "main-processing-v1"
inputMapping:
data: <start.input>
timeout: 180000
connections:
success: main-complete
error: error-recovery-workflow
error-recovery-workflow:
type: workflow
name: "Error Recovery Workflow"
inputs:
workflowId: "error-recovery-v1"
inputMapping:
originalInput: <start.input>
errorDetails: <main-workflow.error>
failureTimestamp: "{{new Date().toISOString()}}"
environmentVariables:
RECOVERY_MODE: "automatic"
FALLBACK_ENABLED: "true"
connections:
success: recovery-complete
error: manual-intervention-required输入映射
将数据从父工作流映射到子工作流：
inputMapping:
# Static values
mode: "production"
version: "1.0"
# References to parent workflow data
userData: <user-processor.profile>
settings: <config-loader.settings>
# Complex object mapping
requestData:
id: <start.requestId>
timestamp: "{{new Date().toISOString()}}"
source: "parent-workflow"输出引用
在工作流块完成后，您可以引用其输出：
# In subsequent blocks
next-block:
inputs:
workflowResult: <workflow-name.output> # Sub-workflow output
executionTime: <workflow-name.duration> # Execution duration
status: <workflow-name.status> # Execution status最佳实践
- 使用描述性工作流 ID 以提高清晰度
- 仅映射必要的数据到子工作流
- 根据工作流的复杂性设置适当的超时时间
- 包含错误处理以确保执行的稳健性
- 安全地传递环境变量
- 先独立测试子工作流
- 监控嵌套工作流的性能
- 使用版本化的工作流 ID 以确保稳定性