会话 · Cloudflare Agents 文档

Sessions 是支撑 Think harness 的会话存储。你也可以在构建自己的 harness时用它存储对话。

agents/sessions 中的 Sessions 会将对话历史记录存储在 Durable Object 的 SQLite 数据库中。消息以树状结构组织，因此对话可以分支。读取采用流式方式，因此长历史记录无需全部载入内存。压缩会在读取时用摘要替换旧消息，但不会删除这些消息；全文搜索则可按文本查找消息。

选择会话存储

应使用哪些对话 API，取决于你的 harness：

起点	指南
Pi	使用 Pi 的会话 API。Pi 会存储自己的对话记录，因此不需要 `Sessions`。
OpenCode	使用 OpenCode 的会话 API。对话存储由 OpenCode 管理。
Think	`Sessions` 已安装。请从 Think 的对话 API 入手；需要底层存储操作时，再参考此文档。
`AIChatAgent`	`Sessions` 已安装。请从聊天 Agent API 入手；不要重复安装。
在普通 Durable Object 上自行实现循环	如果需要历史记录、分支、搜索和压缩 API，请安装 `Sessions`。你也可以使用自己的存储。

安装 capability

对于普通 Durable Object，请构造 Sessions 并将其注册到 Lifecycle：

src/index.jsjs

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions } from "agents/sessions";

export class ConversationAgent extends DurableObject {
	sessions = new Sessions();
	lifecycle = Lifecycle.install(this).use(this.sessions);
	session = this.sessions.session();

	async onRequest(request) {
		if (request.method === "POST") {
			const message = await request.json();
			const result = await this.session.appendMessage(message, {
				source: "client",
			});
			return Response.json(result, { status: result.inserted ? 201 : 200 });
		}

		return Response.json(await this.session.getRecentHistory(1024 * 1024));
	}
}

src/index.tsts

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions, type SessionMessage } from "agents/sessions";

export class ConversationAgent extends DurableObject<Env> {
	readonly sessions = new Sessions();
	readonly lifecycle = Lifecycle.install(this).use(this.sessions);
	readonly session = this.sessions.session();

	async onRequest(request: Request): Promise<Response> {
		if (request.method === "POST") {
			const message = await request.json<SessionMessage>();
			const result = await this.session.appendMessage(message, {
				source: "client",
			});
			return Response.json(result, { status: result.inserted ? 201 : 200 });
		}

		return Response.json(await this.session.getRecentHistory(1024 * 1024));
	}
}

sessions.session() 会返回一个对话的句柄。如果不提供 ID，它会返回默认会话，其 ID 是空字符串。如果一个 Durable Object 只保存一个对话，请使用默认会话。

在实际应用中，追加消息前请先验证消息结构。source: "client" 选项详见不可信输入。

写入消息

SessionMessage 包含 id、role 和 parts 数组。AI SDK 的 UIMessage 具有相同结构，因此可以直接存储 AI SDK 消息：

async function saveUserMessage(session, text) {
	const result = await session.appendMessage({
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text }],
	});
	// result.inserted is false when a message with this ID already existed.
	return result.message;
}

import type { Session } from "agents/sessions";

async function saveUserMessage(session: Session, text: string) {
	const result = await session.appendMessage({
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text }],
	});
	// result.inserted is false when a message with this ID already existed.
	return result.message;
}

appendMessage() 会解析为包含 inserted 和已存储 message 的 AppendResult。若追加的消息 ID 已存在，则不会重复存储，因此安全地重试追加操作。

其他写入方法包括：

方法	说明
`updateMessage(message)`	替换已存储的消息。若 ID 不在会话中，则解析为 `null`；否则解析为存储后的消息。
`upsertMessage(message)`	若消息存在则更新，否则追加。
`deleteMessages(ids)`	删除消息。已删除消息的子消息会连接到其父消息，因此后续历史记录仍可访问。
`clearMessages()`	删除会话中的所有消息。

不可信输入

对于来自客户端的消息，请传入 source: "client"。此时，能力会移除仅应由服务器设置的元数据键。构造能力时请列出这些键：

import { Sessions } from "agents/sessions";

const sessions = new Sessions({
	reservedMetadataKeys: ["channel", "turnMetadata"],
});

import { Sessions } from "agents/sessions";

const sessions = new Sessions({
	reservedMetadataKeys: ["channel", "turnMetadata"],
});

Sessions 不会限制消息大小。无法放入单个 SQLite 行的消息会拆分到额外行中，并在读取时原样还原，因此一次大型写入可能占用 Durable Object 存储空间的很大一部分。追加客户端发送的内容前，请检查其大小。

图片、音频、PDF 和其他以内嵌字节形式携带数据的非文本部分会存储在消息行之外，并在读取时还原。相同的数据只存储一次。Sessions 仍然不是文件存储服务：处理文件的应用应将文件保存到 R2，并在消息中存放引用。

读取历史记录

历史记录可能很长时，请使用流式读取。history() 会从根消息到最新消息逐条生成消息，并且只在内存中保留一个较小的窗口：

async function countToolCalls(session) {
	let count = 0;
	for await (const message of session.history()) {
		count += message.parts.filter((part) =>
			part.type.startsWith("tool-"),
		).length;
	}
	return count;
}

import type { Session } from "agents/sessions";

async function countToolCalls(session: Session): Promise<number> {
	let count = 0;
	for await (const message of session.history()) {
		count += message.parts.filter((part) =>
			part.type.startsWith("tool-"),
		).length;
	}
	return count;
}

传入 newestFirst: true 可从最新消息开始倒序读取。提前退出循环会使较旧消息保持未读，因此这是查找近期内容最省资源的方式。

读取方法包括：

方法	说明
`history(options?)`	以流的形式读取消息。选项：`leafId`、`newestFirst`、`signal`。
`historyBatches(options?)`	以流的形式读取消息数组。另有 `batchSize` 和 `maxBatchBytes` 选项。
`getRecentHistory(maxBytes, options?)`	返回可容纳在 `maxBytes` 中的最长近期消息序列，并始终包含最新消息。
`getHistory(options?)`	以数组形式返回完整路径。
`getMessage(id)`	返回一条消息，若不存在则返回 `null`。
`getLatestLeaf()`	返回最新消息。
`getHistoryRowStats(leafId?)`	返回每条消息存储的大小和令牌估算值，不会加载内容。

getRecentHistory() 会解析为 { messages, truncated, totalContentBytes }。其字节预算会统计每条消息的完整存储大小，包括拆分行和附件，因此可以限制读取占用的内存。可在启动期间使用它加载最近的一段消息。

对于没有上限的对话，请避免使用 getHistory()，因为它会一次性将所有消息加载到内存中。

分支

一条消息可以有多个子消息。默认情况下，追加操作会将消息连接到最新消息。传入 parentId 可将消息连接到其他位置；传入 null 则会创建新的根消息：

flowchart TB
    accTitle: 两条回复从同一条用户消息分支
    accDescr: 较早的消息之后是用户消息 u1。原始回复 a1 和重新生成的回复 a2 都是 u1 的子消息。后续用户消息 u2 仅延续原始分支。
    M["m1：较早的消息"] --> U["u1：用户消息"]
    U --> A["a1：原始回复"]
    U --> B["a2：重新生成的回复"]
    A --> F["u2：后续消息"]

使用 parentId: "u1" 追加 a2，即可创建另一条回复分支。getBranches("u1") 会返回 a1 和 a2。getHistory({ leafId: "a2" }) 会返回路径 m1 → u1 → a2。若要继续原回复并写入 u2，请使用 parentId: "a1" 追加。

例如，重新生成对用户消息的回复：

async function regenerate(session, userMessageId, reply) {
	// Add a second reply to the same user message.
	await session.appendMessage(reply, { parentId: userMessageId });

	const replies = await session.getBranches(userMessageId);
	const path = await session.getHistory({ leafId: reply.id });
	return { replies, path };
}

import type { Session, SessionMessage } from "agents/sessions";

async function regenerate(
	session: Session,
	userMessageId: string,
	reply: SessionMessage,
) {
	// Add a second reply to the same user message.
	await session.appendMessage(reply, { parentId: userMessageId });

	const replies = await session.getBranches(userMessageId);
	const path = await session.getHistory({ leafId: reply.id });
	return { replies, path };
}

getBranches(id) 会返回一条消息的子消息。所有读取方法都接受 leafId，以读取以该消息结尾的路径。

如需将对话复制到另一个 Durable Object，请使用 history() 读取，然后对每条消息调用 importMessage() 写入。它会按原样存储消息，包括你提供的父消息和时间戳，并且不会发送变更事件：

async function copy(source, destination) {
	let parentId = null;
	for await (const message of source.history()) {
		await destination.importMessage(message, {
			parentId,
			createdAt: message.createdAt?.getTime() ?? Date.now(),
		});
		parentId = message.id;
	}
}

import type { Session } from "agents/sessions";

async function copy(source: Session, destination: Session) {
	let parentId: string | null = null;
	for await (const message of source.history()) {
		await destination.importMessage(message, {
			parentId,
			createdAt: message.createdAt?.getTime() ?? Date.now(),
		});
		parentId = message.id;
	}
}

压缩

压缩可让长对话保持在模型的上下文窗口内。读取时，系统会存储一份摘要并用它替换一段消息。原始消息不会被删除。

createCompactFunction() 会根据 summarize 回调构建压缩函数。使用 onCompaction() 注册该函数，并通过 compactAfter() 设置触发压缩的令牌数：

src/index.jsjs

import { DurableObject } from "cloudflare:workers";
import { generateText } from "ai";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/ai-sdk";
import { Sessions, createCompactFunction } from "agents/sessions";

export class ConversationAgent extends DurableObject {
	ai = createAI({ binding: this.env.AI });
	sessions = new Sessions();
	lifecycle = Lifecycle.install(this).use(this.sessions);
	session = this.sessions
		.session()
		.onCompaction(
			createCompactFunction({
				summarize: async (prompt) => {
					const { text } = await generateText({
						model: this.ai("@cf/zai-org/glm-5.3-flash"),
						prompt,
					});
					return text;
				},
				keepRecentTokens: 20_000,
			}),
		)
		.compactAfter(80_000);
}

src/index.tsts

import { DurableObject } from "cloudflare:workers";
import { generateText } from "ai";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/ai-sdk";
import { Sessions, createCompactFunction } from "agents/sessions";

export class ConversationAgent extends DurableObject<Env> {
	readonly ai = createAI({ binding: this.env.AI });
	readonly sessions = new Sessions();
	readonly lifecycle = Lifecycle.install(this).use(this.sessions);
	readonly session = this.sessions
		.session()
		.onCompaction(
			createCompactFunction({
				summarize: async (prompt) => {
					const { text } = await generateText({
						model: this.ai("@cf/zai-org/glm-5.3-flash"),
						prompt,
					});
					return text;
				},
				keepRecentTokens: 20_000,
			}),
		)
		.compactAfter(80_000);
}

keepRecentTokens 指逐字保留的近期消息令牌数，默认值为 20,000。前三条消息始终保留，至少保留最后两条消息，并且工具调用及其结果不会被拆开。

Sessions 会为每条消息存储令牌估算值，因此检查阈值时无需读取对话。如果自动压缩失败，系统会记录错误并保持对话不变。

你也可以按需压缩：

方法	说明
`compact(leafId?)`	立即运行已注册的压缩函数。
`addCompaction(summary, fromId, toId)`	为一段消息存储你自行生成的摘要。
`getCompactions()`	列出已存储的摘要。

搜索

search(query) 会按文本查找消息：

async function find(session, query) {
	const results = await session.search(query, { limit: 10 });
	return results.map(({ id, role, content }) => ({ id, role, content }));
}

import type { Session } from "agents/sessions";

async function find(session: Session, query: string) {
	const results = await session.search(query, { limit: 10 });
	return results.map(({ id, role, content }) => ({ id, role, content }));
}

第一次调用 search() 时会为已有消息构建全文索引。此后，每次写入都会更新索引。从未执行搜索的对象无需承担索引开销。索引仅包含文本部分。

观察变更

sessions.subscribe() 会报告当前隔离区中所有会话的每次写入。每个事件都包含 type 和 sessionId：

类型	触发时机
`append`, `update`	写入了一条消息。事件中包含该消息。
`delete`, `clear`	删除了消息。
`compact`	`compact()` 存储了一份摘要。
`compaction`	`addCompaction()` 存储了一份摘要。
`import`	`importMessage()` 存储了一条消息。

subscribe() 会返回一个用于取消订阅的函数。

若要让内存中的消息数组与存储保持同步，请使用 session.mirror()。它会自动将追加、更新、删除和清空操作应用到数组：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions } from "agents/sessions";

export class ConversationAgent extends DurableObject {
	sessions = new Sessions();
	lifecycle = Lifecycle.install(this).use(this.sessions);
	session = this.sessions.session();
	messages = [];

	async onStart() {
		this.messages = (await this.session.getRecentHistory(1024 * 1024)).messages;
		this.session.mirror({
			get: () => this.messages,
			set: (messages) => {
				this.messages = messages;
			},
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions, type SessionMessage } from "agents/sessions";

export class ConversationAgent extends DurableObject<Env> {
	readonly sessions = new Sessions();
	readonly lifecycle = Lifecycle.install(this).use(this.sessions);
	readonly session = this.sessions.session();
	messages: SessionMessage[] = [];

	async onStart(): Promise<void> {
		this.messages = (await this.session.getRecentHistory(1024 * 1024)).messages;
		this.session.mirror({
			get: () => this.messages,
			set: (messages) => {
				this.messages = messages;
			},
		});
	}
}

传入 transform 可存储其他形态的数据；传入 intercept 可自行处理事件，例如在压缩或创建分支后重新加载。mirror() 会返回一个用于停止同步的函数。

变更馈送仅在当前隔离区内有效，不是跨对象的事件日志。

在一个对象中使用多个会话

向 session() 传入 ID，即可在一个 Durable Object 中保存多个对话：

import { Sessions } from "agents/sessions";

const sessions = new Sessions();
const support = sessions.session("support");
const sales = sessions.session("sales");

import { Sessions } from "agents/sessions";

const sessions = new Sessions();
const support = sessions.session("support");
const sales = sessions.session("sales");

句柄会被缓存，因此压缩阈值等设置会在多次调用 session() 时保留。系统不提供会话列表；如有需要，请自行维护目录。

对于面向用户的聊天，建议每个对话使用一个 Durable Object。这样每个对话都有自己的存储、独立的故障边界，并且各自休眠。对于草稿、本地分支或自定义命名空间，可以在一个对象中使用多个会话。

存储

Sessions 会将数据存储在名称以 cf_agents_session_ 开头的表中。对于从未执行搜索的对象，追加一条文本消息会写入一行。无法放入单行的消息以及内嵌媒体会使用额外的行。

错误来自你写入的值。无法序列化为 JSON 的消息会抛出 JSON.stringify 产生的 TypeError。消息过大不会触发错误。

在 Agent 类上使用

Think 和 AIChatAgent 使用 Sessions 存储消息。如需为 Agent 添加 Sessions，请将其实例构造为字段，并在构造函数中使用 this.lifecycle.use() 安装。