人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆【免费下载链接】voltagentAI Agent Engineering Platform built on an Open Source TypeScript AI Agent Framework项目地址https://gitcode.com/gh_mirrors/vo/voltagent点击查看免费下载本文围绕 VoltAgentTypeScript 开源 AI Agent 框架中工作流Workflow的核心抽象——execute函数展开系统讲解每个工作流步骤收到的上下文对象context中data、state、workflowState、getStepData、suspend、resumeData、retryCount等全部属性并结合仓库源码验证其底层实现。读完本文你将能够编写跨步骤共享数据、挂起等待人工审批、自动重试、条件分支等真实场景下的工作流步骤并理解这些能力在框架内部的执行机制。快速开始每个步骤的execute函数在 VoltAgent 的工作流框架中每个工作流步骤step都拥有一个execute函数它接收一个上下文对象context object用来处理数据、访问工作流状态并控制执行流程。这是整个工作流中最核心的 API.andThen({ id: my-step, execute: async ({ data, state, workflowState, setWorkflowState, getStepData, suspend, resumeData, retryCount, }) { // Your logic here return { result: processed }; } })提示使用VoltAgent实例注册你的工作流这样挂起/恢复suspend/resume以及 REST 路由才能定位到对应的执行记录import { VoltAgent } from voltagent/core; new VoltAgent({ workflows: { myWorkflow } });从源码层面看VoltAgent的构造函数会把这些工作流注册进内部的WorkflowRegistry。其registerWorkflows方法见 packages/core/src/voltagent.ts会遍历传入的工作流对象若传入的是WorkflowChain实例带toWorkflow方法先转换为标准Workflow再应用默认内存Memory并注册。这意味着工作流一旦注册其执行记录、挂起检查点checkpoint等都可以通过框架的注册中心被检索到为 suspend/resume、REST 路由和重启restart能力提供了基础设施。上下文对象里到底有什么execute函数只接收一个参数——上下文对象。它的 TypeScript 定义可以在源码中找到完整形态interface ExecuteContextTData, TSuspendData any, TResumeData any { data: TData; state: WorkflowState; workflowState: Recordstring, unknown; setWorkflowState: ( update: | Recordstring, unknown | ((previous: Recordstring, unknown) Recordstring, unknown) ) void; getStepData: (stepId: string) { input: any; output: any } | undefined; suspend: (reason?: string, data?: TSuspendData) Promisenever; resumeData?: TResumeData; retryCount?: number; }这一接口在框架内部对应WorkflowExecuteContext见 packages/core/src/workflow/internal/types.ts除了文档中列出的属性外源码中还补充暴露了getStepResult、getInitData、bail、abort、logger、writer等进阶工具后文会逐一展开。下面我们按属性逐个深入。1.data—— 步骤的输入数据data是流入当前步骤的数据第一个步骤拿到的是工作流的初始输入workflows initial input后续步骤拿到的是前一个步骤的返回值outputconst workflow createWorkflowChain({ input: z.object({ name: z.string() }), // ... }) .andThen({ id: step-1, execute: async ({ data }) { console.log(data.name); // Original input return { ...data, step1: done }; }, }) .andThen({ id: step-2, execute: async ({ data }) { console.log(data.name); // Still there! console.log(data.step1); // done - from previous step return { ...data, step2: also done }; }, });从实现上看data正是工作流状态机中state.data的透传。工作流核心循环在每个步骤执行后都会调用stateManager.update({ data: result, result })把步骤返回值写回状态见 packages/core/src/workflow/core.ts下一次迭代时又通过createStepExecutionContext(stateManager.state.data, ...)把它作为下一个步骤的data传入见 packages/core/src/workflow/core.ts。因此data的“接力传递”天然支持展开合并spread merge模式例如return { ...data, step1: done }让早期步骤的字段在整个链路中保持可见。2.state—— 工作流执行元信息state包含当前工作流执行execution的元数据.andThen({ id: log-info, execute: async ({ data, state }) { console.log(state.executionId); // Unique ID for this run console.log(state.userId); // Whos running it console.log(state.conversationId); // Conversation context console.log(state.input); // Original workflow input console.log(state.startAt); // When it started // context is a Map for custom data const userRole state.context?.get(role); return data; } })这些字段与运行选项WorkflowRunOptions一一对应executionId默认为uuidv4userId、conversationId、context都来自调用run(input, options)时传入的选项。底层状态结构定义在 packages/core/src/workflow/internal/state.ts包含executionId、conversationId、userId、context、active、startAt、endAt、status、input、data、workflowState、result、error、suspension、cancellation、usage等字段。state.context是一个Mapstring | symbol, unknown专门用于存放自定义上下文数据。关于跨步骤共享状态可进一步阅读 Workflow State 文档。3.workflowState与setWorkflowState—— 共享状态当需要在步骤之间共享上下文、但又不想改动主数据载荷main data payload时使用共享状态.andThen({ id: stash-user, execute: async ({ data, setWorkflowState }) { setWorkflowState((previous) ({ ...previous, userId: data.userId, })); return data; } })setWorkflowState接受两种形式的更新直接对象Recordstring, unknown或基于前值计算的函数(previous) next。底层实现见 packages/core/src/workflow/core.ts更新器会读取当前state.workflowState计算nextState同步写入步骤上下文、执行上下文和状态管理器保证同一个执行内部所有步骤读到一致的共享状态快照。4.getStepData—— 访问任意历史步骤的数据获取任意一个已经执行过的步骤的数据.andThen({ id: combine-results, execute: async ({ data, getStepData }) { // Get data from a specific step const step1Data getStepData(step-1); if (step1Data) { console.log(step1Data.input); // What went INTO step-1 console.log(step1Data.output); // What came OUT of step-1 } return data; } })在源码中getStepData(stepId)的实现是executionContext?.stepData.get(stepId)见 packages/core/src/workflow/internal/utils.ts。executionContext.stepData是一个Mapstring, WorkflowStepData在执行循环中每个步骤完成后会写入{ input, output, status, error }快照见 packages/core/src/workflow/core.ts。也就是说getStepData返回的input/output是“步骤级别”的数据快照即便工作流继续推进历史步骤的数据仍然可查。返回类型是{ input: any; output: any } | undefined因此访问前应先做存在性判断。此外源码还额外提供了getStepResult(stepId)返回步骤的output无输出时返回null和getInitData()返回工作流的初始输入前者适合“只需要某个步骤的结果、不关心输入”的场景。5.suspend—— 挂起工作流暂停执行并等待外部输入例如人工审批.andThen({ id: wait-for-approval, execute: async ({ data, suspend }) { if (data.amount 1000) { // This stops execution immediately await suspend(Manager approval required); // Code below never runs during suspension } return { ...data, approved: true }; } })suspend的签名是(reason?: string, data?: TSuspendData) Promisenever返回类型never意味着调用后函数不会正常返回——挂起会立即中止当前步骤的执行。从实现看andThen包装器见 packages/core/src/workflow/steps/and-then.ts会捕获挂起信号当execute抛出消息为WORKFLOW_SUSPENDED的错误时会原样向上抛出而非当作错误处理工作流核心的 catch 分支见 packages/core/src/workflow/core.ts识别到该信号后调用handleStepSuspension将状态置为suspended并保存挂起检查点checkpoint整个执行结果以status: suspended返回。这也是为什么挂起点之后的代码永远不会执行。6.resumeData—— 恢复时携带的数据当挂起的工作流恢复执行时resumeData包含恢复时传入的数据.andThen({ id: approval-step, execute: async ({ data, suspend, resumeData }) { // Check if were resuming if (resumeData) { // Were resuming! Use the approval decision return { ...data, approved: resumeData.approved, approvedBy: resumeData.managerId }; } // First time through - suspend for approval if (data.amount 1000) { await suspend(Needs approval); } // Auto-approve small amounts return { ...data, approved: true, approvedBy: auto }; } })关键在于恢复时挂起所在的步骤会从头重新执行并且此时resumeData可用。源码中恢复数据的注入有严格的条件——isResumingThisStep options?.resumeFrom index startStepIndex resumeInputData ! undefined见 packages/core/src/workflow/core.ts只有“被挂起的那个步骤”在恢复时会拿到resumeInputData即通过resume(input)传入的、经过resumeSchema校验的数据。这保证了不会把恢复数据错误地注入到其他步骤。这也是execute内部最常见的“双路径”模式先检查resumeData处理恢复分支再判断是否需要挂起。7.retryCount—— 重试次数如果某个步骤配置了retries该值表示当前是第几次尝试.andThen({ id: fetch-user, retries: 2, execute: async ({ data, retryCount }) { console.log(Attempt:, retryCount); // 0, 1, 2 return await fetchUser(data.userId); } })retryCount同样会在工作流级别启用重试时递增const workflow createWorkflowChain({ id: retry-defaults, retryConfig: { attempts: 2, delayMs: 250 }, }).andThen({ id: fetch-user, execute: async ({ data, retryCount }) { console.log(Attempt:, retryCount); return fetchUser(data.userId); }, });retryCount从 0 开始计数因此配置retries: 2时日志依次输出0, 1, 2。源码中的重试循环见 packages/core/src/workflow/core.ts每个步骤的重试上限stepRetryLimit优先取步骤自身的step.retries未配置时回退到工作流级retryConfig.attempts见 packages/core/src/workflow/core.ts步骤抛错后若retryCount stepRetryLimit则retryCount 1并进入下一次尝试若配置了delayMs还会先等待对应毫秒数期间可响应取消/挂起信号。retryCount通过createStepExecutionContext的最后一个参数注入步骤上下文见 packages/core/src/workflow/internal/utils.ts并在每次尝试前更新 OpenTelemetry 步长 span 的workflow.step.retry.count属性便于观测。完整示例订单处理工作流下面是一个使用全部上下文属性的真实示例——订单处理流程包含校验、库存检查、支付审批支持人工审批挂起和发货import { createWorkflowChain } from voltagent/core; import { z } from zod; const orderWorkflow createWorkflowChain({ id: order-processor, name: Order Processing, input: z.object({ orderId: z.string(), amount: z.number(), items: z.array(z.string()), }), result: z.object({ status: z.string(), trackingNumber: z.string(), }), }) .andThen({ id: validate-order, execute: async ({ data, state }) { console.log(Processing order ${data.orderId} for user ${state.userId}); const isValid data.items.length 0 data.amount 0; return { ...data, isValid }; }, }) .andThen({ id: check-inventory, execute: async ({ data, getStepData }) { // Only check if validation passed const validation getStepData(validate-order); if (!validation?.output?.isValid) { return { ...data, inStock: false }; } // Check inventory for each item const inStock await checkInventory(data.items); return { ...data, inStock }; }, }) .andThen({ id: approve-payment, execute: async ({ data, suspend, resumeData }) { // Handle resume from suspension if (resumeData) { return { ...data, paymentApproved: resumeData.approved, approvedBy: resumeData.approver, }; } // Auto-approve small amounts if (data.amount 100) { return { ...data, paymentApproved: true, approvedBy: auto }; } // Suspend for manual approval await suspend(Payment approval needed for $${data.amount}); }, }) .andThen({ id: ship-order, execute: async ({ data, state }) { if (!data.paymentApproved) { return { status: cancelled, trackingNumber: N/A, }; } // Ship the order const tracking await createShipment(data.orderId); // Log completion console.log(Order ${data.orderId} shipped after ${Date.now() - state.startAt.getTime()}ms); return { status: shipped, trackingNumber: tracking, }; }, });观察这个示例的典型模式validate-order通过data读取输入、通过state.userId获取执行者信息并采用展开合并返回新对象check-inventory通过getStepData(validate-order)回读上一步的output.isValid实现步骤间的条件依赖approve-payment是标准的双路径挂起模式resumeData优先处理恢复分支小金额自动通过大金额suspend等待人工ship-order用state.startAt计算端到端耗时并依据前面步骤累积的数据决定发货或取消。不同类型的步骤谁在接收上下文execute上下文并非andThen独有多种步骤类型都会收到上下文或其变体。步骤类型常量定义在 packages/core/src/workflow/steps/types.ts共 14 种agent、func、tap、workflow、guardrail、conditional-when、parallel-all、parallel-race、sleep、sleep-until、foreach、loop、branch、map。基础步骤andThen转换数据或执行操作.andThen({ id: calculate-total, execute: async ({ data }) { const total data.items.reduce((sum, item) sum item.price, 0); return { ...data, total }; } })andThen生成的步骤类型为func见 packages/core/src/workflow/steps/and-then.ts并支持inputSchema、outputSchema、suspendSchema、resumeSchema四个可选 schema 用于运行时校验。此外它还保留了originalExecute的引用用于步骤序列化/反序列化场景。AI Agent 步骤andAgenttask函数同样能拿到上下文.andAgent( async ({ data }) Summarize this order: ${JSON.stringify(data.items)}, myAgent, { schema: z.object({ summary: z.string() }) } )andAgent生成的步骤类型为agent见 packages/core/src/workflow/steps/and-agent.ts。当task是函数时框架会先调用task(context)得到最终的提示词字符串见 packages/core/src/workflow/steps/and-agent.tsschema也可以是一个接收上下文的函数用于动态决定结构化输出格式。条件步骤andWhen仅在条件满足时执行.andWhen({ id: apply-discount, condition: async ({ data }) data.total 50, step: andThen({ id: discount, execute: async ({ data }) ({ ...data, total: data.total * 0.9, discountApplied: true }) }) })andWhen生成conditional-when步骤见 packages/core/src/workflow/steps/and-when.ts先执行condition(context)判断为真才执行嵌套的step条件为假时嵌套步骤会被跳过核心循环通过“输出等于输入”来识别跳过状态并标记为skipped见 packages/core/src/workflow/core.ts。副作用步骤andTap运行代码但不改变数据.andTap({ id: send-notification, execute: async ({ data, state }) { await sendEmail(state.userId, Order ${data.orderId} processed); // Return value ignored - data passes through } })andTap生成tap步骤见 packages/core/src/workflow/steps/and-tap.ts。从类型定义看packages/core/src/workflow/steps/types.tstap 步骤的输出类型恒为输入类型DATA即返回值被忽略、数据原样透传——适合日志、通知、埋点等纯副作用场景。Suspend Resume 深度解析对于 human-in-the-loop人机协作类工作流挂起是关键能力.andThen({ id: review-step, execute: async ({ data, suspend, resumeData }) { // Step 1: Check if were resuming if (resumeData) { console.log(Resuming with:, resumeData); return { ...data, reviewed: true, reviewer: resumeData.userId }; } // Step 2: Check if we need to suspend if (data.requiresReview) { // This immediately stops execution await suspend(Document needs review, { documentId: data.id, reason: High risk score }); // Never reaches here during suspension } // Step 3: Continue if no suspension needed return { ...data, reviewed: true, reviewer: auto }; } })重要恢复时被挂起的步骤会从开头重新执行此时resumeData可用。从框架实现看挂起/恢复链路如下suspend(reason, data)内部通过typedSuspendFn触发挂起信号最终抛出WORKFLOW_SUSPENDED错误Promisenever的类型签名正是这一设计的外在体现核心循环捕获该信号packages/core/src/workflow/core.ts调用handleStepSuspension状态管理器将执行状态置为suspended记录suspendedAt、reason、suspendedStepIndex、lastEventSequence等挂起元数据packages/core/src/workflow/internal/state.ts并保存 checkpoint 到 Memory恢复时调用执行结果上的resume(input, options)框架从 checkpoint 读取resumeStepIndex从该步骤重新开始并把校验过的resumeData注入packages/core/src/workflow/core.ts。关于挂起/恢复的更多模式参见 Suspend Resume 文档。最佳实践1. 始终返回新对象// ✅ Good - creates new object return { ...data, processed: true }; // ❌ Bad - mutates existing object data.processed true; return data;原因在本文data一节已经点明工作流状态机会把步骤返回值写回state.data并作为下一步的data传入。直接修改并返回原对象会引入共享引用导致后续步骤拿到被污染的数据且不利于 checkpoint 持久化与时间旅行time-travel重放的可信度。2. 先检查步骤数据是否存在const previousStep getStepData(step-id); if (previousStep) { // Safe to use previousStep.output }因为getStepData的返回类型包含undefined目标步骤可能尚未执行或被跳过访问.output前务必判空。3. 使用清晰、描述性的步骤 ID// ✅ Good - descriptive id: validate-payment; // ❌ Bad - unclear id: step2;步骤 ID 是getStepData、恢复定位resume(..., { stepId })、时间旅行重放timeTravel({ stepId })以及日志/追踪OpenTelemetry span 的stepId属性的索引键。从源码看步骤 ID 还被持久化到执行检查点与历史记录中清晰的 ID 直接影响可观测性与排障效率。4. 优雅地处理错误execute: async ({ data }) { try { const result await riskyOperation(data); return { ...data, result }; } catch (error) { return { ...data, error: error.message, success: false }; } };注意被suspend触发的WORKFLOW_SUSPENDED错误和取消信号WORKFLOW_CANCELLED会被框架识别为控制流信号而非业务错误packages/core/src/workflow/steps/and-then.ts所以不建议在execute中吞掉所有异常以免干扰挂起/取消机制。5. 记录关键事件execute: async ({ data, state }) { console.log([${state.executionId}] Processing ${data.id}); const result await process(data); console.log([${state.executionId}] Completed with status: ${result.status}); return result; };state.executionId是贯穿整个执行的唯一 ID将其放进日志前缀可以方便地把一次执行的多条日志串联起来如果使用框架的loggerexecute 上下文中可直接访问它还会自动带上userId、conversationId、executionId等执行级上下文见 packages/core/src/workflow/context.ts。TypeScript 类型安全execute函数是完全类型安全的interface ExecuteContextTData, TSuspendData any, TResumeData any { data: TData; state: WorkflowState; workflowState: Recordstring, unknown; setWorkflowState: ( update: | Recordstring, unknown | ((previous: Recordstring, unknown) Recordstring, unknown) ) void; getStepData: (stepId: string) { input: any; output: any } | undefined; suspend: (reason?: string, data?: TSuspendData) Promisenever; resumeData?: TResumeData; retryCount?: number; }类型会沿着工作流自动流转——TypeScript 能够在每个步骤知道当前可用的data形态。这是通过createWorkflow/createWorkflowChain的泛型重载实现的框架针对 1 到 20 个步骤逐一声明了函数重载见 packages/core/src/workflow/core.ts每个重载都用前一步的输出类型S1、S2……作为下一步的输入类型约束从而让 IDE 在编写第 N 个步骤时能精确提示data的字段。配合 zod 的input/result/suspendSchema/resumeSchema声明输入、输出、挂起数据与恢复数据的类型都会被静态校验从编译期就杜绝字段拼写错误。总结VoltAgent 的 Execute Function API 是整个工作流引擎的“心脏”一个统一且类型安全的上下文对象把步骤输入data、执行元信息state、共享状态workflowState/setWorkflowState、历史步骤访问getStepData、人机协作挂起suspend/resumeData与弹性重试retryCount全部收敛到execute函数内部。无论你编写的是纯函数步骤andThen、AI Agent 步骤andAgent、条件分支andWhen还是副作用步骤andTap这套上下文契约都保持一致。理解了它背后的状态机、信号机制与 checkpoint 持久化原理你就能设计出可靠、可观测、支持人工介入的复杂 AI Agent 工作流。赞分享人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆【免费下载链接】voltagentAI Agent Engineering Platform built on an Open Source TypeScript AI Agent Framework项目地址https://gitcode.com/gh_mirrors/vo/voltagent点击查看免费下载相关推荐get-shit-done 的 execute-phase 工作流基于波次的并行执行编排与安全门详解get shit done 的 execute phase 工作流基于波次的并行执行编排与安全门详解 本指南深入解析 get shit doneGSD中负人工智能AI 应用提示工程开发工具工作流自动化AI AgentGemini API 结构化输出与工具调用实战JSON Schema、Function Calling、搜索接地、代码执行与 URL 上下文Gemini API 结构化输出与工具调用实战JSON Schema、Function Calling、搜索接地、代码执行与 URL 上下文 导读 本文是 AAI 技能人工智能大模型Acey Ducey纸牌游戏拆解basic-computer-games中的经典概率游戏与原版隐藏BugAcey Ducey纸牌游戏拆解basic computer games中的经典概率游戏与原版隐藏Bug 在 basic computer games 开源项示例工程上一篇Detect It Easy 完整上手指南三分钟看懂任意文件的底细下一篇视频硬字幕去不掉本地免费的Video-subtitle-remover实测AI补画、无损分辨率还能批量去水印创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考