Skip to content

Refactor error handling for streaming chat flow #117

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

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Refactor error handling for streaming chat flow #117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc4321c
Refactor error handling for streaming chat flow
nomisRev Apr 11, 2025
47c9b6a
Merge branch 'main' into avoid-dropping-messages
nomisRev Apr 17, 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
20 changes: 3 additions & 17 deletions ...n/kotlin/me/kpavlov/langchain4j/kotlin/model/chat/StreamingChatLanguageModelExtensions.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ import kotlinx.coroutines.channels.trySendBlocking
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.flow
import me.kpavlov.langchain4j.kotlin.model.chat.request.ChatRequestBuilder
import me.kpavlov.langchain4j.kotlin.model.chat.request.chatRequest
Expand Down Expand Up @@ -50,19 +49,6 @@ sealed interface StreamingChatLanguageModelReply {
data class CompleteResponse(
val response: ChatResponse,
) : StreamingChatLanguageModelReply

/**
* Represents an error that occurred during the streaming process
* when generating a reply from the AI language model. This type
* of reply is used to indicate a failure in the operation and
* provides details about the cause of the error.
*
* @property cause The underlying exception or error that caused the failure.
* @see StreamingChatResponseHandler.onError
*/
data class Error(
val cause: Throwable,
) : StreamingChatLanguageModelReply
}

/**
Expand All @@ -83,7 +69,7 @@ sealed interface StreamingChatLanguageModelReply {
*/
fun StreamingChatLanguageModel.chatFlow(
block: ChatRequestBuilder.() -> Unit,
): Flow<StreamingChatLanguageModelReply> =
): Flow<StreamingChatLanguageModelReply> = flow {
callbackFlow {
val model = this@chatFlow
val chatRequest = chatRequest(block)
Expand Down Expand Up @@ -114,7 +100,6 @@ fun StreamingChatLanguageModel.chatFlow(
error.message,
error,
)
trySend(StreamingChatLanguageModelReply.Error(error))
close(error)
}
}
Expand All @@ -127,7 +112,8 @@ fun StreamingChatLanguageModel.chatFlow(
// cleanup
logger.info("Flow is canceled")
}
}
}.buffer(Channel.UNLIMITED).collect(this)
}

fun TokenStream.asFlow(): Flow<String> = flow {
callbackFlow {
Expand Down
1 change: 0 additions & 1 deletion .../src/test/kotlin/me/kpavlov/langchain4j/kotlin/model/chat/StreamingChatLanguageModelIT.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,6 @@ internal class StreamingChatLanguageModelIT {
}

is CompleteResponse -> responseRef.set(it.response)
is StreamingChatLanguageModelReply.Error -> fail("Error", it.cause)
else -> fail("Unsupported event: $it")
}
}
Expand Down
Loading