ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Cloudflare Workers 作为 AI Agent 运行时:边缘计算与多AI协作实践

Cloudflare Workers 作为 AI Agent 运行时:边缘计算与多AI协作实践

1. 从“cloudflare-os”这个标题说起:它到底想解决什么问题

第一次看到“cloudflare-os”这个标题,很多人会误以为Cloudflare要发布一款桌面操作系统,或者一个类似Linux发行版的东西。实际上,从Cloudflare近两年的技术布局和公开的工程实践来看,这个标题背后指向的是一个更务实的方向:把Cloudflare Workers当作一个轻量级的“运行时操作系统”,用它来承载AI Agent的调度、编排、状态管理和工具调用。换句话说,它不是传统意义上的OS,而是一套以边缘计算节点为底座、以Worker为执行单元、以AI Agent为“应用程序”的分布式运行环境。

这个思路为什么值得关注?因为传统AI Agent的部署方式通常依赖一台常驻服务器或容器,需要自己维护进程、管理内存、处理并发、配置网络。而Cloudflare Workers的模型是事件驱动、按需执行、全球分布、冷启动极快。把Agent跑在Workers上,意味着你不需要关心服务器在哪里、不需要预置容量、不需要为闲置时间付费。对于个人开发者和中小团队来说,这是一种极低门槛的Agent落地路径。

这篇文章适合三类人看:第一类是对AI Agent感兴趣但不想折腾服务器运维的开发者;第二类是想了解边缘计算与Agent结合方式的技术爱好者;第三类是在做多AI协作、Agent记忆、Agent安全相关项目,需要找一个稳定运行时的工程师。我会从整体设计思路、核心细节、实操过程、常见问题四个维度展开,尽量把每个选择背后的“为什么”讲清楚,让你看完能直接动手搭一个属于自己的Agent运行环境。

2. 整体设计与思路拆解:为什么把Agent跑在Workers上

2.1 传统Agent部署的痛点与边缘运行时的优势

先说说传统做法。你要跑一个AI Agent,通常需要一台云服务器,装好Python环境,拉取Agent框架(比如LangChain、AutoGen或者自己写的调度逻辑),然后启动一个常驻进程。这个进程要监听请求、维护对话状态、调用大模型API、执行工具函数。问题在于:服务器要一直开着,哪怕没有请求也在烧钱;并发一上来就要考虑扩容;全球用户访问时延迟不均匀;状态管理要么放内存(重启就丢),要么自己搭Redis或数据库。

Cloudflare Workers的模型完全不同。它没有“常驻进程”的概念,每个请求触发一次执行,执行完就释放。状态可以通过Durable Objects、KV、D1、R2这些存储服务来持久化。全球300多个节点意味着用户在哪里,代码就在离他最近的节点跑。冷启动时间在毫秒级,对于Agent这种“请求-推理-响应”的交互模式非常合适。你不需要管理任何服务器,只需要写Worker代码,部署上去,按请求量付费。

注意:Workers的执行时间有限制,免费版CPU时间10毫秒,付费版可到30秒甚至更长。对于需要长时间推理的Agent任务,需要把任务拆成多个Worker调用,或者用Durable Objects来维持长连接状态。

2.2 “OS”这个隐喻背后的架构分层

把Workers当OS用,需要自己补上OS该有的几个核心能力。我在实际搭建时把它分成四层:

  • 调度层:用Worker的fetch handler作为入口,根据请求类型分发给不同的Agent逻辑。这相当于OS的进程调度器。
  • 状态层:用Durable Objects存对话历史、Agent记忆、任务队列。Durable Objects有强一致性和事务能力,适合做Agent的“内存”。
  • 工具层:把外部API、数据库查询、文件操作封装成Worker可调用的函数。这相当于OS的系统调用。
  • 安全层:用Workers的Environment Variables存密钥,用CORS和Token校验控制访问。这相当于OS的权限管理。

这个分层不是拍脑袋想的,而是因为Workers本身只提供计算和网络能力,不提供状态和调度。你要跑Agent,就必须自己把这些补上。好处是每一层都可以独立替换和扩展,坏处是初期搭建比直接用一台服务器要麻烦一些。

2.3 选型对比:Workers vs 传统容器 vs 其他边缘平台

维度Cloudflare Workers传统容器(Docker+K8s)其他边缘函数平台
冷启动毫秒级秒级到分钟级通常百毫秒级
状态管理需搭配Durable Objects/KV可挂载卷,但需自己管理通常无内置状态方案
全球分布300+节点自动分布需自己部署多区域节点数较少
运维成本几乎为零需要专人维护较低但仍有配置
执行时长限制有,需拆分任务无硬限制各有不同
适合场景短任务、高并发、全球访问长任务、复杂依赖中等复杂度任务

我选Workers的核心原因是:Agent的交互模式通常是“用户发一句话,Agent推理几秒,返回结果”,这种短任务高并发的场景正好是Workers的强项。如果你的Agent需要跑几分钟的视频处理或者大规模爬虫,那还是老老实实用容器。

3. 核心细节解析与实操要点:把Agent跑起来的关键环节

3.1 Worker入口与Agent路由设计

一个Worker的入口就是一个fetch函数。你要做的第一件事是设计路由,让不同的请求打到不同的Agent逻辑上。我通常用URL路径来区分,比如/agent/chat走对话Agent,/agent/tool走工具调用Agent,/agent/memory走记忆管理。

export default { async fetch(request, env, ctx) { const url = new URL(request.url); const path = url.pathname; if (path === '/agent/chat') { return handleChat(request, env); } else if (path === '/agent/tool') { return handleTool(request, env); } else if (path === '/agent/memory') { return handleMemory(request, env); } return new Response('Not Found', { status: 404 }); } };

这个路由设计的好处是每个Agent逻辑独立,方便单独测试和部署。你可以在本地用wrangler dev启动开发服务器,然后逐个调试。

实操心得:路由路径不要用太深的层级,Workers的URL匹配是精确匹配,层级太深容易写错。我习惯用两级,第一级固定为/agent,第二级是功能名。

3.2 Durable Objects做Agent记忆的实操细节

Agent记忆是绕不开的问题。用户昨天说“我喜欢喝美式”,今天问“推荐什么咖啡”,Agent应该记得。用KV存的话,写入有延迟,读取可能读到旧值。用D1的话,每次读写都要走SQL,对于高频对话来说太重。Durable Objects是更好的选择,因为它有强一致性,而且每个Object是一个单线程执行的实例,天然适合做对话状态的串行处理。

创建一个Durable Object类:

export class AgentMemory { constructor(state, env) { this.state = state; this.env = env; } async fetch(request) { const url = new URL(request.url); const userId = url.searchParams.get('userId'); if (request.method === 'POST') { const body = await request.json(); let history = await this.state.storage.get(userId) || []; history.push(body.message); if (history.length > 50) history = history.slice(-50); await this.state.storage.put(userId, history); return new Response('OK'); } if (request.method === 'GET') { const history = await this.state.storage.get(userId) || []; return new Response(JSON.stringify(history)); } } }

然后在Worker里调用:

const id = env.AGENT_MEMORY.idFromName(userId); const stub = env.AGENT_MEMORY.get(id); const response = await stub.fetch('https://memory/get?userId=' + userId);

这里的关键点是idFromName,它保证同一个userId总是打到同一个Durable Object实例上,这样记忆就不会错乱。

注意:Durable Objects的存储有大小限制,单个Object建议不要超过128KB。如果对话历史很长,需要做摘要压缩,只保留最近几轮和关键信息。

3.3 工具调用的封装与安全控制

Agent要干活,就得调用外部工具。比如查天气、搜网页、发邮件。在Workers里,工具就是一个普通的异步函数。但要注意两点:一是密钥不能硬编码在代码里,要用env变量;二是工具调用要有超时和错误处理,不能让一个失败的调用卡死整个Agent。

async function callWeatherAPI(city, env) { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), 5000); try { const response = await fetch( `https://api.weather.com/v1/current?city=${city}&key=${env.WEATHER_API_KEY}`, { signal: controller.signal } ); clearTimeout(timeout); if (!response.ok) throw new Error('Weather API failed'); return await response.json(); } catch (e) { clearTimeout(timeout); return { error: 'Weather service unavailable' }; } }

超时设置5秒是因为Workers的CPU时间有限,如果外部API响应太慢,不如直接返回错误让Agent换一种方式回答。

实操心得:所有工具函数都要返回结构化数据,不要返回原始字符串。这样Agent在解析结果时不容易出错。我习惯返回{ success: boolean, data: any, error: string }这种格式。

3.4 多AI协作的编排模式

“多AI协作”是热词里频繁出现的一个方向。在Workers上实现多Agent协作,我试过两种模式:一种是串行流水线,Agent A的输出作为Agent B的输入;另一种是并行投票,多个Agent同时处理同一个问题,然后由一个汇总Agent选最佳答案。

串行流水线适合有明确步骤的任务,比如“先翻译再润色再格式化”。并行投票适合开放性问题,比如“写一个 slogan”,让三个Agent各写一个,然后选最好的。

async function pipeline(input, env) { const step1 = await callAgent('translator', input, env); const step2 = await callAgent('polisher', step1, env); const step3 = await callAgent('formatter', step2, env); return step3; } async function voting(input, env) { const [a, b, c] = await Promise.all([ callAgent('agentA', input, env), callAgent('agentB', input, env), callAgent('agentC', input, env) ]); return await callAgent('judge', { a, b, c }, env); }

并行投票的延迟取决于最慢的那个Agent,所以每个Agent的超时设置要一致。我通常设10秒,超过就返回已有结果。

4. 实操过程与核心环节实现:从零搭一个Agent运行环境

4.1 环境准备与项目初始化

你需要一个Cloudflare账号,安装Node.js和npm,然后安装Wrangler CLI。

npm install -g wrangler wrangler login

登录后会打开浏览器授权。然后创建项目:

npm create cloudflare@latest my-agent-os cd my-agent-os

选择“Hello World Worker”模板,语言选JavaScript或TypeScript。我建议用TypeScript,因为Agent逻辑复杂时类型检查能帮你避免很多低级错误。

接下来配置wrangler.toml:

name = "my-agent-os" main = "src/index.ts" compatibility_date = "2024-01-01" [durable_objects] bindings = [ { name = "AGENT_MEMORY", class_name = "AgentMemory" } ] [[migrations]] tag = "v1" new_classes = ["AgentMemory"] [vars] ENVIRONMENT = "production"

durable_objects的绑定和迁移配置是必须的,否则Durable Objects无法使用。

4.2 大模型API的接入与流式响应处理

Agent的核心是调用大模型。我通常用OpenAI兼容的API,因为切换模型方便。在Worker里调用大模型要注意:请求体可能很大,响应可能是流式的。

async function callLLM(prompt: string, env: Env): Promise<string> { const response = await fetch('https://api.openai.com/v1/chat/completions', { method: 'POST', headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${env.OPENAI_API_KEY}` }, body: JSON.stringify({ model: 'gpt-4o-mini', messages: [{ role: 'user', content: prompt }], stream: false }) }); if (!response.ok) { const error = await response.text(); throw new Error(`LLM API error: ${error}`); } const data = await response.json(); return data.choices[0].message.content; }

如果要流式响应,需要把stream设为true,然后用TransformStream把SSE数据转发给客户端。流式的好处是用户能更快看到部分结果,体验更好。但流式处理会增加代码复杂度,建议先用非流式跑通,再改成流式。

注意:API密钥一定要用wrangler secret put OPENAI_API_KEY来设置,不要写在wrangler.toml里。vars里的变量是明文存储的,只适合放非敏感配置。

4.3 Agent主循环的实现与状态流转

Agent的主循环是“接收输入 -> 调用LLM -> 解析工具调用 -> 执行工具 -> 把结果喂回LLM -> 生成最终回答”。这个循环在Worker里要控制好轮数,避免无限循环。

async function agentLoop(userInput: string, userId: string, env: Env): Promise<string> { const memoryId = env.AGENT_MEMORY.idFromName(userId); const memory = env.AGENT_MEMORY.get(memoryId); const historyResponse = await memory.fetch('https://memory/get?userId=' + userId); const history = await historyResponse.json(); let messages = [ { role: 'system', content: '你是一个有用的助手。' }, ...history, { role: 'user', content: userInput } ]; for (let i = 0; i < 5; i++) { const response = await callLLMWithTools(messages, env); if (response.toolCalls && response.toolCalls.length > 0) { for (const toolCall of response.toolCalls) { const result = await executeTool(toolCall.name, toolCall.arguments, env); messages.push({ role: 'tool', content: JSON.stringify(result) }); } } else { await memory.fetch('https://memory/post?userId=' + userId, { method: 'POST', body: JSON.stringify({ message: { role: 'assistant', content: response.content } }) }); return response.content; } } return '抱歉,我无法在限定步骤内完成这个任务。'; }

最多5轮是为了防止Agent陷入死循环。实际测试中,大部分任务2到3轮就能完成。

4.4 部署与自定义域名配置

部署很简单:

wrangler deploy

部署成功后会得到一个*.workers.dev的域名。如果要绑自定义域名,在Cloudflare控制台的Workers设置里添加Custom Domain,然后配置DNS记录。

实操心得:自定义域名建议用子域名,比如agent.yourdomain.com,不要用根域名。根域名通常已经绑了网站,改起来麻烦。另外,Workers的自定义域名需要域名也在Cloudflare托管,否则要改NS记录。

5. 常见问题与排查技巧实录

5.1 Worker执行超时与CPU时间限制

这是最常见的问题。免费版Worker的CPU时间限制是10毫秒,付费版是30秒。注意,CPU时间不等于墙钟时间。等待网络请求的时间不算CPU时间,但JSON解析、字符串处理、循环计算算CPU时间。

如果你发现Worker报“Worker exceeded CPU time limit”,说明你的代码里有大量同步计算。解决办法是把大计算拆成多个Worker调用,或者用Durable Objects的alarm机制异步处理。

问题现象可能原因解决方法
请求返回1102错误CPU时间超限拆分任务,减少同步计算
请求返回1101错误Worker抛出未捕获异常加try-catch,检查日志
请求返回1104错误子请求数量超限合并请求,减少fetch调用
Durable Object无响应Object被阻塞检查是否有死循环或长同步操作

5.2 Durable Objects的并发与一致性陷阱

Durable Objects是单线程执行的,这意味着同一个Object的请求是串行处理的。如果你在一个请求里做了很慢的操作,后面的请求就会排队。我踩过一次坑:在Object里调用了一个响应很慢的外部API,导致所有对话请求都卡住。

解决办法是把慢操作移到Worker里做,Object只负责读写状态。或者用blockConcurrencyWhile来初始化,但不要在里面做网络请求。

注意:Durable Objects的存储操作是异步的,但同一个Object内的存储操作有事务保证。如果你需要跨Object的事务,那就要自己实现两阶段提交,比较复杂。

5.3 大模型API调用的网络错误与重试策略

调用外部API时,网络抖动是常态。我遇到过OpenAI API返回429(限流)、500(服务器错误)、超时等各种情况。如果不做重试,用户体验会很差。

async function fetchWithRetry(url: string, options: RequestInit, maxRetries = 3): Promise<Response> { for (let i = 0; i < maxRetries; i++) { try { const response = await fetch(url, options); if (response.status === 429 || response.status >= 500) { await new Promise(r => setTimeout(r, Math.pow(2, i) * 1000)); continue; } return response; } catch (e) { if (i === maxRetries - 1) throw e; await new Promise(r => setTimeout(r, Math.pow(2, i) * 1000)); } } throw new Error('Max retries exceeded'); }

指数退避是标准做法,第一次等1秒,第二次等2秒,第三次等4秒。这样既能应对临时故障,又不会把请求堆积得太厉害。

5.4 密钥管理与环境变量安全

密钥泄露是Agent项目最常见的安全事故。我见过有人把API密钥直接写在Worker代码里然后推到GitHub,结果被人扫到,一夜之间跑掉几百美元。

正确的做法是:

  • 用wrangler secret put KEY_NAME设置密钥,密钥会加密存储。
  • 在代码里通过env.KEY_NAME读取。
  • 不要把.dev.vars文件提交到Git。
  • 定期轮换密钥,尤其是怀疑泄露时。

实操心得:我习惯给每个环境(开发、生产)用不同的密钥,这样开发环境的密钥泄露不会影响生产。另外,在Cloudflare控制台可以设置Usage Limits,防止意外超额。

5.5 Agent安全:输入校验与输出过滤

Agent安全是热词里反复出现的主题。你的Agent可能会被用户诱导执行危险操作,比如删除数据、发送垃圾邮件。防护措施包括:

  • 输入长度限制,防止超长输入消耗资源。
  • 工具调用白名单,只允许调用预定义的工具。
  • 输出过滤,检查是否包含敏感信息。
  • 速率限制,防止滥用。
function validateInput(input: string): boolean { if (input.length > 2000) return false; if (/<script|DROP TABLE|rm -rf/i.test(input)) return false; return true; }

这个校验很粗糙,但能挡住大部分低级攻击。更严格的方案是用另一个LLM来做安全审核,但那样会增加延迟和成本。

6. 从个人实践看Agent运行时的未来形态

我在过去半年里用Workers跑了三个不同规模的Agent项目,最大的一个每天处理约两万次对话。整体感受是:Workers作为Agent运行时的优势在于极低的运维成本和全球分布,劣势在于执行时长限制和状态管理的复杂度。如果你的Agent任务能在30秒内完成,且状态可以拆分成小块,那Workers是非常合适的选择。

一个值得尝试的扩展方向是把Agent的“技能”做成独立的Worker,通过Service Bindings互相调用。这样每个技能可以独立部署、独立扩缩容,整个系统就像一个微服务架构的OS。另一个方向是用Workers AI做本地推理,减少对外部API的依赖,但目前Workers AI的模型能力还有限,适合做简单的分类和摘要任务。

最后分享一个小技巧:在开发阶段,用wrangler tail实时查看日志,比在控制台翻日志快得多。部署后如果发现异常,第一时间wrangler tail,能看到请求的完整生命周期,包括CPU时间、子请求数、异常堆栈。这个命令我几乎每天都要用。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进