Cloudflare Agents 状态管理实战指南持久化、双向同步与类型安全【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agentsAgent 内置的状态管理State Management是构建实时协作应用的核心能力状态自动持久化到 SQLite、通过 WebSocket 广播给所有已连接客户端并且支持服务端与客户端双向更新。本文以 docs/agents/state.md 为骨架结合 packages/agents/src/index.ts 的源码实现与 packages/agents/src/tests/state.test.ts 的测试用例系统讲解如何在 Cloudflare Agents 上定义、读取、更新、校验与同步状态并给出避免无限循环、乐观更新、State 与 SQL 分工等实战最佳实践。状态管理的核心特性Agent 的状态体系围绕四个设计目标展开持久化Persistent状态自动写入 SQLiteAgent 实例本质是 Durable Object在重启、休眠hibernation后依然能恢复完整状态同步Synchronized每次状态变更都会实时广播给所有连接的 WebSocket 客户端双向Bidirectional服务端通过setState()更新客户端通过 WebSocket 推送更新二者共享同一状态视图类型安全Type-safeAgentEnv, State的第二个泛型参数让this.state、this.setState()全程具备 TypeScript 类型推断。一个典型的多人在线游戏示例import { Agent } from agents; type GameState { players: string[]; score: number; status: waiting | playing | finished; }; export class GameAgent extends AgentEnv, GameState { // Default state for new agents initialState: GameState { players: [], score: 0, status: waiting }; // React to state changes onStateChanged(state: GameState, source: Connection | server) { if (source ! server state.players.length 2) { // Client added a player, start the game this.setState({ ...state, status: playing }); } } addPlayer(name: string) { this.setState({ ...this.state, players: [...this.state.players, name] }); } }底层实现一行 SQL 的状态表从源码可以看到状态持久化依赖 Agent 内部 SQLite 表cf_agents_state通过固定的行 IDcf_state_row_id存储整份 JSON 状态。核心写入逻辑在_setStateInternal()packages/agents/src/index.tsthis.sql INSERT OR REPLACE INTO cf_agents_state (id, state) VALUES (${STATE_ROW_ID}, ${JSON.stringify(nextState)}) ;INSERT OR REPLACE意味着整份状态以单行 JSON 的形式整体覆盖不存在增量合并这正是文档中反复强调更新具体字段时必须先展开已有状态spread的根本原因。定义初始状态使用initialState属性为新的 Agent 实例提供默认值type State { messages: Message[]; settings: UserSettings; lastActive: string | null; }; export class ChatAgent extends AgentEnv, State { initialState: State { messages: [], settings: { theme: dark, notifications: true }, lastActive: null }; }类型安全AgentEnv, State的第二个泛型参数即为状态类型this.state与this.setState()全程有类型约束// State is fully typed export class MyAgent extends AgentEnv, MyState { initialState: MyState { count: 0 }; increment() { // TypeScript knows this.state is MyState this.setState({ count: this.state.count 1 }); } }初始状态的生效时机初始状态是懒加载的——在首次访问this.state时才生效而不是每次唤醒都重新应用新 AgentinitialState被使用并持久化已有 Agent从 SQLite 加载已持久化的状态未定义initialStatethis.state返回undefined。async onStart() { // Safe to access - returns initialState if new, or persisted state console.log(Current count:, this.state.count); }从源码stategetterpackages/agents/src/index.ts可以看清这段懒加载逻辑若_state内存缓存已初始化直接返回否则查询cf_agents_state表中cf_state_row_id行——行的存在性即状态曾被设置过的信号这一设计使得null、0、false、等假值也能被正确区分源码注释明确说明Row existence is the signal that state was previously set首次访问且定义了initialState时才执行_setStateInternal(this.initialState)落库并返回。值得注意的是源码为损坏状态提供了自动恢复机制若持久化的 JSON 解析失败会打印错误并回退到initialState重新持久化若连initialState也未定义则删除损坏行返回undefined避免无限重试循环。对应测试用例 packages/agents/src/tests/state.test.tsshould recover from corrupted state JSON by falling back to initialState验证了该行为。读取状态通过this.stategetter 读取当前状态async onRequest(request: Request) { // Read current state const { players, status } this.state; if (status waiting players.length 2) { return new Response(Waiting for players...); } return new Response(JSON.stringify(this.state)); }undefined 状态如果不定义initialStatethis.state返回undefined此时需要在首次访问时自行初始化export class MinimalAgent extends AgentEnv { // No initialState defined async onConnect(connection: Connection) { if (!this.state) { // First time - initialize state this.setState({ initialized: true }); } } }测试用例 packages/agents/src/tests/state.test.tsshould return undefined when no initialState defined与 packages/agents/src/tests/state.test.tsshould allow setting state when no initialState defined共同验证了这一行为路径。更新状态使用setState()更新状态其完整链路为校验同步调用validateStateChange()抛错则中止本次更新持久化INSERT OR REPLACE写入 SQLitecf_agents_state表广播向所有协议启用的连接广播CF_AGENT_STATE消息排除发起方连接通知通过waitUntil异步调用onStateChanged()非阻塞、best-effort。// Replace entire state this.setState({ players: [Alice, Bob], score: 0, status: playing }); // Update specific fields (spread existing state) this.setState({ ...this.state, score: this.state.score 10 });从源码 packages/agents/src/index.ts 可见服务端调用setState()时还会检查当前连接上下文是否为只读连接readonly connection若为只读则直接抛出Connection is readonly异常。相关机制详见 docs/agents/readonly-connections.md。状态必须可序列化状态以 JSON 形式存储于 SQLite因此必须是可序列化的// Good - plain objects, arrays, primitives this.setState({ items: [a, b, c], count: 42, active: true, metadata: { key: value } }); // Bad - functions, classes, circular references this.setState({ callback: () {}, // Functions dont serialize date: new Date(), // Becomes string, loses methods self: this // Circular reference }); // For dates, use ISO strings this.setState({ createdAt: new Date().toISOString() });客户端推送更新的处理路径当客户端通过 WebSocket 推送状态时服务端在消息循环中识别CF_AGENT_STATE类型的消息类型守卫isStateUpdateMessage见 packages/agents/src/index.ts随后依次检查只读权限、执行_setStateInternal(state, connection)此时source为该连接。若validateStateChange或其它同步校验抛出异常服务端会向发起方连接发送CF_AGENT_STATE_ERROR错误信息为通用的 State update rejected对应处理代码见 packages/agents/src/index.ts测试覆盖见 packages/agents/src/tests/state.test.ts。响应状态变化重写onStateChanged()来响应状态变化通知 / 副作用onStateChanged(state: GameState, source: Connection | server) { console.log(State updated:, state); console.log(Updated by:, source server ? server : source.id); }onStateChanged()在状态持久化并广播之后被调用且通过ctx.waitUntil()异步执行——它的异常不会影响状态落库或广播而是被路由到onError()源码注释明确onStateChanged/onStateUpdate errors should not affect state or broadcasts。测试用例 packages/agents/src/tests/state.test.tsshould still broadcast state even if onStateChanged throws专门验证了这一隔离保证。迁移提示onStateChanged取代了已废弃的onStateUpdate服务端钩子。若你的 Agent 类仍在使用onStateUpdate请重命名为onStateChanged——签名与行为完全一致。每个类在重命名前会触发一次控制台警告。源码层面_callStatePersistenceHook()会根据构造函数缓存的_persistenceHookMode分发到新旧钩子packages/agents/src/index.ts并在同一类同时重写两个钩子时报错测试见 packages/agents/src/tests/state.test.ts。校验状态更新若需校验或拒绝状态更新重写validateStateChange()在持久化与广播之前执行必须同步源码 JSDoc 明确This hook must be synchronous抛错即中止本次更新。validateStateChange(nextState: GameState, source: Connection | server) { // Example: reject negative scores if (nextState.score 0) { throw new Error(score cannot be negative); } }onStateChanged()不用于校验它是通知钩子不应阻塞广播。validateStateChange()的调用发生在_setStateInternal()的第一行packages/agents/src/index.ts因此只要抛错后续的持久化与广播都会被跳过。测试用例分别验证了抛错时不广播packages/agents/src/tests/state.test.ts与通过时正常广播packages/agents/src/tests/state.test.ts两条路径。source参数source指明更新的发起方ValueMeaningserverAgent 调用了setState()Connection客户端通过 WebSocket 推送了状态该参数常用于避免无限循环不要响应自己发起的更新校验客户端输入仅对客户端操作触发副作用。onStateChanged(state: State, source: Connection | server) { // Ignore server-initiated updates if (source server) return; // A client updated state - validate and process const connection source; console.log(Client ${connection.id} updated state); // Maybe trigger something based on the change if (state.status submitted) { this.processSubmission(state); } }常见模式客户端驱动的动作onStateChanged(state: State, source: Connection | server) { if (source server) return; // Client added a message const lastMessage state.messages[state.messages.length - 1]; if (lastMessage !lastMessage.processed) { // Process and update this.setState({ ...state, messages: state.messages.map(m m.id lastMessage.id ? { ...m, processed: true } : m ) }); } }客户端状态同步状态自动与已连接客户端同步useAgent与AgentClient均暴露state属性跟踪当前状态。完整客户端文档见 docs/agents/client-sdk.md。ReactuseAgentimport { useAgent } from agents/react; function GameUI() { const agent useAgent({ agent: game-agent, name: room-123 }); // Read state directly — reactive, triggers re-render on change // Push state to agent with spread for partial updates const addPlayer (name: string) { agent.setState({ ...agent.state, players: [...(agent.state?.players ?? []), name] }); }; return divPlayers: {agent.state?.players.join(, )}/div; }从 packages/agents/src/react.tsx 可以看到agent.setState的客户端实现先通过活跃 socket 发送CF_AGENT_STATE消息随后立即更新本地 React 状态setAgentState(newState)并调用options.onStateUpdate——这正是乐观更新的基础同时服务端广播到达时会再次setAgentState覆盖为服务端权威值packages/agents/src/react.tsx保证最终一致性。原生 JSAgentClientimport { AgentClient } from agents/client; const client new AgentClient({ agent: game-agent, name: room-123, host: your-worker.workers.dev }); await client.ready; // Read state directly console.log(Score:, client.state?.score); // Push state update with spread for partial updates client.setState({ ...client.state, score: 100 });客户端底层实现见 packages/agents/src/client.ts收到CF_AGENT_STATE广播时更新this.state并以server为 source 触发onStateUpdate调用setState()时发送消息、立即更新本地state并以client为 source 回调packages/agents/src/client.ts。若被服务端拒绝如只读连接会收到CF_AGENT_STATE_ERROR并触发onStateUpdateError。状态流转示意┌─────────────────────────────────────────────────────────────┐ │ Agent │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ this.state │ │ │ │ (persisted in SQLite) │ │ │ └─────────────────────────────────────────────────────┘ │ │ ▲ │ │ │ │ setState() │ broadcast │ │ │ ▼ │ └───────────┼──────────────────────────────┼──────────────────┘ │ │ │ │ WebSocket │ │ ┌───────────┴──────────────────────────────┴───────────────────┐ │ Clients │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Client 1 │ │ Client 2 │ │ Client 3 │ │ │ │ state │ │ state │ │ state │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ Any client can call setState() to push updates │ └──────────────────────────────────────────────────────────────┘从 Workflow 更新状态当使用 docs/agents/workflows.md 时可以在工作流步骤中更新 Agent 状态// In your workflow async run(event: AgentWorkflowEventParams, step: AgentWorkflowStep) { // Replace entire state await step.updateAgentState({ status: processing, progress: 0 }); // Merge partial updates (preserves other fields) await step.mergeAgentState({ progress: 50 }); // Reset to initialState await step.resetAgentState(); return result; }这些操作是持久化操作——即使工作流发生重试状态变更也会保留。从源码实现看三个方法最终都通过step.do()的幂等步骤封装调用 Agent 内部的_workflow_updateState(action, state)packages/agents/src/workflows.ts后者再映射为setState/ 展开合并 /setState(initialState)packages/agents/src/index.ts。接口签名定义在 packages/agents/src/workflow-types.ts测试覆盖见 packages/agents/src/tests/test-workflow.ts。模式与最佳实践保持状态精简状态每次变更都会广播给所有客户端。对于大数据量场景// Bad - storing large arrays in state initialState { allMessages: [] // Could grow to thousands of items }; // Good - store in SQL, keep state light initialState { messageCount: 0, lastMessageId: null }; // Query SQL for full data async getMessages(limit 50) { return this.sqlSELECT * FROM messages ORDER BY created_at DESC LIMIT ${limit}; }乐观更新为了获得响应式 UI可以先更新客户端状态// Client-side function sendMessage(text: string) { const optimisticMessage { id: crypto.randomUUID(), text, pending: true }; // Update immediately — agent.state updates optimistically agent.setState({ ...agent.state, messages: [...(agent.state?.messages ?? []), optimisticMessage] }); // Server will confirm/update } // Server-side onStateChanged(state: State, source: Connection | server) { if (source server) return; const pendingMessages state.messages.filter(m m.pending); for (const msg of pendingMessages) { // Validate and confirm this.setState({ ...state, messages: state.messages.map(m m.id msg.id ? { ...m, pending: false, timestamp: Date.now() } : m ) }); } }如前文源码分析所示乐观更新的可行性来自客户端setState的先本地生效、后服务端确认设计以及服务端广播回传权威状态的最终一致性保证。状态 vs SQL使用 State使用 SQLUI 状态loading、选中项历史数据实时计数器大规模集合活跃会话数据关系型数据配置可查询数据export class ChatAgent extends AgentEnv, State { // State: current UI state initialState { typing: [], unreadCount: 0, activeUsers: [] }; // SQL: message history async getMessages(limit 100) { return this.sql SELECT * FROM messages ORDER BY created_at DESC LIMIT ${limit} ; } async saveMessage(message: Message) { this.sql INSERT INTO messages (id, text, user_id, created_at) VALUES (${message.id}, ${message.text}, ${message.userId}, ${Date.now()}) ; // Update state for real-time UI this.setState({ ...this.state, unreadCount: this.state.unreadCount 1 }); } }避免无限循环注意不要在响应自身更新时再次触发状态更新// Bad - infinite loop onStateChanged(state: State) { this.setState({ ...state, lastUpdated: Date.now() }); } // Good - check source onStateChanged(state: State, source: Connection | server) { if (source server) return; // Dont react to own updates this.setState({ ...state, lastUpdated: Date.now() }); }由于服务端setState()触发的onStateChanged回调中source server而客户端推送触发的回调中source是该Connection对象因此通过判断 source 即可天然切断写入-回调-再写入的递归链。API 参考属性PropertyTypeDescriptionstateState当前状态getterinitialStateState新 Agent 的默认状态方法MethodSignatureDescriptionsetState(state: State) void更新状态、持久化并广播onStateChanged(state: State, source: Connection \| server) void状态持久化并广播后被调用validateStateChange(nextState: State, source: Connection \| server) void持久化与广播前执行校验抛错则中止更新同步Workflow 步骤方法MethodDescriptionstep.updateAgentState(state)从工作流替换 Agent 状态step.mergeAgentState(partial)从工作流合并部分状态step.resetAgentState()从工作流重置为initialState延伸阅读Readonly Connections限制哪些连接可以更新状态Client SDK完整的客户端状态同步文档Workflows从工作流进行持久化状态更新状态管理测试集packages/agents/src/tests/state.test.ts覆盖初始状态懒加载、SQLite 持久化、客户端发起更新、校验拒绝、广播隔离与钩子迁移等全部关键路径。【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考