Return ContentThinking objects from stream_text()

DESCRIPTION

@@ -55,6 +55,7 @@ Suggests:
     withr
 VignetteBuilder:
     knitr
+Remotes: posit-dev/shinychat/pkg-r#167
 Config/Needs/website: tidyverse/tidytemplate, rmarkdown
 Config/testthat/edition: 3
 Config/testthat/parallel: true

NEWS.md

@@ -1,5 +1,6 @@
 # ellmer (development version)
 
+* `stream_text()` now returns `ContentThinking` objects for thinking/reasoning content from Anthropic, OpenAI, and Google Gemini, allowing downstream packages like shinychat to distinguish thinking from regular text during streaming (@simonpcouch, #909).
 * `chat_github()` now uses `chat_openai_compatible()` for improved compatibility, and `models_github()` now supports custom `base_url` configuration (@D-M4rk, #877).
 * `chat_ollama()` now contains a slot for `top_k` within the `params` argument (@frankiethull).
 

R/chat.R

@@ -606,14 +606,11 @@ Chat <- R6::R6Class(
       if (stream) {
         result <- NULL
         for (chunk in response) {
-          text <- stream_text(private$provider, chunk)
-          if (!is.null(text)) {
+          content <- stream_content(private$provider, chunk)
+          if (!is.null(content)) {
+            text <- content_text(content)
             emit(text)
-            if (yield_as_content) {
-              yield(ContentText(text))
-            } else {
-              yield(text)
-            }
+            yield(if (yield_as_content) content else text)
             any_text <- TRUE
           }
 
@@ -688,14 +685,11 @@ Chat <- R6::R6Class(
       if (stream) {
         result <- NULL
         for (chunk in await_each(response)) {
-          text <- stream_text(private$provider, chunk)
-          if (!is.null(text)) {
+          content <- stream_content(private$provider, chunk)
+          if (!is.null(content)) {
+            text <- content_text(content)
             emit(text)
-            if (yield_as_content) {
-              yield(ContentText(text))
-            } else {
-              yield(text)
-            }
+            yield(if (yield_as_content) content else text)
             any_text <- TRUE
           }
 

R/provider-aws.R

@@ -266,9 +266,13 @@ method(stream_parse, ProviderAWSBedrock) <- function(provider, event) {
   body
 }
 
-method(stream_text, ProviderAWSBedrock) <- function(provider, event) {
+method(stream_content, ProviderAWSBedrock) <- function(provider, event) {
   if (event$event_type == "contentBlockDelta") {
-    event$delta$text
+    text <- event$delta$text
+    if (is.null(text)) {
+      return(NULL)
+    }
+    ContentText(text)
   }
 }
 

R/provider-claude.R

@@ -273,9 +273,16 @@ method(stream_parse, ProviderAnthropic) <- function(provider, event) {
 
   data
 }
-method(stream_text, ProviderAnthropic) <- function(provider, event) {
+method(stream_content, ProviderAnthropic) <- function(provider, event) {
   if (event$type == "content_block_delta") {
-    event$delta$text %||% event$delta$thinking
+    if (identical(event$delta$type, "thinking_delta")) {
+      return(ContentThinking(event$delta$thinking))
+    }
+    text <- event$delta$text
+    if (is.null(text)) {
+      return(NULL)
+    }
+    ContentText(text)
   }
 }
 method(stream_merge_chunks, ProviderAnthropic) <- function(

R/provider-google.R

@@ -272,8 +272,18 @@ method(stream_parse, ProviderGoogleGemini) <- function(provider, event) {
     jsonlite::parse_json(event$data)
   }
 }
-method(stream_text, ProviderGoogleGemini) <- function(provider, event) {
-  event$candidates[[1]]$content$parts[[1]]$text
+method(stream_content, ProviderGoogleGemini) <- function(provider, event) {
+  parts <- event$candidates[[1]]$content$parts
+  if (is.null(parts) || length(parts) == 0) {
+    return(NULL)
+  }
+
+  part <- parts[[1]]
+  if (isTRUE(part$thought) && !is.null(part$text)) {
+    ContentThinking(part$text)
+  } else if (!is.null(part$text)) {
+    ContentText(part$text)
+  }
 }
 method(stream_merge_chunks, ProviderGoogleGemini) <- function(
   provider,
@@ -320,7 +330,9 @@ method(value_turn, ProviderGoogleGemini) <- function(
   message <- result$candidates[[1]]$content
 
   contents <- lapply(message$parts, function(content) {
-    if (has_name(content, "text")) {
+    if (isTRUE(content$thought) && has_name(content, "text")) {
+      ContentThinking(content$text)
+    } else if (has_name(content, "text")) {
       if (has_type) {
         ContentJson(string = content$text)
       } else {
@@ -404,6 +416,15 @@ method(as_json, list(ProviderGoogleGemini, ContentText)) <- function(
   }
 }
 
+method(as_json, list(ProviderGoogleGemini, ContentThinking)) <- function(
+  provider,
+  x,
+  ...
+) {
+  # https://ai.google.dev/gemini-api/docs/thinking
+  list(thought = TRUE, text = x@thinking)
+}
+
 method(as_json, list(ProviderGoogleGemini, ContentPDF)) <- function(
   provider,
   x,

R/provider-openai-compatible.R

@@ -238,12 +238,15 @@ method(stream_parse, ProviderOpenAICompatible) <- function(provider, event) {
 
   jsonlite::parse_json(event$data)
 }
-method(stream_text, ProviderOpenAICompatible) <- function(provider, event) {
+method(stream_content, ProviderOpenAICompatible) <- function(provider, event) {
   if (length(event$choices) == 0) {
-    NULL
-  } else {
-    event$choices[[1]]$delta[["content"]]
+    return(NULL)
+  }
+  text <- event$choices[[1]]$delta[["content"]]
+  if (is.null(text)) {
+    return(NULL)
   }
+  ContentText(text)
 }
 method(stream_merge_chunks, ProviderOpenAICompatible) <- function(
   provider,

R/provider-openai.R

@@ -241,16 +241,19 @@ method(chat_params, ProviderOpenAI) <- function(provider, params) {
 
 # OpenAI -> ellmer --------------------------------------------------------------
 
-method(stream_text, ProviderOpenAI) <- function(provider, event) {
+method(stream_content, ProviderOpenAI) <- function(provider, event) {
   if (event$type == "response.output_text.delta") {
     # https://platform.openai.com/docs/api-reference/responses-streaming/response/output_text/delta
-    event$delta
+    if (is.null(event$delta)) {
+      return(NULL)
+    }
+    ContentText(event$delta)
   } else if (event$type == "response.reasoning_summary_text.delta") {
     # https://platform.openai.com/docs/api-reference/responses-streaming/response/reasoning_summary_text/delta
-    event$delta
+    ContentThinking(event$delta)
   } else if (event$type == "response.reasoning_summary_text.done") {
     # https://platform.openai.com/docs/api-reference/responses-streaming/response/reasoning_summary_text/done
-    "\n\n"
+    NULL
   }
 }
 method(stream_merge_chunks, ProviderOpenAI) <- function(

R/provider.R

@@ -144,13 +144,30 @@ stream_parse <- new_generic(
     S7_dispatch()
   }
 )
-stream_text <- new_generic(
-  "stream_text",
+stream_content <- new_generic(
+  "stream_content",
   "provider",
   function(provider, event) {
     S7_dispatch()
   }
 )
+
+stream_text <- function(provider, event) {
+  content <- stream_content(provider, event)
+  if (is.null(content)) {
+    return(NULL)
+  }
+  content_text(content)
+}
+
+content_text <- function(content) {
+  switch(
+    class(content)[1],
+    "ellmer::ContentThinking" = content@thinking,
+    "ellmer::ContentText" = content@text,
+    format(content)
+  )
+}
 stream_merge_chunks <- new_generic(
   "stream_merge_chunks",
   "provider",