diff --git a/.yarnrc.yml b/.yarnrc.yml index a1c07901ea..549cd156a8 100644 --- a/.yarnrc.yml +++ b/.yarnrc.yml @@ -16,7 +16,7 @@ npmPreapprovedPackages: - stream-chat - react-native-teleport - react-native-nitro-sound - - '@stream-io/calypso' + - '@stream-io/stream-agents' - '@a2ui/web_core' npmPublishProvenance: true diff --git a/examples/ExpoChatGPTSample/README.md b/examples/ExpoChatGPTSample/README.md index eb804dbc29..3eabfc3bbb 100644 --- a/examples/ExpoChatGPTSample/README.md +++ b/examples/ExpoChatGPTSample/README.md @@ -10,18 +10,18 @@ Ported from the `ExpoChatGPTSample` of the deprecated `@stream-io/chat-react-nat ## How it works The agent runs on a [Vision-Agents](https://github.com/GetStream/Vision-Agents) router, which the app talks to through -[`@stream-io/calypso`](https://www.npmjs.com/package/@stream-io/calypso) (`agent/`). +[`@stream-io/stream-agents`](https://www.npmjs.com/package/@stream-io/stream-agents) (`agent/`). - Each conversation is a router session, kept in an `agent` channel. The drawer lists the user's `agent` channels. `+` starts a new session, and one that is never asked anything is deleted when you leave it. -- The conversation is `AIConversation`, fed from `agent/agentChat.ts`: a small store that watches the channel, copies - its messages on every `message.*` event, loads older pages, and keeps each reply's full reasoning from its live - updates. There's no ``, so the store also watches the channel again after `connection.recovered`, to pick up - what was written while the connection was down. +- The conversation is `AIConversation`, over an `AIConversationController` that `contexts/AppContext.tsx` builds from + the SDK's `AgentChat` and `agentChatTransport` (`@stream-io/stream-agents/chat`). `AgentChat` watches the channel, + loads older pages, keeps each reply's full reasoning from its live updates, and follows what the agent is doing on + the session's socket. The controller offers no edit or regenerate: the router can't rewind a conversation. - Asking goes through the session, not the chat composer. The router writes the question and the streamed reply into the channel, the reply's steps as `ai_reasoning` and `ai_tool_call` attachments, and titles the conversation itself. - A plain question goes to `respond` with this install's id, so the agent can call the phone's tools - (`agent/deviceTools.ts`). A question with photos or "Think harder" goes to `responses.create`, which takes neither. + Photos go as data URIs, and "Think harder" as a high reasoning effort. The phone's tools (`agent/deviceTools.ts`) are + declared on the session, which runs them. - The keyboard is handled by the AI components through `react-native-keyboard-controller`; the app only wraps itself in `KeyboardProvider`. The starters and the voice button sit in the composer's `header`, so they ride the keyboard. diff --git a/examples/ExpoChatGPTSample/agent/agentChat.ts b/examples/ExpoChatGPTSample/agent/agentChat.ts deleted file mode 100644 index 5d9a03571d..0000000000 --- a/examples/ExpoChatGPTSample/agent/agentChat.ts +++ /dev/null @@ -1,180 +0,0 @@ -import { LiveReasoning, type Session } from '@stream-io/calypso'; -import { type Channel, type Event, type LocalMessage, StateStore } from 'stream-chat'; - -/** How many messages are read at a time. */ -const PAGE_SIZE = 30; - -export type AgentChatState = { - /** The channel's messages, oldest first. */ - messages: readonly LocalMessage[]; - /** Whether the channel holds messages older than those loaded. */ - hasOlderMessages: boolean; - /** Whether the first page of messages is in. A new conversation has none to wait for. */ - isLoaded: boolean; - /** The session the conversation is held in, once something here needed it. */ - session: Session | undefined; -}; - -/** - * Opens a conversation's session: carries on `sessionId`, or starts a new conversation without one. - */ -export type OpenSession = (sessionId: string | undefined) => Promise; - -/** - * A written conversation as the AI components want it: the messages of the channel the router - * keeps it in, and the whole of each reply's thinking. - * - * The router writes everything there. Asking goes through the session, which records the question - * in the channel and answers it there: the reply's text, and its steps (each round of reasoning and - * each tool call) as attachments. - * - * The channel and the session open apart. Reading a conversation needs only its channel, which - * Stream serves in a fraction of the time the session takes to come back, and carrying a session - * on marks the conversation active, which moves it up the router's list. So the session opens only - * once something needs it, such as a question, and a new conversation has none until it is asked - * something: the router never lists one that was only looked at. - */ -export class AgentChat { - /** Each reply's thinking in full, from its live updates. A stored step only keeps its opening. */ - readonly reasoning = new LiveReasoning(); - - /** Whether the person asked something here, which may not have arrived in `messages` yet. */ - wasAsked = false; - - readonly state: StateStore; - - private channel: Channel | undefined; - private stops: (() => void)[] = []; - private opening: Promise | undefined; - private closed = false; - - /** - * @param key Tells this conversation apart from the one shown before it, such as for a React key. - * @param sessionId The conversation's session, or `undefined` for a new conversation. - */ - constructor( - readonly key: string, - private readonly sessionId: string | undefined, - private readonly openSession: OpenSession, - ) { - this.state = new StateStore({ - hasOlderMessages: false, - isLoaded: sessionId === undefined, - messages: [], - session: undefined, - }); - } - - /** The conversation's session id: the one it was opened with, or the one asking created. */ - get id() { - return this.state.getLatestValue().session?.id ?? this.sessionId; - } - - /** Watches the channel and loads its latest messages. */ - async follow(channel: Channel) { - if (this.channel || this.closed) return; - this.channel = channel; - // Every event rather than the list kept for us: each live update carries its own window of the - // reply's thinking, and folding them into one would lose all but the last. - const subscription = channel.on((event: Event) => { - if (event.type === 'message.updated' && event.message) { - this.reasoning.read(event.message as { id: string; reasoning?: unknown }); - } - if ( - event.type === 'message.new' || - event.type === 'message.updated' || - event.type === 'message.deleted' - ) { - // A new array, in which only the messages that changed are new objects. - this.state.partialNext({ messages: [...channel.state.messages] }); - } - }); - this.stops.push(() => subscription.unsubscribe()); - - const reconnected = channel.getClient().on('connection.changed', (event: Event) => { - if (event.online) void this.watchLatest().catch(() => {}); - }); - this.stops.push(() => reconnected.unsubscribe()); - - // A channel this client read before shows what it had at once, while the latest page loads. - if (channel.state.messages.length > 0) { - this.state.partialNext({ isLoaded: true, messages: [...channel.state.messages] }); - } - await this.watchLatest(); - } - - /** - * The conversation's session, opened the first time it is needed. A new conversation is created - * then, and its channel followed. Rejects when it can't open, to be asked again. - */ - session(): Promise { - this.opening ??= this.open().catch((error: unknown) => { - this.opening = undefined; - throw error; - }); - return this.opening; - } - - /** - * Deletes the session a new conversation was given when the question that created it failed - * before anything was written, so the router doesn't list an empty conversation. The next - * question creates another. - */ - async discardUnusedSession() { - const { messages, session } = this.state.getLatestValue(); - if (this.sessionId !== undefined || !session || messages.length > 0) return; - this.opening = undefined; - this.state.partialNext({ session: undefined }); - await session.delete().catch(() => {}); - } - - /** Loads the messages before the oldest one loaded. Throws when it can't, to be asked again. */ - loadOlderMessages = async () => { - const oldest = this.state.getLatestValue().messages[0]?.id; - if (!oldest || !this.channel) return; - const page = await this.channel.query({ messages: { id_lt: oldest, limit: PAGE_SIZE } }); - this.state.partialNext({ - hasOlderMessages: page.messages.length >= PAGE_SIZE, - messages: [...this.channel.state.messages], - }); - }; - - /** Stops following the channel, and lets the session go. */ - async close() { - this.closed = true; - for (const stop of this.stops.splice(0)) stop(); - const opened = await this.opening?.catch(() => undefined); - await opened?.close(); - } - - private async open() { - const session = await this.openSession(this.sessionId); - if (this.closed) { - // The person moved on while it opened. A conversation created for them was never used. - await (this.sessionId === undefined ? session.delete() : session.close()).catch(() => {}); - throw new Error('The conversation was closed'); - } - if (!this.channel) { - const { channel } = await session.chat(); - await this.follow(channel as Channel); - } - this.state.partialNext({ session }); - return session; - } - - /** Watches the channel again and takes its latest page, in place of what was loaded. */ - private async watchLatest() { - const channel = this.channel; - if (!channel) return; - const watched = await channel.watch({ messages: { limit: PAGE_SIZE } }); - if (this.closed) return; - this.state.partialNext({ - hasOlderMessages: (watched.messages?.length ?? 0) >= PAGE_SIZE, - isLoaded: true, - messages: [...channel.state.messages], - }); - } -} - -/** A reply the agent wrote. */ -export const isAgentReply = (message: LocalMessage) => message.ai_generated === true; diff --git a/examples/ExpoChatGPTSample/agent/agentTransport.ts b/examples/ExpoChatGPTSample/agent/agentTransport.ts deleted file mode 100644 index 0972ac1d16..0000000000 --- a/examples/ExpoChatGPTSample/agent/agentTransport.ts +++ /dev/null @@ -1,109 +0,0 @@ -import type { ConversationSnapshot, Session, UIAction } from '@stream-io/calypso'; -import { - type AIAgentActivity, - AIAgentActivityStates, - type AIConversationTransport, - createFromStreamMessage, -} from 'stream-chat-expo/ai-components'; - -import type { AgentChat, AgentChatState } from './agentChat'; -import { toDataUri } from './photos'; - -import { THINK_HARDER } from '../chatOptions'; - -/** What the person did in a reply's UI, which a request carries in its `custom` data. */ -export type AgentRequestCustom = { uiAction?: UIAction }; - -export type AgentTransportOptions = { - /** Asks the agent in `session`, with photos as data URIs, a reasoning effort and a UI action. */ - respond: ( - session: Session, - text: string, - options: { images: { url: string }[]; reasoningEffort?: string; uiAction?: UIAction }, - ) => Promise; - /** Called once asking created the conversation, so the drawer lists it. */ - onCreated: () => void; -}; - -// What the agent is doing, from the session's socket: thinking while it responds, and the skills -// it runs while it works. -const activityOf = (snapshot: ConversationSnapshot): AIAgentActivity | undefined => { - if (snapshot.kind === 'responding') return { state: AIAgentActivityStates.thinking }; - if (snapshot.kind === 'working') { - return { - state: AIAgentActivityStates.checkingExternalSources, - text: `Working on ${snapshot.skills.map((skill) => skill.replace(/_/g, ' ')).join(' and ')}`, - }; - } - return undefined; -}; - -/** - * The router's conversation as an `AIConversationController` follows it: the messages of the - * channel `AgentChat` reads, and what the agent is doing from the session's socket. Asking goes - * through the session, which records the question in the channel and answers it there. The - * router has no regenerate or edit of its own: those ask again, as a new question. - */ -export const createAgentTransport = ( - chat: AgentChat, - { onCreated, respond }: AgentTransportOptions, -): AIConversationTransport => ({ - connect: (feed) => { - // The same message object for each Stream message that didn't change, so only the messages - // an update touched render again. The live thinking is read through `chat.reasoning`. - const fromStreamMessage = createFromStreamMessage({ - reasoningText: (messageId, stepId) => chat.reasoning.get(messageId, stepId), - }); - let followedSession: Session | undefined; - let stopFollowingSession: (() => void) | undefined; - const update = ({ hasOlderMessages, messages, session }: AgentChatState) => { - feed.updateMessages(messages.map(fromStreamMessage)); - feed.updateHasOlderMessages(hasOlderMessages); - if (session === followedSession) return; - stopFollowingSession?.(); - followedSession = session; - stopFollowingSession = session?.state.subscribe(() => - feed.updateActivity(activityOf(session.state.get())), - ); - if (session) feed.updateActivity(activityOf(session.state.get())); - }; - update(chat.state.getLatestValue()); - const unsubscribe = chat.state.subscribe(update); - return () => { - unsubscribe(); - stopFollowingSession?.(); - }; - }, - loadOlderMessages: () => chat.loadOlderMessages(), - send: async ({ attachments, chatOption, custom, text }) => { - // From now on the conversation is used, even before Stream's copy of the question arrives. - chat.wasAsked = true; - const photos = attachments - .filter((asset) => !asset.type || asset.type.startsWith('image')) - .map((asset) => asset.uri); - try { - // A new conversation is created here, with its first question. - const wasNew = chat.id === undefined; - const [session, images] = await Promise.all([ - chat.session(), - Promise.all(photos.map(async (uri) => ({ url: await toDataUri(uri) }))), - ]); - if (wasNew) onCreated(); - await respond(session, text, { - images, - reasoningEffort: chatOption?.id === THINK_HARDER.id ? 'high' : undefined, - uiAction: (custom as AgentRequestCustom | undefined)?.uiAction, - }); - } catch (error) { - // A conversation created for this question, which it never reached, isn't left listed. - await chat.discardUnusedSession(); - throw error; - } - }, - // A reply that was being written when the conversation was opened stops through its session, - // which opens for it. - stop: async () => { - const session = await chat.session(); - await session.interrupt(); - }, -}); diff --git a/examples/ExpoChatGPTSample/agent/agents.ts b/examples/ExpoChatGPTSample/agent/agents.ts index b7b6c55ea9..9d557feecb 100644 --- a/examples/ExpoChatGPTSample/agent/agents.ts +++ b/examples/ExpoChatGPTSample/agent/agents.ts @@ -1,6 +1,6 @@ -import { Client } from '@stream-io/calypso'; +import { Client } from '@stream-io/stream-agents'; // Lets `session.chat()` open the conversation's channel on the app's Stream Chat client. -import '@stream-io/calypso/chat'; +import '@stream-io/stream-agents/chat'; import { AgentConfig } from './config'; diff --git a/examples/ExpoChatGPTSample/agent/conversation.ts b/examples/ExpoChatGPTSample/agent/conversation.ts new file mode 100644 index 0000000000..95c6483e94 --- /dev/null +++ b/examples/ExpoChatGPTSample/agent/conversation.ts @@ -0,0 +1,49 @@ +import type { AgentChat, AgentChatState } from '@stream-io/stream-agents/chat'; +import type { LocalMessage, StateStore } from 'stream-chat'; + +/** A written conversation with the agent, over the messages of the channel it is kept in. */ +export type Chat = AgentChat; + +export type ChatState = AgentChatState; + +/** + * The chat's state for `useStateStore`. The SDK's store has the shape of stream-chat's + * `StateStore` (`getLatestValue` and `subscribeWithSelector`, which is all `useStateStore` calls) + * without depending on stream-chat, so the type is only told so here. + */ +export const chatStore = (chat: Chat) => chat.state as unknown as StateStore; + +let keys = 0; +const chatKeys = new WeakMap(); + +/** Tells a conversation apart from the one shown before it, for a React key. */ +export const keyOf = (chat: Chat) => { + let key = chatKeys.get(chat); + if (!key) { + key = `chat-${++keys}`; + chatKeys.set(chat, key); + } + return key; +}; + +/** A reply the agent wrote. */ +export const isAgentReply = (message: LocalMessage) => message.ai_generated === true; + +/** + * Calls `used` when a new conversation is used, such as asked something: the router lists it from + * then on. A first question that never reached the router has its session deleted, and the next + * one creates another, which is listed in its turn. A conversation carried on is listed already, + * and isn't waited for. Returns a function that stops waiting. + */ +export const whenUsed = (chat: Chat, used: () => void) => { + if (chat.id !== undefined) return () => {}; + let listed: string | undefined; + return chat.state.subscribeWithSelector( + ({ usedSessionId }) => ({ usedSessionId }), + ({ usedSessionId }) => { + if (usedSessionId === undefined || usedSessionId === listed) return; + listed = usedSessionId; + used(); + }, + ); +}; diff --git a/examples/ExpoChatGPTSample/agent/deviceTools.ts b/examples/ExpoChatGPTSample/agent/deviceTools.ts index 205bbe8890..9ade250922 100644 --- a/examples/ExpoChatGPTSample/agent/deviceTools.ts +++ b/examples/ExpoChatGPTSample/agent/deviceTools.ts @@ -1,4 +1,4 @@ -import { Tools } from '@stream-io/calypso'; +import { Tools } from '@stream-io/stream-agents'; import { type AIClientTool, @@ -51,16 +51,14 @@ class Reports { else this.reported.set(callId, result); } - wait(callId: string, signal: AbortSignal) { - return new Promise((resolve, reject) => { + // A call that was cancelled stops being waited on by the session's tools, which leave its + // result unread. + wait(callId: string) { + return new Promise((resolve) => { const reported = this.reported.get(callId); this.reported.delete(callId); if (reported) return resolve(reported); this.waiting.set(callId, resolve); - signal.addEventListener('abort', () => { - this.waiting.delete(callId); - reject(signal.reason); - }); }); } } @@ -107,9 +105,9 @@ export const createDeviceTools = ({ clientId, userId }: { clientId: string; user executor: 'client', name, parameters, - run: async (input, signal, call) => { + run: async (input, call) => { const { failure, output } = deviceTools.addressed - ? await reports.wait(call.id, signal) + ? await reports.wait(call.id) : await runUnaddressed(name, { arguments: input, executor: 'client', diff --git a/examples/ExpoChatGPTSample/agent/photos.ts b/examples/ExpoChatGPTSample/agent/photos.ts index 3dc3d8d0bf..7dac92f52e 100644 --- a/examples/ExpoChatGPTSample/agent/photos.ts +++ b/examples/ExpoChatGPTSample/agent/photos.ts @@ -1,4 +1,5 @@ import { manipulateAsync, SaveFormat } from 'expo-image-manipulator'; +import type { AIConversationRequest } from 'stream-chat-expo/ai-components'; // The router takes photos inline, so they are shrunk first. const MAX_IMAGE_WIDTH = 1280; @@ -12,3 +13,14 @@ export const toDataUri = async (uri: string) => { }); return `data:image/jpeg;base64,${base64}`; }; + +/** + * The photos among the files the person picked, as the images the agent's transport sends. + * Anything else they picked isn't sent: the router takes images only. + */ +export const imagesOf = (attachments: readonly unknown[]) => + Promise.all( + (attachments as AIConversationRequest['attachments']) + .filter((asset) => !asset.type || asset.type.startsWith('image')) + .map(async (asset) => ({ url: await toDataUri(asset.uri) })), + ); diff --git a/examples/ExpoChatGPTSample/contexts/AppContext.tsx b/examples/ExpoChatGPTSample/contexts/AppContext.tsx index ff6b7b347b..ca47262e37 100644 --- a/examples/ExpoChatGPTSample/contexts/AppContext.tsx +++ b/examples/ExpoChatGPTSample/contexts/AppContext.tsx @@ -1,22 +1,35 @@ import type { PropsWithChildren } from 'react'; import React, { useEffect, useMemo, useRef, useState } from 'react'; -import { type ConversationSnapshot, type Session, type UIAction } from '@stream-io/calypso'; -import type { StreamChat } from 'stream-chat'; +import { type ConversationSnapshot, type Session } from '@stream-io/stream-agents'; +import { + AgentChat, + agentChatTransport, + type OpenAgentSession, +} from '@stream-io/stream-agents/chat'; +import type { LocalMessage, StreamChat } from 'stream-chat'; import { useStableCallback, useStateStore } from 'stream-chat-expo'; import { AIClientIdentity, AIConversationController, type AIToolApprover, createAuthSessionAuthorizer, + createFromStreamMessage, parseMessageParts, } from 'stream-chat-expo/ai-components'; -import { AgentChat, type AgentChatState, isAgentReply, type OpenSession } from '../agent/agentChat'; import { agents, agentSessions, streamToken } from '../agent/agents'; -import { createAgentTransport } from '../agent/agentTransport'; import { AgentConfig } from '../agent/config'; +import { + type Chat, + type ChatState, + chatStore, + isAgentReply, + whenUsed, +} from '../agent/conversation'; import { createDeviceTools } from '../agent/deviceTools'; +import { imagesOf } from '../agent/photos'; +import { THINK_HARDER } from '../chatOptions'; const IDLE: ConversationSnapshot = { kind: 'idle', skills: [] }; @@ -31,7 +44,7 @@ const deviceTools = createDeviceTools({ clientId, userId: AgentConfig.user.id }) const SIGN_IN_SCHEME = 'io.getstream.expochatgptsample'; const SIGN_IN_RETURN_URL = `${SIGN_IN_SCHEME}://sign-in`; -// The person's answer to a call that asks first. calypso's `decide` answers only a call this +// The person's answer to a call that asks first. stream-agents' `decide` answers only a call this // session was asked on its own socket, and leaves any other alone, so a question shown from the // channel, as after reopening the conversation or reloading the app, would wait forever. Such a // call is answered with the frames `decide` sends, which the router matches by the call's id: an @@ -54,26 +67,9 @@ const answerApproval = (session: Session, toolCallId: string, allowed: boolean) } }; -// Asks, with calypso's `responses.create`, which takes photos, a reasoning effort and what the -// person did in a reply's UI. The router's `respond`, which addresses a client tool's call to the -// install a question came from, is only for an app's backend, and this is a phone: the calls reach -// the device's tools without an install, and the phone runs them itself. -const respond = async ( - session: Session, - text: string, - options: { images: { url: string }[]; reasoningEffort?: string; uiAction?: UIAction }, -) => { - deviceTools.addressed = false; - await session.responses.create(text, { - ...(options.images.length > 0 ? { images: options.images } : {}), - ...(options.reasoningEffort ? { reasoningEffort: options.reasoningEffort } : {}), - ...(options.uiAction ? { uiAction: options.uiAction } : {}), - }); -}; - // Answers go over the session's socket, which opens for them in a conversation that was only // read so far. Each throws when it can't send, so the card lets the person answer again. -const approverFor = (chat: AgentChat): AIToolApprover => ({ +const approverFor = (chat: Chat): AIToolApprover => ({ authorizer: createAuthSessionAuthorizer({ callbackScheme: SIGN_IN_SCHEME, decline: async (call) => (await chat.session()).declineLogin(call.id), @@ -87,7 +83,7 @@ export type AppContextValue = { /** What the agent is doing in the open conversation, from the session's socket. */ agentState: ConversationSnapshot; /** The open conversation's messages and live reasoning. */ - chat: AgentChat | undefined; + chat: Chat | undefined; /** The open conversation's state, and what the person does in it. */ conversation: AIConversationController | undefined; /** Bumped when the drawer's list of conversations may have changed. */ @@ -120,13 +116,13 @@ const messageOf = (error: unknown) => (error instanceof Error ? error.message : // What the provider follows of the messages: the latest reply, whose calls this phone runs, and // the session, once one is open. -const selector = ({ messages, session }: AgentChatState) => ({ +const selector = ({ messages, session }: ChatState) => ({ lastReply: [...messages].reverse().find(isAgentReply), session, }); // Carries a conversation on, or creates one, with the phone's tools declared on it. -const openSession: OpenSession = (sessionId) => +const openSession: OpenAgentSession = (sessionId) => sessionId ? agentSessions().resume(sessionId, { tools: deviceTools.sessionTools() }) : agentSessions().create({ @@ -135,10 +131,37 @@ const openSession: OpenSession = (sessionId) => tools: deviceTools.sessionTools(), }); -let chats = 0; -const newChat = (sessionId?: string) => new AgentChat(`chat-${++chats}`, sessionId, openSession); +const newChat = (sessionId?: string): Chat => + new AgentChat({ open: openSession, sessionId }); -type Opened = { chat: AgentChat; conversation: AIConversationController }; +// The conversation's state over the chat. Asking goes through the session, which records the +// question in the channel and answers it there; the router takes photos as data URIs, and +// thinking harder as a high reasoning effort. +const conversationOf = (chat: Chat) => { + // The same message object for each Stream message that didn't change, so only the messages an + // update touched render again. The live thinking is read through `chat.reasoning`. + const fromStreamMessage = createFromStreamMessage({ + reasoningText: (messageId, stepId) => chat.reasoning.get(messageId, stepId), + }); + return new AIConversationController({ + // The agents router keeps the conversation in Stream Chat and can't rewind it, so a reply + // can't be written again or a message edited: either would only be asked again as new. + actions: { edit: false, regenerate: false }, + approver: approverFor(chat), + transport: agentChatTransport(chat, { + adapt: (message) => fromStreamMessage(message), + images: imagesOf, + reasoningEffort: (option) => (option?.id === THINK_HARDER.id ? 'high' : undefined), + }), + }); +}; + +type Opened = { + chat: Chat; + conversation: AIConversationController; + /** Stops waiting for a new conversation to be used. */ + stopWaiting: () => void; +}; export const AppProvider = ({ client, children }: PropsWithChildren<{ client: StreamChat }>) => { const [opened, setOpened] = useState(); @@ -148,7 +171,7 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St const current = useRef(undefined); const chat = opened?.chat; - const followed = useStateStore(chat?.state, selector); + const followed = useStateStore(chat ? chatStore(chat) : undefined, selector); const session = followed?.session; // Runs the calls of the latest reply that wait for this phone, each once. @@ -161,21 +184,17 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St }, [lastReply]); // Shows a conversation in place of the one before it, at once, and lets the one before go. - const show = (next: AgentChat) => { + const show = (next: Chat) => { const previous = current.current; - const conversation = new AIConversationController({ - // The agents router keeps the conversation in Stream Chat and can't rewind it, so a reply - // can't be written again or a message edited: either would only be asked again as new. - actions: { edit: false, regenerate: false }, - approver: approverFor(next), - transport: createAgentTransport(next, { - onCreated: () => setConversationsVersion((version) => version + 1), - respond, - }), - }); - current.current = { chat: next, conversation }; + current.current = { + chat: next, + conversation: conversationOf(next), + // The router lists a new conversation once its first question created it. + stopWaiting: whenUsed(next, () => setConversationsVersion((version) => version + 1)), + }; setFailure(undefined); setOpened(current.current); + previous?.stopWaiting(); previous?.conversation.dispose(); void previous?.chat.close().catch(() => {}); }; diff --git a/examples/ExpoChatGPTSample/package.json b/examples/ExpoChatGPTSample/package.json index 9768628c79..49a86ed266 100644 --- a/examples/ExpoChatGPTSample/package.json +++ b/examples/ExpoChatGPTSample/package.json @@ -12,7 +12,7 @@ "dependencies": { "@react-native-community/netinfo": "12.0.1", "@shopify/react-native-skia": "^2.14.0", - "@stream-io/calypso": "^0.3.0", + "@stream-io/stream-agents": "^0.4.1", "expo": "^58.0.8", "expo-clipboard": "~58.0.3", "expo-constants": "~58.0.11", diff --git a/examples/ExpoChatGPTSample/screens/ChatContent.tsx b/examples/ExpoChatGPTSample/screens/ChatContent.tsx index efcea65fff..f85ce9e6b7 100644 --- a/examples/ExpoChatGPTSample/screens/ChatContent.tsx +++ b/examples/ExpoChatGPTSample/screens/ChatContent.tsx @@ -28,8 +28,7 @@ import { } from 'stream-chat-expo/ai-components'; import { AIConversation } from 'stream-chat-expo/ai-components/conversation'; -import type { AgentChat, AgentChatState } from '../agent/agentChat'; -import type { AgentRequestCustom } from '../agent/agentTransport'; +import { type Chat, type ChatState, chatStore, keyOf } from '../agent/conversation'; import { chatOptions, STARTERS } from '../chatOptions'; import { tripComponents } from '../components/TripComponents'; import { VoiceMode } from '../components/VoiceMode'; @@ -47,7 +46,7 @@ const ComposerArea = ({ conversation, failure, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; failure: string | undefined; }) => { @@ -90,7 +89,7 @@ const ComposerArea = ({ const EmptyState = () => What can I help with ?; // A conversation still loading isn't empty: it shows no starters meanwhile. -const isEmptySelector = ({ isLoaded, messages }: AgentChatState) => ({ +const isEmptySelector = ({ isLoaded, messages }: ChatState) => ({ isEmpty: isLoaded && messages.length === 0, }); @@ -106,10 +105,10 @@ const Starters = ({ chat, conversation, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; }) => { - const { isEmpty } = useStateStore(chat.state, isEmptySelector); + const { isEmpty } = useStateStore(chatStore(chat), isEmptySelector); const { pendingMessage } = useAIConversationState(conversation, pendingSelector); if (!isEmpty || pendingMessage) return null; const onSelect = (text: string) => { @@ -119,18 +118,18 @@ const Starters = ({ return ; }; -const loadedSelector = ({ isLoaded }: AgentChatState) => ({ isLoaded }); +const loadedSelector = ({ isLoaded }: ChatState) => ({ isLoaded }); /** The open conversation, as its controller has it. */ const Conversation = ({ chat, conversation, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; }) => { const { agentState } = useAppContext(); - const { isLoaded } = useStateStore(chat.state, loadedSelector); + const { isLoaded } = useStateStore(chatStore(chat), loadedSelector); // The UI the agent writes for its replies, in A2UI, with the trip planner's own components. One // host per conversation, for as long as it shows, so what a reply's UI holds survives scrolling. @@ -144,7 +143,7 @@ const Conversation = ({ if (isAIConversationAnswering(conversation.state.getLatestValue().status)) { throw new Error('The agent is still answering'); } - const custom: AgentRequestCustom = { uiAction: aiUIActionJSON(action) }; + const custom = { uiAction: aiUIActionJSON(action) }; conversation.send({ custom, text: action.label }); }, }), @@ -182,7 +181,7 @@ export const ChatContent = () => { return ( } - | { event: string } - | { voiceMemoUrl: string }; - -/** How many messages are read at a time. */ -const PAGE_SIZE = 30; - -export type AgentChatState = { - /** The channel's messages, oldest first. */ - messages: readonly LocalMessage[]; - /** Whether the channel holds messages older than those loaded. */ - hasOlderMessages: boolean; - /** Whether the first page of messages is in. A new conversation has none to wait for. */ - isLoaded: boolean; - /** The session the conversation is held in, once something here needed it. */ - session: Session | undefined; - /** Whether the agent was asked to open the conversation and hasn't written yet. */ - awaitingAgent: boolean; -}; - -/** - * Opens a conversation's session: carries on `sessionId`, or starts a new conversation without one. - */ -export type OpenSession = (sessionId: string | undefined) => Promise; - -/** - * A written conversation as the AI components want it: the messages of the channel the router - * keeps it in, and the whole of each reply's thinking. - * - * The router writes everything there. Asking goes through the session, which records the question - * in the channel and answers it there: the reply's text, and its steps (each round of reasoning and - * each tool call) as attachments. - * - * The channel and the session open apart. Reading a conversation needs only its channel, which - * Stream serves in a fraction of the time the session takes to come back, and carrying a session - * on marks the conversation active, which moves it up the router's list. So the session opens only - * once something needs it, such as a question, and a new conversation has none until it is asked - * something: the router never lists one that was only looked at. - */ -export class AgentChat { - /** Each reply's thinking in full, from its live updates. A stored step only keeps its opening. */ - readonly reasoning = new LiveReasoning(); - - /** Whether the person asked something here, which may not have arrived in `messages` yet. */ - wasAsked = false; - - /** What was last asked of the agent here. */ - lastRequest: AgentRequest | undefined; - - readonly state: StateStore; - - private channel: Channel | undefined; - private stops: (() => void)[] = []; - private opening: Promise | undefined; - private closed = false; - - /** - * @param key Tells this conversation apart from the one shown before it, such as for a React key. - * @param sessionId The conversation's session, or `undefined` for a new conversation. - */ - constructor( - readonly key: string, - private readonly sessionId: string | undefined, - private readonly openSession: OpenSession, - ) { - this.state = new StateStore({ - awaitingAgent: false, - hasOlderMessages: false, - isLoaded: sessionId === undefined, - messages: [], - session: undefined, - }); - } - - /** The channel the conversation is kept in, once it is followed. */ - get conversationChannel() { - return this.channel; - } - - /** The conversation's session id: the one it was opened with, or the one asking created. */ - get id() { - return this.state.getLatestValue().session?.id ?? this.sessionId; - } - - /** Watches the channel and loads its latest messages. */ - async follow(channel: Channel) { - if (this.channel || this.closed) return; - this.channel = channel; - // Every event rather than the list kept for us: each live update carries its own window of the - // reply's thinking, and folding them into one would lose all but the last. - const subscription = channel.on((event: Event) => { - if (event.type === 'message.updated' && event.message) { - this.reasoning.read(event.message as { id: string; reasoning?: unknown }); - } - if ( - event.type === 'message.new' || - event.type === 'message.updated' || - event.type === 'message.deleted' - ) { - // A new array, in which only the messages that changed are new objects. - const messages = [...channel.state.messages]; - this.state.partialNext({ - messages, - ...(messages.some(showsAgentReply) ? { awaitingAgent: false } : {}), - }); - } - }); - this.stops.push(() => subscription.unsubscribe()); - - const reconnected = channel.getClient().on('connection.changed', (event: Event) => { - if (event.online) void this.watchLatest().catch(() => {}); - }); - this.stops.push(() => reconnected.unsubscribe()); - - const foregrounded = AppState.addEventListener('change', (state) => { - if (state === 'active' && this.dropped()) void this.session().catch(() => {}); - }); - this.stops.push(() => foregrounded.remove()); - - // A channel this client read before shows what it had at once, while the latest page loads. - if (channel.state.messages.length > 0) { - this.state.partialNext({ isLoaded: true, messages: [...channel.state.messages] }); - } - await this.watchLatest(); - } - - /** - * The conversation's session, opened the first time it is needed. A new conversation is created - * then, and its channel followed. A session whose socket dropped, as it does once the app has been - * in the background, is carried on in a new one. Rejects when it can't open, to be asked again. - */ - session(): Promise { - if (this.dropped()) this.opening = undefined; - this.opening ??= this.open().catch((error: unknown) => { - this.opening = undefined; - throw error; - }); - return this.opening; - } - - /** - * Deletes the session a new conversation was given when the question that created it failed - * before anything was written, so the router doesn't list an empty conversation. The next - * question creates another. - */ - async discardUnusedSession() { - const { messages, session } = this.state.getLatestValue(); - if (this.sessionId !== undefined || !session || messages.length > 0) return; - this.opening = undefined; - this.state.partialNext({ session: undefined }); - await session.delete().catch(() => {}); - } - - /** Loads the messages before the oldest one loaded. Throws when it can't, to be asked again. */ - loadOlderMessages = async () => { - const oldest = this.state.getLatestValue().messages[0]?.id; - if (!oldest || !this.channel) return; - const page = await this.channel.query({ messages: { id_lt: oldest, limit: PAGE_SIZE } }); - this.state.partialNext({ - hasOlderMessages: page.messages.length >= PAGE_SIZE, - messages: [...this.channel.state.messages], - }); - }; - - /** Stops following the channel, and lets the session go. */ - async close() { - this.closed = true; - for (const stop of this.stops.splice(0)) stop(); - const opened = await this.opening?.catch(() => undefined); - await opened?.close(); - } - - /** Whether the open session's socket has stopped carrying the conversation. */ - private dropped() { - const { session } = this.state.getLatestValue(); - return !this.closed && session !== undefined && !session.live; - } - - private async open() { - const opened = this.id; - const session = await this.openSession(opened); - if (this.closed) { - // The person moved on while it opened. A conversation created for them was never used. - await (opened === undefined ? session.delete() : session.close()).catch(() => {}); - throw new Error('The conversation was closed'); - } - if (!this.channel) { - const { channel } = await session.chat(); - await this.follow(channel as Channel); - } - this.state.partialNext({ session }); - return session; - } - - /** Watches the channel again and takes its latest page, in place of what was loaded. */ - private async watchLatest() { - const channel = this.channel; - if (!channel) return; - const watched = await channel.watch({ messages: { limit: PAGE_SIZE } }); - if (this.closed) return; - this.state.partialNext({ - hasOlderMessages: (watched.messages?.length ?? 0) >= PAGE_SIZE, - isLoaded: true, - messages: [...channel.state.messages], - }); - } -} - -/** A reply the agent wrote. */ -export const isAgentReply = (message: LocalMessage) => message.ai_generated === true; - -/** - * A reply of the agent's that has something to show: the router writes it empty first and fills - * it in, and the conversation hides it until then. - */ -const showsAgentReply = (message: LocalMessage) => - isAgentReply(message) && - (message.generating !== true || !!message.text || (message.attachments?.length ?? 0) > 0); diff --git a/examples/ExpoCrewAssistant/agent/agentTransport.ts b/examples/ExpoCrewAssistant/agent/agentTransport.ts deleted file mode 100644 index adf27a9818..0000000000 --- a/examples/ExpoCrewAssistant/agent/agentTransport.ts +++ /dev/null @@ -1,147 +0,0 @@ -import type { ConversationSnapshot, Session, UIAction } from '@stream-io/calypso'; -import { - type AIAgentActivity, - AIAgentActivityStates, - type AIConversationTransport, - type AIMessage, - type AIMessagePartData, - type AIStreamMessage, - createFromStreamMessage, -} from 'stream-chat-expo/ai-components'; - -import type { AgentChat, AgentChatState } from './agentChat'; -import { DISPLAY_TITLES } from './crewTools'; -import { toDataUri } from './photos'; - -/** What the person did in a reply's UI, which a request carries in its `custom` data. */ -export type AgentRequestCustom = { uiAction?: UIAction }; - -export type AgentTransportOptions = { - /** Asks the agent in `session`, with photos as data URIs, a reasoning effort and a UI action. */ - respond: ( - session: Session, - text: string, - options: { images: { url: string }[]; reasoningEffort?: string; uiAction?: UIAction }, - ) => Promise; - /** Called once asking created the conversation, so the drawer lists it. */ - onCreated: () => void; -}; - -// What the agent is doing, from the session's socket: thinking while it responds, and the skills -// it runs while it works. -const activityOf = (snapshot: ConversationSnapshot): AIAgentActivity | undefined => { - if (snapshot.kind === 'responding') return { state: AIAgentActivityStates.thinking }; - if (snapshot.kind === 'working') { - return { - state: AIAgentActivityStates.checkingExternalSources, - text: `Working on ${snapshot.skills.map((skill) => skill.replace(/_/g, ' ')).join(' and ')}`, - }; - } - return undefined; -}; - -/** - * Titles the steps of the backend's tools, which the router shows without one, from - * `DISPLAY_TITLES`. The same message gives back the same object, as `adapt` does. - */ -const withTitles = (adapt: (message: AIStreamMessage) => AIMessage) => { - const titled = new WeakMap(); - return (message: AIStreamMessage) => { - const adapted = adapt(message); - let shown = titled.get(adapted); - if (!shown) { - const untitled = (part: AIMessagePartData) => - part.toolCall && !part.toolCall.displayTitle && DISPLAY_TITLES[part.toolCall.name]; - shown = adapted.parts?.some(untitled) - ? { - ...adapted, - parts: adapted.parts.map((part) => - untitled(part) && part.toolCall - ? { - ...part, - toolCall: { - ...part.toolCall, - displayTitle: DISPLAY_TITLES[part.toolCall.name], - }, - } - : part, - ), - } - : adapted; - titled.set(adapted, shown); - } - return shown; - }; -}; - -/** - * The router's conversation as an `AIConversationController` follows it: the messages of the - * channel `AgentChat` reads, and what the agent is doing from the session's socket. Asking goes - * through the session, which records the question in the channel and answers it there. The - * router has no regenerate or edit of its own: those ask again, as a new question. - */ -export const createAgentTransport = ( - chat: AgentChat, - { onCreated, respond }: AgentTransportOptions, -): AIConversationTransport => ({ - connect: (feed) => { - // The same message object for each Stream message that didn't change, so only the messages - // an update touched render again. The live thinking is read through `chat.reasoning`. - const fromStreamMessage = withTitles( - createFromStreamMessage({ - reasoningText: (messageId, stepId) => chat.reasoning.get(messageId, stepId), - }), - ); - let followedSession: Session | undefined; - let stopFollowingSession: (() => void) | undefined; - const update = ({ hasOlderMessages, messages, session }: AgentChatState) => { - feed.updateMessages(messages.map(fromStreamMessage)); - feed.updateHasOlderMessages(hasOlderMessages); - if (session === followedSession) return; - stopFollowingSession?.(); - followedSession = session; - stopFollowingSession = session?.state.subscribe(() => - feed.updateActivity(activityOf(session.state.get())), - ); - if (session) feed.updateActivity(activityOf(session.state.get())); - }; - update(chat.state.getLatestValue()); - const unsubscribe = chat.state.subscribe(update); - return () => { - unsubscribe(); - stopFollowingSession?.(); - }; - }, - loadOlderMessages: () => chat.loadOlderMessages(), - send: async ({ attachments, custom, text }) => { - // From now on the conversation is used, even before Stream's copy of the question arrives. - chat.wasAsked = true; - chat.lastRequest = { custom: custom as Record | undefined, text }; - const photos = attachments - .filter((asset) => !asset.type || asset.type.startsWith('image')) - .map((asset) => asset.uri); - try { - // A new conversation is created here, with its first question. - const wasNew = chat.id === undefined; - const [session, images] = await Promise.all([ - chat.session(), - Promise.all(photos.map(async (uri) => ({ url: await toDataUri(uri) }))), - ]); - if (wasNew) onCreated(); - await respond(session, text, { - images, - uiAction: (custom as AgentRequestCustom | undefined)?.uiAction, - }); - } catch (error) { - // A conversation created for this question, which it never reached, isn't left listed. - await chat.discardUnusedSession(); - throw error; - } - }, - // A reply that was being written when the conversation was opened stops through its session, - // which opens for it. - stop: async () => { - const session = await chat.session(); - await session.interrupt(); - }, -}); diff --git a/examples/ExpoCrewAssistant/agent/agents.ts b/examples/ExpoCrewAssistant/agent/agents.ts index e3e73c672f..cce11a0f87 100644 --- a/examples/ExpoCrewAssistant/agent/agents.ts +++ b/examples/ExpoCrewAssistant/agent/agents.ts @@ -1,6 +1,6 @@ -import { Client } from '@stream-io/calypso'; +import { Client } from '@stream-io/stream-agents'; // Lets `session.chat()` open the conversation's channel on the app's Stream Chat client. -import '@stream-io/calypso/chat'; +import '@stream-io/stream-agents/chat'; import { AgentConfig } from './config'; diff --git a/examples/ExpoCrewAssistant/agent/conversation.ts b/examples/ExpoCrewAssistant/agent/conversation.ts new file mode 100644 index 0000000000..97febf67f1 --- /dev/null +++ b/examples/ExpoCrewAssistant/agent/conversation.ts @@ -0,0 +1,94 @@ +import type { AgentChat, AgentChatState } from '@stream-io/stream-agents/chat'; +import type { LocalMessage, StateStore } from 'stream-chat'; +import type { AIMessage, AIMessagePartData, AIStreamMessage } from 'stream-chat-expo/ai-components'; + +import { DISPLAY_TITLES } from './crewTools'; + +/** A written conversation with the agent, over the messages of the channel it is kept in. */ +export type Chat = AgentChat; + +export type ChatState = AgentChatState; + +/** + * The chat's state for `useStateStore`. The SDK's store has the shape of stream-chat's + * `StateStore` (`getLatestValue` and `subscribeWithSelector`, which is all `useStateStore` calls) + * without depending on stream-chat, so the type is only told so here. + */ +export const chatStore = (chat: Chat) => chat.state as unknown as StateStore; + +let keys = 0; +const chatKeys = new WeakMap(); + +/** Tells a conversation apart from the one shown before it, for a React key. */ +export const keyOf = (chat: Chat) => { + let key = chatKeys.get(chat); + if (!key) { + key = `chat-${++keys}`; + chatKeys.set(chat, key); + } + return key; +}; + +/** A reply the agent wrote. */ +export const isAgentReply = (message: LocalMessage) => message.ai_generated === true; + +/** + * Calls `used` when a new conversation is used, such as asked something: the router lists it from + * then on. A first question that never reached the router has its session deleted, and the next + * one creates another, which is listed in its turn. A conversation carried on is listed already, + * and isn't waited for. Returns a function that stops waiting. + */ +export const whenUsed = (chat: Chat, used: () => void) => { + if (chat.id !== undefined) return () => {}; + let listed: string | undefined; + return chat.state.subscribeWithSelector( + ({ usedSessionId }) => ({ usedSessionId }), + ({ usedSessionId }) => { + if (usedSessionId === undefined || usedSessionId === listed) return; + listed = usedSessionId; + used(); + }, + ); +}; + +/** + * A reply of the agent's that has something to show: the router writes it empty first and fills + * it in, and the conversation hides it until then. + */ +export const showsAgentReply = (message: LocalMessage) => + isAgentReply(message) && + (message.generating !== true || !!message.text || (message.attachments?.length ?? 0) > 0); + +/** + * Titles the steps of the backend's tools, which the router shows without one, from + * `DISPLAY_TITLES`. The same message gives back the same object, as `adapt` does. + */ +export const withTitles = (adapt: (message: AIStreamMessage) => AIMessage) => { + const titled = new WeakMap(); + return (message: AIStreamMessage) => { + const adapted = adapt(message); + let shown = titled.get(adapted); + if (!shown) { + const untitled = (part: AIMessagePartData) => + part.toolCall && !part.toolCall.displayTitle && DISPLAY_TITLES[part.toolCall.name]; + shown = adapted.parts?.some(untitled) + ? { + ...adapted, + parts: adapted.parts.map((part) => + untitled(part) && part.toolCall + ? { + ...part, + toolCall: { + ...part.toolCall, + displayTitle: DISPLAY_TITLES[part.toolCall.name], + }, + } + : part, + ), + } + : adapted; + titled.set(adapted, shown); + } + return shown; + }; +}; diff --git a/examples/ExpoCrewAssistant/agent/crewTools.ts b/examples/ExpoCrewAssistant/agent/crewTools.ts index 7093e61aef..d40e67da40 100644 --- a/examples/ExpoCrewAssistant/agent/crewTools.ts +++ b/examples/ExpoCrewAssistant/agent/crewTools.ts @@ -1,4 +1,4 @@ -import type { ToolApproval } from '@stream-io/calypso'; +import type { ToolApproval } from '@stream-io/stream-agents'; import { type AIClientTool, completedToolResult } from 'stream-chat-expo/ai-components'; diff --git a/examples/ExpoCrewAssistant/agent/deviceTools.ts b/examples/ExpoCrewAssistant/agent/deviceTools.ts index 7c4d551db0..25cb559e96 100644 --- a/examples/ExpoCrewAssistant/agent/deviceTools.ts +++ b/examples/ExpoCrewAssistant/agent/deviceTools.ts @@ -1,4 +1,4 @@ -import { Tools } from '@stream-io/calypso'; +import { Tools } from '@stream-io/stream-agents'; import { type AIClientToolResult, @@ -28,16 +28,14 @@ class Reports { else this.reported.set(callId, result); } - wait(callId: string, signal: AbortSignal) { - return new Promise((resolve, reject) => { + // A call that was cancelled stops being waited on by the session's tools, which leave its + // result unread. + wait(callId: string) { + return new Promise((resolve) => { const reported = this.reported.get(callId); this.reported.delete(callId); if (reported) return resolve(reported); this.waiting.set(callId, resolve); - signal.addEventListener('abort', () => { - this.waiting.delete(callId); - reject(signal.reason); - }); }); } } @@ -86,9 +84,9 @@ export const createDeviceTools = ({ clientId, userId }: { clientId: string; user executor: 'client', name, parameters, - run: async (input, signal, call) => { + run: async (input, call) => { const { failure, output } = deviceTools.addressed - ? await reports.wait(call.id, signal) + ? await reports.wait(call.id) : await runUnaddressed(name, { arguments: input, executor: 'client', diff --git a/examples/ExpoCrewAssistant/agent/photos.ts b/examples/ExpoCrewAssistant/agent/photos.ts index 3dc3d8d0bf..7dac92f52e 100644 --- a/examples/ExpoCrewAssistant/agent/photos.ts +++ b/examples/ExpoCrewAssistant/agent/photos.ts @@ -1,4 +1,5 @@ import { manipulateAsync, SaveFormat } from 'expo-image-manipulator'; +import type { AIConversationRequest } from 'stream-chat-expo/ai-components'; // The router takes photos inline, so they are shrunk first. const MAX_IMAGE_WIDTH = 1280; @@ -12,3 +13,14 @@ export const toDataUri = async (uri: string) => { }); return `data:image/jpeg;base64,${base64}`; }; + +/** + * The photos among the files the person picked, as the images the agent's transport sends. + * Anything else they picked isn't sent: the router takes images only. + */ +export const imagesOf = (attachments: readonly unknown[]) => + Promise.all( + (attachments as AIConversationRequest['attachments']) + .filter((asset) => !asset.type || asset.type.startsWith('image')) + .map(async (asset) => ({ url: await toDataUri(asset.uri) })), + ); diff --git a/examples/ExpoCrewAssistant/contexts/AppContext.tsx b/examples/ExpoCrewAssistant/contexts/AppContext.tsx index 465daa881d..eee1158ca6 100644 --- a/examples/ExpoCrewAssistant/contexts/AppContext.tsx +++ b/examples/ExpoCrewAssistant/contexts/AppContext.tsx @@ -1,22 +1,37 @@ import type { PropsWithChildren } from 'react'; import React, { useEffect, useMemo, useRef, useState } from 'react'; +import { AppState } from 'react-native'; -import { type ConversationSnapshot, type Session, type UIAction } from '@stream-io/calypso'; -import type { StreamChat } from 'stream-chat'; +import { type ConversationSnapshot, type Session } from '@stream-io/stream-agents'; +import { + AgentChat, + agentChatTransport, + type OpenAgentSession, +} from '@stream-io/stream-agents/chat'; +import type { Channel, LocalMessage, StreamChat } from 'stream-chat'; import { useStableCallback, useStateStore } from 'stream-chat-expo'; import { AIClientIdentity, AIConversationController, type AIToolApprover, createAuthSessionAuthorizer, + createFromStreamMessage, parseMessageParts, } from 'stream-chat-expo/ai-components'; -import { AgentChat, type AgentChatState, isAgentReply, type OpenSession } from '../agent/agentChat'; import { agents, agentSessions, crewBackend, streamToken } from '../agent/agents'; -import { createAgentTransport } from '../agent/agentTransport'; import { AgentConfig } from '../agent/config'; +import { + type Chat, + type ChatState, + chatStore, + isAgentReply, + showsAgentReply, + whenUsed, + withTitles, +} from '../agent/conversation'; import { createDeviceTools } from '../agent/deviceTools'; +import { imagesOf } from '../agent/photos'; const IDLE: ConversationSnapshot = { kind: 'idle', skills: [] }; @@ -31,7 +46,7 @@ const deviceTools = createDeviceTools({ clientId, userId: AgentConfig.user.id }) const SIGN_IN_SCHEME = 'io.getstream.expocrewassistant'; const SIGN_IN_RETURN_URL = `${SIGN_IN_SCHEME}://sign-in`; -// The person's answer to a call that asks first. calypso's `decide` answers only a call this +// The person's answer to a call that asks first. stream-agents' `decide` answers only a call this // session was asked on its own socket, and leaves any other alone, so a question shown from the // channel, as after reopening the conversation or reloading the app, would wait forever. Such a // call is answered with the frames `decide` sends, which the router matches by the call's id: an @@ -54,26 +69,9 @@ const answerApproval = (session: Session, toolCallId: string, allowed: boolean) } }; -// Asks, with calypso's `responses.create`, which takes photos, a reasoning effort and what the -// person did in a reply's UI. The router's `respond`, which addresses a client tool's call to the -// install a question came from, is only for an app's backend, and this is a phone: the calls reach -// the device's tools without an install, and the phone runs them itself. -const respond = async ( - session: Session, - text: string, - options: { images: { url: string }[]; reasoningEffort?: string; uiAction?: UIAction }, -) => { - deviceTools.addressed = false; - await session.responses.create(text, { - ...(options.images.length > 0 ? { images: options.images } : {}), - ...(options.reasoningEffort ? { reasoningEffort: options.reasoningEffort } : {}), - ...(options.uiAction ? { uiAction: options.uiAction } : {}), - }); -}; - // Answers go over the session's socket, which opens for them in a conversation that was only // read so far. Each throws when it can't send, so the card lets the person answer again. -const approverFor = (chat: AgentChat): AIToolApprover => ({ +const approverFor = (chat: Chat): AIToolApprover => ({ authorizer: createAuthSessionAuthorizer({ callbackScheme: SIGN_IN_SCHEME, decline: async (call) => (await chat.session()).declineLogin(call.id), @@ -87,7 +85,9 @@ export type AppContextValue = { /** What the agent is doing in the open conversation, from the session's socket. */ agentState: ConversationSnapshot; /** The open conversation's messages and live reasoning. */ - chat: AgentChat | undefined; + chat: Chat | undefined; + /** Whether the agent was asked to open the open conversation and hasn't written yet. */ + awaitingAgent: boolean; /** The open conversation's state, and what the person does in it. */ conversation: AIConversationController | undefined; /** Bumped when the drawer's list of conversations may have changed. */ @@ -123,6 +123,7 @@ export type AppContextValue = { export const AppContext = React.createContext({ agentState: IDLE, + awaitingAgent: false, chat: undefined, conversation: undefined, conversationsVersion: 0, @@ -140,6 +141,12 @@ export const AppContext = React.createContext({ /** What the restaurant's systems tell the agent about, so it speaks first. */ export type CrewEvent = 'clock_in' | 'clock_out' | 'back_from_time_off' | 'onboarding'; +/** What was last asked of the agent here, so it can be asked again when it couldn't answer. */ +type AgentRequest = + | { text: string; custom?: Record } + | { event: CrewEvent } + | { voiceMemoUrl: string }; + // Tells the backend what happened, which has the agent open the conversation about it. const tellBackend = (kind: CrewEvent, sessionId: string) => crewBackend('events', { @@ -150,7 +157,7 @@ const tellBackend = (kind: CrewEvent, sessionId: string) => }, }); -// Asks with a voice memo at `url`. calypso's `responses.create` has no voice note yet, so this is +// Asks with a voice memo at `url`. stream-agents' `responses.create` has no voice note yet, so this is // the request it makes, with the voice note the router takes beside the text. const askWithVoiceMemo = async (session: Session, url: string) => { deviceTools.addressed = false; @@ -160,20 +167,35 @@ const askWithVoiceMemo = async (session: Session, url: string) => { }); }; +// Tells the backend of an event, or asks with a voice memo, in `chat`, and only then counts the +// conversation as used, as a question does: one where neither got through isn't listed, and is +// deleted when the person leaves it. +const tellAbout = async (chat: Chat, kind: CrewEvent) => { + const session = await chat.session(); + await tellBackend(kind, session.id); + await chat.use(); +}; + +const askVoiceMemo = async (chat: Chat, url: string) => { + await askWithVoiceMemo(await chat.session(), url); + await chat.use(); +}; + // What the recorder wrote, by its file's extension. const audioTypeOf = (uri: string) => (/\.(m4a|mp4)$/i.test(uri) ? 'audio/mp4' : 'audio/aac'); const messageOf = (error: unknown) => (error instanceof Error ? error.message : String(error)); -// What the provider follows of the messages: the latest reply, whose calls this phone runs, and -// the session, once one is open. -const selector = ({ messages, session }: AgentChatState) => ({ +// What the provider follows of the messages: the latest reply, whose calls this phone runs, +// whether the agent has written anything yet, and the session, once one is open. +const selector = ({ messages, session }: ChatState) => ({ + hasShownReply: messages.some(showsAgentReply), lastReply: [...messages].reverse().find(isAgentReply), session, }); // Carries a conversation on, or creates one, with the phone's tools declared on it. -const openSession: OpenSession = (sessionId) => +const openSession: OpenAgentSession = (sessionId) => sessionId ? agentSessions().resume(sessionId, { tools: deviceTools.sessionTools() }) : agentSessions().create({ @@ -181,10 +203,54 @@ const openSession: OpenSession = (sessionId) => tools: deviceTools.sessionTools(), }); -let chats = 0; -const newChat = (sessionId?: string) => new AgentChat(`chat-${++chats}`, sessionId, openSession); +const newChat = (sessionId?: string): Chat => + new AgentChat({ open: openSession, sessionId }); + +// The conversation's state over the chat. Asking goes through the session, which records the +// question in the channel and answers it there, with photos as data URIs. `asked` hears each +// request as it goes out: the question, or `undefined` for a retry of the last one. +const conversationOf = (chat: Chat, asked: (request: AgentRequest | undefined) => void) => { + // The same message object for each Stream message that didn't change, so only the messages an + // update touched render again. The live thinking is read through `chat.reasoning`, and every + // tool's step is titled, the backend's too. + const fromStreamMessage = withTitles( + createFromStreamMessage({ + reasoningText: (messageId, stepId) => chat.reasoning.get(messageId, stepId), + }), + ); + const transport = agentChatTransport(chat, { + adapt: (message) => fromStreamMessage(message), + images: imagesOf, + }); + return new AIConversationController({ + // The agents router keeps the conversation in Stream Chat and can't rewind it, so a reply + // can't be written again or a message edited: either would only be asked again as new. + actions: { edit: false, regenerate: false }, + approver: approverFor(chat), + transport: { + ...transport, + send: (request) => { + asked( + // A retry asks the same question again without what it went out with, such as a + // generated-UI action, which the request it retries still has. + request.kind === 'regenerate' + ? undefined + : { custom: request.custom, text: request.text }, + ); + return transport.send(request); + }, + }, + }); +}; -type Opened = { chat: AgentChat; conversation: AIConversationController }; +type Opened = { + chat: Chat; + conversation: AIConversationController; + /** Stops waiting for a new conversation to be used. */ + stopWaiting: () => void; + /** What was last asked of the agent here. */ + lastRequest?: AgentRequest; +}; export const AppProvider = ({ client, children }: PropsWithChildren<{ client: StreamChat }>) => { const [opened, setOpened] = useState(); @@ -192,11 +258,14 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St const [isReady, setIsReady] = useState(false); const [agentState, setAgentState] = useState(IDLE); const [conversationsVersion, setConversationsVersion] = useState(0); + // The conversation the agent was asked to open, about an event. + const [awaitedChat, setAwaitedChat] = useState(); const current = useRef(undefined); const chat = opened?.chat; - const followed = useStateStore(chat?.state, selector); + const followed = useStateStore(chat ? chatStore(chat) : undefined, selector); const session = followed?.session; + const awaitingAgent = chat !== undefined && chat === awaitedChat && !followed?.hasShownReply; // Runs the calls of the latest reply that wait for this phone, each once. const lastReply = followed?.lastReply; @@ -208,96 +277,80 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St }, [lastReply]); // Shows a conversation in place of the one before it, at once, and lets the one before go. - const show = (next: AgentChat) => { + const show = (next: Chat) => { const previous = current.current; - const conversation = new AIConversationController({ - // The agents router keeps the conversation in Stream Chat and can't rewind it, so a reply - // can't be written again or a message edited: either would only be asked again as new. - actions: { edit: false, regenerate: false }, - approver: approverFor(next), - transport: createAgentTransport(next, { - onCreated: () => setConversationsVersion((version) => version + 1), - respond: (...asked) => { - setFailure(undefined); - return respond(...asked); - }, + const shown: Opened = { + chat: next, + conversation: conversationOf(next, (request) => { + if (request) shown.lastRequest = request; + setFailure(undefined); }), - }); - current.current = { chat: next, conversation }; + // The router lists a new conversation once something used it. + stopWaiting: whenUsed(next, () => setConversationsVersion((version) => version + 1)), + }; + current.current = shown; setFailure(undefined); - setOpened(current.current); + setOpened(shown); + previous?.stopWaiting(); previous?.conversation.dispose(); void previous?.chat.close().catch(() => {}); + return shown; }; // A new conversation is only shown: the router creates it with its first question, so one // that is never asked anything isn't left in the drawer. - const startConversation = useStableCallback(() => show(newChat())); + const startConversation = useStableCallback(() => { + show(newChat()); + }); const startConversationWith = useStableCallback((text: string) => { - const next = newChat(); - show(next); - void current.current?.conversation.send({ text }); + void show(newChat()).conversation.send({ text }); }); const startConversationAbout = useStableCallback((kind: CrewEvent) => { const next = newChat(); - show(next); - next.state.partialNext({ awaitingAgent: true }); - next.lastRequest = { event: kind }; - void (async () => { - try { - const opened = await next.session(); - setConversationsVersion((version) => version + 1); - await tellBackend(kind, opened.id); - } catch (error) { - next.state.partialNext({ awaitingAgent: false }); - if (current.current?.chat === next) setFailure(messageOf(error)); - } - })(); + show(next).lastRequest = { event: kind }; + setAwaitedChat(next); + // Nothing is asked here, but once the backend has the event the conversation is used: the + // agent writes first, and it is kept and listed. + tellAbout(next, kind).catch((error: unknown) => { + setAwaitedChat((awaited) => (awaited === next ? undefined : awaited)); + if (current.current?.chat === next) setFailure(messageOf(error)); + }); }); const sendVoiceMemo = useStableCallback(async (uri: string) => { - const opened = current.current; - if (!opened) return; - const { chat } = opened; + const shown = current.current; + if (!shown) return; setFailure(undefined); try { - const wasNew = chat.id === undefined; - const session = await chat.session(); - if (wasNew) setConversationsVersion((version) => version + 1); - const channel = chat.conversationChannel; - if (!channel) throw new Error('The conversation has no channel yet'); + const session = await shown.chat.session(); + const { channel } = await session.chat(); const name = uri.split('/').pop() ?? 'voice-memo.aac'; - const { file } = await channel.sendFile(uri, name, audioTypeOf(uri)); - chat.wasAsked = true; - chat.lastRequest = { voiceMemoUrl: file }; - await askWithVoiceMemo(session, file); + const { file } = await (channel as Channel).sendFile(uri, name, audioTypeOf(uri)); + shown.lastRequest = { voiceMemoUrl: file }; + await askVoiceMemo(shown.chat, file); } catch (error) { setFailure(messageOf(error)); } }); const retry = useStableCallback(() => { - const opened = current.current; - const last = opened?.chat.lastRequest; - if (!opened || !last) return; + const shown = current.current; + const last = shown?.lastRequest; + if (!shown || !last) return; setFailure(undefined); if ('voiceMemoUrl' in last) { - void opened.chat - .session() - .then((session) => askWithVoiceMemo(session, last.voiceMemoUrl)) - .catch((error: unknown) => setFailure(messageOf(error))); + askVoiceMemo(shown.chat, last.voiceMemoUrl).catch((error: unknown) => + setFailure(messageOf(error)), + ); return; } if ('event' in last) { - void opened.chat - .session() - .then((session) => tellBackend(last.event as CrewEvent, session.id)) - .catch((error: unknown) => setFailure(messageOf(error))); + tellAbout(shown.chat, last.event).catch((error: unknown) => setFailure(messageOf(error))); return; } - void opened.conversation.send({ custom: last.custom, text: last.text }); + void shown.conversation.send({ custom: last.custom, text: last.text }); }); const openConversation = useStableCallback((sessionId: string, conversationId?: string) => { @@ -317,6 +370,18 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St }); }); + // The session's socket drops once the app has been in the background a while. Coming back, the + // open conversation is carried on in a new one, so what the agent does shows again. + useEffect(() => { + const subscription = AppState.addEventListener('change', (state) => { + const shown = current.current?.chat; + if (state !== 'active' || !shown) return; + const { session: held } = shown.state.getLatestValue(); + if (held && !held.live) void shown.prepare().catch(() => {}); + }); + return () => subscription.remove(); + }, []); + // The open conversation: what the agent is doing, from the session's socket, once one is open. useEffect(() => { if (!session) { @@ -357,6 +422,7 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St const contextValue = useMemo( () => ({ agentState, + awaitingAgent, chat, conversation, conversationsVersion, @@ -372,6 +438,7 @@ export const AppProvider = ({ client, children }: PropsWithChildren<{ client: St }), [ agentState, + awaitingAgent, chat, conversation, conversationsVersion, diff --git a/examples/ExpoCrewAssistant/package.json b/examples/ExpoCrewAssistant/package.json index 71896ec1c9..579d6e0e47 100644 --- a/examples/ExpoCrewAssistant/package.json +++ b/examples/ExpoCrewAssistant/package.json @@ -12,7 +12,7 @@ "dependencies": { "@react-native-community/netinfo": "12.0.1", "@shopify/react-native-skia": "^2.14.0", - "@stream-io/calypso": "^0.3.0", + "@stream-io/stream-agents": "^0.4.1", "expo": "^58.0.8", "expo-audio": "patch:expo-audio@npm%3A58.0.9#~/.yarn/patches/expo-audio-npm-58.0.9-5c313f7a3f.patch", "expo-clipboard": "~58.0.3", diff --git a/examples/ExpoCrewAssistant/screens/AgentNotices.tsx b/examples/ExpoCrewAssistant/screens/AgentNotices.tsx index 9eeb5997a8..f95b2787e2 100644 --- a/examples/ExpoCrewAssistant/screens/AgentNotices.tsx +++ b/examples/ExpoCrewAssistant/screens/AgentNotices.tsx @@ -1,16 +1,16 @@ import { useSyncExternalStore } from 'react'; import { Pressable, StyleSheet, Text, View } from 'react-native'; -import type { Session, ToolApprovalRequest } from '@stream-io/calypso'; +import type { Session, ToolApprovalRequest } from '@stream-io/stream-agents'; import { useStateStore } from 'stream-chat-expo'; import { AIToolApprovalCard } from 'stream-chat-expo/ai-components'; -import type { AgentChat, AgentChatState } from '../agent/agentChat'; +import { type Chat, type ChatState, chatStore } from '../agent/conversation'; import { useAppContext } from '../contexts/AppContext'; const NONE: readonly ToolApprovalRequest[] = []; -const sessionSelector = ({ session }: AgentChatState) => ({ session }); +const sessionSelector = ({ session }: ChatState) => ({ session }); // What the call would do, in a line, from the arguments the model wrote. const detailsOf = ({ arguments: written }: ToolApprovalRequest) => { @@ -70,8 +70,8 @@ const Requests = ({ session }: { session: Session }) => { * phone asks through `responses.create`, so those wait on its session rather than on the reply's * step, and the person answers them here. */ -export const AgentNotices = ({ chat }: { chat: AgentChat }) => { - const { session } = useStateStore(chat.state, sessionSelector); +export const AgentNotices = ({ chat }: { chat: Chat }) => { + const { session } = useStateStore(chatStore(chat), sessionSelector); return session ? ( <> diff --git a/examples/ExpoCrewAssistant/screens/ChatContent.tsx b/examples/ExpoCrewAssistant/screens/ChatContent.tsx index 51de24037a..c2c245cb55 100644 --- a/examples/ExpoCrewAssistant/screens/ChatContent.tsx +++ b/examples/ExpoCrewAssistant/screens/ChatContent.tsx @@ -57,9 +57,8 @@ import { import { AgentNotices } from './AgentNotices'; import { VoiceMemoBar, VoiceMemoButton } from './VoiceMemo'; -import type { AgentChat, AgentChatState } from '../agent/agentChat'; import { crewBackend } from '../agent/agents'; -import type { AgentRequestCustom } from '../agent/agentTransport'; +import { type Chat, type ChatState, chatStore, keyOf } from '../agent/conversation'; import { VoiceMode } from '../components/VoiceMode'; import { useAppContext } from '../contexts/AppContext'; import { type Flow, FLOWS } from '../flows'; @@ -75,7 +74,7 @@ const ComposerArea = ({ conversation, failure, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; failure: string | undefined; }) => { @@ -155,19 +154,17 @@ const Flows = ({ conversation }: { conversation: AIConversationController }) => ); }; -const emptySelector = ({ awaitingAgent, isLoaded }: AgentChatState) => ({ - awaitingAgent, - isLoaded, -}); +const emptySelector = ({ isLoaded }: ChatState) => ({ isLoaded }); const EmptyContent = ({ chat, conversation, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; }) => { - const { awaitingAgent, isLoaded } = useStateStore(chat.state, emptySelector); + const { awaitingAgent } = useAppContext(); + const { isLoaded } = useStateStore(chatStore(chat), emptySelector); if (awaitingAgent) { return ( @@ -189,19 +186,19 @@ const sourcesOf = (message: LocalMessage | undefined): Source[] => { }; /** The chat the open conversation is kept in, for the views of its messages. */ -const ConversationChatContext = createContext(undefined); +const ConversationChatContext = createContext(undefined); /** The message of the conversation with `messageId`, following only that message. */ const useChatMessage = (messageId: string | undefined) => { const chat = useContext(ConversationChatContext); const selector = useMemo( () => - ({ messages }: AgentChatState) => ({ + ({ messages }: ChatState) => ({ message: messages.find(({ id }) => id === messageId), }), [messageId], ); - return useStateStore(chat?.state, selector)?.message; + return useStateStore(chat ? chatStore(chat) : undefined, selector)?.message; }; const SOURCES_OPEN = { duration: 300, easing: Easing.out(Easing.ease) }; @@ -444,7 +441,7 @@ const Conversation = ({ chat, conversation, }: { - chat: AgentChat; + chat: Chat; conversation: AIConversationController; }) => { const { agentState } = useAppContext(); @@ -472,7 +469,7 @@ const Conversation = ({ throw error; } } - const custom: AgentRequestCustom = { uiAction: aiUIActionJSON(sent) }; + const custom = { uiAction: aiUIActionJSON(sent) }; conversation.send({ custom, text: sent.label }); }, }), @@ -515,7 +512,7 @@ export const ChatContent = () => { return ( =1.0.0" - stream-chat: ">=9.0.0" - peerDependenciesMeta: - "@stream-io/video-client": - optional: true - stream-chat: - optional: true - checksum: 10c0/8a331b8e5c3fc77c672b268fc0db937a5f0b23297d9238096602cf28f09efb41e9cdf35e11535532c92f1e1431f53cc5af427ab546f552f169a001af5059ec20 - languageName: node - linkType: hard - "@stream-io/state-store@npm:^1.1.6": version: 1.1.6 resolution: "@stream-io/state-store@npm:1.1.6" @@ -6478,6 +6463,21 @@ __metadata: languageName: node linkType: hard +"@stream-io/stream-agents@npm:^0.4.1": + version: 0.4.1 + resolution: "@stream-io/stream-agents@npm:0.4.1" + peerDependencies: + "@stream-io/video-client": ^1.0.0 + stream-chat: ^9.0.0 || ^10.0.2 + peerDependenciesMeta: + "@stream-io/video-client": + optional: true + stream-chat: + optional: true + checksum: 10c0/f04554aa230702d8e137574e7951a07fa88128d5d67a48ebfa0d2854af1c10ac2192fd4688866b04c5eab1a83e238ebe496331c97365ddb6e53dd2e97dcd8257 + languageName: node + linkType: hard + "@stream-io/typescript-config@workspace:^, @stream-io/typescript-config@workspace:configs/typescript-config": version: 0.0.0-use.local resolution: "@stream-io/typescript-config@workspace:configs/typescript-config" @@ -11498,7 +11498,7 @@ __metadata: "@react-native-community/netinfo": "npm:12.0.1" "@rnx-kit/metro-config": "npm:^2.1.2" "@shopify/react-native-skia": "npm:^2.14.0" - "@stream-io/calypso": "npm:^0.3.0" + "@stream-io/stream-agents": "npm:^0.4.1" "@stream-io/typescript-config": "workspace:^" "@types/react": "npm:~19.3.0" expo: "npm:^58.0.8" @@ -11535,7 +11535,7 @@ __metadata: "@react-native-community/netinfo": "npm:12.0.1" "@rnx-kit/metro-config": "npm:^2.1.2" "@shopify/react-native-skia": "npm:^2.14.0" - "@stream-io/calypso": "npm:^0.3.0" + "@stream-io/stream-agents": "npm:^0.4.1" "@stream-io/typescript-config": "workspace:^" "@types/react": "npm:~19.3.0" expo: "npm:^58.0.8"