Bedrock Runtime API: Problems with hot versus cold flows

[jg-develop]

I am struggling getting the Nova Sonic API to work. Is there any example Kotlin code available to show me the right patterns to use? I'm trying to use the Kotlin SDK on Android, and using the Java sample as a guide, but struggling. I have a problem with Flows versus MutableSharedFlows. The following code works, but isn't what I need. I need to change the Flow to a MutableSharedFlow to be able to emit new values in the future as the conversation with bedrock progresses. But when I use a MutableSharedFlow, it doesn't work.

val events = flow {
        emit(InvokeModelWithBidirectionalStreamInput.Chunk(
            value = BidirectionalInputPayloadPart {
                this.bytes = SessionStart.toByteArray()
                }
        ))}

val request = InvokeModelWithBidirectionalStreamRequest {
    this.modelId = "amazon.nova-sonic-v1:0"
    this.body = events
}

bedrockClient.invokeModelWithBidirectionalStream(request) { response ->
    println("Response: $response")
}

When I call invokeModelWithBidirectionalStream above, I see lots of HTTP traffic and the SDK doing its stuff. Whereas the code below doesn't do anything:

val events: MutableSharedFlow<InvokeModelWithBidirectionalStreamInput> = MutableSharedFlow(
    replay = 1,
    extraBufferCapacity = 10
)

events.emit(InvokeModelWithBidirectionalStreamInput.Chunk(
    value = BidirectionalInputPayloadPart {
    this.bytes = SessionStart.toByteArray()
}))

/* Same code below as for the cold flow, but invokeModelWithBidirectionalStream doesn't do anything. */
val request = InvokeModelWithBidirectionalStreamRequest {
    this.modelId = "amazon.nova-sonic-v1:0"
    this.body = events
}

bedrockClient.invokeModelWithBidirectionalStream(request) { response ->
    println("Response: $response")
}

Even though I have replay set in the hot flow, it doesn't work, and I have tried wrapping code in launch, and making sure the emit is called after collect by using onSubscription. And I have put delays in the code. In all cases the AWS SDK doesn't perform any HTTP transactions, whereas it does with the cold flow? I have also tried callbackFlows etc, and still no success.

Any help would be gratefully received.

[ianbotsf]

Hi @jg-develop, we haven't tested hot flows with our event streaming APIs yet. I've been able to reproduce your results with the code samples you provided and I'm digging into what it'll take to properly support hot flows. I'll post back here when I have an update.

[ianbotsf]

Yes, you should generally call invokeModelWithBidirectionalStream once for a given session which means using a single flow for all of the events. I think hot flows may work but challenge is signaling closure. What's the source of your events? Is something producing them asynchronously and you're looking to bridge that to the API input flow? And how will you know when you're finished and you definitely won't be sending any more events?

Depending on the answers to those questions, you may consider using a Channel for the bridging mechanism and then converting the receive side to a flow via consumeAsFlow. It still produces a hot flow but one that supports clear closure semantics. For example:

val eventChannel = Channel<InvokeModelWithBidirectionalStreamInput>(Channel.BUFFERED)

launch {
    channel.send(
        InvokeModelWithBidirectionalStreamInput.Chunk(
            value = BidirectionalInputPayloadPart {
                this.bytes = sessionStartJson.toByteArray()
            }
        )
    )
    
    // Simulate a pause before/during receiving other events
    delay(5.seconds)
    
    // This part is critical because otherwise the API input flow will never terminate
    channel.close()
}

val eventFlow = eventChannel.consumeAsFlow()

val request = InvokeModelWithBidirectionalStreamRequest {
    this.modelId = "amazon.nova-sonic-v1:0"
    this.body = eventFlow
}

bedrockClient.invokeModelWithBidirectionalStream(request) { response ->
    println("Response: $response")
}

[jg-develop]

Thanks again for your reply. To give some context, I am attempting to port the Nova Sonic examples from Python/Java to Android (sample 1 and sample 2).

In both samples they create an object once, and then use it to send/receive a stream of events for the duration of the conversation. The main type of events is a continual stream of bidirectional audio samples. The Kotlin API doesn't seem to support the same semantics as the Python+Java APIs; and in particular the closure semantics that you mentioned seem to be fundamentally at odds with it. I'm clearly missing something about the intended pattern of usage, either that or the API has a fundamental problem which seems doubtful.

[ianbotsf]

I'm not an expert on the Python or Java SDKs but it appears that both of the linked samples explicitly close the input to signal when they're finished. The Python code closes the input stream when the user presses the Enter key and the Java code closes the input publisher when the web socket closes.

My toy example above omits a lot of the complexity of the Python/Java samples by putting the close call in the same execution context as the event emission, which may not be realistic for your application. What will the source of your audio events be and how will you "subscribe" to source? What would indicate to your application that the session is complete and no more audio will be sent?

[ianbotsf]

Any update here @jg-develop? Have you been able to test using channels to accumulate/send input event streams?

[jg-develop]

I decided to spend some time porting the Python sample code to iOS using the AWs Swift SDK to see if it offers any clues. The shape of the Swift API is very similar to the Kotlin one (and both are slightly different to Python and Java).

I have the Swift example working, and it has indeed shown me the way forward. I am soon going to apply those lessons to Android/Kotlin. The main discovery is that the SDK "eats" my initial messages and sends nothing on the wire until a certain point is reached. But if my code continues sending blindly, it does all work. Ie. after sending my initial messages, if I then move on to sending audio samples, it does work as expected. My development approach was to get the initial messages working, before moving on to the audio, but it looks like I just needed to bite the bullet and try it all at once.

So I suspect that hot flows have been working all along, and I just needed to carry on with my development and send the audio. If it works, then I am sorry for the false alarm. I guess my only suggestion is that it would have been helpful if Bedrock had some sort of positive ack to my initial messages to confirm to me that I was doing the right thing. I never like writing the whole application in one go before being able to run it. But this is not anything that the Kotlin API can solve, it's the way Bedrock works.