Skip to content

Add support for streaming to Omi Chat #1661

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
beastoin merged 6 commits into from
Jan 10, 2025
Merged

Add support for streaming to Omi Chat #1661

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8c009be
Add support streaming to chat on backend
beastoin Jan 8, 2025
edcfc4e
Add support streaming chat on app
beastoin Jan 9, 2025
8b27c9b
Add support streaming on omi question
beastoin Jan 9, 2025
acd89fb
Voice chat - button click - with stream
beastoin Jan 10, 2025
9c9d16f
Clean code on llm > prompt
beastoin Jan 10, 2025
2846b0b
Clean code > app
beastoin Jan 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 102 additions & 0 deletions app/lib/backend/http/api/messages.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,55 @@ Future<ServerMessage> sendMessageServer(String text, {String? appId}) {
});
}

Stream<ServerMessageChunk> sendMessageStreamServer(String text, {String? appId}) async* {
var url = '${Env.apiBaseUrl}v2/messages?plugin_id=$appId';
if (appId == null || appId.isEmpty || appId == 'null' || appId == 'no_selected') {
url = '${Env.apiBaseUrl}v2/messages';
}

try {
final request = await HttpClient().postUrl(Uri.parse(url));
request.headers.set('Authorization', await getAuthHeader());
request.headers.contentType = ContentType.json;
request.write(jsonEncode({'text': text}));

final response = await request.close();

if (response.statusCode != 200) {
Logger.error('Failed to send message: ${response.statusCode}');
yield ServerMessageChunk.failedMessage();
return;
}

var messageId = "1000"; // default new message
await for (var data in response.transform(utf8.decoder)) {
var lines = data.split('\n\n');
for (var line in lines.where((line) => line.isNotEmpty)) {
if (line.startsWith('think: ')) {
yield ServerMessageChunk(messageId, line.substring(7).replaceAll("__CRLF__", "\n"), MessageChunkType.think);
continue;
}

if (line.startsWith('data: ')) {
yield ServerMessageChunk(messageId, line.substring(6).replaceAll("__CRLF__", "\n"), MessageChunkType.data);
continue;
}

if (line.startsWith('done: ')) {
var text = line.substring(6);
debugPrint(text);
yield ServerMessageChunk(messageId, text, MessageChunkType.done,
message: ServerMessage.fromJson(json.decode(text)));
continue;
}
}
}
} catch (e) {
Logger.error('Error sending message: $e');
yield ServerMessageChunk.failedMessage();
}
}

Future<ServerMessage> getInitialAppMessage(String? appId) {
return makeApiCall(
url: '${Env.apiBaseUrl}v1/initial-message?plugin_id=$appId',
Expand All @@ -95,6 +144,59 @@ Future<ServerMessage> getInitialAppMessage(String? appId) {
});
}

Stream<ServerMessageChunk> sendVoiceMessageStreamServer(List<File> files) async* {
var request = http.MultipartRequest(
'POST',
Uri.parse('${Env.apiBaseUrl}v2/voice-messages'),
);
for (var file in files) {
request.files.add(await http.MultipartFile.fromPath('files', file.path, filename: basename(file.path)));
}
request.headers.addAll({'Authorization': await getAuthHeader()});

try {
var response = await request.send();
if (response.statusCode != 200) {
Logger.error('Failed to send message: ${response.statusCode}');
yield ServerMessageChunk.failedMessage();
return;
}

var messageId = "1000"; // default new message
await for (var data in response.stream.transform(utf8.decoder)) {
var lines = data.split('\n\n');
for (var line in lines.where((line) => line.isNotEmpty)) {
if (line.startsWith('think: ')) {
yield ServerMessageChunk(messageId, line.substring(7).replaceAll("__CRLF__", "\n"), MessageChunkType.think);
continue;
}

if (line.startsWith('data: ')) {
yield ServerMessageChunk(messageId, line.substring(6).replaceAll("__CRLF__", "\n"), MessageChunkType.data);
continue;
}

if (line.startsWith('done: ')) {
var text = line.substring(6);
yield ServerMessageChunk(messageId, text, MessageChunkType.done,
message: ServerMessage.fromJson(json.decode(text)));
continue;
}

if (line.startsWith('message: ')) {
var text = line.substring(9);
yield ServerMessageChunk(messageId, text, MessageChunkType.message,
message: ServerMessage.fromJson(json.decode(text)));
continue;
}
}
}
} catch (e) {
Logger.error('Error sending message: $e');
yield ServerMessageChunk.failedMessage();
}
}

Future<List<ServerMessage>> sendVoiceMessageServer(List<File> files) async {
var request = http.MultipartRequest(
'POST',
Expand Down
37 changes: 37 additions & 0 deletions app/lib/backend/schema/message.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,8 @@ class ServerMessage {
List<MessageConversation> memories;
bool askForNps = false;

List<String> thinkings = [];

ServerMessage(
this.id,
this.createdAt,
Expand Down Expand Up @@ -140,3 +142,38 @@ class ServerMessage {

bool get isEmpty => id == '0000';
}

enum MessageChunkType {
think('think'),
data('data'),
done('done'),
error('error'),
message('message'),
;

final String value;

const MessageChunkType(this.value);
}

class ServerMessageChunk {
String messageId;
MessageChunkType type;
String text;
ServerMessage? message;

ServerMessageChunk(
this.messageId,
this.text,
this.type, {
this.message,
});

static ServerMessageChunk failedMessage() {
return ServerMessageChunk(
const Uuid().v4(),
'Looks like we are having issues with the server. Please try again later.',
MessageChunkType.error,
);
}
}
20 changes: 10 additions & 10 deletions app/lib/pages/chat/page.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -340,22 +340,22 @@ class ChatPageState extends State<ChatPage> with AutomaticKeepAliveClientMixin {
);
}

_sendMessageUtil(String message) async {
MixpanelManager().chatMessageSent(message);
context.read<MessageProvider>().setSendingMessage(true);
String? appId = context.read<MessageProvider>().appProvider?.selectedChatAppId;
_sendMessageUtil(String text) async {
var provider = context.read<MessageProvider>();
MixpanelManager().chatMessageSent(text);
provider.setSendingMessage(true);
String? appId = provider.appProvider?.selectedChatAppId;
if (appId == 'no_selected') {
appId = null;
}
var newMessage = ServerMessage(
const Uuid().v4(), DateTime.now(), message, MessageSender.human, MessageType.text, appId, false, []);
context.read<MessageProvider>().addMessage(newMessage);
var message =
ServerMessage(const Uuid().v4(), DateTime.now(), text, MessageSender.human, MessageType.text, appId, false, []);
provider.addMessage(message);
scrollToBottom();
textController.clear();
await context.read<MessageProvider>().sendMessageToServer(message, appId);
// TODO: restore streaming capabilities, with initial empty message
await provider.sendMessageStreamToServer(text, appId);
scrollToBottom();
context.read<MessageProvider>().setSendingMessage(false);
provider.setSendingMessage(false);
}

sendInitialAppMessage(App? app) async {
Expand Down
110 changes: 74 additions & 36 deletions app/lib/pages/chat/widgets/ai_message.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import 'package:friend_private/utils/analytics/mixpanel.dart';
import 'package:friend_private/utils/other/temp.dart';
import 'package:friend_private/widgets/extensions/string.dart';
import 'package:provider/provider.dart';
import 'package:shimmer/shimmer.dart';

class AIMessage extends StatefulWidget {
final bool showTypingIndicator;
Expand Down Expand Up @@ -95,21 +96,14 @@ class _AIMessageState extends State<AIMessage> {
),
const SizedBox(width: 16.0),
Expanded(
child: Column(
mainAxisSize: MainAxisSize.max,
crossAxisAlignment: CrossAxisAlignment.start,
children: [
const SizedBox(height: 6),
buildMessageWidget(
widget.message,
widget.sendMessage,
widget.showTypingIndicator,
widget.displayOptions,
widget.appSender,
widget.updateConversation,
widget.setMessageNps,
),
],
child: buildMessageWidget(
widget.message,
widget.sendMessage,
widget.showTypingIndicator,
widget.displayOptions,
widget.appSender,
widget.updateConversation,
widget.setMessageNps,
),
),
],
Expand Down Expand Up @@ -147,6 +141,7 @@ Widget buildMessageWidget(
} else {
return NormalMessageWidget(
showTypingIndicator: showTypingIndicator,
thinkings: message.thinkings,
messageText: message.text.decodeString,
message: message,
setMessageNps: sendMessageNps,
Expand Down Expand Up @@ -353,6 +348,7 @@ class DaySummaryWidget extends StatelessWidget {
class NormalMessageWidget extends StatelessWidget {
final bool showTypingIndicator;
final String messageText;
final List<String> thinkings;
final ServerMessage message;
final Function(int) setMessageNps;
final DateTime createdAt;
Expand All @@ -364,36 +360,78 @@ class NormalMessageWidget extends StatelessWidget {
required this.message,
required this.setMessageNps,
required this.createdAt,
this.thinkings = const [],
});

@override
Widget build(BuildContext context) {
var previousThinkingText = message.thinkings.length > 1
? message.thinkings
.sublist(message.thinkings.length - 2 >= 0 ? message.thinkings.length - 2 : 0)
.first
.decodeString
: null;
var thinkingText = message.thinkings.isNotEmpty ? message.thinkings.last.decodeString : null;
return Column(
crossAxisAlignment: CrossAxisAlignment.start,
mainAxisAlignment: MainAxisAlignment.start,
mainAxisSize: MainAxisSize.min,
children: [
Padding(
padding: const EdgeInsets.only(bottom: 4.0),
child: Text(
formatChatTimestamp(createdAt),
style: TextStyle(
color: Colors.grey.shade500,
fontSize: 12,
),
),
),
SelectionArea(
child: showTypingIndicator
? const Row(
mainAxisSize: MainAxisSize.min,
crossAxisAlignment: CrossAxisAlignment.start,
mainAxisAlignment: MainAxisAlignment.start,
showTypingIndicator && messageText.isEmpty
? Container(
margin: EdgeInsets.only(top: previousThinkingText != null ? 0 : 8),
child: Row(
children: [
SizedBox(width: 4),
TypingIndicator(),
Spacer(),
thinkingText != null
? Expanded(
child: Column(
crossAxisAlignment: CrossAxisAlignment.start,
mainAxisAlignment: MainAxisAlignment.start,
children: [
previousThinkingText != null
? Text(
overflow: TextOverflow.fade,
maxLines: 1,
softWrap: false,
previousThinkingText,
style: const TextStyle(color: Colors.white60, fontSize: 14),
)
: const SizedBox.shrink(),
Shimmer.fromColors(
baseColor: Colors.white,
highlightColor: Colors.grey,
child: Text(
overflow: TextOverflow.fade,
maxLines: 1,
softWrap: false,
thinkingText,
style: const TextStyle(color: Colors.white, fontSize: 14),
),
)
],
),
)
: const SizedBox(
height: 16,
child: TypingIndicator(),
),
],
)
: _getMarkdownWidget(context, messageText),
))
: const SizedBox.shrink(),
!(showTypingIndicator && messageText.isEmpty)
? Container(
margin: const EdgeInsets.only(bottom: 4.0),
child: Text(
formatChatTimestamp(createdAt),
style: TextStyle(
color: Colors.grey.shade500,
fontSize: 12,
),
),
)
: const SizedBox.shrink(),
SelectionArea(
child: messageText.isEmpty ? const SizedBox.shrink() : _getMarkdownWidget(context, messageText),
),
_getNpsWidget(context, message, setMessageNps),
if (!showTypingIndicator) CopyButton(messageText: messageText),
Expand Down
Loading