存储与同步状态

Agent 提供内置状态管理，自动持久化并实时同步到所有已连接客户端。

概览

Agent 中的状态具有以下特点：

持久化 — 自动保存到 SQLite，在重启和休眠后仍然保留
同步 — 变更即时广播到所有已连接的 WebSocket 客户端
双向 — 服务端和客户端均可更新状态
类型安全 — 通过泛型提供完整 TypeScript 支持
即时一致 — 读取自己的写入
线程安全 — 可安全并发更新
快速 — 状态与 Agent 运行位置同址存放

Agent state 存储在每个 Agent 实例内嵌的 SQL 数据库中。可使用更高级的 this.setState API（推荐）同步 state 并在变更时触发事件，或直接用 this.sql 查询数据库。

import { Agent } from "agents";

export class GameAgent extends Agent {
	// Default state for new agents
	initialState = {
		players: [],
		score: 0,
		status: "waiting",
	};

	// React to state changes
	onStateChanged(state, source) {
		if (source !== "server" && state.players.length >= 2) {
			// Client added a player, start the game
			this.setState({ ...state, status: "playing" });
		}
	}

	addPlayer(name) {
		this.setState({
			...this.state,
			players: [...this.state.players, name],
		});
	}
}

import { Agent } from "agents";

type GameState = {
	players: string[];
	score: number;
	status: "waiting" | "playing" | "finished";
};

export class GameAgent extends Agent<Env, 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],
		});
	}
}

定义初始 state

使用 initialState 属性为新建 Agent 实例定义默认值：

export class ChatAgent extends Agent {
	initialState = {
		messages: [],
		settings: { theme: "dark", notifications: true },
		lastActive: null,
	};
}

type State = {
	messages: Message[];
	settings: UserSettings;
	lastActive: string | null;
};

export class ChatAgent extends Agent<Env, State> {
	initialState: State = {
		messages: [],
		settings: { theme: "dark", notifications: true },
		lastActive: null,
	};
}

类型安全

Agent 的第二个泛型参数定义 state 类型：

// State is fully typed
export class MyAgent extends Agent {
	initialState = { count: 0 };

	increment() {
		// TypeScript knows this.state is MyState
		this.setState({ count: this.state.count + 1 });
	}
}

// State is fully typed
export class MyAgent extends Agent<Env, MyState> {
	initialState: MyState = { count: 0 };

	increment() {
		// TypeScript knows this.state is MyState
		this.setState({ count: this.state.count + 1 });
	}
}

初始 state 何时生效

初始 state 在首次访问时懒加载，而非每次唤醒：

新 Agent — 使用 initialState 并持久化
已有 Agent — 从 SQLite 加载已持久化 state
未定义 initialState — this.state 为 undefined

class MyAgent extends Agent {
	initialState = { count: 0 };
	async onStart() {
		// Safe to access - returns initialState if new, or persisted state
		console.log("Current count:", this.state.count);
	}
}

class MyAgent extends Agent<Env, { count: number }> {
	initialState = { count: 0 };
	async onStart() {
		// Safe to access - returns initialState if new, or persisted state
		console.log("Current count:", this.state.count);
	}
}

读取 state

通过 this.state getter 访问当前 state：

class MyAgent extends Agent {
	async onRequest(request) {
		// Read current state
		const { players, status } = this.state;

		if (status === "waiting" && players.length < 2) {
			return new Response("Waiting for players...");
		}

		return Response.json(this.state);
	}
}

class MyAgent extends Agent<
	Env,
	{ players: string[]; status: "waiting" | "playing" | "finished" }
> {
	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 Response.json(this.state);
	}
}

未定义的 state

若未定义 initialState，this.state 返回 undefined：

export class MinimalAgent extends Agent {
	// No initialState defined

	async onConnect(connection) {
		if (!this.state) {
			// First time - initialize state
			this.setState({ initialized: true });
		}
	}
}

export class MinimalAgent extends Agent {
	// No initialState defined

	async onConnect(connection: Connection) {
		if (!this.state) {
			// First time - initialize state
			this.setState({ initialized: true });
		}
	}
}

更新 state

使用 setState() 更新状态。这会：

保存到 SQLite（持久化）
广播到所有已连接客户端（排除 shouldSendProtocolMessages 返回 false 的连接）
触发 onStateChanged()（广播之后；尽力而为）

// 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,
});

// 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,
});

State 须可序列化

State 以 JSON 存储，须可序列化：

// Good - plain objects, arrays, primitives
this.setState({
	items: ["a", "b", "c"],
	count: 42,
	active: true,
	metadata: { key: "value" },
});

// Bad - functions, classes, circular references
// Functions do not serialize
// Dates become strings, lose methods
// Circular references fail

// For dates, use ISO strings
this.setState({
	createdAt: new Date().toISOString(),
});

// Good - plain objects, arrays, primitives
this.setState({
	items: ["a", "b", "c"],
	count: 42,
	active: true,
	metadata: { key: "value" },
});

// Bad - functions, classes, circular references
// Functions do not serialize
// Dates become strings, lose methods
// Circular references fail

// For dates, use ISO strings
this.setState({
	createdAt: new Date().toISOString(),
});

响应 state 变更

覆盖 onStateChanged() 以在 state 变更时反应（通知/副作用）：

class MyAgent extends Agent {
	onStateChanged(state, source) {
		console.log("State updated:", state);
		console.log("Updated by:", source === "server" ? "server" : source.id);
	}
}

class MyAgent extends Agent<Env, GameState> {
	onStateChanged(state: GameState, source: Connection | "server") {
		console.log("State updated:", state);
		console.log("Updated by:", source === "server" ? "server" : source.id);
	}
}

source 参数

source 表示谁触发了更新：

值	含义
`"server"`	Agent 调用了 `setState()`
`Connection`	客户端经 WebSocket 推送 state

适用于：

避免无限循环（不响应自己的更新）
校验客户端输入
仅在客户端操作时触发副作用

class MyAgent extends Agent {
	onStateChanged(state, source) {
		// 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);
		}
	}
}

class MyAgent extends Agent<
	Env,
	{ status: "waiting" | "playing" | "finished" }
> {
	onStateChanged(state: GameState, 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);
		}
	}
}

常见模式：客户端驱动操作

class MyAgent extends Agent {
	onStateChanged(state, source) {
		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,
				),
			});
		}
	}
}

class MyAgent extends Agent<Env, { messages: Message[] }> {
	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,
				),
			});
		}
	}
}

校验状态更新

若要校验或拒绝状态更新，覆盖 validateStateChange()：

在持久化与广播之前运行
须同步
抛出异常则中止更新

class MyAgent extends Agent {
	validateStateChange(nextState, source) {
		// Example: reject negative scores
		if (nextState.score < 0) {
			throw new Error("score cannot be negative");
		}

		// Example: only allow certain status transitions
		if (this.state.status === "finished" && nextState.status !== "finished") {
			throw new Error("Cannot restart a finished game");
		}
	}
}

class MyAgent extends Agent<Env, GameState> {
	validateStateChange(nextState: GameState, source: Connection | "server") {
		// Example: reject negative scores
		if (nextState.score < 0) {
			throw new Error("score cannot be negative");
		}

		// Example: only allow certain status transitions
		if (this.state.status === "finished" && nextState.status !== "finished") {
			throw new Error("Cannot restart a finished game");
		}
	}
}

客户端 state 同步

State 与已连接客户端自动同步。

React（useAgent）

import { useAgent } from "agents/react";

function GameUI() {
	const agent = useAgent({
		agent: "game-agent",
		name: "room-123",
		onStateUpdate: (state, source) => {
			console.log("State updated:", state);
		},
	});

	// Push state to agent
	const addPlayer = (name) => {
		agent.setState({
			...agent.state,
			players: [...agent.state.players, name],
		});
	};

	return <div>Players: {agent.state?.players.join(", ")}</div>;
}

import { useAgent } from "agents/react";

function GameUI() {
  const agent = useAgent({
    agent: "game-agent",
    name: "room-123",
    onStateUpdate: (state, source) => {
      console.log("State updated:", state);
    }
  });

  // Push state to agent
  const addPlayer = (name: string) => {
    agent.setState({
      ...agent.state,
      players: [...agent.state.players, name]
    });
  };

  return <div>Players: {agent.state?.players.join(", ")}</div>;
}

原生 JS（AgentClient）

import { AgentClient } from "agents/client";

const client = new AgentClient({
	agent: "game-agent",
	name: "room-123",
	onStateUpdate: (state) => {
		document.getElementById("score").textContent = state.score;
	},
});

// Push state update
client.setState({ ...client.state, score: 100 });

import { AgentClient } from "agents/client";

const client = new AgentClient({
	agent: "game-agent",
	name: "room-123",
	onStateUpdate: (state) => {
		document.getElementById("score").textContent = state.score;
	},
});

// Push state update
client.setState({ ...client.state, score: 100 });

State 流转

flowchart TD
    subgraph Agent
        S["this.state<br/>(persisted in SQLite)"]
    end
    subgraph Clients
        C1["Client 1"]
        C2["Client 2"]
        C3["Client 3"]
    end
    C1 & C2 & C3 -->|setState| S
    S -->|broadcast via WebSocket| C1 & C2 & C3

来自 Workflow 的 state

使用 Workflows 时，可从 workflow 步骤更新 Agent state：

// In your workflow
class MyWorkflow extends Workflow {
	async run(event, step) {
		// 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;
	}
}

// In your workflow
class MyWorkflow extends Workflow<Env> {
	async run(event: AgentWorkflowEvent, 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;
	}
}

这些是持久操作——Workflow 重试时仍会保留。

SQL API

每个 Agent 实例拥有在 Agent 同址运行的 SQL（SQLite）数据库。在 Agent 内插入或查询数据几乎零延迟——Agent 无需跨洲或跨洋访问自己的数据。

可在 Agent 任意方法中通过 this.sql 访问 SQL API。SQL API 接受模板字面量：

export class MyAgent extends Agent {
	async onRequest(request) {
		let userId = new URL(request.url).searchParams.get("userId");

		// 'users' is just an example here: you can create arbitrary tables and define your own schemas
		// within each Agent's database using SQL (SQLite syntax).
		let [user] = this.sql`SELECT * FROM users WHERE id = ${userId}`;
		return Response.json(user);
	}
}

export class MyAgent extends Agent {
	async onRequest(request: Request) {
		let userId = new URL(request.url).searchParams.get("userId");

		// 'users' is just an example here: you can create arbitrary tables and define your own schemas
		// within each Agent's database using SQL (SQLite syntax).
		let [user] = this.sql`SELECT * FROM users WHERE id = ${userId}`;
		return Response.json(user);
	}
}

也可为查询提供 TypeScript 类型参数，用于推断结果类型：

export class MyAgent extends Agent {
	async onRequest(request) {
		let userId = new URL(request.url).searchParams.get("userId");
		// Supply the type parameter to the query when calling this.sql
		// This assumes the results returns one or more User rows with "id", "name", and "email" columns
		const [user] = this.sql`SELECT * FROM users WHERE id = ${userId}`;
		return Response.json(user);
	}
}

type User = {
	id: string;
	name: string;
	email: string;
};

export class MyAgent extends Agent {
	async onRequest(request: Request) {
		let userId = new URL(request.url).searchParams.get("userId");
		// Supply the type parameter to the query when calling this.sql
		// This assumes the results returns one or more User rows with "id", "name", and "email" columns
		const [user] = this.sql<User>`SELECT * FROM users WHERE id = ${userId}`;
		return Response.json(user);
	}
}

无需指定数组类型（User[] 或 Array<User>），this.sql 始终返回指定类型的数组。

Agent 暴露的 SQL API 与 Durable Objects 内类似。可在 Agent 数据库中使用相同 SQL 查询——创建表、查询数据，与 Durable Objects 或 D1 相同。

最佳实践

保持状态精简

状态在每次变更时广播到所有客户端。大数据：

// 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.sql`SELECT * FROM messages ORDER BY created_at DESC LIMIT ${limit}`;
}

乐观更新

为响应式 UI，立即更新客户端 state：

// Client-side
function sendMessage(text) {
	const optimisticMessage = {
		id: crypto.randomUUID(),
		text,
		pending: true,
	};

	// Update immediately
	agent.setState({
		...agent.state,
		messages: [...agent.state.messages, optimisticMessage],
	});

	// Server will confirm/update
}

// Server-side
class MyAgent extends Agent {
	onStateChanged(state, source) {
		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,
				),
			});
		}
	}
}

// Client-side
function sendMessage(text: string) {
	const optimisticMessage = {
		id: crypto.randomUUID(),
		text,
		pending: true,
	};

	// Update immediately
	agent.setState({
		...agent.state,
		messages: [...agent.state.messages, optimisticMessage],
	});

	// Server will confirm/update
}

// Server-side
class MyAgent extends Agent<Env, { messages: Message[] }> {
	onStateChanged(state: GameState, 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,
				),
			});
		}
	}
}

State 与 SQL 对比

使用 State	使用 SQL
UI state（loading、选中项）	历史数据
实时计数	大型集合
活跃会话数据	关系
配置	可查询数据

export class ChatAgent extends Agent {
	// 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) {
		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,
		});
	}
}

export class ChatAgent extends Agent {
	// 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,
		});
	}
}

避免无限循环

注意不要因自己的更新而再次触发 state 更新：

// 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; // Do not react to own updates
  this.setState({ ...state, lastUpdated: Date.now() });
}

将 Agent state 用作 model 上下文

可将 state 与 SQL API 同 Agent 调用 AI 模型的能力结合，将历史上下文纳入 model prompt。现代 LLM 常有极大 context window（可达数百万 token），可直接把相关上下文拉入 prompt。

例如，用 Agent 内置 SQL 拉取历史、带历史查询 model，并在下次调用前追加到历史中：

export class ReasoningAgent extends Agent {
	async callReasoningModel(prompt) {
		let result = this
			.sql`SELECT * FROM history WHERE user = ${prompt.userId} ORDER BY timestamp DESC LIMIT 1000`;
		let context = [];
		for (const row of result) {
			context.push(row.entry);
		}

		const systemPrompt = prompt.system || "You are a helpful assistant.";
		const userPrompt = `${prompt.user}\n\nUser history:\n${context.join("\n")}`;

		try {
			const response = await this.env.AI.run("@cf/zai-org/glm-4.7-flash", {
				messages: [
					{ role: "system", content: systemPrompt },
					{ role: "user", content: userPrompt },
				],
			});

			// Store the response in history
			this
				.sql`INSERT INTO history (timestamp, user, entry) VALUES (${new Date()}, ${prompt.userId}, ${response.response})`;

			return response.response;
		} catch (error) {
			console.error("Error calling reasoning model:", error);
			throw error;
		}
	}
}

interface Env {
	AI: Ai;
}

export class ReasoningAgent extends Agent<Env> {
	async callReasoningModel(prompt: Prompt) {
		let result = this
			.sql<History>`SELECT * FROM history WHERE user = ${prompt.userId} ORDER BY timestamp DESC LIMIT 1000`;
		let context = [];
		for (const row of result) {
			context.push(row.entry);
		}

		const systemPrompt = prompt.system || "You are a helpful assistant.";
		const userPrompt = `${prompt.user}\n\nUser history:\n${context.join("\n")}`;

		try {
			const response = await this.env.AI.run("@cf/zai-org/glm-4.7-flash", {
				messages: [
					{ role: "system", content: systemPrompt },
					{ role: "user", content: userPrompt },
				],
			});

			// Store the response in history
			this
				.sql`INSERT INTO history (timestamp, user, entry) VALUES (${new Date()}, ${prompt.userId}, ${response.response})`;

			return response.response;
		} catch (error) {
			console.error("Error calling reasoning model:", error);
			throw error;
		}
	}
}

这是因为每个 Agent 实例有独立数据库，其中 state 对该 Agent 私有——无论代表单用户、房间/频道还是深度研究工具。默认无需管理争用或访问中心化数据库来获取与存储 state。

API 参考

属性

属性	类型	描述
`state`	`State`	当前 state（getter）
`initialState`	`State`	新 Agent 的默认 state

方法

方法	Signature	描述
`setState`	`(state: State) => void`	更新状态、持久化并广播
`onStateChanged`	`(state: State, source: Connection \| "server") => void`	状态变更时调用
`validateStateChange`	`(nextState: State, source: Connection \| "server") => void`	持久化前校验（抛出则拒绝）

Workflow 步骤方法

方法	描述
`step.updateAgentState(state)`	从 workflow 替换 Agent state
`step.mergeAgentState(partial)`	从 workflow 合并部分 state
`step.resetAgentState()`	从 workflow 重置为 `initialState`