diff --git a/examples/pub-sub-message-annotations/javascript/README.md b/examples/pub-sub-message-annotations/javascript/README.md index 7a2a50c046..18773b9ad1 100644 --- a/examples/pub-sub-message-annotations/javascript/README.md +++ b/examples/pub-sub-message-annotations/javascript/README.md @@ -67,7 +67,7 @@ This example demonstrates: yarn install ``` -6. Enable the "Annotations, updates and deletes" channel rule that matches the channel name you'll be using (by default we use a channel name of `annotation:pub-sub-message-annotations`, so if using this, [create this rule](https://ably.com/docs/channels#rules) for the "annotation" channel namespace). +6. Enable the "Annotations, updates, deletes, and appends" channel rule that matches the channel name you'll be using (by default we use a channel name of `annotation:pub-sub-message-annotations`, so if using this, [create this rule](https://ably.com/docs/channels#rules) for the "annotation" channel namespace). 7. Run the server: diff --git a/examples/pub-sub-message-annotations/javascript/src/config.ts b/examples/pub-sub-message-annotations/javascript/src/config.ts index 5a0ffcf0ae..32996b5600 100644 --- a/examples/pub-sub-message-annotations/javascript/src/config.ts +++ b/examples/pub-sub-message-annotations/javascript/src/config.ts @@ -1,6 +1,6 @@ export const urlParams = new URLSearchParams(window.location.search); export const clientId = urlParams.get('clientId') || 'user1'; -// Remember to enable the "Annotations, updates, and deletes" channel rule for the channel +// Remember to enable the "Annotations, updates, deletes, and appends" channel rule for the channel // namespace you're using (the first colon-delimited segment, here, "annotation") export const channelName = `annotation:${import.meta.env.VITE_NAME ?? 'annotation:pub-sub-message-annotations'}`; diff --git a/examples/pub-sub-message-annotations/javascript/src/script.ts b/examples/pub-sub-message-annotations/javascript/src/script.ts index 924db6e11c..67af06ca0d 100644 --- a/examples/pub-sub-message-annotations/javascript/src/script.ts +++ b/examples/pub-sub-message-annotations/javascript/src/script.ts @@ -27,7 +27,9 @@ async function main() { // Regular messages will be received as message.create events. getChannel().subscribe((message) => { if (!hasSerial(message)) { - console.error('Received message without serial (this indicates that you need to enable the "Annotations, updates, and deletes" feature in channel rules)'); + console.error( + 'Received message without serial (this indicates that you need to enable the "Annotations, updates, deletes, and appends" feature in channel rules)', + ); return; } if (message.action === 'message.summary') { diff --git a/src/components/Examples/ExamplesGrid.tsx b/src/components/Examples/ExamplesGrid.tsx index c372f8ee84..1fde8d68dc 100644 --- a/src/components/Examples/ExamplesGrid.tsx +++ b/src/components/Examples/ExamplesGrid.tsx @@ -32,6 +32,8 @@ const ExamplesGrid = ({ return 'text-blue-600'; case 'liveObjects': return 'text-green-600'; + case 'aiTransport': + return 'text-cyan-500'; default: return 'text-orange-700'; } diff --git a/src/data/content/homepage.ts b/src/data/content/homepage.ts index 71c113c8ea..f25ece224d 100644 --- a/src/data/content/homepage.ts +++ b/src/data/content/homepage.ts @@ -38,6 +38,10 @@ export default { name: 'liveSync', link: '/docs/livesync', }, + { + name: 'aiTransport', + link: '/docs/ai-transport', + }, ], }, examples: { diff --git a/src/data/examples/index.ts b/src/data/examples/index.ts index ce75f4d31a..780f96c66b 100644 --- a/src/data/examples/index.ts +++ b/src/data/examples/index.ts @@ -287,6 +287,9 @@ export const products = { spaces: { label: 'Spaces', }, + aitransport: { + label: 'AI Transport', + }, }; const useCasesList = [ diff --git a/src/data/index.ts b/src/data/index.ts index b7bd28c49a..c36331ceb1 100644 --- a/src/data/index.ts +++ b/src/data/index.ts @@ -1,4 +1,12 @@ -import { chatNavData, liveObjectsNavData, liveSyncNavData, platformNavData, pubsubNavData, spacesNavData } from './nav'; +import { + aiTransportNavData, + chatNavData, + liveObjectsNavData, + liveSyncNavData, + platformNavData, + pubsubNavData, + spacesNavData, +} from './nav'; import { languageData } from './languages'; import { PageData, ProductData } from './types'; import homepageContentData from './content/homepage'; @@ -16,6 +24,10 @@ export const productData = { nav: chatNavData, languages: languageData.chat, }, + aiTransport: { + nav: aiTransportNavData, + languages: languageData.aiTransport, + }, spaces: { nav: spacesNavData, languages: languageData.spaces, diff --git a/src/data/languages/languageData.ts b/src/data/languages/languageData.ts index 605d7eac9d..cd416f9c07 100644 --- a/src/data/languages/languageData.ts +++ b/src/data/languages/languageData.ts @@ -29,6 +29,9 @@ export default { swift: '1.0', kotlin: '1.0', }, + aiTransport: { + javascript: '2.11', + }, spaces: { javascript: '0.4', react: '0.4', diff --git a/src/data/nav/aitransport.ts b/src/data/nav/aitransport.ts new file mode 100644 index 0000000000..cb8eeb5f11 --- /dev/null +++ b/src/data/nav/aitransport.ts @@ -0,0 +1,70 @@ +import { NavProduct } from './types'; + +export default { + name: 'Ably AI Transport', + link: '/docs/ai-transport', + icon: { + closed: 'icon-gui-prod-ai-transport-outline', + open: 'icon-gui-prod-ai-transport-solid', + }, + content: [ + { + name: 'Introduction', + pages: [ + { + name: 'About AI Transport', + link: '/docs/ai-transport', + index: true, + }, + ], + }, + { + name: 'Token streaming', + pages: [ + { + name: 'Message per response', + link: '/docs/ai-transport/features/token-streaming/message-per-response', + }, + { + name: 'Message per token', + link: '/docs/ai-transport/features/token-streaming/message-per-token', + }, + ], + }, + { + name: 'Sessions & Identity', + pages: [ + { + name: 'Overview', + link: '/docs/ai-transport/features/sessions-identity', + }, + { + name: 'Identifying users and agents', + link: '/docs/ai-transport/features/sessions-identity/identifying-users-and-agents', + }, + { + name: 'Online status', + link: '/docs/ai-transport/features/sessions-identity/online-status', + }, + { + name: 'Resuming sessions', + link: '/docs/ai-transport/features/sessions-identity/resuming-sessions', + }, + ], + }, + { + name: 'Guides', + pages: [ + { + name: 'OpenAI token streaming - message per token', + link: '/docs/guides/ai-transport/openai-message-per-token', + }, + { + name: 'OpenAI token streaming - message per response', + link: '/docs/guides/ai-transport/openai-message-per-response', + }, + ], + }, + ], + api: [], +} satisfies NavProduct; diff --git a/src/data/nav/index.ts b/src/data/nav/index.ts index aac3975f22..e5ed49fddb 100644 --- a/src/data/nav/index.ts +++ b/src/data/nav/index.ts @@ -1,8 +1,17 @@ import platformNavData from './platform'; import pubsubNavData from './pubsub'; import chatNavData from './chat'; +import aiTransportNavData from './aitransport'; import liveObjectsNavData from './liveobjects'; import spacesNavData from './spaces'; import liveSyncNavData from './livesync'; -export { platformNavData, pubsubNavData, chatNavData, liveObjectsNavData, spacesNavData, liveSyncNavData }; +export { + platformNavData, + pubsubNavData, + chatNavData, + aiTransportNavData, + liveObjectsNavData, + spacesNavData, + liveSyncNavData, +}; diff --git a/src/data/types.ts b/src/data/types.ts index 0884a04a8b..a9c2b2977b 100644 --- a/src/data/types.ts +++ b/src/data/types.ts @@ -3,7 +3,7 @@ import { LanguageData } from './languages/types'; import { NavProduct } from './nav/types'; const pageKeys = ['homepage'] as const; -const productKeys = ['platform', 'pubsub', 'chat', 'spaces', 'liveObjects', 'liveSync'] as const; +const productKeys = ['platform', 'pubsub', 'chat', 'aiTransport', 'spaces', 'liveObjects', 'liveSync'] as const; export type ProductKey = (typeof productKeys)[number]; type PageKey = (typeof pageKeys)[number]; diff --git a/src/images/content/diagrams/ai-transport-before-and-after.png b/src/images/content/diagrams/ai-transport-before-and-after.png new file mode 100644 index 0000000000..d29ae4a6f1 Binary files /dev/null and b/src/images/content/diagrams/ai-transport-before-and-after.png differ diff --git a/src/pages/docs/ai-transport/features/sessions-identity/identifying-users-and-agents.mdx b/src/pages/docs/ai-transport/features/sessions-identity/identifying-users-and-agents.mdx new file mode 100644 index 0000000000..4614747840 --- /dev/null +++ b/src/pages/docs/ai-transport/features/sessions-identity/identifying-users-and-agents.mdx @@ -0,0 +1,419 @@ +--- +title: "Identifying users and agents" +meta_description: "Establish trusted identity and roles in decoupled AI sessions" +meta_keywords: "user authentication, agent identity, JWT authentication, token authentication, verified identity, capabilities, authorization, user claims, RBAC, role-based access control, API key authentication, message attribution" +--- + +Secure AI applications require agents to trust who sent each message and understand what that sender is authorized to do. Ably's identity system uses token-based authentication to provide cryptographically-verified identities with custom attributes that you can access throughout your applications. + +## Why identity matters + +In decoupled architectures, identity serves several critical purposes: + +- Prevent spoofing: Without verified identity, malicious users could impersonate others by claiming to be someone else. Ably supports cryptographically binding each client's identity to their credentials, making spoofing impossible. +- Message attribution: Agents need to know whether messages come from users or other agents. This is essential for conversation flows in which agent responses should be securely distinguished from user prompts. +- Personalized behavior: Different users may have different privileges or attributes. A premium user might get access to more capable models, while a free user gets basic functionality. Ably allows your trusted authentication server to embed this information in the client's credentials, allowing this information to be securely passed to agents. +- Authorization decisions: Some operations should only be performed for specific users. For example, human-in-the-loop (HITL) tool calls that access sensitive data might require admin privileges. Ably allows agents to verify the privilege level and role of the user resolving the tool call. + +## Authenticating users + +Use [token authentication](/docs/auth/token) to authenticate users securely. Your authentication server generates a token that is signed with the secret part of your Ably API key. Clients use this token to connect to Ably, and the token signature ensures it cannot be tampered with. + +The following examples use [JWT authentication](/docs/auth/token#jwt) for its simplicity and standard tooling support. For other approaches, see [token authentication](/docs/auth/token). + +Create a server endpoint that generates signed JWTs after verifying user authentication: + + +```javascript +// Server code +import express from "express"; +import jwt from "jsonwebtoken"; + +const app = express(); + +// Mock authentication middleware. +// This should be replaced with your actual authentication logic. +function authenticateUser(req, res, next) { + // Assign a mock user ID for demonstration + req.session = { userId: "user123" }; + next(); +} + +// Return the claims payload to embed in the signed JWT. +function getJWTClaims(userId) { + // Returns an empty payload, so the token + // inherits the capabilities of the signing key. + return {}; +} + +// Define an auth endpoint used by the client to obtain a signed JWT +// which it can use to authenticate with the Ably service. +app.get("/api/auth/token", authenticateUser, (req, res) => { + const [keyName, keySecret] = "{{API_KEY}}".split(":"); + + // Sign a JWT using the secret part of the Ably API key. + const token = jwt.sign(getJWTClaims(req.session.userId), keySecret, { + algorithm: "HS256", + keyid: keyName, + expiresIn: "1h", + }); + + res.type("application/jwt").send(token); +}); + +app.listen(3001); +``` + + + + +The JWT is signed with the secret part of your Ably API key using [HMAC-SHA-256](https://datatracker.ietf.org/doc/html/rfc4868). This example does not embed any claims in the JWT payload, so by default the token inherits the capabilities of the Ably API key used to sign the token. + +Configure your client to obtain a signed JWT from your server endpoint using an [`authCallback`](/docs/auth/token#auth-callback). The client obtains a signed JWT from the callback and uses it to authenticate requests to Ably. The client automatically makes a request for a new token before it expires. + + + + +```javascript +// Client code +import * as Ably from "ably"; + +const ably = new Ably.Realtime({ + authCallback: async (tokenParams, callback) => { + try { + const response = await fetch("/api/auth/token"); + const token = await response.text(); + callback(null, token); + } catch (error) { + callback(error, null); + } + } +}); + +ably.connection.on("connected", () => { + console.log("Connected to Ably"); +}); +``` + + +## Authenticating agents + +Agents typically run on servers in trusted environments where API keys can be securely stored. Use [API key authentication](/docs/auth#basic-authentication) to authenticate agents directly with Ably. + + +```javascript +// Agent code +import * as Ably from "ably"; + +const ably = new Ably.Realtime({ + key: "{{API_KEY}}" +}); + +ably.connection.on("connected", () => { + console.log("Connected to Ably"); +}); +``` + + + + + + +## Specifying capabilities + +Use [capabilities](/docs/auth/capabilities) to specify which operations clients can perform on which channels. This applies to both users and agents, allowing you to enforce fine-grained permissions. + +### User capabilities + +Add the [`x-ably-capability`](/docs/api/realtime-sdk/authentication#ably-jwt) claim to your JWT to specify the allowed capabilities of a client. This allows you to enforce fine-grained permissions, such as restricting some users to only subscribe to messages while allowing others to publish. + +Update your `getJWTClaims` function to specify the allowed capabilities for the authenticated user: + + +```javascript +// Server code + +// Return the claims payload to embed in the signed JWT. +// Includes the `x-ably-capabilities` claim, which controls +// which operations the user can perform on which channels. +function getJWTClaims(userId) { + const orgId = "acme"; // Mock organization ID for demonstration + const capabilities = { + // The user can publish and subscribe to channels within the organization, + // that is, any channel matching `org:acme:*`. + [`org:${orgId}:*`]: ["publish", "subscribe"], + // The user can only subscribe to the `announcements` channel. + announcements: ["subscribe"], + }; + return { + "x-ably-capability": JSON.stringify(capabilities), + }; +} +``` + + +When a client authenticates with this token, Ably enforces these capabilities server-side. Any attempt to perform unauthorized operations will be rejected. For example, a client with the capabilities above can publish to channels prefixed with `org:acme:`, but an attempt to publish to a channel prefixed with `org:foobar:` will fail with error code [`40160`](/docs/platform/errors/codes#40160): + + +```javascript +// Client code +const acmeChannel = ably.channels.get("org:acme:{{RANDOM_CHANNEL_NAME}}"); +await acmeChannel.publish("prompt", "What is the weather like today?"); // succeeds + +const foobarChannel = ably.channels.get("org:foobar:{{RANDOM_CHANNEL_NAME}}"); +await foobarChannel.publish("prompt", "What is the weather like today?"); // fails + +const announcementsChannel = ably.channels.get("announcements"); +await announcementsChannel.publish("prompt", "What is the weather like today?"); // fails +await announcementsChannel.subscribe((msg) => console.log(msg)); // succeeds +``` + + + + +### Agent capabilities + +When using API key authentication, provision API keys through the [Ably dashboard](https://ably.com/dashboard) or [Control API](/docs/account/control-api) with only the capabilities required by the agent. + +The following example uses the Control API to create an API key with specific capabilities for a weather agent: + + + + +```shell +curl --location --request POST 'https://control.ably.net/v1/apps/{{APP_ID}}/keys' \ +--header 'Content-Type: application/json' \ +--header 'Authorization: Bearer ${ACCESS_TOKEN}' \ +--data-raw '{ + "name": "weather-agent-key", + "capability": { + "org:acme:weather:*": ["publish", "subscribe"] + } +}' +``` + + +This creates an API key that can only publish and subscribe on channels matching `org:acme:weather:*`. The agent can then use this key to authenticate: + + +```javascript +// Agent code +const weatherChannel = ably.channels.get("org:acme:weather:{{RANDOM_CHANNEL_NAME}}"); +await weatherChannel.subscribe((msg) => console.log(msg)); // succeeds +await weatherChannel.publish("update", "It's raining in London"); // succeeds + +const otherChannel = ably.channels.get("org:acme:other:{{RANDOM_CHANNEL_NAME}}"); +await otherChannel.subscribe((msg) => console.log(msg)); // fails +await otherChannel.publish("update", "It's raining in London"); // fails +``` + + + + +## Establishing verified identity + +Use the [`clientId`](/docs/messages#properties) to identify the user or agent that published a message. The method for setting `clientId` depends on your authentication approach: + +- When using [basic authentication](/docs/auth/identified-clients#basic), specify the `clientId` directly in the client options when instantiating the client instance. +- When using [token authentication](/docs/auth/identified-clients#token), specify an explicit `clientId` when issuing the token. + +### User identity + +Users typically authenticate using [token authentication](/docs/auth/identified-clients#token). Add the [`x-ably-clientId`](/docs/api/realtime-sdk/authentication#ably-jwt) claim to your JWT to establish a verified identity for each user client. This identity appears as the [`clientId`](/docs/messages#properties) in all messages the user publishes, and subscribers can trust this identity because only your server can issue JWTs with specific `clientId` values. + +As with all clients, the method for setting `clientId` depends on your [authentication approach](#identity). + +Update your `getJWTClaims` function to specify a `clientId` for the user: + + +```javascript +// Return the claims payload to embed in the signed JWT. +function getJWTClaims(userId) { + // Returns a payload with the `x-ably-clientId` claim, which ensures + // that the user's ID appears as the `clientId` on all messages + // published by the client using this token. + return { "x-ably-clientId": userId }; +} +``` + + +When a client authenticates using this token, Ably's servers automatically attach the `clientId` specified in the token to every message the user publishes: + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Publish a message - the clientId is automatically attached +await channel.publish("prompt", "What is the weather like today?"); +``` + + +Agents can then access this verified identity to identify the sender: + + +```javascript +// Agent code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Subscribe to messages from clients +await channel.subscribe("prompt", (message) => { + // Access the verified clientId from the message + const userId = message.clientId; + const prompt = message.data; + + console.log(`Received message from user: ${userId}`); + console.log(`Prompt:`, prompt); +}); +``` + + +The `clientId` in the message can be trusted, so agents can use this identity to make decisions about what actions the user can take. For example, agents can check user permissions before executing tool calls, route messages to appropriate AI models based on subscription tiers, or maintain per-user conversation history and context. + +### Agent identity + +Agent code typically runs in a trusted environment, so you can use [basic authentication](/docs/auth/identified-clients#basic) and directly specify the `clientId` when instantiating the agent client. This identity appears as the [`clientId`](/docs/messages#properties) in all messages the agent publishes, allowing subscribers to identify the agent which published a message. + + +```javascript +// Agent code +import * as Ably from "ably"; + +const ably = new Ably.Realtime({ + key: "{{API_KEY}}", + // Specify an identity for this agent + clientId: "weather-agent" +}); +``` + + +When subscribers receive messages, they can use the `clientId` to determine which agent published the message: + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +await channel.subscribe((message) => { + if (message.clientId === "weather-agent") { + console.log("Weather agent response:", message.data); + } +}); +``` + + + + +## Adding roles and attributes + +Embed custom roles and attributes in messages to enable role-based access control (RBAC) and convey additional context about users and agents. This enables agents to make authorization decisions without additional database lookups. + +### User claims + +Use [authenticated claims for users](/docs/auth/capabilities#custom-restrictions-on-channels-) to embed custom claims in JWTs that represent user roles or attributes. + +Add claims with names matching the `ably.channel.*` pattern to your JWT to specify user claims for specific channels. Claims can be scoped to individual channels or to [namespaces](/docs/channels#namespaces) of channels. The most specific user claim matching the channel is automatically included under `extras.userClaim` in all messages the client publishes. + +Update your `getJWTClaims` function to specify some user claims: + + +```javascript +// Return the claims payload to embed in the signed JWT. +function getJWTClaims(userId) { + // Returns a payload with `ably.channel.*` claims, which ensures that + // the most specific claim appears as the `message.extras.userClaim` + // on all messages published by the client using this token. + return { + // The user is an editor on all acme channels. + "ably.channel.org:acme:*": "editor", + // The user is a guest on all other channels. + "ably.channel.*": "guest", + }; +} +``` + + +When a client authenticates with a JWT containing `ably.channel.*` claims, Ably automatically includes the most specific matching claim value in the `message.extras.userClaim` field on messages published by the client: + + +```javascript +// Agent code +const channel = ably.channels.get("org:acme:{{RANDOM_CHANNEL_NAME}}"); + +// Subscribe to user prompts +await channel.subscribe("prompt", async (message) => { + // Access the user's role from the user claim in message extras + const role = message.extras?.userClaim; + + console.log(`Message from user with role: ${role}`); +}); +``` + + +The `message.extras.userClaim` in the message can be trusted, so agents can rely on this information to make decisions about what actions the user can take. For example, an agent could allow users with an "editor" role to execute tool calls that modify documents, while restricting users with a "guest" role to read-only operations. + +### Agent metadata + +Use [`message.extras.headers`](/docs/api/realtime-sdk/types#extras) to include custom metadata in agent messages, such as agent roles or attributes. + +Agents can directly specify metadata in `message.extras.headers`. Since agents run as trusted code in server environments, this metadata can be trusted by subscribers. This is useful for communicating agent characteristics, such as which model the agent uses, the agent's role in a multi-agent system, or version information. + + + + +```javascript +// Agent code +import * as Ably from "ably"; + +const ably = new Ably.Realtime({ + key: "{{API_KEY}}", + clientId: "weather-agent" +}); + +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +await channel.publish({ + name: "update", + data: "It's raining in London", + extras: { + headers: { + model: "gpt-4" + } + } +}); +``` + + +Clients and other agents can access this metadata when messages are received: + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +await channel.subscribe((message) => { + if (message.clientId === "weather-agent") { + const model = message.extras?.headers?.model; + console.log(`Response from weather agent using ${model}:`, message.data); + } +}); +``` + diff --git a/src/pages/docs/ai-transport/features/sessions-identity/index.mdx b/src/pages/docs/ai-transport/features/sessions-identity/index.mdx new file mode 100644 index 0000000000..e1b9be5a05 --- /dev/null +++ b/src/pages/docs/ai-transport/features/sessions-identity/index.mdx @@ -0,0 +1,63 @@ +--- +title: "Sessions & identity overview" +meta_description: "Manage session lifecycle and identity in decoupled AI architectures" +meta_keywords: "AI sessions, session management, channel-oriented sessions, connection-oriented sessions, session persistence, session lifecycle, identity management, decoupled architecture, session resumption, multi-device, multi-user" +--- + +Ably AI Transport provides robust session management and identity capabilities designed for modern AI applications. Sessions persist beyond individual connections, enabling agents and clients to connect independently through shared channels. Built-in token-based authentication provides verified user identity and fine-grained authorization for channel operations. + +## What is a session? + +A session is an interaction between a user (or multiple users) and an AI agent where messages and data are exchanged, building up shared context over time. In AI Transport, sessions are designed to persist beyond the boundaries of individual connections, enabling modern AI experiences where users expect to: + +- Resume conversations across devices: Start a conversation on mobile and seamlessly continue on desktop with full context preserved +- Return to long-running work: Close the browser while agents continue processing in the background, delivering results when you return +- Recover from interruptions: Experience connection drops, browser refreshes, or network instability without losing conversation progress +- Collaborate in shared sessions: Multiple users can participate in the same conversation simultaneously and remain in sync + +These capabilities represent a fundamental shift from traditional request/response AI experiences to continuous, resumable interactions that remain accessible across all user devices and locations. Sessions have a lifecycle: they begin when a user starts interacting with an agent, remain active while the interaction continues, and can persist even when users disconnect - enabling truly asynchronous AI workflows. + +Managing this lifecycle in AI Transport's decoupled architecture involves detecting when users are present, deciding when to stop or continue agent work, and handling scenarios where users disconnect and return. + +## Connection-oriented vs channel-oriented sessions + +In traditional connection-oriented architectures, sessions are bound to the lifecycle of a WebSocket or SSE connection: + +1. Client opens connection to agent server to establish a session +2. Agent streams response over the connection +3. When the connection closes, the session ends + +This tight coupling means network interruptions terminate sessions, agents cannot continue work after disconnections, and supporting multiple devices or users introduces significant complexity. + +AI Transport uses a channel-oriented model where sessions persist independently of individual connections. Clients and agents communicate through [Channels](/docs/channels): + +1. Client sends a single request to agent server to establish a session +2. Server responds with a unique ID for the session, which is used to identify the channel +3. All further communication happens over the channel + +In this model, sessions are associated with the channel, enabling seamless reconnection, background agent work, and multi-device access without additional complexity. + +![AI Transport architecture](../../../../../images/content/diagrams/ai-transport-before-and-after.png) + + +The channel-oriented model provides key benefits for modern AI applications: sessions maintain continuity in the face of disconnections, users can refresh or navigate back to the ongoing session, multiple users or devices can participate in the same session, and agents can continue long-running or asynchronous workloads even when clients disconnect. + +The following table compares how each architecture addresses the engineering challenges of delivering these capabilities: + +| Challenge | Connection-oriented sessions | Channel-oriented sessions | +|-----------|------------------------------|---------------------------| +| Routing | Agents must track which instance holds each session. Reconnecting clients need routing logic to find the correct agent instance across your infrastructure. | Agents and clients only need the channel name. Ably handles message delivery to all subscribers without agents tracking sessions or implementing routing logic. | +| Message resume | Agents must buffer sent messages and implement replay logic. When clients reconnect, agents must determine what was missed and retransmit without duplicates or gaps, distinctly for each connection. | When clients reconnect, they automatically receive messages published while disconnected. The channel maintains history without agents implementing buffering or replay logic, eliminating the need for server-side session state. | +| Abandonment detection | Agents must implement logic to distinguish between brief network interruptions and users who have actually left, so they can decide whether to continue work or clean up resources. | Built-in presence tracking signals when users enter and leave channels, providing clear lifecycle events to agents without custom detection logic. | +| Multi-user and multi-device | Agents must manage multiple concurrent connections from the same user across devices, or from multiple users in collaborative sessions. This requires tracking connections, synchronizing state, and ensuring all participants receive consistent updates. | Multiple users and devices can connect to the same channel. The channel handles message delivery to all participants, simplifying agent logic for multi-user and multi-device scenarios. | + +## Identity in channel-oriented sessions + +In connection-oriented architectures, the agent server handles authentication directly when establishing the connection. When the connection is opened, the server verifies credentials and associates the authenticated user identity with that specific connection. + +In channel-oriented sessions, agents don't manage connections or handle authentication directly. Instead, your server authenticates users and issues tokens that control their access to channels. Ably enforces these authorization rules and provides verified identity information to agents, giving you powerful capabilities for managing who can participate in sessions and what they can do: + +- Verified identity: Agents automatically receive the authenticated identity of message senders, with cryptographic guarantees that identities cannot be spoofed +- Fine-grained authorization: Control precisely what operations each user can perform on specific channels through fine-grained capabilities +- Rich user attributes: Pass authenticated user data to agents for personalized behavior without building custom token systems +- Role-based participation: Distinguish between different types of participants, such as users and agents, to customize behaviour based on their role diff --git a/src/pages/docs/ai-transport/features/sessions-identity/online-status.mdx b/src/pages/docs/ai-transport/features/sessions-identity/online-status.mdx new file mode 100644 index 0000000000..7a852d482f --- /dev/null +++ b/src/pages/docs/ai-transport/features/sessions-identity/online-status.mdx @@ -0,0 +1,280 @@ +--- +title: "Online status" +meta_description: "Use Ably Presence to show which users and agents are currently connected to an AI session" +meta_keywords: "presence, online status, multi-device, multi-user, session abandonment, async workflows" +--- + +Modern AI applications require agents to know when users are online, when they've fully disconnected, and how to handle users connected across multiple devices. Ably's [Presence](/docs/presence-occupancy/presence) feature provides realtime online status with automatic lifecycle management, allowing agents to decide when to continue processing, when to wait for user input, and when to clean up resources. Presence detects which users and agents are currently connected to a session, distinguishes between a single device disconnecting and a user going completely offline, and enables responsive online/offline indicators. + +## Why online status matters + +In channel-oriented sessions, online status serves several critical purposes: + +- Session abandonment detection: Agents need to know when users have fully disconnected to decide whether to continue processing, pause work, or clean up resources. Presence provides reliable signals when all of a user's devices have left the session. +- Multi-device coordination: A single user can connect from multiple devices simultaneously. Presence tracks each connection separately while maintaining stable identity across devices, allowing you to distinguish between "one device left" and "user completely offline". +- Agent availability signaling: Clients need to know when agents are online and ready to process requests. Agents can enter presence to advertise availability and leave when they complete work or shut down. +- Collaborative session awareness: In sessions with multiple users, participants can see who else is currently present. This enables realtime collaboration features and helps users understand the current session context. + +## Going online + +Use the [`enter()`](/docs/presence-occupancy/presence#enter) method to signal that a user or agent is online. When a client enters presence, they are added to the presence set and identified by their `clientId`. You can optionally include data when entering presence to communicate additional context. + + + +You have flexibility in when to enter presence. For example, an agent might choose to appear as online only while processing a specific task, or remain present for the duration of the entire session. Users typically enter presence when they connect to a session and remain present until they disconnect. + + + +For example, a user client can enter presence when joining a session: + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Enter presence with metadata about the user's device +await channel.presence.enter({ + device: "mobile", + platform: "ios" +}); +``` + + +Similarly, an agent can enter presence to signal that it's online: + + +```javascript +// Agent code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Enter presence with metadata about the agent +await channel.presence.enter({ + model: "gpt-4" +}); +``` + + +### Going online from multiple devices + +A single user can be present on a channel from multiple devices simultaneously. Ably tracks each connection separately using a unique [`connectionId`](/docs/connect#connection-ids), while maintaining the same [`clientId`](/docs/auth/identified-clients#assign) across all connections. + +When a user connects from multiple devices, each device enters presence independently. All connections share the same `clientId` but have different `connectionId` values. + +For example, when the user connects from their desktop browser: + + +```javascript +// Client code (device 1: desktop browser) +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); +await channel.presence.enter({ device: "desktop" }); +``` + + +And then connects from their mobile app while still connected on desktop: + + +```javascript +// Client code (device 2: mobile app) +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); +await channel.presence.enter({ device: "mobile" }); +``` + + +Both devices are now members of the presence set with the same `clientId` but different `connectionId` values. When you query the presence set, you'll see two separate entries: + + +```javascript +// Query presence to see both devices +const members = await channel.presence.get(); +for (const { clientId, connectionId, data } of members) { + console.log(clientId, connectionId, data); +} +// Example output: +// user-123 hd67s4!abcdef-0 { device: "desktop" } +// user-123 hd67s4!ghijkl-1 { device: "mobile" } +``` + + +When either device leaves or disconnects, the other device remains in the presence set. + +## Going offline + +Clients can go offline in two ways: explicitly by calling the leave method, or automatically when Ably detects a disconnection. + +### Explicitly going offline + +Use the [`leave()`](/docs/presence-occupancy/presence#leave) method when a user or agent wants to mark themselves as offline. This immediately notifies presence subscribers on the channel and removes the entry from the presence set, even if they remain connected to Ably. + + + +For example, a user client can explicitly leave presence: + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Leave presence when the user marks themselves offline +await channel.presence.leave(); +``` + + +Similarly, an agent can leave presence when it completes its work or shuts down: + + +```javascript +// Agent code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Leave presence when the agent shuts down +await channel.presence.leave(); +``` + + +Optionally include data when leaving presence to communicate the reason for going offline. This data is delivered to presence subscribers listening to `leave` events and is also available in [presence history](/docs/presence-occupancy/presence#history): + + +```javascript +// Leave with a reason +await channel.presence.leave({ + reason: "session-completed", + timestamp: Date.now() +}); +``` + + +Subscribers receive the `leave` data in the presence message: + + +```javascript +// Subscribe to leave events to see why members left +await channel.presence.subscribe("leave", (presenceMessage) => { + console.log(`${presenceMessage.clientId} left`); + if (presenceMessage.data) { + console.log(`Reason: ${presenceMessage.data.reason}`); + } +}); +``` + + +### Going offline after disconnection + +When a client loses connection unexpectedly, Ably detects the lost connection and automatically leaves the client from the presence set. + +By default, clients remain present for 15 seconds after an abrupt disconnection. This prevents excessive enter/leave events during brief network interruptions. If the client reconnects within this window, they remain in the presence set without triggering leave and reenter events. + +Use the `transportParams` [client option](/docs/api/realtime-sdk#client-options) to configure disconnection detection and presence lifecycle behaviour. After an abrupt disconnection, the `heartbeatInterval` transport parameter controls how quickly Ably detects the dead connection, while the `remainPresentFor` option controls how long the member is kept in presence before Ably emits the leave event. + + + +For example, if implementing resumable agents using techniques such as durable execution, configure a longer `remainPresentFor` period to allow time for the new agent instance to come online and resume processing before the previous instance appears as offline. This provides a seamless handoff: + + +```javascript +// Agent code +const ably = new Ably.Realtime({ + key: "{{API_KEY}}", + clientId: "weather-agent", + // Allow 30 seconds for agent resume and reconnection + transportParams: { + remainPresentFor: 30000 + } +}); +``` + + +## Viewing who is online + +Participants in a session can query the current presence set or subscribe to presence events to see who else is online and react to changes in realtime. Users might want to see which agents are processing work, while agents might want to detect when specific users are offline to pause or cancel work. + + + +### Retrieving current presence members + +Use [`presence.get()`](/docs/api/realtime-sdk/presence#get) to retrieve the current list of users and agents in the session. Each presence member is uniquely identified by the combination of their `clientId` and `connectionId`. This is useful for showing who is currently available or checking if a specific participant is online before taking action. + + +```javascript +// Get all currently present members +const members = await channel.presence.get(); + +// Display each member - the same user will appear once per distinct connection +members.forEach((member) => { + console.log(`${member.clientId} (connection: ${member.connectionId})`); +}); +``` + + +### Subscribing to presence changes + +Use [`presence.subscribe()`](/docs/api/realtime-sdk/presence#subscribe) to receive realtime notifications when users or agents enter or leave the session. This enables building responsive UIs that show online users, or implementing agent logic that reacts to user connectivity changes. + + +```javascript +// Client code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +// Subscribe to changes to the presence set +await channel.presence.subscribe(async (presenceMessage) => { + // Get the current synced presence set after any change + const members = await channel.presence.get(); + + // Display each member - the same user will appear once per distinct connection + members.forEach((member) => { + console.log(`${member.clientId} (connection: ${member.connectionId})`); + }); +}); +``` + + +You can also subscribe to specific presence events: + + +```javascript +// Subscribe only to enter events +await channel.presence.subscribe("enter", (presenceMessage) => { + console.log(`${presenceMessage.clientId} joined on connection ${presenceMessage.connectionId}`); +}); + +// Subscribe only to leave events +await channel.presence.subscribe("leave", (presenceMessage) => { + console.log(`${presenceMessage.clientId} left on connection ${presenceMessage.connectionId}`); +}); +``` + + +### Detecting when a user is offline on all devices + +Agents can monitor presence changes to detect when a specific user has gone completely offline across all devices. This is useful for deciding whether to pause expensive operations, cancel ongoing work, deprioritize tasks, or schedule work for later. + + +```javascript +// Agent code +const channel = ably.channels.get("{{RANDOM_CHANNEL_NAME}}"); + +await channel.presence.subscribe(async (presenceMessage) => { + // Get the current synced presence set + const members = await channel.presence.get(); + + // Check if all clients are offline + if (members.length === 0) { + console.log(`All clients are offline`); + } + + // Check if a specific client is offline + if (!members.map(m => m.clientId).includes(targetUserId)) { + console.log(`${targetUserId} is now offline on all devices`); + } +}); +``` + diff --git a/src/pages/docs/ai-transport/features/sessions-identity/resuming-sessions.mdx b/src/pages/docs/ai-transport/features/sessions-identity/resuming-sessions.mdx new file mode 100644 index 0000000000..6883634eee --- /dev/null +++ b/src/pages/docs/ai-transport/features/sessions-identity/resuming-sessions.mdx @@ -0,0 +1,140 @@ +--- +title: Resuming sessions +description: How clients and agents reconnect to ongoing AI Transport sessions after network interruptions or service restarts +meta_keywords: "session resumption, reconnection, hydration, presence sync, conversation history, channel history, untilAttach, durable execution, agent restart, message recovery, failover" +--- + +AI Transport uses a channel-oriented model where sessions persist independently of individual connections. Both users and agents can disconnect and rejoin without ending the session. When users or agents rejoin, they need to resume the session from where they left off. + +An agent or user might resume an existing session when: + +- A user goes offline or navigates away before returning, expecting to see the latest conversation state +- An agent goes offline and comes back online when the user returns +- An agent resumes after a failover or service restart + +## Hydrating presence + +When you attach to a channel, Ably automatically syncs the complete current presence set to your client. You can then query the presence set or subscribe to presence events without any additional hydration steps. This works the same way for both users and agents. + +For details on obtaining the synced presence set, see [Viewing who is online](/docs/ai-transport/sessions-and-identity/online-status#viewing-presence). + +## User resumes a session + +Users resume by reattaching to the same session channel and hydrating the conversation transcript, in-progress model output, or other session state. + +### Hydrating conversation history + +The hydration strategy you choose depends on your application model and your chosen approach to token streaming. Clients typically hydrate conversation state using one of these patterns: + +- Hydrate entirely from the channel: Use [rewind](/docs/channels/options/rewind) or [history](/docs/storage-history/history) to obtain previous messages on the channel. +- Hydrate in-progress responses from the channel: Load completed messages from your database and catch up on any in-progress responses from the channel. + +For detailed examples of hydrating the token stream, see the token streaming documentation: +- [Message-per-response hydration](/docs/ai-transport/features/token-streaming/message-per-response#hydration) +- [Message-per-token hydration](/docs/ai-transport/features/token-streaming/message-per-token#hydration) + +## Agent resumes a session + +When an agent restarts, it needs to resume from where it left off. This involves two distinct concerns: + +1. **Recovering the agent's execution state**: The current step in the workflow, local variables, function call results, pending operations, and any other state needed to continue execution. This state is internal to the agent and typically not visible to users. + +2. **Catching up on session activity**: Any user messages, events, or other activity that occurred while the agent was offline. + +These are separate problems requiring different solutions. Agent execution state is handled by your application and you choose how to persist and restore the internal state your agent needs to resume. + + + +Ably provides access to channel message history, enabling agents to retrieve any messages sent while they were offline. When your agent comes back online, it reattaches to the same channel and catches up on messages it missed. This channel-oriented model provides several key benefits: + +- Guaranteed message delivery: Clients can continue publishing messages even while the agent faults and relocates since the channel exists independently of the agent +- Reliable catch-up: The agent can retrieve any messages published during the interim when it comes back online +- Ordered delivery: Messages are delivered in the order they were published, ensuring agents process events in the correct sequence +- Channel-based addressing: The agent only needs the channel name to reconnect, no need to track individual client connections or manage connection state + +### Catching up on messages using history + +When an agent resumes, it needs to retrieve messages published while it was offline. Use [channel history](/docs/storage-history/history) with the [`untilAttach` option](/docs/storage-history/history#continuous-history) to catch up on historical messages while preserving continuity with live message delivery. + + + +#### Persisted session state + +Your agent should persist the following state to enable resumption: + +- Channel name: The channel the agent was processing +- Last processed timestamp: The timestamp of the last message successfully processed by the agent + +This state allows the agent to reconnect to the correct channel and retrieve only the messages it missed. + +#### Catching up with continuity + +The recommended pattern uses `untilAttach` to paginate backwards through history while maintaining continuity with live message delivery. This ensures no messages are lost between history retrieval and subscription. + + + + +```javascript +// Agent code +import * as Ably from 'ably'; + +const ably = new Ably.Realtime({ + key: process.env.ABLY_API_KEY, + clientId: 'agent:assistant' +}); + +// Load persisted session state +const channelName = await loadChannelName(); +const lastProcessedTimestamp = await loadLastProcessedTimestamp(); + +// Use a channel in a namespace with persistence enabled +// to access more than 2 minutes of message history +const channel = ably.channels.get(channelName); + +// Subscribe to live messages (implicitly attaches the channel) +await channel.subscribe('prompt', (message) => { + // Process the live message + processMessage(message); + + // Persist the timestamp after successful processing + saveLastProcessedTimestamp(message.timestamp); +}); + +// Fetch history up until the point of attachment, starting from last checkpoint +let page = await channel.history({ + untilAttach: true, + start: lastProcessedTimestamp, + direction: 'forwards' +}); + +// Paginate through all missed messages +while (page) { + for (const message of page.items) { + // Process the historical message + await processMessage(message); + + // Persist the timestamp after successful processing + await saveLastProcessedTimestamp(message.timestamp); + } + + // Move to next page if available + page = page.hasNext() ? await page.next() : null; +} +``` + + + + +This pattern provides guaranteed continuity between historical and live message processing by ensuring that: + +1. The subscription starts receiving live messages immediately when you subscribe +2. History retrieval stops exactly at the point the channel attached +3. No messages are lost between the end of history and the start of live delivery diff --git a/src/pages/docs/ai-transport/features/token-streaming/message-per-response.mdx b/src/pages/docs/ai-transport/features/token-streaming/message-per-response.mdx new file mode 100644 index 0000000000..a501f562a6 --- /dev/null +++ b/src/pages/docs/ai-transport/features/token-streaming/message-per-response.mdx @@ -0,0 +1,412 @@ +--- +title: Message per response +meta_description: "Stream individual tokens from AI models into a single message over Ably." +--- + +Token streaming with message-per-response is a pattern where every token generated by your model for a given response is appended to a single Ably message. Each complete AI response then appears as one message in the channel history while delivering live tokens in realtime. This uses [Ably Pub/Sub](/docs/basics) for realtime communication between agents and clients. + +This pattern is useful for chat-style applications where you want each complete AI response stored as a single message in history, making it easy to retrieve and display multi-response conversation history. Each agent response becomes a single message that grows as tokens are appended, allowing clients joining mid-stream to catch up efficiently without processing thousands of individual tokens. + +## How it works + +1. **Initial message**: When an agent response begins, publish an initial message with `message.create` action to the Ably channel with an empty or the first token as content. +2. **Token streaming**: Append subsequent tokens to the original message by publishing those tokens with the `message.append` action. +3. **Live delivery**: Clients subscribed to the channel receive each appended token in realtime, allowing them to progressively render the response. +4. **Compacted history**: The channel history contains only one message per agent response, which includes all tokens appended to it concatenated together. + +You do not need to mark the message or token stream as completed; the final message content will automatically include the full response constructed from all appended tokens. + + + +## Enable appends + +Message append functionality requires "Message annotations, updates, deletes and appends" to be enabled in a [channel rule](/docs/channels#rules) associated with the channel. + + + +To enable the channel rule: + +1. Go to the [Ably dashboard](https://www.ably.com/dashboard) and select your app. +2. Navigate to the "Configuration" > "Rules" section from the left-hand navigation bar. +3. Choose "Add new rule". +4. Enter a channel name or namespace pattern (e.g. `ai:*` for all channels starting with `ai:`). +5. Select the "Message annotations, updates, deletes and appends" option from the list. +6. Click "Create channel rule". + +The examples on this page use the `ai:` namespace prefix, which assumes you have configured the rule for `ai:*`. + +## Publishing tokens + +Publish tokens from a [Realtime](/docs/api/realtime-sdk) client, which maintains a persistent connection to the Ably service. This allows you to publish at very high message rates with the lowest possible latencies, while preserving guarantees around message delivery order. For more information, see [Realtime and REST](/docs/basics#realtime-and-rest). + +[Channels](/docs/channels) separate message traffic into different topics. For token streaming, each conversation or session typically has its own channel. + +Use the [`get()`](/docs/api/realtime-sdk/channels#get) method to create or retrieve a channel instance: + + +```javascript +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); +``` + + +To start streaming an AI response, publish the initial message. The message is identified by a server-assigned identifier called a [`serial`](/docs/messages#properties). Use the `serial` to append each subsequent token to the message as it arrives from the AI model: + + +```javascript +// Publish initial message and capture the serial for appending tokens +const { serials: [msgSerial] } = await channel.publish('response', { data: '' }); + +// Example: stream returns events like { type: 'token', text: 'Hello' } +for await (const event of stream) { + // Append each token as it arrives + if (event.type === 'token') { + channel.appendMessage(msgSerial, event.text); + } +} +``` + + +When publishing tokens, don't await the `channel.appendMessage()` call. Ably rolls up acknowledgments and debounces them for efficiency, which means awaiting each append would unnecessarily slow down your token stream. Messages are still published in the order that `appendMessage()` is called, so delivery order is not affected. + + +```javascript +// ✅ Do this - append without await for maximum throughput +for await (const event of stream) { + if (event.type === 'token') { + channel.appendMessage(msgSerial, event.text); + } +} + +// ❌ Don't do this - awaiting each append reduces throughput +for await (const event of stream) { + if (event.type === 'token') { + await channel.appendMessage(msgSerial, event.text); + } +} +``` + + + + +This pattern allows publishing append operations for multiple concurrent model responses on the same channel. As long as you append to the correct message serial, tokens from different responses will not interfere with each other, and the final concatenated message for each response will contain only the tokens from that response. + +## Subscribing to token streams + +Subscribers receive different message actions depending on when they join and how they're retrieving messages. Each message has an `action` field that indicates how to process it, and a `serial` field that identifies which message the action relates to: + +- `message.create`: Indicates a new response has started (i.e. a new message was created). The message `data` contains the initial content (often empty or the first token). Store this as the beginning of a new response using `serial` as the identifier. +- `message.append`: Contains a single token fragment to append. The message `data` contains only the new token, not the full concatenated response. Append this token to the existing response identified by `serial`. +- `message.update`: Contains the whole response up to that point. The message `data` contains the full concatenated text so far. Replace the entire response content with this data for the message identified by `serial`. This action occurs when the channel needs to resynchronize the full message state, such as after a client [resumes](/docs/connect/states#resume) from a transient disconnection. + + +```javascript +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); + +// Track responses by message serial +const responses = new Map(); + +// Subscribe to live messages (implicitly attaches the channel) +await channel.subscribe((message) => { + switch (message.action) { + case 'message.create': + // New response started + responses.set(message.serial, message.data); + break; + case 'message.append': + // Append token to existing response + const current = responses.get(message.serial) || ''; + responses.set(message.serial, current + message.data); + break; + case 'message.update': + // Replace entire response content + responses.set(message.serial, message.data); + break; + } +}); +``` + + +## Client hydration + +When clients connect or reconnect, such as after a page refresh, they often need to catch up on complete responses and individual tokens that were published while they were offline or before they joined. + +The message per response pattern enables efficient client state hydration without needing to process every individual token and supports seamlessly transitioning from historical responses to live tokens. + + + +### Using rewind for recent history + +The simplest approach is to use Ably's [rewind](/docs/channels/options/rewind) channel option to attach to the channel at some point in the recent past, and automatically receive all messages since that point. Historical messages are delivered as `message.update` events containing the complete concatenated response, which then seamlessly transition to live `message.append` events for any ongoing responses: + + +```javascript +// Use rewind to receive recent historical messages +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}', { + params: { rewind: '2m' } // or rewind: '10' for message count +}); + +// Track responses by message serial +const responses = new Map(); + +// Subscribe to receive both recent historical and live messages, +// which are delivered in order to the subscription +await channel.subscribe((message) => { + switch (message.action) { + case 'message.create': + // New response started + responses.set(message.serial, message.data); + break; + case 'message.append': + // Append token to existing response + const current = responses.get(message.serial) || ''; + responses.set(message.serial, current + message.data); + break; + case 'message.update': + // Replace entire response content + responses.set(message.serial, message.data); + break; + } +}); +``` + + +Rewind supports two formats: + +- **Time-based**: Use a time interval like `'30s'` or `'2m'` to retrieve messages from that time period +- **Count-based**: Use a number like `10` or `50` to retrieve the most recent N messages (maximum 100) + + + +### Using history for older messages + +Use [channel history](/docs/storage-history/history) with the [`untilAttach` option](/docs/storage-history/history#continuous-history) to paginate back through history to obtain historical responses, while preserving continuity with the delivery of live tokens: + + +```javascript +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); + +// Track responses by message serial +const responses = new Map(); + +// Subscribe to live messages (implicitly attaches the channel) +await channel.subscribe((message) => { + switch (message.action) { + case 'message.create': + // New response started + responses.set(message.serial, message.data); + break; + case 'message.append': + // Append token to existing response + const current = responses.get(message.serial) || ''; + responses.set(message.serial, current + message.data); + break; + case 'message.update': + // Replace entire response content + responses.set(message.serial, message.data); + break; + } +}); + +// Fetch history up until the point of attachment +let page = await channel.history({ untilAttach: true }); + +// Paginate backwards through history +while (page) { + // Messages are newest-first + for (const message of page.items) { + // message.data contains the full concatenated text + responses.set(message.serial, message.data); + } + + // Move to next page if available + page = page.hasNext() ? await page.next() : null; +} +``` + + +### Hydrating an in-progress response + +A common pattern is to persist complete model responses in your database while using Ably for streaming in-progress responses. + +The client loads completed responses from your database, then uses Ably to catch up on any response that was still in progress. + +You can hydrate in-progress responses using either the [rewind](#rewind) or [history](#history) pattern. + +#### Publishing with correlation metadata + +To correlate Ably messages with your database records, include the `responseId` in the message [extras](/docs/messages#properties) when publishing: + + +```javascript +// Publish initial message with responseId in extras +const { serials: [msgSerial] } = await channel.publish({ + name: 'response', + data: '', + extras: { + headers: { + responseId: 'resp_abc123' // Your database response ID + } + } +}); + +// Append tokens, including extras to preserve headers +for await (const event of stream) { + if (event.type === 'token') { + channel.appendMessage(msgSerial, event.text, { + extras: { + headers: { + responseId: 'resp_abc123' + } + } + }); + } +} +``` + + + + +#### Hydrate using rewind + +When hydrating, load completed responses from your database, then use rewind to catch up on any in-progress response. Check the `responseId` from message extras to skip responses already loaded from your database: + + +```javascript +// Load completed responses from your database +// completedResponses is a Set of responseIds +const completedResponses = await loadResponsesFromDatabase(); + +// Use rewind to receive recent historical messages +const channel = realtime.channels.get('ai:responses', { + params: { rewind: '2m' } +}); + +// Track in-progress responses by responseId +const inProgressResponses = new Map(); + +await channel.subscribe((message) => { + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Message missing responseId'); + return; + } + + // Skip messages for responses already loaded from database + if (completedResponses.has(responseId)) { + return; + } + + switch (message.action) { + case 'message.create': + // New response started + inProgressResponses.set(responseId, message.data); + break; + case 'message.append': + // Append token to existing response + const current = inProgressResponses.get(responseId) || ''; + inProgressResponses.set(responseId, current + message.data); + break; + case 'message.update': + // Replace entire response content + inProgressResponses.set(responseId, message.data); + break; + } +}); +``` + + + + +#### Hydrate using history + +Load completed responses from your database, then use [channel history](/docs/storage-history/history) with the [`untilAttach` option](/docs/storage-history/history#continuous-history) to catch up on any in-progress responses. Use the timestamp of the last completed response to start pagination from that point forward, ensuring continuity with live message delivery. + + +```javascript +// Load completed responses from database (sorted by timestamp, oldest first) +const completedResponses = await loadResponsesFromDatabase(); + +// Get the timestamp of the latest completed response +const latestTimestamp = completedResponses.latest().timestamp; + +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); + +// Track in progress responses by ID +const inProgressResponses = new Map(); + +// Subscribe to live messages (implicitly attaches) +await channel.subscribe((message) => { + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Message missing responseId'); + return; + } + + // Skip messages for responses already loaded from database + if (completedResponses.has(responseId)) { + return; + } + + switch (message.action) { + case 'message.create': + // New response started + inProgressResponses.set(responseId, message.data); + break; + case 'message.append': + // Append token to existing response + const current = inProgressResponses.get(responseId) || ''; + inProgressResponses.set(responseId, current + message.data); + break; + case 'message.update': + // Replace entire response content + inProgressResponses.set(responseId, message.data); + break; + } +}); + +// Fetch history from the last completed response until attachment +let page = await channel.history({ + untilAttach: true, + start: latestTimestamp, + direction: 'forwards' +}); + +// Paginate through all missed messages +while (page) { + for (const message of page.items) { + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Message missing responseId'); + continue; + } + + // message.data contains the full concatenated text so far + inProgressResponses.set(responseId, message.data); + } + + // Move to next page if available + page = page.hasNext() ? await page.next() : null; +} +``` + + + diff --git a/src/pages/docs/ai-transport/features/token-streaming/message-per-token.mdx b/src/pages/docs/ai-transport/features/token-streaming/message-per-token.mdx new file mode 100644 index 0000000000..66232ff5a2 --- /dev/null +++ b/src/pages/docs/ai-transport/features/token-streaming/message-per-token.mdx @@ -0,0 +1,433 @@ +--- +title: Message per token +meta_description: "Stream individual tokens from AI models as separate messages over Ably." +--- + +Token streaming with message-per-token is a pattern where every token generated by your model is published as an independent Ably message. Each token then appears as one message in the channel history. This uses [Ably Pub/Sub](/docs/basics) for realtime communication between agents and clients. + +This pattern is useful when clients only care about the most recent part of a response and you are happy to treat the channel history as a short sliding window rather than a full conversation log. For example: + +- **Backend-stored responses**: The backend writes complete responses to a database and clients load those full responses from there, while Ably is used only to deliver live tokens for the current in-progress response. +- **Live transcription, captioning, or translation**: A viewer who joins a live stream only needs sufficient tokens for the current "frame" of subtitles, not the entire transcript so far. +- **Code assistance in an editor**: Streamed tokens become part of the file on disk as they are accepted, so past tokens do not need to be replayed from Ably. +- **Autocomplete**: A fresh response is streamed for each change a user makes to a document, with only the latest suggestion being relevant. + +## Publishing tokens + +Publish tokens from a [Realtime](/docs/api/realtime-sdk) client, which maintains a persistent connection to the Ably service. This allows you to publish at very high message rates with the lowest possible latencies, while preserving guarantees around message delivery order. For more information, see [Realtime and REST](/docs/basics#realtime-and-rest). + +[Channels](/docs/channels) separate message traffic into different topics. For token streaming, each conversation or session typically has its own channel. + +Use the [`get()`](/docs/api/realtime-sdk/channels#get) method to create or retrieve a channel instance: + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); +``` + + +When publishing tokens, don't await the `channel.publish()` call. Ably rolls up acknowledgments and debounces them for efficiency, which means awaiting each publish would unnecessarily slow down your token stream. Messages are still published in the order that `publish()` is called, so delivery order is not affected. + + +```javascript +// ✅ Do this - publish without await for maximum throughput +for await (const event of stream) { + if (event.type === 'token') { + channel.publish('token', event.text); + } +} + +// ❌ Don't do this - awaiting each publish reduces throughput +for await (const event of stream) { + if (event.type === 'token') { + await channel.publish('token', event.text); + } +} +``` + + +This approach maximizes throughput while maintaining ordering guarantees, allowing you to stream tokens as fast as your AI model generates them. + +## Streaming patterns + +Ably is a pub/sub messaging platform, so you can structure your messages however works best for your application. Below are common patterns for streaming tokens, each showing both agent-side publishing and client-side subscription. Choose the approach that fits your use case, or create your own variation. + +### Continuous token stream + +For simple streaming scenarios such as live transcription, where all tokens are part of a continuous stream, simply publish each token as a message. + +#### Publish tokens + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Example: stream returns events like { type: 'token', text: 'Hello' } +for await (const event of stream) { + if (event.type === 'token') { + channel.publish('token', event.text); + } +} +``` + + +#### Subscribe to tokens + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Subscribe to token messages +await channel.subscribe('token', (message) => { + const token = message.data; + console.log(token); // log each token as it arrives +}); +``` + + +This pattern is simple and works well when you're displaying a single, continuous stream of tokens. + +### Token stream with multiple responses + +For applications with multiple responses, such as chat conversations, include a `responseId` in message [extras](/docs/messages#properties) to correlate tokens together that belong to the same response. + +#### Publish tokens + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Example: stream returns events like { type: 'token', text: 'Hello', responseId: 'resp_abc123' } +for await (const event of stream) { + if (event.type === 'token') { + channel.publish({ + name: 'token', + data: event.text, + extras: { + headers: { + responseId: event.responseId + } + } + }); + } +} +``` + + +#### Subscribe to tokens + +Use the `responseId` header in message extras to correlate tokens. The `responseId` allows you to group tokens belonging to the same response and correctly handle token delivery for distinct responses, even when delivered concurrently. + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Track responses by ID +const responses = new Map(); + +await channel.subscribe('token', (message) => { + const token = message.data; + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Token missing responseId'); + return; + } + + // Create an empty response + if (!responses.has(responseId)) { + responses.set(responseId, ''); + } + + // Append token to response + responses.set(responseId, responses.get(responseId) + token); +}); +``` + + +### Token stream with explicit start/stop events + +In some cases, your AI model response stream may include explicit events to mark response boundaries. You can indicate the event type, such as a response start/stop event, using the Ably message name. + +#### Publish tokens + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Example: stream returns events like: +// { type: 'message_start', responseId: 'resp_abc123' } +// { type: 'message_delta', responseId: 'resp_abc123', text: 'Hello' } +// { type: 'message_stop', responseId: 'resp_abc123' } + +for await (const event of stream) { + if (event.type === 'message_start') { + // Publish response start + channel.publish({ + name: 'start', + extras: { + headers: { + responseId: event.responseId + } + } + }); + } else if (event.type === 'message_delta') { + // Publish tokens + channel.publish({ + name: 'token', + data: event.text, + extras: { + headers: { + responseId: event.responseId + } + } + }); + } else if (event.type === 'message_stop') { + // Publish response stop + channel.publish({ + name: 'stop', + extras: { + headers: { + responseId: event.responseId + } + } + }); + } +} +``` + + +#### Subscribe to tokens + +Handle each event type to manage response lifecycle: + + +```javascript +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +const responses = new Map(); + +// Handle response start +await channel.subscribe('start', (message) => { + const responseId = message.extras?.headers?.responseId; + responses.set(responseId, ''); +}); + +// Handle tokens +await channel.subscribe('token', (message) => { + const responseId = message.extras?.headers?.responseId; + const token = message.data; + + const currentText = responses.get(responseId) || ''; + responses.set(responseId, currentText + token); +}); + +// Handle response stop +await channel.subscribe('stop', (message) => { + const responseId = message.extras?.headers?.responseId; + const finalText = responses.get(responseId); + console.log('Response complete:', finalText); +}); +``` + + +## Client hydration + +When clients connect or reconnect, such as after a page refresh, they often need to catch up on tokens that were published while they were offline or before they joined. Ably provides several approaches to hydrate client state depending on your application's requirements. + + + + + +### Using rewind for recent history + +The simplest approach is to use Ably's [rewind](/docs/channels/options/rewind) channel option to attach to the channel at some point in the recent past, and automatically receive all tokens since that point: + + +```javascript +// Use rewind to receive recent historical messages +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}', { + params: { rewind: '2m' } // or rewind: 100 for message count +}); + +// Subscribe to receive both recent historical and live messages, +// which are delivered in order to the subscription +await channel.subscribe('token', (message) => { + const token = message.data; + + // Process tokens from both recent history and live stream + console.log('Token received:', token); +}); +``` + + +Rewind supports two formats: + +- **Time-based**: Use a time interval like `'30s'` or `'2m'` to retrieve messages from that time period +- **Count-based**: Use a number like `50` or `100` to retrieve the most recent N messages (maximum 100) + + + +By default, rewind is limited to the last 2 minutes of messages. This is usually sufficient for scenarios where clients need only recent context, such as for continuous token streaming, or when the response stream from a given model request does not exceed 2 minutes. If you need more than 2 minutes of history, see [Using history for longer persistence](#history). + +### Using history for older messages + +For applications that need to retrieve tokens beyond the 2-minute rewind window, enable [persistence](/docs/storage-history/storage#all-message-persistence) on your channel. Use [channel history](/docs/storage-history/history) with the [`untilAttach` option](/docs/storage-history/history#continuous-history) to paginate back through history to obtain historical tokens, while preserving continuity with the delivery of live tokens: + + +```javascript +// Use a channel in a namespace called 'persisted', which has persistence enabled +const channel = realtime.channels.get('persisted:{{RANDOM_CHANNEL_NAME}}'); + +let response = ''; + +// Subscribe to live messages (implicitly attaches the channel) +await channel.subscribe('token', (message) => { + // Append the token to the end of the response + response += message.data; +}); + +// Fetch history up until the point of attachment +let page = await channel.history({ untilAttach: true }); + +// Paginate backwards through history +while (page) { + // Messages are newest-first, so prepend them to response + for (const message of page.items) { + response = message.data + response; + } + + // Move to next page if available + page = page.hasNext() ? await page.next() : null; +} +``` + + +### Hydrating an in-progress response + +A common pattern is to persist complete model responses in your database while using Ably for live token delivery of the in-progress response. + +The client loads completed responses from your database, then reaches back into Ably channel history until it encounters a token for a response it's already loaded. + +You can retrieve partial history using either the [rewind](#rewind) or [history](#history) pattern. + +#### Hydrate using rewind + +Load completed responses from your database, then use rewind to catch up on any in-progress responses, skipping any tokens that belong to a response that was already loaded: + + +```javascript +// Load completed responses from database +const completedResponses = await loadResponsesFromDatabase(); + +// Use rewind to receive recent historical messages +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}', { + params: { rewind: '2m' } +}); + +// Track in progress responses by ID +const inProgressResponses = new Map(); + +// Subscribe to receive both recent historical and live messages, +// which are delivered in order to the subscription +await channel.subscribe('token', (message) => { + const token = message.data; + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Token missing responseId'); + return; + } + + // Skip tokens for responses already hydrated from database + if (completedResponses.has(responseId)) { + return; + } + + // Create an empty in-progress response + if (!inProgressResponses.has(responseId)) { + inProgressResponses.set(responseId, ''); + } + + // Append tokens for new responses + inProgressResponses.set(responseId, inProgressResponses.get(responseId) + token); +}); +``` + + +#### Hydrate using history + +Load completed responses from your database, then paginate backwards through history to catch up on in-progress responses until you reach a token that belongs to a response you've already loaded: + + +```javascript +// Load completed responses from database +const completedResponses = await loadResponsesFromDatabase(); + +// Use a channel in a namespace called 'persisted', which has persistence enabled +const channel = realtime.channels.get('persisted:{{RANDOM_CHANNEL_NAME}}'); + +// Track in progress responses by ID +const inProgressResponses = new Map(); + +// Subscribe to live tokens (implicitly attaches) +await channel.subscribe('token', (message) => { + const token = message.data; + const responseId = message.extras?.headers?.responseId; + + if (!responseId) { + console.warn('Token missing responseId'); + return; + } + + // Skip tokens for responses already hydrated from database + if (completedResponses.has(responseId)) { + return; + } + + // Create an empty in-progress response + if (!inProgressResponses.has(responseId)) { + inProgressResponses.set(responseId, ''); + } + + // Append live tokens for in-progress responses + inProgressResponses.set(responseId, inProgressResponses.get(responseId) + token); +}); + +// Paginate backwards through history until we encounter a hydrated response +let page = await channel.history({ untilAttach: true }); + +// Paginate backwards through history +let done = false; +while (page && !done) { + // Messages are newest-first, so prepend them to response + for (const message of page.items) { + const token = message.data; + const responseId = message.extras?.headers?.responseId; + + // Stop when we reach a response already loaded from database + if (completedResponses.has(responseId)) { + done = true; + break; + } + + // Create an empty in-progress response + if (!inProgressResponses.has(responseId)) { + inProgressResponses.set(responseId, ''); + } + + // Prepend historical tokens for in-progress responses + inProgressResponses.set(responseId, token + inProgressResponses.get(responseId)); + } + + // Move to next page if available + page = page.hasNext() ? await page.next() : null; +} +``` + diff --git a/src/pages/docs/ai-transport/index.mdx b/src/pages/docs/ai-transport/index.mdx new file mode 100644 index 0000000000..fb2f2b271e --- /dev/null +++ b/src/pages/docs/ai-transport/index.mdx @@ -0,0 +1,6 @@ +--- +title: About AI Transport +meta_description: "Learn more about Ably's AI Transport and the features that enable you to quickly build functionality into new and existing applications." +redirect_from: + - /docs/products/ai-transport +--- diff --git a/src/pages/docs/api/realtime-sdk/channels.mdx b/src/pages/docs/api/realtime-sdk/channels.mdx index 75a60eba21..a3a1209039 100644 --- a/src/pages/docs/api/realtime-sdk/channels.mdx +++ b/src/pages/docs/api/realtime-sdk/channels.mdx @@ -1942,7 +1942,7 @@ The action type of the message, one of the [`MessageAction`](/docs/api/realtime- ### serial -A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, and deletes in [channel rules](/docs/channels#rules).
_Type: `String`_ +A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, deletes, and appends in [channel rules](/docs/channels#rules).
_Type: `String`_ ### annotations diff --git a/src/pages/docs/api/realtime-sdk/messages.mdx b/src/pages/docs/api/realtime-sdk/messages.mdx index 7118cd20df..6325aecc38 100644 --- a/src/pages/docs/api/realtime-sdk/messages.mdx +++ b/src/pages/docs/api/realtime-sdk/messages.mdx @@ -56,7 +56,7 @@ The action type of the message, one of the [`MessageAction`](/docs/api/realtime- ### serial -A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, and deletes in [channel rules](/docs/channels#rules).
_Type: `String`_ +A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, deletes, and appends in [channel rules](/docs/channels#rules).
_Type: `String`_ ### annotations diff --git a/src/pages/docs/api/rest-api.mdx b/src/pages/docs/api/rest-api.mdx index 455091a272..3d44ac9aff 100644 --- a/src/pages/docs/api/rest-api.mdx +++ b/src/pages/docs/api/rest-api.mdx @@ -686,7 +686,7 @@ An unsuccessful request returns an error. A 404 error is returned if a message w #### PATCH main.realtime.ably.net/channels/\/messages/\ -Update an existing message on a channel, the message with the specified serial. This endpoint requires that the channel is configured with the "Message annotations, updates, and deletes" channel rule. +Update an existing message on a channel, the message with the specified serial. This endpoint requires that the channel is configured with the "Message annotations, updates, deletes, and appends" channel rule. See [Updating and deleting messages](/docs/messages/updates-deletes#update) for more information about message updates and field semantics. @@ -770,7 +770,7 @@ An unsuccessful request returns an error. #### POST main.realtime.ably.net/channels/\/messages/\/delete -Delete a message on a channel, the message with the specified serial. This is a 'soft' delete that publishes a new version of the message with an action of `message.delete`. The full message history remains accessible through the [message versions](message-versions) endpoint. This endpoint requires that the channel is configured with the "Message annotations, updates, and deletes" channel rule. +Delete a message on a channel, the message with the specified serial. This is a 'soft' delete that publishes a new version of the message with an action of `message.delete`. The full message history remains accessible through the [message versions](message-versions) endpoint. This endpoint requires that the channel is configured with the "Message annotations, updates, deletes, and appends" channel rule. See [Updating and deleting messages](/docs/messages/updates-deletes#update) for more information about message updates and field semantics. @@ -855,7 +855,7 @@ An unsuccessful request returns an error. #### GET main.realtime.ably.net/channels/\/messages/\/versions -Retrieve all historical versions of a specific message, including the original and all subsequent updates or delete operations. This endpoint requires that the channel is configured with the "Message annotations, updates, and deletes" channel rule. +Retrieve all historical versions of a specific message, including the original and all subsequent updates or delete operations. This endpoint requires that the channel is configured with the "Message annotations, updates, deletes, and appends" channel rule. Example request: diff --git a/src/pages/docs/api/rest-sdk/channels.mdx b/src/pages/docs/api/rest-sdk/channels.mdx index f57747d395..c6c87a077d 100644 --- a/src/pages/docs/api/rest-sdk/channels.mdx +++ b/src/pages/docs/api/rest-sdk/channels.mdx @@ -594,7 +594,7 @@ The action type of the message, one of the [`MessageAction`](/docs/api/realtime- ### serial -A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, and deletes in [channel rules](/docs/channels#rules).
_Type: `String`_ +A server-assigned identifier that will be the same in all future updates of this message. It can be used to add [annotations](/docs/messages/annotations) to a message or to [update or delete](/docs/messages/updates-deletes) it. Serial will only be set if you enable annotations, updates, deletes, and appends in [channel rules](/docs/channels#rules).
_Type: `String`_ ### annotations diff --git a/src/pages/docs/channels/index.mdx b/src/pages/docs/channels/index.mdx index 908607717f..1d340f49c4 100644 --- a/src/pages/docs/channels/index.mdx +++ b/src/pages/docs/channels/index.mdx @@ -200,7 +200,7 @@ The channel rules related to enabling features are: | Push notifications enabled | If checked, publishing messages with a push payload in the `extras` field is permitted. This triggers the delivery of a [Push Notification](/docs/push) to devices registered for push on the channel. | | Server-side batching | If enabled, messages are grouped into batches before being sent to subscribers. [Server-side batching](/docs/messages/batch#server-side) reduces the overall message count, lowers costs, and mitigates the risk of hitting rate limits during high-throughput scenarios. | | Message conflation | If enabled, messages are aggregated over a set period of time and evaluated against a conflation key. All but the latest message for each conflation key value will be discarded, and the resulting message, or messages, will be delivered to subscribers as a single batch once the period of time elapses. [Message conflation](/docs/messages#conflation) reduces costs in high-throughput scenarios by removing redundant and outdated messages. | -| Message annotations, updates, and deletes | If enabled, allows message "annotations":/docs/messages/annotations to be used, as well as updates and deletes to be published to messages. Note that these features are currently Experimental, still in development, and subject to change. When this feature is enabled, messages will be "persisted":/docs/storage-history/storage#all-message-persistence (necessary in order from them later be annotated or updated), and "continuous history":/docs/storage-history/history#continuous-history features will not work. +| Message annotations, updates, deletes, and appends | If enabled, allows message "annotations":/docs/messages/annotations to be used, as well as updates, deletes, and appends to be published to messages. Note that these features are currently Experimental, still in development, and subject to change. When this feature is enabled, messages will be "persisted":/docs/storage-history/storage#all-message-persistence (necessary in order from them later be annotated or updated), and "continuous history":/docs/storage-history/history#continuous-history features will not work. To set a channel rule in the Ably dashboard: diff --git a/src/pages/docs/chat/integrations.mdx b/src/pages/docs/chat/integrations.mdx index 404f47e97a..cd7140fbd9 100644 --- a/src/pages/docs/chat/integrations.mdx +++ b/src/pages/docs/chat/integrations.mdx @@ -60,4 +60,4 @@ Message reactions are a Chat feature based on the Pub/Sub [annotations](/docs/me Messages arriving from the realtime channel or integrations will show the annotation type under `message.annotations.summary`. The Chat SDKs automatically map the annotation type to the message reaction type and provide convenient `message.reactions.unique`, `message.reactions.distinct`, and `message.reactions.multiple` fields. When working with Pub/Sub integrations, you will need to either use the annotations directly or apply the mappping in your own code. -All messages on a chat channel (or any channel with the **Message annotations, updates, and deletes** rule enabled) contain a `message.annotations.summary` field if the summary isn't empty. This is true even if the action is not `message.summary`. +All messages on a chat channel (or any channel with the **Message annotations, updates, deletes, and appends** rule enabled) contain a `message.annotations.summary` field if the summary isn't empty. This is true even if the action is not `message.summary`. diff --git a/src/pages/docs/guides/ai-transport/openai-message-per-response.mdx b/src/pages/docs/guides/ai-transport/openai-message-per-response.mdx new file mode 100644 index 0000000000..5683b2195f --- /dev/null +++ b/src/pages/docs/guides/ai-transport/openai-message-per-response.mdx @@ -0,0 +1,437 @@ +--- +title: "Guide: Stream OpenAI responses using the message-per-response pattern" +meta_description: "Stream tokens from the OpenAI Responses API over Ably in realtime using message appends." +meta_keywords: "AI, token streaming, OpenAI, Responses API, AI transport, Ably, realtime, message appends" +--- + +This guide shows you how to stream AI responses from OpenAI's [Responses API](https://platform.openai.com/docs/api-reference/responses) over Ably using the [message-per-response pattern](/docs/ai-transport/features/token-streaming/message-per-response). Specifically, it appends each response token to a single Ably message, creating a complete AI response that grows incrementally while delivering tokens in realtime. + +Using Ably to distribute tokens from the OpenAI SDK enables you to broadcast AI responses to thousands of concurrent subscribers with reliable message delivery and ordering guarantees. This approach stores each complete response as a single message in channel history, making it easy to retrieve conversation history without processing thousands of individual token messages. + + + +## Prerequisites + +To follow this guide, you need: +- Node.js 20 or higher +- An OpenAI API key +- An Ably API key + +Useful links: +- [OpenAI developer quickstart](https://platform.openai.com/docs/quickstart) +- [Ably JavaScript SDK getting started](/docs/getting-started/javascript) + +Create a new NPM package, which will contain the publisher and subscriber code: + + +```shell +mkdir ably-openai-example && cd ably-openai-example +npm init -y +``` + + +Install the required packages using NPM: + + +```shell +npm install openai@^4 ably@^2 +``` + + + + +Export your OpenAI API key to the environment, which will be used later in the guide by the OpenAI SDK: + + +```shell +export OPENAI_API_KEY="your_api_key_here" +``` + + +## Step 1: Enable message appends + +Message append functionality requires "Message annotations, updates, deletes and appends" to be enabled in a [channel rule](/docs/channels#rules) associated with the channel. + + + +To enable the channel rule: + +1. Go to the [Ably dashboard](https://www.ably.com/dashboard) and select your app. +2. Navigate to the "Configuration" > "Rules" section from the left-hand navigation bar. +3. Choose "Add new rule". +4. Enter a channel name or namespace pattern (e.g. `ai:*` for all channels starting with `ai:`). +5. Select the "Message annotations, updates, deletes and appends" option from the list. +6. Click "Create channel rule". + +The examples in this guide use the `ai:` namespace prefix, which assumes you have configured the rule for `ai:*`. + + + +## Step 2: Get a streamed response from OpenAI + +Initialize an OpenAI client and use the [Responses API](https://platform.openai.com/docs/api-reference/responses) to stream model output as a series of events. + +Create a new file `publisher.mjs` with the following contents: + + +```javascript +import OpenAI from 'openai'; + +// Initialize OpenAI client +const openai = new OpenAI(); + +// Process each streaming event +function processEvent(event) { + console.log(JSON.stringify(event)); + // This function is updated in the next sections +} + +// Create streaming response from OpenAI +async function streamOpenAIResponse(prompt) { + const stream = await openai.responses.create({ + model: "gpt-5", + input: prompt, + stream: true, + }); + + // Iterate through streaming events + for await (const event of stream) { + processEvent(event); + } +} + +// Usage example +streamOpenAIResponse("Tell me a short joke"); +``` + + +### Understand OpenAI streaming events + +OpenAI's Responses API [streams](https://platform.openai.com/docs/guides/streaming-responses) model output as a series of events when you set `stream: true`. Each streamed event includes a `type` property which describes the [event type](https://platform.openai.com/docs/api-reference/responses-streaming). A complete text response can be constructed from the following event types: + +- [`response.created`](https://platform.openai.com/docs/api-reference/responses-streaming/response/created): Signals the start of a response. Contains `response.id` to correlate subsequent events. + +- [`response.output_item.added`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_item/added): Indicates a new output item. If `item.type === "message"` the item contains model response text; other types may be specified, such as `"reasoning"` for internal reasoning tokens. The `output_index` indicates the position of this item in the response's [`output`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed#responses_streaming-response-completed-response-output) array. + +- [`response.content_part.added`](https://platform.openai.com/docs/api-reference/responses-streaming/response/content_part/added): Indicates a new content part within an output item. If `part.type === "output_text"` the part contains model response text; other types may be specified, such as `"reasoning_text"` for internal reasoning tokens. The `content_index` indicates the position of this item in the output items's [`content`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed#responses_streaming-response-completed-response-output-output_message-content) array. + +- [`response.output_text.delta`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_text/delta): Contains a single token in the `delta` field. Use the `item_id`, `output_index`, and `content_index` to correlate tokens relating to a specific content part. + +- [`response.content_part.done`](https://platform.openai.com/docs/api-reference/responses-streaming/response/content_part/done): Signals completion of a content part. Contains the complete `part` object with full text, along with `item_id`, `output_index`, and `content_index`. + +- [`response.output_item.done`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_item/done): Signals completion of an output item. Contains the complete `item` object and `output_index`. + +- [`response.completed`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed): Signals the end of the response. Contains the complete `response` object. + +The following example shows the event sequence received when streaming a response: + + +```json +// 1. Response starts +{"type":"response.created","response":{"id":"resp_abc123","status":"in_progress"}} + +// 2. First output item (reasoning) is added +{"type":"response.output_item.added","output_index":0,"item":{"id":"rs_456","type":"reasoning"}} +{"type":"response.output_item.done","output_index":0,"item":{"id":"rs_456","type":"reasoning"}} + +// 3. Second output item (message) is added +{"type":"response.output_item.added","output_index":1,"item":{"id":"msg_789","type":"message"}} +{"type":"response.content_part.added","item_id":"msg_789","output_index":1,"content_index":0} + +// 4. Text tokens stream in as delta events +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"Why"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" don"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"'t"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" scientists"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" trust"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" atoms"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"?"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" Because"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" they"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" make"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" up"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" everything"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"."} + +// 5. Content part and output item complete +{"type":"response.content_part.done","item_id":"msg_789","output_index":1,"content_index":0,"part":{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}} +{"type":"response.output_item.done","output_index":1,"item":{"id":"msg_789","type":"message","status":"completed","content":[{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}]}} + +// 6. Response completes +{"type":"response.completed","response":{"id":"resp_abc123","status":"completed","output":[{"id":"rs_456","type":"reasoning"},{"id":"msg_789","type":"message","status":"completed","content":[{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}]}]}} +``` + + + + +## Step 3: Publish streaming tokens to Ably + +Publish OpenAI streaming events to Ably using message appends to reliably and scalably distribute them to subscribers. + +Each AI response is stored as a single Ably message that grows as tokens are appended. + +### Initialize the Ably client + +Add the Ably client initialization to your `publisher.mjs` file: + + +```javascript +import Ably from 'ably'; + +// Initialize Ably Realtime client +const realtime = new Ably.Realtime({ key: '{{API_KEY}}' }); + +// Create a channel for publishing streamed AI responses +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); +``` + + +The Ably Realtime client maintains a persistent connection to the Ably service, which allows you to publish tokens at high message rates with low latency. + +### Publish initial message and append tokens + +When a new response begins, publish an initial message to create it. Ably assigns a [`serial`](/docs/messages#properties) identifier to the message. Use this `serial` to append each token to the message as it arrives from the OpenAI model. + + + +Update your `publisher.mjs` file to publish the initial message and append tokens: + + +```javascript +// Track state across events +let msgSerial = null; +let messageItemId = null; + +// Process each streaming event and publish to Ably +async function processEvent(event) { + switch (event.type) { + case 'response.created': + // Publish initial empty message when response starts + const result = await channel.publish({ + name: 'response', + data: '' + }); + + // Capture the message serial for appending tokens + msgSerial = result.serials[0]; + break; + + case 'response.output_item.added': + // Capture message item ID when a message output item is added + if (event.item.type === 'message') { + messageItemId = event.item.id; + } + break; + + case 'response.output_text.delta': + // Append tokens from message output items only + if (event.item_id === messageItemId && msgSerial) { + channel.appendMessage({ + serial: msgSerial, + data: event.delta + }); + } + break; + + case 'response.completed': + console.log('Stream completed!'); + break; + } +} +``` + + +This implementation: + +- Publishes an initial empty message when the response begins and captures the `serial` +- Filters for `response.output_text.delta` events from `message` type output items +- Appends each token to the original message + + + + + +Run the publisher to see tokens streaming to Ably: + + +```shell +node publisher.mjs +``` + + +## Step 4: Subscribe to streaming tokens + +Create a subscriber that receives the streaming tokens from Ably and reconstructs the response in realtime. + +Create a new file `subscriber.mjs` with the following contents: + + +```javascript +import Ably from 'ably'; + +// Initialize Ably Realtime client +const realtime = new Ably.Realtime({ key: '{{API_KEY}}' }); + +// Get the same channel used by the publisher +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}'); + +// Track responses by message serial +const responses = new Map(); + +// Subscribe to receive messages +await channel.subscribe((message) => { + switch (message.action) { + case 'message.create': + // New response started + console.log('\n[Response started]', message.serial); + responses.set(message.serial, message.data); + break; + + case 'message.append': + // Append token to existing response + const current = responses.get(message.serial) || ''; + responses.set(message.serial, current + message.data); + + // Display token as it arrives + process.stdout.write(message.data); + break; + + case 'message.update': + // Replace entire response content + responses.set(message.serial, message.data); + console.log('\n[Response updated with full content]'); + break; + } +}); + +console.log('Subscriber ready, waiting for tokens...'); +``` + + +Subscribers receive different message actions depending on when they join and how they're retrieving messages: + +- `message.create`: Indicates a new response has started (i.e. a new message was created). The message `data` contains the initial content (often empty or the first token). Store this as the beginning of a new response using `serial` as the identifier. + +- `message.append`: Contains a single token fragment to append. The message `data` contains only the new token, not the full concatenated response. Append this token to the existing response identified by `serial`. + +- `message.update`: Contains the whole response up to that point. The message `data` contains the full concatenated text so far. Replace the entire response content with this data for the message identified by `serial`. This action occurs when the channel needs to resynchronize the full message state, such as after a client [resumes](/docs/connect/states#resume) from a transient disconnection. + +Run the subscriber in a separate terminal: + + +```shell +node subscriber.mjs +``` + + +With the subscriber running, run the publisher in another terminal. The tokens stream in realtime as they are generated by the OpenAI model. + +## Step 5: Stream with multiple publishers and subscribers + +Ably's [channel-oriented sessions](/docs/ai-transport/features/sessions-identity#connection-oriented-vs-channel-oriented-sessions) enables multiple AI agents to publish responses and multiple users to receive them on a single channel simultaneously. Ably handles message delivery to all participants, eliminating the need to implement routing logic or manage state synchronization across connections. + +### Broadcasting to multiple subscribers + +Each subscriber receives the complete stream of tokens independently, enabling you to build collaborative experiences or multi-device applications. + +Run a subscriber in multiple separate terminals: + + +```shell +# Terminal 1 +node subscriber.mjs + +# Terminal 2 +node subscriber.mjs + +# Terminal 3 +node subscriber.mjs +``` + + +All subscribers receive the same stream of tokens in realtime. + +### Publishing concurrent responses + +Multiple publishers can stream different responses concurrently on the same channel. Each response is a distinct message with its own unique `serial` identifier, so tokens from different responses are isolated to distinct messages and don't interfere with each other. + +To demonstrate this, run a publisher in multiple separate terminals: + + +```shell +# Terminal 1 +node publisher.mjs + +# Terminal 2 +node publisher.mjs + +# Terminal 3 +node publisher.mjs +``` + + +All running subscribers receive tokens from all responses concurrently. Each subscriber correctly reconstructs each response separately using the `serial` to correlate tokens. + +## Step 6: Retrieve complete responses from history + +One key advantage of the message-per-response pattern is that each complete AI response is stored as a single message in channel history. This makes it efficient to retrieve conversation history without processing thousands of individual token messages. + +Use Ably's [rewind](/docs/channels/options/rewind) channel option to attach to the channel at some point in the recent past and automatically receive complete responses from history. Historical messages are delivered as `message.update` events containing the complete concatenated response, which then seamlessly transition to live `message.append` events for any ongoing responses: + + +```javascript +// Use rewind to receive recent historical messages +const channel = realtime.channels.get('ai:{{RANDOM_CHANNEL_NAME}}', { + params: { rewind: '2m' } // Retrieve messages from the last 2 minutes +}); + +const responses = new Map(); + +await channel.subscribe((message) => { + switch (message.action) { + case 'message.create': + responses.set(message.serial, message.data); + break; + + case 'message.append': + const current = responses.get(message.serial) || ''; + responses.set(message.serial, current + message.data); + process.stdout.write(message.data); + break; + + case 'message.update': + // Historical messages contain full concatenated response + responses.set(message.serial, message.data); + console.log('\n[Historical response]:', message.data); + break; + } +}); +``` + + + + +## Next steps + +- Learn more about the [message-per-response pattern](/docs/ai-transport/features/token-streaming/message-per-response) used in this guide +- Learn about [client hydration strategies](/docs/ai-transport/features/token-streaming/message-per-response#hydration) for handling late joiners and reconnections +- Understand [sessions and identity](/docs/ai-transport/features/sessions-identity) in AI enabled applications +- Explore the [message-per-token pattern](/docs/ai-transport/features/token-streaming/message-per-token) for explicit control over individual token messages diff --git a/src/pages/docs/guides/ai-transport/openai-message-per-token.mdx b/src/pages/docs/guides/ai-transport/openai-message-per-token.mdx new file mode 100644 index 0000000000..adc9fdfb87 --- /dev/null +++ b/src/pages/docs/guides/ai-transport/openai-message-per-token.mdx @@ -0,0 +1,378 @@ +--- +title: "Guide: Stream OpenAI responses using the message-per-token pattern" +meta_description: "Stream tokens from the OpenAI Responses API over Ably in realtime." +meta_keywords: "AI, token streaming, OpenAI, Responses API, AI transport, Ably, realtime" +--- + +This guide shows you how to stream AI responses from OpenAI's [Responses API](https://platform.openai.com/docs/api-reference/responses) over Ably using the [message-per-token pattern](/docs/ai-transport/features/token-streaming/message-per-token). Specifically, it implements the [explicit start/stop events approach](/docs/ai-transport/features/token-streaming/message-per-token#explicit-events), which publishes each response token as an individual message, along with explicit lifecycle events to signal when responses begin and end. + +Using Ably to distribute tokens from the OpenAI SDK enables you to broadcast AI responses to thousands of concurrent subscribers with reliable message delivery and ordering guarantees, ensuring that each client receives the complete response stream with all tokens delivered in order. This approach decouples your AI inference from client connections, enabling you to scale agents independently and handle reconnections gracefully. + + + +## Prerequisites + +To follow this guide, you need: +- Node.js 20 or higher +- An OpenAI API key +- An Ably API key + +Useful links: +- [OpenAI developer quickstart](https://platform.openai.com/docs/quickstart) +- [Ably JavaScript SDK getting started](/docs/getting-started/javascript) + +Create a new NPM package, which will contain the publisher and subscriber code: + + +```shell +mkdir ably-openai-example && cd ably-openai-example +npm init -y +``` + + +Install the required packages using NPM: + + +```shell +npm install openai@^4 ably@^2 +``` + + + + +Export your OpenAI API key to the environment, which will be used later in the guide by the OpenAI SDK: + + +```shell +export OPENAI_API_KEY="your_api_key_here" +``` + + +## Step 1: Get a streamed response from OpenAI + +Initialize an OpenAI client and use the [Responses API](https://platform.openai.com/docs/api-reference/responses) to stream model output as a series of events. + +Create a new file `publisher.mjs` with the following contents: + + +```javascript +import OpenAI from 'openai'; + +// Initialize OpenAI client +const openai = new OpenAI(); + +// Process each streaming event +function processEvent(event) { + console.log(JSON.stringify(event)); + // This function is updated in the next sections +} + +// Create streaming response from OpenAI +async function streamOpenAIResponse(prompt) { + const stream = await openai.responses.create({ + model: "gpt-5", + input: prompt, + stream: true, + }); + + // Iterate through streaming events + for await (const event of stream) { + processEvent(event); + } +} + +// Usage example +streamOpenAIResponse("Tell me a short joke"); +``` + + +### Understand OpenAI streaming events + +OpenAI's Responses API [streams](https://platform.openai.com/docs/guides/streaming-responses) model output as a series of events when you set `stream: true`. Each streamed event includes a `type` property which describes the [event type](https://platform.openai.com/docs/api-reference/responses-streaming). A complete text response can be constructed from the following event types: + +- [`response.created`](https://platform.openai.com/docs/api-reference/responses-streaming/response/created): Signals the start of a response. Contains `response.id` to correlate subsequent events. + +- [`response.output_item.added`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_item/added): Indicates a new output item. If `item.type === "message"` the item contains model response text; other types may be specified, such as `"reasoning"` for internal reasoning tokens. The `output_index` indicates the position of this item in the response's [`output`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed#responses_streaming-response-completed-response-output) array. + +- [`response.content_part.added`](https://platform.openai.com/docs/api-reference/responses-streaming/response/content_part/added): Indicates a new content part within an output item. If `part.type === "output_text"` the part contains model response text; other types may be specified, such as `"reasoning_text"` for internal reasoning tokens. The `content_index` indicates the position of this item in the output items's [`content`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed#responses_streaming-response-completed-response-output-output_message-content) array. + +- [`response.output_text.delta`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_text/delta): Contains a single token in the `delta` field. Use the `item_id`, `output_index`, and `content_index` to correlate tokens relating to a specific content part. + +- [`response.content_part.done`](https://platform.openai.com/docs/api-reference/responses-streaming/response/content_part/done): Signals completion of a content part. Contains the complete `part` object with full text, along with `item_id`, `output_index`, and `content_index`. + +- [`response.output_item.done`](https://platform.openai.com/docs/api-reference/responses-streaming/response/output_item/done): Signals completion of an output item. Contains the complete `item` object and `output_index`. + +- [`response.completed`](https://platform.openai.com/docs/api-reference/responses-streaming/response/completed): Signals the end of the response. Contains the complete `response` object. + +The following example shows the event sequence received when streaming a response: + + +```json +// 1. Response starts +{"type":"response.created","response":{"id":"resp_abc123","status":"in_progress"}} + +// 2. First output item (reasoning) is added +{"type":"response.output_item.added","output_index":0,"item":{"id":"rs_456","type":"reasoning"}} +{"type":"response.output_item.done","output_index":0,"item":{"id":"rs_456","type":"reasoning"}} + +// 3. Second output item (message) is added +{"type":"response.output_item.added","output_index":1,"item":{"id":"msg_789","type":"message"}} +{"type":"response.content_part.added","item_id":"msg_789","output_index":1,"content_index":0} + +// 4. Text tokens stream in as delta events +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"Why"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" don"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"'t"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" scientists"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" trust"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" atoms"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"?"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" Because"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" they"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" make"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" up"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":" everything"} +{"type":"response.output_text.delta","item_id":"msg_789","output_index":1,"content_index":0,"delta":"."} + +// 5. Content part and output item complete +{"type":"response.content_part.done","item_id":"msg_789","output_index":1,"content_index":0,"part":{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}} +{"type":"response.output_item.done","output_index":1,"item":{"id":"msg_789","type":"message","status":"completed","content":[{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}]}} + +// 6. Response completes +{"type":"response.completed","response":{"id":"resp_abc123","status":"completed","output":[{"id":"rs_456","type":"reasoning"},{"id":"msg_789","type":"message","status":"completed","content":[{"type":"output_text","text":"Why don't scientists trust atoms? Because they make up everything."}]}]}} +``` + + + + +## Step 2: Publish streaming events to Ably + +Publish OpenAI streaming events to Ably to reliably and scalably distribute them to subscribers. + +This implementation follows the [explicit start/stop events pattern](/docs/ai-transport/features/token-streaming/message-per-token#explicit-events), which provides clear response boundaries. + +### Initialize the Ably client + +Add the Ably client initialization to your `publisher.mjs` file: + + +```javascript +import Ably from 'ably'; + +// Initialize Ably Realtime client +const realtime = new Ably.Realtime({ key: '{{API_KEY}}' }); + +// Create a channel for publishing streamed AI responses +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); +``` + + +The Ably Realtime client maintains a persistent connection to the Ably service, which allows you to publish tokens at high message rates with low latency. + +### Map OpenAI streaming events to Ably messages + +Choose how to map [OpenAI streaming events](#understand-streaming-events) to Ably messages. You can choose any mapping strategy that suits your application's needs. This guide uses the following pattern as an example: + +- `start`: Signals the beginning of a response +- `token`: Contains the incremental text content for each delta +- `stop`: Signals the completion of a response + + + +Update your `publisher.mjs` file to initialize the Ably client and update the `processEvent()` function to publish events to Ably: + + +```javascript +// Track state across events +let responseId = null; +let messageItemId = null; + +// Process each streaming event and publish to Ably +function processEvent(event) { + switch (event.type) { + case 'response.created': + // Capture response ID when response starts + responseId = event.response.id; + + // Publish start event + channel.publish({ + name: 'start', + extras: { + headers: { responseId } + } + }); + break; + + case 'response.output_item.added': + // Capture message item ID when a message output item is added + if (event.item.type === 'message') { + messageItemId = event.item.id; + } + break; + + case 'response.output_text.delta': + // Publish tokens from message output items only + if (event.item_id === messageItemId) { + channel.publish({ + name: 'token', + data: event.delta, + extras: { + headers: { responseId } + } + }); + } + break; + + case 'response.completed': + // Publish stop event when response completes + channel.publish({ + name: 'stop', + extras: { + headers: { responseId } + } + }); + break; + } +} +``` + + +This implementation: + +- Publishes a `start` event when the response begins +- Filters for `response.output_text.delta` events from `message` type output items and publishes them as `token` events +- Publishes a `stop` event when the response completes +- All published events include the `responseId` in message `extras` to allow the client to correlate events relating to a particular response + + + +Run the publisher to see tokens streaming to Ably: + + +```shell +node publisher.mjs +``` + + +## Step 3: Subscribe to streaming tokens + +Create a subscriber that receives the streaming events from Ably and reconstructs the response. + +Create a new file `subscriber.mjs` with the following contents: + + +```javascript +import Ably from 'ably'; + +// Initialize Ably Realtime client +const realtime = new Ably.Realtime({ key: '{{API_KEY}}' }); + +// Get the same channel used by the publisher +const channel = realtime.channels.get('{{RANDOM_CHANNEL_NAME}}'); + +// Track responses by ID +const responses = new Map(); + +// Handle response start +await channel.subscribe('start', (message) => { + const responseId = message.extras?.headers?.responseId; + console.log('\n[Response started]', responseId); + responses.set(responseId, ''); +}); + +// Handle tokens +await channel.subscribe('token', (message) => { + const responseId = message.extras?.headers?.responseId; + const token = message.data; + + // Append token to response + const currentText = responses.get(responseId) || ''; + responses.set(responseId, currentText + token); + + // Display token as it arrives + process.stdout.write(token); +}); + +// Handle response stop +await channel.subscribe('stop', (message) => { + const responseId = message.extras?.headers?.responseId; + const finalText = responses.get(responseId); + console.log('\n[Response completed]', responseId); +}); + +console.log('Subscriber ready, waiting for tokens...'); +``` + + +Run the subscriber in a separate terminal: + + +```shell +node subscriber.mjs +``` + + +With the subscriber running, run the publisher in another terminal. The tokens stream in realtime as they are generated by the OpenAI model. + +## Step 4: Stream with multiple publishers and subscribers + +Ably's [channel-oriented sessions](/docs/ai-transport/features/sessions-identity#connection-oriented-vs-channel-oriented-sessions) enables multiple AI agents to publish responses and multiple users to receive them on a single channel simultaneously. Ably handles message delivery to all participants, eliminating the need to implement routing logic or manage state synchronization across connections. + +### Broadcasting to multiple subscribers + +Each subscriber receives the complete stream of tokens independently, enabling you to build collaborative experiences or multi-device applications. + +Run a subscriber in multiple separate terminals: + + +```shell +# Terminal 1 +node subscriber.mjs + +# Terminal 2 +node subscriber.mjs + +# Terminal 3 +node subscriber.mjs +``` + + +All subscribers receive the same stream of tokens in realtime. + +### Publishing concurrent responses + +The implementation uses `responseId` in message `extras` to correlate tokens with their originating response. This enables multiple publishers to stream different responses concurrently on the same channel, with each subscriber correctly tracking all responses independently. + +To demonstrate this, run a publisher in multiple separate terminals: + + +```shell +# Terminal 1 +node publisher.mjs + +# Terminal 2 +node publisher.mjs + +# Terminal 3 +node publisher.mjs +``` + + +All running subscribers receive tokens from all responses concurrently. Each subscriber correctly reconstructs each response separately using the `responseId` to correlate tokens. + +## Next steps + +- Learn more about the [message-per-token pattern](/docs/ai-transport/features/token-streaming/message-per-token) used in this guide +- Learn about [client hydration strategies](/docs/ai-transport/features/token-streaming/message-per-token#hydration) for handling late joiners and reconnections +- Understand [sessions and identity](/docs/ai-transport/features/sessions-identity) in AI enabled applications +- Explore the [message-per-response pattern](/docs/ai-transport/features/token-streaming/message-per-response) for storing complete AI responses as single messages in history diff --git a/src/pages/docs/messages/annotations.mdx b/src/pages/docs/messages/annotations.mdx index ef24b54124..25b3554d67 100644 --- a/src/pages/docs/messages/annotations.mdx +++ b/src/pages/docs/messages/annotations.mdx @@ -19,18 +19,18 @@ When clients publish or delete an annotation, Ably automatically creates a [summ ## Enable annotations -Annotations can be enabled for a channel or channel namespace with the *Message annotations, updates, and deletes* channel rule. +Annotations can be enabled for a channel or channel namespace with the *Message annotations, updates, deletes, and appends* channel rule. 1. Go to the **Settings** tab of an app in your dashboard. 3. Under [channel rules](/docs/channels#rules), click **Add new rule**. 4. Enter the channel name or channel namespace on which to enable message annotations. -5. Check **Message annotations, updates, and deletes** to enable message annotations. +5. Check **Message annotations, updates, deletes, and appends** to enable message annotations. 6. Click **Create channel rule** to save. ## Annotation types diff --git a/src/pages/docs/messages/index.mdx b/src/pages/docs/messages/index.mdx index d3ecbad18a..f148ff6eee 100644 --- a/src/pages/docs/messages/index.mdx +++ b/src/pages/docs/messages/index.mdx @@ -34,7 +34,7 @@ The following are the properties of a message: | **extras** | A JSON object of arbitrary key-value pairs that may contain metadata, and/or ancillary payloads. Valid payloads include those related to [Push Notifications](/docs/push), [deltas](/docs/channels/options/deltas) and headers | | **encoding** | This is typically empty, as all messages received from Ably are automatically decoded client-side using this value. However, if the message encoding cannot be processed, this attribute contains the remaining transformations not applied to the data payload | | **action** | An [enum](/docs/api/realtime-sdk/types#message-action) telling you whether this is a normal ('create') message, an update to a previous message, an annotation summary, etc. | -| **serial** | The message's serial (a server-assigned identifier that will be the same in all future updates of this message, and can be used to add [annotations](/docs/messages/annotations) or to [update or delete](/docs/messages/updates-deletes) the message). This will only be set if you enable annotations, updates, and deletes in [channel rules](/docs/channels#rules) | +| **serial** | The message's serial (a server-assigned identifier that will be the same in all future updates of this message, and can be used to add [annotations](/docs/messages/annotations) or to [update or delete](/docs/messages/updates-deletes) the message). This will only be set if you enable annotations, updates, deletes, and appends in [channel rules](/docs/channels#rules) | | **annotations** | An object containing a summary of any [annotations](/docs/messages/annotations) that have been made to the message | | **version** | An object containing [version metadata](/docs/messages/updates-deletes#version-structure) for messages that have been updated or deleted | diff --git a/src/pages/docs/messages/updates-deletes.mdx b/src/pages/docs/messages/updates-deletes.mdx index 9aed8f97e5..25921fa44b 100644 --- a/src/pages/docs/messages/updates-deletes.mdx +++ b/src/pages/docs/messages/updates-deletes.mdx @@ -23,18 +23,18 @@ You can access the full version history of any given message. ## Enable message updates and deletes -Message updates and deletes can be enabled for a channel or channel namespace with the *Message annotations, updates, and deletes* channel rule. +Message updates and deletes can be enabled for a channel or channel namespace with the *Message annotations, updates, deletes, and appends* channel rule. 1. Go to the **Settings** tab of an app in your dashboard. 2. Under [channel rules](/docs/channels#rules), click **Add new rule**. 3. Enter the channel name or channel namespace on which to enable message updates and deletes. -4. Check **Message annotations, updates, and deletes** to enable the feature. +4. Check **Message annotations, updates, deletes, and appends** to enable the feature. 5. Click **Create channel rule** to save. ## Update a message diff --git a/src/pages/docs/platform/errors/codes.mdx b/src/pages/docs/platform/errors/codes.mdx index 538ae2279e..0e6eb53a13 100644 --- a/src/pages/docs/platform/errors/codes.mdx +++ b/src/pages/docs/platform/errors/codes.mdx @@ -149,7 +149,7 @@ Resolution: The following steps can help resolve this issue: This error occurs when the requested operation is only available in a newer version of the Ably protocol than the one specified in the request. This can happen if the client is using an older version of the Ably SDK that does not support the requested operation. -Examples of this error include attempting to use [channel objects](/docs/liveobjects) or message annotations, updates and deletes from an Ably SDK that does not support protocol version 2. +Examples of this error include attempting to use [channel objects](/docs/liveobjects) or message annotations, updates, deletes, and appends from an Ably SDK that does not support protocol version 2. Resolution: The following steps can help resolve this issue: * Upgrade the Ably SDK to a version that supports the requested operation. @@ -773,12 +773,12 @@ This error occurs when attempting to [subscribe to individual annotations](/docs **Resolution:** * Ensure that `annotation_subscribe` mode is specified in the client [channel options](/docs/channels/options) before subscribing to individual annotations. -## 93002: Annotations are only supported on channels with message annotations, updates, and deletes enabled +## 93002: Annotations are only supported on channels with message annotations, updates, deletes, and appends enabled This error occurs when attempting to use [message annotations](/docs/messages/annotations) on a channel that does not have them enabled. **Resolution:** -* Create a [channel rule](/docs/channels#rules) for the channel or channel namespace with **Message annotations, updates, and deletes** enabled. +* Create a [channel rule](/docs/channels#rules) for the channel or channel namespace with **Message annotations, updates, deletes, and appends** enabled. ## 101000: Space name missing