Conversation State

AIConversationView shows a conversation, and ChatComposer takes what the user writes. Between them and your agent sits the state of the conversation: its messages, whether the agent is thinking or writing, the message on its way, and why the last request failed. AIConversation holds that state and acts on it. You tell it how to reach your agent with a transport, and give it to both views:

class AssistantScreen extends StatefulWidget {
  const AssistantScreen({super.key, required this.channel});

  // A channel you've watched, with `channel.watch()`.
  final Channel channel;

  @override
  State<AssistantScreen> createState() => _AssistantScreenState();
}

class _AssistantScreenState extends State<AssistantScreen> {
  late final conversation = AIConversation<StreamAIMessage>(
    transport: ChannelTransport(widget.channel),
  );

  @override
  void dispose() {
    conversation.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Column(
      children: [
        Expanded(child: AIConversationView.conversation(conversation: conversation)),
        ChatComposer.conversation(conversation: conversation),
      ],
    );
  }
}

StreamAIMessage is the message type from Messages from Stream Chat, and ChannelTransport is written below. With these two views, the conversation:

• Shows the user's message as soon as they send it, and the agent thinking until its reply starts.
• Turns the composer's send button into a stop button while the agent answers, and stops the agent when it's tapped.
• Loads older messages as the user scrolls up.
• Offers to copy, share and rate each finished reply, and to regenerate the latest one.
• Offers to edit the user's latest message, which is then answered again.
• Keeps the error of a request that failed, so the user can try it again.

If you know useChat from Vercel's AI SDK on the web, AIConversation plays the same part in a Flutter app.

Make a conversation

AIConversation is a ChangeNotifier. Make one per conversation and keep it for as long as its screen shows, for example in a late final field of your State, and dispose it with the screen, as above. It connects to its transport straight away.

It notifies its listeners each time what the views show changes, so a ListenableBuilder of your own that reads it updates with it:

status is submitted from the moment the user sends a message until the reply shows something, which is when the conversation view shows the agent thinking. It's streaming while the agent writes the reply, and ready again once the reply is finished or stopped.

The views call its methods, and you can call them too, for example from a button of your own: send(text, attachments:, chatOption:), stop(), regenerate(reply), edit(message), cancelEditing(), retry(), clearError(), rate(reply, rating) and loadOlderMessages().

Show what the agent is doing and errors

The view's footer shows under the latest turn's messages, or under the empty content while there are none. Read the conversation through a ListenableBuilder there, so the footer follows it as it changes:

AIConversationView.conversation(
  conversation: conversation,
  footer: ListenableBuilder(
    listenable: conversation,
    builder: (context, _) => Column(
      mainAxisSize: MainAxisSize.min,
      crossAxisAlignment: CrossAxisAlignment.start,
      children: [
        if (conversation.activity?.text case final text? when conversation.status.isAnswering)
          AITypingIndicatorView(text: text),
        if (conversation.error case final error?)
          Row(
            children: [
              Flexible(
                child: Text(switch (error) {
                  final AIAgentError error => error.localizedMessage(AITranslations.of(context)),
                  _ => "Your message couldn't be sent.",
                }),
              ),
              TextButton(onPressed: conversation.retry, child: const Text('Try again')),
            ],
          ),
      ],
    ),
  ),
  emptyContent: const Text('How can I help?'),
)

The error is one of two kinds:

• The request failed, such as when the message couldn't be sent. It's the error your transport's send threw. retry() sends the same request again, with the same message id.
• The agent couldn't answer. It's an AIAgentError, when the agent reports the error state (AI_STATE_ERROR) or its reply doesn't start within replyTimeout. Its message is what the agent said, if anything, and its localizedMessage falls back to "The agent couldn't answer. Try again." (AITranslations.agentError). retry() asks the agent for the reply again with a regenerate request.

clearError() forgets the error and the request that failed.

When the user stops the agent, what it wrote so far stays, and the reply counts as finished even while the channel still marks it as generating. A reply that showed nothing yet is hidden, and the conversation ignores what the agent says it's doing until the agent clears it.

Write a transport

The transport is the part that knows your agent. It follows what the conversation holds and tells the conversation through a feed, and it sends what the user asks:

Mix AIConversationTransport into your class. Its AIConversationFeed has three updates, and holds the conversation weakly, so keep the feed for as long as the transport follows the conversation:

• updateMessages(messages) - the conversation's messages, oldest first, each time they change.
• updateHasOlderMessages(hasOlderMessages) - whether there are messages before the first one.
• updateActivity(activity) - what the agent says it's doing, as an AIAgentActivity, or null once it stops saying. Its state uses the raw values of Stream Chat's AI indicator, such as AIAgentActivityState.thinking or checkingExternalSources, and its text is what the agent says about it.

Cancel your subscriptions in disconnect, which the conversation calls as it's disposed. isConnected also tells whether the conversation is still there, for a transport that outlives it.

Requests

send receives an AIConversationRequest:

A retry sends the same request again, with the same messageId. If the first attempt reached your backend, or you keep a copy of the failed message, use the id so the message isn't added twice.

While a regenerate or an edit is on its way, the conversation leaves out the messages it replaces. Once the new reply starts, it shows the messages as the feed has them again, so update the conversation on your backend before the new reply arrives:

• For a regenerate, remove the request's message and the messages after it, or write the new reply in place of message, under its id. Without a message, it asks for a reply to the latest message, which has none, such as after the agent couldn't answer it.
• For an edit, change the user's message to say the request's text, under its id, and remove the messages after it.

A transport over a Stream Chat channel

With Stream Chat, the conversation is a channel. This transport follows the channel's messages and its AI indicator events, sends what the user asks as a message in the channel, and stops the agent with channel.stopAIResponse():

import 'dart:async';

import 'package:stream_chat_flutter/stream_chat_flutter.dart';
import 'package:stream_chat_flutter_ai/stream_chat_flutter_ai.dart';

class ChannelTransport with AIConversationTransport<StreamAIMessage> {
  ChannelTransport(this.channel);

  // A channel you've watched, whose state the transport reads.
  final Channel channel;

  AIConversationFeed<StreamAIMessage>? _feed;
  final _subscriptions = <StreamSubscription<Object?>>[];
  var _hasOlderMessages = true;

  @override
  void connect(AIConversationFeed<StreamAIMessage> feed) {
    _feed = feed;
    _subscriptions
      ..add(channel.state!.messagesStream.listen(_channelChanged))
      // `channel.on` delivers only this channel's events.
      ..add(channel.on(EventType.aiIndicatorUpdate, EventType.aiIndicatorClear).listen(_received));
    _channelChanged(channel.state!.messages);
  }

  @override
  Future<void> send(AIConversationRequest<StreamAIMessage> request) async {
    final attachments = [
      for (final file in request.attachments)
        Attachment(
          // The composer's photos may have no media type.
          type: file.mimeType?.startsWith('image/') ?? true ? AttachmentType.image : AttachmentType.file,
          file: AttachmentFile(bytes: await file.readAsBytes(), name: file.name, size: await file.length()),
        ),
    ];
    // A regenerate asks the question again, and an edit sends the new text as a new message.
    await channel.sendMessage(Message(id: request.messageId, text: request.text, attachments: attachments));
  }

  @override
  Future<void> stop() => channel.stopAIResponse();

  @override
  void disconnect() => _cancel();

  @override
  Future<void> loadOlderMessages() async {
    const limit = 30;
    final oldest = channel.state!.messages.firstOrNull;
    if (oldest == null) return;
    final state = await channel.query(
      messagesPagination: PaginationParams(lessThan: oldest.id, limit: limit),
    );
    // The older messages join the channel's state, and so the conversation.
    _hasOlderMessages = (state.messages?.length ?? 0) >= limit;
    _channelChanged(channel.state!.messages);
  }

  void _channelChanged(List<Message> messages) {
    final feed = _feed;
    if (feed == null || !feed.isConnected) {
      _cancel();
      return;
    }
    // The channel's state lists its messages oldest first.
    feed.updateMessages([
      for (final message in messages)
        if (message.type == MessageType.regular) StreamAIMessage(message),
    ]);
    feed.updateHasOlderMessages(_hasOlderMessages && messages.isNotEmpty);
  }

  void _received(Event event) {
    final feed = _feed;
    if (feed == null || !feed.isConnected) {
      _cancel();
      return;
    }
    final state = switch (event.aiState) {
      AITypingState.thinking => AIAgentActivityState.thinking,
      AITypingState.checkingSources => AIAgentActivityState.checkingExternalSources,
      AITypingState.generating => AIAgentActivityState.generating,
      AITypingState.error => AIAgentActivityState.error,
      // `ai_indicator.clear`, or a state the SDK does not know.
      AITypingState.idle || null => null,
    };
    feed.updateActivity(state == null ? null : AIAgentActivity(state: state, text: event.aiMessage));
  }

  // The conversation is gone: stop following the channel.
  void _cancel() {
    for (final subscription in _subscriptions) {
      unawaited(subscription.cancel());
    }
    _subscriptions.clear();
  }
}

It follows the channel through its state's messagesStream and channel.on, so it needs a channel you've watched, and leaves the rest of the channel to your app. A few things to keep in mind:

• channel.sendMessage adds the message to the channel's state straight away, and completes once it's sent, so a message that couldn't be sent shows as the conversation's error. The message stays in the channel, marked as failed, and a retry sends it again under the same id.
• A channel has no way to write a reply again or to edit a message, so this transport asks the question again for a regenerate, and sends an edit as a new message. The messages they replace come back once the new reply starts. If your backend can't remove or replace them, leave those actions out: AIConversation(transport: transport, actions: AIConversationActions.rate). A retry() after the agent couldn't answer is a regenerate too, whatever the actions, so this transport sends the question to the channel again.
• Ratings stay on the device unless the transport implements rate. To show the ratings your backend keeps, return them from your message's rating, which the conversation shows until the user rates the reply on the device.
• The channel doesn't say whether it has older messages, so the transport assumes it does until a query returns fewer than it asked for.

For an agent with an API of its own, follow its messages the same way, and send the requests to it, where regenerate and edit can replace the messages they're about.

What your backend sends

With Stream Chat, your backend writes the agent's reply into the channel, and the transport shows what it writes. A typical backend posts an empty placeholder message as the agent, updates it as the model streams, and stores the final text. Stream Agents does all of this for you. With a backend of your own, follow the conventions below, which the components, the transport above and Stream's backend SDKs share. The AI message streaming guide covers the whole backend flow.

Mark the agent's replies

Set two custom fields on the agent's reply:

• ai_generated: true, when the backend creates the placeholder. Your app reads it to tell the agent's replies apart, for example for your message's role, and your backend checks it to skip the agent's own replies instead of answering them.
• generating, true on every update while the reply streams, and false on the final one. StreamAIMessage reads it as isGenerating. A reply types itself out only while it's true, so a stored reply, saved with false, doesn't type itself out again when the user scrolls back to it.

const { message } = await channel.sendMessage({
  text: "",
  user_id: botId,
  ai_generated: true,
});

placeholder = channel.send_message(
    MessageRequest(text="", user_id=bot_id, custom={"ai_generated": True}),
).data.message

In Flutter, the fields are message.extraData['ai_generated'] and message.extraData['generating'].

Say what the agent is doing

Before the first word of the reply arrives, the model can spend a while thinking or calling tools, and the message is still empty. The backend says what it's doing with ai_indicator.* custom events on the channel, which the Flutter SDK decodes, and the transport above turns into the conversation's activity:

The state of an update is an AITypingState:

Both spellings of the "checking sources" state are in use: the reference agents send AI_STATE_EXTERNAL_SOURCES, and other backends send AI_STATE_CHECKING_SOURCES. From stream_chat 10.6.0, the SDK reads either as checkingSources, and a state it doesn't know as idle. Earlier versions read AI_STATE_EXTERNAL_SOURCES as idle.

A typical reply sends AI_STATE_THINKING right after the placeholder, AI_STATE_EXTERNAL_SOURCES while a tool call runs, AI_STATE_GENERATING with the first token, and ai_indicator.clear once the final text is stored. On failure the backend sends an ai_indicator.update with AI_STATE_ERROR instead of a clear, so treat error as the end of a reply too.

The indicator events reach only the clients connected and watching the channel when they're sent, and aren't stored, so a user who opens the channel halfway through a reply never gets the earlier ones. Use them for the indicator only, and rely on the message's text and generating, which arrive with every update, for the reply itself.

Stop the reply

When the user stops the reply, the transport calls channel.stopAIResponse(), which sends ai_indicator.stop. It carries only the channel's cid, not a message id, so on the backend treat it as "stop whatever reply is in flight in this channel". Abort the model's stream, then finish the way a normal reply finishes: store the text written so far with a regular partial update and generating: false, then send ai_indicator.clear. Without that final update, the message stays as it was last stored.

Stream with ephemeral updates

A regular partial update writes the message to the database on every call. A long reply flushed every few tokens turns into hundreds of writes, and with several conversations at once, you can hit the rate limits for message updates. While the reply streams, update it with ephemeralUpdateMessage instead. It takes the same payload and sends the same message.updated event, so the reply types itself out the same way, but it stores nothing.

The last update must not be ephemeral. When the model finishes, store the complete text with a regular partial update, and generating: false. Otherwise anyone who opens the channel later loads the last stored version of the message: the empty placeholder, still generating. The same goes for whatever else ends the reply, such as an error or the user stopping it.

// While the model streams: send the latest text without storing it.
await client.ephemeralUpdateMessage(
  messageId,
  { set: { text, generating: true } },
  botId,
);

// Once the model is done: store the final text.
await client.partialUpdateMessage(
  messageId,
  { set: { text, generating: false } },
  botId,
);

Ephemeral updates are server-side only, and every server-side SDK has them: the AI message streaming guide shows each one. Even ephemeral updates reach every device watching the channel, so don't send one per token: flush every 20 or so chunks, or at most every 50 to 100 ms.

Actions

The conversation's views offer what its actions allow:

• Under each finished reply, an AIReplyActionsView: copy and share while the reply has text, rate (AIConversationActions.rate), and, while the agent isn't answering, regenerate on the latest reply (AIConversationActions.regenerate).
• On the user's latest message, while the agent isn't answering, pressing and holding the bubble, or right-clicking it, offers to edit it (AIConversationActions.edit). The message goes back into the composer, which shows "Editing message" with a button that cancels it, and the next message the user sends is an edit request.

The actions reach your own builders through their props. AIReplyActionsProps.actions holds the reply's AIReplyActions, with the user's rating and the callbacks that regenerate and rate it, and is null only when the list doesn't show an AIConversation. AIUserMessageProps.onEdit puts the message back in the composer, and is null when the message can't be edited. For example, to add a button of your own next to the conversation's:

late final builders = streamAIComponentBuilders(
  replyActions: (context, props) => Row(
    children: [
      if (props.actions case final actions?) AIReplyActionsView(text: props.text, actions: actions),
      ReadAloudButton(text: props.text),
    ],
  ),
);

late final componentBuilders = StreamComponentBuilders(extensions: builders);

@override
Widget build(BuildContext context) {
  return StreamComponentFactory(
    builders: componentBuilders,
    child: AIConversationView.conversation(conversation: conversation),
  );
}

Here ReadAloudButton is a widget of your app. The row is built again when the reply's rating or the actions offered change. The Conversation View guide lists the other builders.

Generative UI

To show the UI your agent writes in place of a reply's text, pass an AIUIHost as ui:

Column(
  children: [
    Expanded(child: AIConversationView.conversation(conversation: conversation, ui: ui)),
    ChatComposer.conversation(conversation: conversation),
  ],
)

Everything else stays as it is: the actions show under each finished reply, also one that answers only with UI, the user's latest message offers to edit it, and the footer and emptyContent work the same. To change the views too, register builders, as in Actions. A reply whose UI none of the host's renderers reads shows its text. Keep the same host for as long as the conversation shows, so each reply's UI keeps its state.

The thinking indicator gives way to a reply's UI as soon as it shows. The conversation's status can't tell a reply from its UI alone, so it stays submitted until the reply has text, steps or attachments, or is finished. Keep that in mind in a footer that reads the status.