用 Cloudflare Agents 构建 Chat SDK 持久化状态agents/chat-sdk完整接入指南【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents导读本指南聚焦 Cloudflare Agents SDK 中的agents/chat-sdk模块当你在 Agent 内部运行 Chat SDK 时它把 Chat SDK 的StateAdapter接口转换为基于 Agents 子代理sub-agent的 Durable Object SQLite 存储实现为订阅、锁、队列、去重键、线程/频道状态、回调查元数据、转录列表与线程历史提供持久化后端。阅读本文后你将掌握从安装、基础接入、迁移配置到分片路由、自定义分片、API 用法与清理机制的完整落地方法并能在 Telegram、Slack、Discord、Teams、Google Chat 等任意 Chat SDK 适配器上复用同一套状态层。模块定位为什么需要agents/chat-sdkChat SDK 是一个跨平台的消息机器人运行时负责把不同即时通讯平台Telegram、Slack、Discord、Teams 等的 Webhook 归一化为统一的Thread与Message对象。为了让机器人状态订阅关系、并发锁、待发送队列、缓存、列表可持久化、可水平扩展Chat SDK 抽象出了StateAdapter接口允许接入不同的存储后端。agents/chat-sdk就是该接口在 Cloudflare Agents 生态中的官方实现位于 packages/agents/src/chat-sdk状态保存在Durable Object SQLite中天然具备持久化、单点强一致与低延迟读写的特性每个状态分片state shard对应一个ChatSdkStateAgent子代理挂在你的入口 Agentingress Agent之下子代理之间通过subAgent()机制路由分片粒度可自定义适配不同平台、不同租户、不同会话规模的流量模型。需要特别指出的是agents/chat-sdk本身不提供 messenger 适配器。它只负责状态层你需要配合任何 Chat SDK 官方适配器Telegram、Slack、Discord、Teams、Google Chat一起使用。从源码结构看该模块由四个文件组成入口与类型定义 index.ts、types.ts、StateAdapter实现 adapter.ts以及状态存储子代理 agent.ts。安装在承载 messenger 入口ingress的 Worker 中同时安装agents与chat两个包npm install agents chatagents提供Agent基类、subAgent()子代理路由、getCurrentAgent()上下文等能力chat提供 Chat SDK 运行时本身Chat、Message、Thread等。基础接入在 Agent 中运行 Chat SDK创建父 Agent 并注入状态适配器创建一个父 Agent 来拥有你的 Chat SDK 运行时把createChatSdkState()作为 Chat SDK 的state选项传入import { Agent } from agents; import { Chat } from chat; import { createChatSdkState } from agents/chat-sdk; import { createTelegramAdapter } from chat-adapter/telegram; export { ChatSdkStateAgent } from agents/chat-sdk; export class MessengerAgent extends AgentEnv { private chat!: Chat; onStart() { const telegram createTelegramAdapter({ botToken: this.env.TELEGRAM_BOT_TOKEN, mode: webhook, secretToken: this.env.TELEGRAM_WEBHOOK_SECRET_TOKEN, userName: my_bot }); this.chat new Chat({ adapters: { telegram }, userName: my_bot, state: createChatSdkState(), concurrency: { strategy: burst, debounceMs: 600 } }); } }要点说明onStart()是 Agent 的生命周期钩子Chat SDK 运行时在 Agent 启动时初始化concurrency: { strategy: burst, debounceMs: 600 }是 Chat SDK 的并发策略配置burst表示同一线程内的消息按 600ms 防抖合并处理其底层的待处理队列pending message queue正由agents/chat-sdk的ChatSdkStateAgent持久化见 agent.ts 中的chat_sdk_state_queue表必须从 Worker 入口导出ChatSdkStateAgent这样子代理路由sub-agent routing才能解析并实例化状态子代理。添加 Durable Object 迁移父 Agent 需要注册为 Durable Object 类并在迁移中使用new_sqlite_classes以获得 SQLite 存储能力{ durable_objects: { bindings: [{ name: MessengerAgent, class_name: MessengerAgent }] }, migrations: [ { tag: v1, new_sqlite_classes: [MessengerAgent] } ] }注意迁移中只需要注册父 AgentMessengerAgent。状态子代理ChatSdkStateAgent由父 Agent 在运行时通过subAgent()按需创建由 Worker 入口的 export 解析类不需要在durable_objects中逐一预注册。底层实现状态分片如何创建从 adapter.ts 的构造函数可以看出createChatSdkState()的实现原理const parent options.parent ?? getCurrentAgent().agent; if (!parent) { throw new Error( ChatSdkStateAdapter requires a parent Agent. Pass parent or create it inside an Agent context. ); }即当你在 Agent 生命周期方法如onStart或请求处理器中调用createChatSdkState()时适配器通过getCurrentAgent()自动取得当前 Agent 作为父代理之后每次读写都会调用parent.subAgent(this.agentClass, name)创建/复用对应名称的状态子代理。如果不在 Agent 上下文中调用且未显式传入parent会直接抛出错误。状态分片默认分片策略Chat SDK 的不同状态类型使用不同的 key 前缀。默认情况下agents/chat-sdk按线程型 key 的前两段冒号分隔片段进行分片。例如telegram:-100123:456与telegram:-100123:789会共享同一个状态分片telegram:-100123因为它们都属于同一个 Telegram 频道会话相关操作可以落在同一个 Durable Object 上保证并发锁与消息顺序的一致性。默认 key 分片器识别的前缀默认 key 分片器defaultKeyShard识别以下 Chat SDK key 前缀常量定义见 adapter.ts前缀用途分片规则thread-state:线程状态取去掉前缀后 key 的线程分片channel-state:频道状态同上msg-history:消息历史同上transcripts:user:跨平台转录取去掉前缀后剩余 key 的线程分片其中线程分片由defaultThreadShard实现export function defaultThreadShard(threadId: string): string { return threadId.split(:).slice(0, 2).join(:) || default; }未知 key无法被识别时会落入适配器默认分片名default对应createChatSdkState的name选项默认值为default。分片路由的测试验证在 packages/agents/src/tests/agents/chat-sdk.ts 中testShardRouting()用具体 key 验证了路由行为thread-state:telegram:123:456→ 分片telegram:123channel-state:telegram:123→ 分片telegram:123msg-history:telegram:123:456→ 分片telegram:123transcripts:user:acme:user-123→ 分片acme:user-123chat:callback:opaque未知前缀→ 返回undefined回退到默认分片自定义分片shardKey控制线程 ID 到分片的映射当默认的取前两段规则不满足你的业务模型时用shardKey(threadId)完全接管线程 ID 到状态子代理名称的映射const state createChatSdkState({ shardKey(threadId) { return threadId.split(:).slice(0, 2).join(:); } });keyShard为非线程型 key 指定分片某些适配器会存放非线程形状的 key例如去重 keydedupe:telegram:...。此时shardKey无法处理需要用keyShard(key)把这类 key 路由到与提供者provider相关的分片const state createChatSdkState({ keyShard(key) { if (!key.startsWith(dedupe:telegram:)) { return undefined; } const chatId key.slice(dedupe:telegram:.length).split(:)[0]; return chatId ? telegram:${chatId} : undefined; } });返回值约定返回undefined表示不接管适配器会回退到内置 key 分片器defaultKeyShard若内置分片器也无法识别则回退到默认分片default。这一优先级链在 adapter.ts 的stateAgentForKey()中体现const name this.keyShard?.(key) ?? defaultKeyShard(key, this.shardKey) ?? this.defaultName;API 详解createChatSdkState(options)创建一个由ChatSdkStateAgent子代理支撑的 Chat SDKStateAdapter。工厂函数的完整实现见 index.ts直接返回一个ChatSdkStateAdapter实例。import { createChatSdkState } from agents/chat-sdk; export { ChatSdkStateAgent } from agents/chat-sdk; const state createChatSdkState({ // parent: this // 可选默认取 getCurrentAgent() 得到的当前 Agent });完整选项定义于 types.ts 的ChatSdkStateAdapterOptions选项类型默认值说明agentSubAgentClassChatSdkStateAgentChatSdkStateAgent自定义的ChatSdkStateAgent子类用于覆盖存储行为parentChatSdkStateParentgetCurrentAgent().agent负责调用subAgent()创建状态分片的父 Agent在 Agent 生命周期方法或请求处理器内调用时可省略namestringdefault无法映射到任何分片的 key 所使用的默认分片名shardKey(threadId)(threadId: string) stringdefaultThreadShard将 Chat SDK 线程 ID 与锁 key 映射为分片名keyShard(key)(key: string) string \| undefined无使用内置分片器将通用的 Chat SDK 缓存/列表 key 映射为分片名返回undefined时回退ChatSdkStateAgent负责把状态写入 SQLite 的子代理类。它的onStart()会执行migrate()建表并调度首次清理见 agent.ts。必须从 Worker 入口导出运行时才能创建它export { ChatSdkStateAgent } from agents/chat-sdk;ChatSdkStateAdaptercreateChatSdkState()返回的具体StateAdapter实现类实现了 Chat SDK 的StateAdapter接口订阅、锁、队列、缓存、列表全部方法。大多数应用无需直接实例化它——通过createChatSdkState()即可。存储内容与底层表结构实现的StateAdapter原语适配器完整实现了 Chat SDKStateAdapter接口的如下原语订阅支撑thread.subscribe()/thread.unsubscribe()查询用isSubscribed()锁支撑按线程/按频道的并发控制acquireLock/releaseLock/extendLock/forceReleaseLock待处理消息队列支撑queue、debounce、burst三种并发策略的挂起消息通用键值缓存get/set/setIfNotExists/delete支持可选 TTL追加型列表appendToList/getList支持最大长度裁剪与列表级 TTL 刷新。SQLite 表结构子代理在onStart()中通过 migrate() 幂等建表共六张表表关键字段用途chat_sdk_state_subscriptionsthread_id主键线程订阅关系chat_sdk_state_locksthread_id主键、token、expires_at并发锁含过期时间chat_sdk_state_cachekey主键、value、expires_at通用缓存chat_sdk_state_queueid自增、thread_id、value、enqueued_at、expires_atFIFO 消息队列chat_sdk_state_listsid自增、key、value、expires_at追加型列表chat_sdk_state_metadatakey主键、value清理调度元数据同时为expires_at、thread_id id等查询路径建立了索引如idx_chat_sdk_state_queue_thread保证按线程取队列、按时间清理的高效执行。基于这些原语的 Chat SDK 特性Chat SDK 构建在这些原语之上的功能包括消息去重deduplication依赖缓存 setIfNotExists线程与频道状态依赖thread-state:/channel-state:前缀的缓存持久化线程历史对选择persistThreadHistory: true的适配器历史消息以msg-history:前缀的列表存储回调 URL token 存储依赖通用缓存Modal 上下文存储依赖通用缓存跨平台转录transcripts以transcripts:user:前缀的列表存储。上述特性路径在 packages/agents/src/tests/agents/chat-sdk.ts 的testChatFeaturePaths()中被端到端验证它启动一个真实Chat实例写入线程状态、频道状态、三条历史消息、两条转录记录并重复投递同一条消息验证去重计数只增加一次。清理行为严格读、惰性写agents/chat-sdk的过期清理遵循读时严格、物理清理惰性的策略。读时严格Strict TTL reads所有读取路径都会先剔除过期数据acquireLock在事务内先DELETE ... WHERE expires_at now再尝试插入agent.tspopQueue同样先删过期条目再取最早一条cacheGet的查询条件带(expires_at IS NULL OR expires_at now)listGet读取前先清理该 key 的过期行。因此已过期的锁、缓存值、队列条目、列表条目在返回前就会被忽略或删除应用永远不会读到过期状态。物理清理惰性Lazy cleanup物理删除通过scheduleCleanupForExpiry()调度完成。ChatSdkStateAgent会记录最早已知过期时间next_cleanup_at并调用this.schedule(delaySeconds, cleanupExpired, { expiresAt })注册一次延迟调度见 agent.tscleanupExpired执行时批量删除四类表中所有expires_at now的行然后调用scheduleNextCleanup()重新计算下一次最早过期时间并再次调度若不存在任何未过期的 TTL 行则取消已注册的调度并清空元数据。这种设计让空闲分片保持安静不产生周期性空跑同时防止过期行无限累积。测试 chat-sdk.test.ts 中的testExpiredLock、testExpiredQueue、testListTtlRefresh分别验证了过期锁可被重新获取、过期队列条目被跳过、以及列表级 TTL 刷新任何一次追加都会刷新整个逻辑列表的过期时间而非仅新行。完整实战示例examples/chat-sdk-messenger仓库中的 examples/chat-sdk-messenger 提供了开箱即用的完整 Telegram 机器人示例直接演示了本文所有概念用createChatSdkState({ agent: ThinkMessengerStateAgent })为 Chat SDK 提供状态其中ThinkMessengerStateAgent是 Think 对ChatSdkStateAgent的包装子类Chat SDK 的 burst/debounce 并发控制由 Think 支撑的 AI 回复运行在受管理的 fibermanaged fiber中waitForCompletion: true保持 Chat SDK 处理器挂起直到可见回复完成从而保留每线程的并发语义与幂等边界。其架构为一个 Chat SDK 入口 Agent 两类子代理详见该示例的 READMEChatIngressAgent Chat({ adapters: { telegram } }) ThinkMessengerStateAgent # Chat SDK 基础设施状态订阅/锁/队列/缓存/列表 ConversationAgent # 每线程的 AI 会话与模型调用本地运行方式在示例目录下npm install # 仓库根目录安装依赖 cp .env.example .env # 填写 TELEGRAM_BOT_TOKEN / TELEGRAM_WEBHOOK_SECRET_TOKEN / TELEGRAM_BOT_USERNAME npm start # 启动本地 Vite/Workers 开发服务器默认开启 Quick Tunnel部署方式wrangler secret put TELEGRAM_BOT_TOKEN wrangler secret put TELEGRAM_WEBHOOK_SECRET_TOKEN wrangler secret put TELEGRAM_BOT_USERNAME npm run deploy该示例的ChatSdkStateAgent只承担基础设施职责锁、队列、订阅、缓存、列表不拥有频道人格、工具或推理逻辑——这与状态层与 AI 层分离的设计原则一致。要迁移到 Slack、Discord 等其他平台只需替换或追加Chat()调用中的adapters并调整 Webhook 路由状态适配器与 AI 子代理完全不用改动。注意事项与适用边界agents/chat-sdk仅覆盖状态层不提供任何 messenger 适配器必须搭配 Chat SDK 适配器使用createChatSdkState()必须在 Agent 上下文生命周期方法或请求处理器中调用否则必须显式传入parent否则抛错状态子代理依赖new_sqlite_classes迁移SQLite 存储且需要从 Worker 入口导出ChatSdkStateAgent或你的自定义子类以供子代理路由解析分片策略决定并发与一致性的平衡同分片内的锁、队列、列表操作落在同一 Durable Object 上天然串行跨分片则并发执行。默认按provider:channel前两段分片对大多数 messenger 场景是合理默认高频场景可参考示例 README 中按租户/机器人/会话路由到不同父 Agent 名称的扩展思路。相关文档与源码索引原始文档docs/agents/chat-sdk.md状态适配器实现packages/agents/src/chat-sdk/adapter.ts状态子代理SQLite 存储与清理packages/agents/src/chat-sdk/agent.ts选项类型定义packages/agents/src/chat-sdk/types.ts入口与工厂函数packages/agents/src/chat-sdk/index.ts单元测试packages/agents/src/tests/chat-sdk.test.ts、packages/agents/src/tests/agents/chat-sdk.ts完整示例examples/chat-sdk-messenger/README.md【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考