Refly AI工作流引擎：3步构建企业级智能体技能基础设施

【免费下载链接】reflyThe first open-source agent skills builder. Define skills by vibe workflow, run on Claude Code, Cursor, Codex & more. Build Clawdbot 🦞· APIs for Lovable · Bots for Slack & Lark/Feishu · Skills are infrastructure, not prompts.项目地址: https://gitcode.com/GitHub_Trending/re/refly

在AI技术快速发展的今天，企业面临的核心挑战已从模型能力转向智能体技能的标准化与规模化。传统AI工作流工具如n8n、Dify虽然提供了可视化编排能力，但存在黑盒执行、难以复用和缺乏治理的痛点。Refly作为首个开源的智能体技能构建平台，通过创新的Vibe工作流和确定性技能基础设施，帮助企业将业务流程转化为可版本控制、可部署的智能体能力。

技术架构解析：从意图到执行的确定性转换

核心设计哲学：技能即基础设施

Refly采用技能优先（Skills-first）架构，将传统的工作流概念升级为原子化技能。每个技能都是独立的、可版本控制的执行单元，支持跨平台部署。这种设计解决了传统AI工作流的三个关键问题：

1. 确定性执行：通过状态机确保每次执行结果一致
2. 可干预运行时：支持执行过程中的暂停、审计和重定向
3. 统一交付层：技能可导出为API、Webhook或MCP工具

架构组件深度剖析

工作流引擎核心模块

Refly的工作流引擎基于NestJS构建，采用微服务架构设计。核心模块位于apps/api/src/modules/workflow/目录：

// 工作流执行状态管理 export const WORKFLOW_EXECUTION_CONSTANTS = { POLL_INTERVAL_MS: 1500, // 轮询间隔 EXECUTION_TIMEOUT_MS: 30 * 60 * 1000, // 工作流执行超时 NODE_EXECUTION_TIMEOUT_MS: 30 * 60 * 1000, // 节点执行超时 POLL_LOCK_TTL_MS: 5000 // 分布式锁TTL };

工作流服务WorkflowService实现了基于BullMQ的队列处理机制，支持并发执行和故障恢复：

@Injectable() export class WorkflowService { constructor( @InjectQueue(QUEUE_RUN_WORKFLOW) private readonly runWorkflowQueue?: Queue<RunWorkflowJobData>, @InjectQueue(QUEUE_POLL_WORKFLOW) private readonly pollWorkflowQueue?: Queue<PollWorkflowJobData>, ) {} // 异步执行工作流 async executeWorkflow(workflowId: string, input: any) { const executionId = genWorkflowExecutionID(); await this.runWorkflowQueue.add('run-workflow', { workflowId, input, executionId }); return { executionId, status: 'running' }; } }

技能模板系统

技能模板位于packages/skill-template/src/，采用面向对象设计模式：

export abstract class BaseSkill { abstract name: string; abstract description: string; abstract configSchema: SkillTemplateConfigDefinition; abstract graphState: StateGraphArgs<BaseSkillState>['channels']; // 转换为LangChain可执行单元 abstract toRunnable(): Runnable; // 事件驱动架构 emitEvent(data: Partial<SkillEvent>, config: SkillRunnableConfig) { const { emitter } = config?.configurable || {}; if (emitter) { // 发送技能执行事件 emitter.emit('skill-event', eventData); } } }

可视化界面与用户体验

Refly采用双模态设计：可视化画布与自然语言描述（Vibe模式）并存，满足不同用户需求。

Refly工作台提供自然语言创建工作流的能力，支持预设模板和自定义需求

工作台界面采用左侧导航+中央操作区布局：

• 左侧导航：工作区、工作流管理、模板库、市场、运行历史
• 中央操作区：自然语言输入框，支持"Build a podcast generation workflow"等复杂需求
• 预设示例：提供播客生成、设计素材自动化等典型用例

部署配置与性能优化

Docker容器化部署

Refly提供完整的Docker Compose部署方案，支持单机到生产环境：

# deploy/docker/docker-compose.yml name: refly include: - path: docker-compose.base.yml services: api: extends: file: docker-compose.services.yml service: api image: reflyai/refly-api:latest web: extends: file: docker-compose.services.yml service: web image: reflyai/refly-web:latest ports: - 5700:80

性能调优参数

数据库连接池配置

// apps/api/src/utils/prisma.ts export const prisma = new PrismaClient({ log: process.env.NODE_ENV === 'development' ? ['query', 'info', 'warn', 'error'] : ['error'], datasources: { db: { url: process.env.DATABASE_URL, }, }, // 连接池优化 connection: { pool: { max: 20, // 最大连接数 min: 5, // 最小连接数 acquire: 30000, // 获取连接超时 idle: 10000 // 空闲连接超时 } } });

Redis缓存策略

// 工作流状态缓存 const WORKFLOW_CACHE_TTL = 300; // 5分钟 const NODE_EXECUTION_CACHE_TTL = 600; // 10分钟 // 分布式锁配置 const LOCK_TTL_MS = 5000; const LOCK_RETRY_DELAY_MS = 100;

监控与可观测性

Refly集成Langfuse进行执行追踪，支持OpenTelemetry标准：

# deploy/docker/trace/docker-compose.yml services: alloy: image: grafana/alloy:latest command: ["run", "/etc/alloy/alloy-local-config.alloy"] ports: - "12345:12345" # Alloy metrics - "4317:4317" # OTLP gRPC - "4318:4318" # OTLP HTTP

企业级集成方案

多平台导出能力

Refly支持将技能导出到多种运行时环境：

Refly支持将确定性技能导出到Claude Code、Cursor、Lovable、Slack等多种平台

API集成示例

// 调用工作流API async function executeWorkflow(workflowId: string, input: any) { const response = await fetch( `${API_BASE}/api/v1/workflows/${workflowId}/execute`, { method: 'POST', headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ input }) } ); const { executionId, status } = await response.json(); // 轮询执行状态 while (status === 'running') { await new Promise(resolve => setTimeout(resolve, 1500)); const statusResponse = await fetch( `${API_BASE}/api/v1/executions/${executionId}`, { headers: { 'Authorization': `Bearer ${API_KEY}` } } ); const result = await statusResponse.json(); status = result.status; } return result; }

Webhook配置

# Lark/Feishu Webhook集成 triggers: - type: webhook config: url: https://your-refly-instance.com/api/v1/webhooks/lark events: - message.receive verification: type: signature secret: ${WEBHOOK_SECRET}

技能注册中心

技能注册中心位于packages/skill-template/src/skills/，支持技能发现、版本管理和依赖解析：

// 技能元数据定义 export interface SkillMetadata { id: string; name: string; version: string; description: string; author: string; dependencies: SkillDependency[]; configSchema: JSONSchema7; runtimeRequirements: RuntimeRequirement[]; } // 技能依赖管理 export class SkillRegistry { private skills: Map<string, SkillMetadata> = new Map(); async register(skill: BaseSkill): Promise<void> { const metadata = this.extractMetadata(skill); await this.validateDependencies(metadata); this.skills.set(metadata.id, metadata); } async resolve(id: string, version?: string): Promise<BaseSkill> { const metadata = this.findSkill(id, version); return this.instantiateSkill(metadata); } }

高级功能与最佳实践

可干预运行时设计

Refly的可干预运行时允许在执行过程中进行人工干预：

// 工作流执行控制 class IntervenableRuntime { private state: WorkflowState; private breakpoints: Set<string> = new Set(); async executeWithIntervention( workflow: Workflow, onBreakpoint?: (state: WorkflowState) => Promise<void> ): Promise<ExecutionResult> { while (!this.state.isCompleted) { const nextNode = this.getNextNode(); // 检查断点 if (this.breakpoints.has(nextNode.id)) { await onBreakpoint?.(this.state); } // 执行节点 const result = await this.executeNode(nextNode); this.state.update(result); // 检查是否需要人工干预 if (result.requiresIntervention) { await this.requestHumanInput(result); } } return this.state.toResult(); } }

技能版本控制

// 技能版本管理 export class SkillVersionManager { private versions: Map<string, SkillVersion[]> = new Map(); async publish( skillId: string, version: string, skill: BaseSkill ): Promise<void> { const skillVersions = this.versions.get(skillId) || []; // 语义化版本检查 if (!this.isValidVersion(version)) { throw new Error(`Invalid version: ${version}`); } // 依赖冲突检测 const conflicts = await this.checkDependencyConflicts(skill); if (conflicts.length > 0) { throw new Error(`Dependency conflicts: ${conflicts.join(', ')}`); } skillVersions.push({ version, skill, publishedAt: new Date(), metadata: this.extractSkillMetadata(skill) }); this.versions.set(skillId, skillVersions); } }

性能优化策略

批量处理与流式响应

// 流式工作流执行 export class StreamingWorkflowExecutor { async executeStreaming( workflowId: string, input: any, onProgress?: (progress: ExecutionProgress) => void ): Promise<AsyncIterable<WorkflowChunk>> { const execution = await this.startExecution(workflowId, input); return { async *[Symbol.asyncIterator]() { let isCompleted = false; while (!isCompleted) { const status = await this.pollStatus(execution.id); // 发送进度更新 onProgress?.({ percentage: status.progress, currentNode: status.currentNode, elapsedTime: status.elapsedTime }); // 发送增量结果 if (status.newOutputs) { for (const output of status.newOutputs) { yield output; } } isCompleted = status.status === 'completed' || status.status === 'failed'; await new Promise(resolve => setTimeout(resolve, 1000)); } } }; } }

缓存策略优化

// 智能缓存层 export class SmartCacheLayer { private cache: Map<string, CacheEntry> = new Map(); async getOrCompute<T>( key: string, compute: () => Promise<T>, options: CacheOptions = {} ): Promise<T> { const entry = this.cache.get(key); // 检查缓存有效性 if (entry && !this.isExpired(entry, options.ttl)) { return entry.value as T; } // 计算新值 const value = await compute(); // 更新缓存 this.cache.set(key, { value, timestamp: Date.now(), ttl: options.ttl || DEFAULT_TTL }); return value; } // 基于工作流特征的缓存键生成 generateCacheKey(workflowId: string, input: any): string { const inputHash = this.hashInput(input); return `workflow:${workflowId}:input:${inputHash}`; } }

故障排除与常见问题

性能问题诊断

工作流执行缓慢

可能原因：

1. 数据库连接池配置不当
2. Redis缓存未命中率过高
3. 外部API调用超时

解决方案：

# 监控数据库性能 docker-compose exec postgres pg_stat_statements # 检查Redis命中率 redis-cli info stats | grep -E "(keyspace_hits|keyspace_misses)" # 调整工作流超时设置 export WORKFLOW_EXECUTION_TIMEOUT_MS=3600000 export WORKFLOW_NODE_EXECUTION_TIMEOUT_MS=600000

内存泄漏排查

// 启用内存监控 import * as v8 from 'v8'; export class MemoryMonitor { private intervals: NodeJS.Timeout[] = []; startMonitoring(intervalMs: number = 60000): void { const interval = setInterval(() => { const heapStats = v8.getHeapStatistics(); const memoryUsage = process.memoryUsage(); console.log({ timestamp: new Date().toISOString(), heapTotal: heapStats.total_heap_size, heapUsed: heapStats.used_heap_size, rss: memoryUsage.rss, external: memoryUsage.external }); }, intervalMs); this.intervals.push(interval); } }

集成问题处理

API调用失败

常见错误：

• 401 Unauthorized：API密钥无效或过期
• 429 Too Many Requests：请求频率超限
• 502 Bad Gateway：后端服务异常

调试步骤：

# 检查API服务状态 curl -X GET http://localhost:3000/health # 验证API密钥 curl -H "Authorization: Bearer YOUR_API_KEY" \ http://localhost:3000/api/v1/auth/verify # 查看详细日志 docker-compose logs -f api

Webhook配置问题

配置验证：

// Webhook签名验证 export class WebhookValidator { static verifySignature( payload: string, signature: string, secret: string ): boolean { const expectedSignature = crypto .createHmac('sha256', secret) .update(payload) .digest('hex'); return crypto.timingSafeEqual( Buffer.from(signature), Buffer.from(expectedSignature) ); } }

应用场景与最佳实践

企业SOP自动化

场景：客户支持工单处理流程

// 客户支持工作流定义 class CustomerSupportWorkflow extends BaseSkill { name = 'customer-support-ticket'; description = '自动化处理客户支持工单'; configSchema = { type: 'object', properties: { priority: { type: 'string', enum: ['low', 'medium', 'high'] }, category: { type: 'string' }, autoEscalate: { type: 'boolean', default: false } } }; async toRunnable(): Runnable { return new CustomerSupportRunnable(this.config); } } // 执行流程 1. 接收工单 → 2. 分类与优先级评估 → 3. AI初步回复 → 4. 人工审核（可干预）→ 5. 发送解决方案 → 6. 满意度调查

内容生成流水线

场景：社交媒体内容批量生成

# 内容生成工作流配置 workflow: id: social-media-generator version: 1.0.0 nodes: - id: topic-research type: web-search config: query: "{{input.topic}} latest trends" max_results: 10 - id: content-outline type: llm config: model: gpt-4 prompt: "基于搜索结果生成内容大纲" inputs: ["topic-research.results"] - id: generate-content type: llm config: model: claude-3-opus prompt: "根据大纲生成完整内容" inputs: ["content-outline.outline"] - id: format-optimization type: llm config: model: gpt-4 prompt: "优化内容格式为社交媒体发布格式" inputs: ["generate-content.content"]

数据ETL处理

场景：多源数据聚合分析

// 数据ETL技能实现 class DataETLWorkflow extends BaseSkill { async execute(input: ETLInput): Promise<ETLResult> { // 阶段1：数据提取 const rawData = await this.extractFromSources(input.sources); // 阶段2：数据转换 const transformedData = await this.transformData(rawData, input.rules); // 阶段3：数据加载 const result = await this.loadToDestination( transformedData, input.destination ); // 阶段4：质量检查（可干预） if (input.enableQualityCheck) { const qualityReport = await this.qualityCheck(result); if (qualityReport.issues.length > 0) { await this.requestHumanReview(qualityReport); } } return result; } }

未来规划与技术路线图

即将推出的功能

1. 技能市场增强：支持技能发现、评分和社区贡献
2. 性能优化：引入WebAssembly运行时，提升执行效率
3. 安全增强：基于角色的访问控制（RBAC）和审计日志

技术演进方向

• 边缘计算支持：在边缘设备上运行轻量级技能
• 联邦学习集成：支持分布式模型训练和更新
• 多模态扩展：增强图像、音频处理能力

生态建设

• 插件市场：第三方开发者可贡献技能插件
• 标准协议：推动智能体技能标准化
• 教育培训：提供技能开发认证体系

总结

Refly通过技能即基础设施的设计理念，解决了企业AI应用中的关键痛点。其可干预运行时确保业务合规性，统一交付层实现跨平台部署，版本控制保障技能稳定性。相比传统工作流工具，Refly提供了更符合企业需求的生产就绪解决方案。

对于技术团队而言，Refly的价值在于：

1. 降低集成成本：一次开发，多平台部署
2. 提升运维效率：集中化的技能管理和监控
3. 保障业务连续性：确定性的执行和故障恢复机制
4. 加速创新周期：快速迭代和测试新技能

随着AI智能体生态的成熟，Refly的技能基础设施定位将使其成为企业AI转型的关键技术栈。通过开源模式和活跃的社区贡献，Refly正在构建下一代企业级AI应用的标准范式。

【免费下载链接】reflyThe first open-source agent skills builder. Define skills by vibe workflow, run on Claude Code, Cursor, Codex & more. Build Clawdbot 🦞· APIs for Lovable · Bots for Slack & Lark/Feishu · Skills are infrastructure, not prompts.项目地址: https://gitcode.com/GitHub_Trending/re/refly