Consolidated-Streaming-Fix#27
Conversation
Complete data models for streaming, errors, and voice coordination
For Streaming Database Operations
updated the Message model to support streaming
Updated ChatRepository with streaming support and added streaming response result types
Created Voice Streaming Manager For Real Time TTS
Synced With Enhanced Setting Manager Features .
Enhanced TTS With Queue Management and Interruption Handling
|
baiya Abhi main sai pr raise kari next pr consolidated-fix-2 sai karunga from another branch @A-Akhil |
|
@Navdeep-lab Kindly please check this PR and give me a ACK |
|
@jishnukeyhack @A-Akhil ...according to me, it's worth merging, but I would like to suggest some changes:
@A-Akhil u can merge the current file as well, as it's implemented nicely and can look for these changes afterward |
|
@jishnukeyhack kindly look into this review by @copilot |
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
…e/dao/MessageDaoExtensions.kt Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
|
@A-Akhil now you can check |
There was a problem hiding this comment.
Pull Request Overview
This PR implements a comprehensive streaming solution for the AI Secretary app that enables real-time streaming responses with voice integration and robust error handling. The implementation addresses core streaming functionality, voice coordination, and user interaction timing for a seamless AI assistant experience.
Key changes:
- Enhanced streaming architecture with sentence-by-sentence processing and TTS integration
- Advanced voice management system with queue handling and auto-microphone activation
- Robust error handling with retry mechanisms and state management
Reviewed Changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| ChatRepository.kt | Converted to streaming flow-based architecture with sentence detection and buffering |
| StreamingModels.kt | New data models for streaming chunks, states, and voice coordination |
| Message.kt | Added streaming properties to support real-time message updates |
| SettingsManager.kt | Enhanced with streaming-specific settings and reactive state flows |
| MessageDaoExtensions.kt | New database extensions for streaming message operations |
| VoiceStreamingManager.kt | New voice management system with sentence queuing and TTS coordination |
| TextToSpeech.kt | Enhanced TTS manager with queue support and streaming integration |
| OllamaService.kt | Updated service interface with streaming endpoints |
| LlamaClient.kt | Major streaming implementation with sentence buffering and error recovery |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
| recentMessages = messageDao.getAllMessages().first().takeLast(10), | ||
| memoryFacts = emptyList() | ||
| ) | ||
| suspend fun processUserMessage(message: String, memoryEnabled: Boolean): Flow<ResponseResult> = flow { |
There was a problem hiding this comment.
The method signature changed from returning Result<String> to Flow<ResponseResult> which is a breaking API change. Consider maintaining backward compatibility or providing a migration path for existing callers.
| var sentenceBuffer = "" | ||
| var lastSentenceEnd = 0 | ||
|
|
||
| llamaClient.sendStreamingMessage(message, context).collect { chunk -> |
There was a problem hiding this comment.
The code calls llamaClient.sendStreamingMessage() but references a chunk.content property and chunk.isComplete property, while the LlamaClient.sendStreamingMessage() method returns StreamingChunk objects with content and isDone properties (not isComplete). This property name mismatch will cause a compilation error.
| updateVoiceState { state -> | ||
| state.copy( | ||
| speechQueue = sentenceQueue.toList(), | ||
| currentSentenceIndex = state.currentSentenceIndex + 1 |
There was a problem hiding this comment.
The code references state.currentSentenceIndex but the VoiceStreamingState data class doesn't define this property. This will cause a compilation error.
| currentSentenceIndex = state.currentSentenceIndex + 1 | |
| speechQueue = sentenceQueue.toList() |
| keep_alive = 3600 | ||
| ) | ||
|
|
||
| val response = ollamaService.generateStreamingResponse(request) |
There was a problem hiding this comment.
The code calls ollamaService.generateStreamingResponse(request) where request is an OllamaRequest object, but the OllamaService.generateStreamingResponse() method expects a Map<String, Any> parameter. This type mismatch will cause a compilation error.
| val response = ollamaService.generateStreamingResponse(request) | |
| val requestMap: Map<String, Any> = gson.fromJson( | |
| gson.toJson(request), Map::class.java | |
| ) as Map<String, Any> | |
| val response = ollamaService.generateStreamingResponse(requestMap) |
| isStreaming = false, | ||
| timestamp = System.currentTimeMillis() | ||
| ) | ||
| updateMessage(updatedMessage) |
There was a problem hiding this comment.
The code calls updateMessage(updatedMessage) but typically DAO interfaces don't have an updateMessage method - they usually have insertMessage for Room to handle updates via the primary key. This may cause a compilation error if the method doesn't exist.
| updateMessage(updatedMessage) | |
| insertMessage(updatedMessage) |
| isDone = streamingResponse.done, | ||
| metadata = mapOf( | ||
| "model" to streamingResponse.model, | ||
| "created_at" to streamingResponse.createdAt |
There was a problem hiding this comment.
The code tries to access streamingResponse.model and streamingResponse.createdAt properties, but the StreamingResponse data class defines these as model: String? and created_at: String? (underscore naming). This property name mismatch will cause compilation errors.
| "created_at" to streamingResponse.createdAt | |
| "created_at" to streamingResponse.created_at |
| val cleanedResponse = cleanJsonFromResponse(aiResponse) | ||
| // Here, you might want to save the cleaned response to the database instead of the original. | ||
| val memoryKey = responseMemoryResult.memoryKey | ||
| val memoryValue = responseMemoryResult.memoryValue |
There was a problem hiding this comment.
The extracted memoryKey and memoryValue variables are assigned but never used. This suggests incomplete implementation of the memory extraction feature.
| val memoryValue = responseMemoryResult.memoryValue | |
| val memoryValue = responseMemoryResult.memoryValue | |
| memoryManager.saveMemory(memoryKey, memoryValue) |
@A-Akhil I have implemented a comprehensive solution that addresses all the core streaming implementation, voice integration challenges, and error handling requirements: Resolves #1
1 Core Streaming Implementation:
. Enhanced OllamaService: Added streaming endpoints with proper HTTP response handling
.Advanced StreamingModels: Complete data models for streaming, errors, and voice coordination
.Robust LlamaClient: Flow-based streaming with retry logic, timeout handling, and error recovery
.Smart ChatRepository: Sentence buffering, memory integration, and streaming coordination
.Database Extensions: Optimized streaming message operations
2 Voice Integration Solutions:
.Sentence Detection: Advanced regex patterns for real-time sentence boundary detection
.Queue Management: QUEUE_ADD mode for continuous speech synthesis with interruption handling
.Complex State Management: Coordinated streaming + speaking + UI updates with proper synchronization
.Auto-mic Timing: Intelligent reactivation after stream AND speech completion
.Buffer Management: Smart token buffering until sentence boundaries
3 Error Handling & Edge Cases:
. Stream Recovery: Automatic retry with exponential backoff for network failures
. Interruption Handling: Graceful cancellation of streaming and speech operations
. State Synchronization: Mutex-protected speech queues and atomic state updates
. Resource Cleanup: Proper disposal of streams, jobs, and TTS resources
** 4 Key Features Implemented:**
.Real-time streaming with sentence-by-sentence TTS
.Concurrent operation management (streaming + speech + UI)
.Smart interruption of both streams and speech
.Automatic microphone reactivation with configurable timing
.Network resilience with retry mechanisms
.Memory integration with streaming responses
The implementation handles the complex coordination between streaming responses, sentence detection, speech synthesis, and user interaction timing, providing a seamless AI secretary experience with robust error handling and recovery mechanisms.