Start sketching out batch API

R/BatchChat.R

@@ -0,0 +1,70 @@
+# Make print method call `$run()`?
+
+BatchChat <- R6::R6Class(
+  "BatchChat",
+  public = list(
+
+    # Need to take existing turns and then add user turns to them
+    # TODO: optionally automatically cache all previous the turns
+    initialize = function(id, provider, turns, type = NULL) {
+      private$provider <- provider
+      private$turns <- turns
+      private$type <- type
+    },
+
+    run = function() {
+      if (status == "new") {
+        private$submitted <- batch_submit(
+          private$provider,
+          turns = private$turns,
+          type = private$type
+        )
+        private$poll()
+      }
+
+      while (status != "completed") {
+        # TODO: update progress spinner
+        tryCatch(
+          {
+            result <- batch_poll(private$provider, private$submitted)
+            private$polled <- result$body
+          },
+          interrupt = function(cnd) {
+            cli::cli_inform(c(
+              x = "Interrupted by user.",
+              i = "Use `$run()` to resume."
+            ))
+            break
+          }
+        )
+
+        if (result$done) {
+          break
+        }
+      }
+
+      if (status == "completed") {
+        cli::cli_inform(c(
+          v = "Completed",
+          i = "Use `$results()` to get results."
+        ))
+      }
+    },
+
+    results = function() {
+      batch_results(private$provider, private$polled)
+    }
+  ),
+  private = list(
+    id = NULL,
+    provider = NULL,
+    turns = NULL,
+    type = NULL,
+
+    # "new" | "submitted" | "completed"
+    status = NULL,
+
+    submitted = NULL,
+    polled = NULL
+  )
+)

R/provider-claude.R

@@ -72,15 +72,8 @@ anthropic_key_exists <- function() {
   key_exists("ANTHROPIC_API_KEY")
 }
 
-method(chat_request, ProviderClaude) <- function(provider,
-                                                 stream = TRUE,
-                                                 turns = list(),
-                                                 tools = list(),
-                                                 type = NULL) {
-
+method(base_request, ProviderClaude) <- function(provider) {
   req <- request(provider@base_url)
-  # https://docs.anthropic.com/en/api/messages
-  req <- req_url_path_append(req, "/messages")
   # <https://docs.anthropic.com/en/api/versioning>
   req <- req_headers(req, `anthropic-version` = "2023-06-01")
   # <https://docs.anthropic.com/en/api/getting-started#authentication>
@@ -102,6 +95,37 @@ method(chat_request, ProviderClaude) <- function(provider,
     }
   })
 
+  req
+}
+
+# Chat ------------------------------------------------------------------------
+
+method(chat_request, ProviderClaude) <- function(provider,
+                                                 stream = TRUE,
+                                                 turns = list(),
+                                                 tools = list(),
+                                                 type = NULL) {
+  req <- base_request(provider)
+  # https://docs.anthropic.com/en/api/messages
+  req <- req_url_path_append(req, "/messages")
+
+  body <- chat_body(
+    provider,
+    stream = stream,
+    turns = turns,
+    tools = tools,
+    type = type
+  )
+  req <- req_body_json(req, body)
+  req
+}
+
+method(chat_body, ProviderClaude) <- function(provider,
+                                              stream = TRUE,
+                                              turns = list(),
+                                              tools = list(),
+                                              type = NULL) {
+
   if (length(turns) >= 1 && is_system_prompt(turns[[1]])) {
     system <- turns[[1]]@text
   } else {
@@ -134,10 +158,69 @@ method(chat_request, ProviderClaude) <- function(provider,
     tools = tools,
     tool_choice = tool_choice,
   ))
-  body <- modify_list(body, provider@extra_args)
-  req <- req_body_json(req, body)
+  modify_list(body, provider@extra_args)
+}
 
-  req
+# Batch chat -------------------------------------------------------------------
+
+# https://docs.anthropic.com/en/api/creating-message-batches
+method(batch_submit, ProviderClaude) <- function(provider, turns, type = NULL) {
+  req <- provider_request(provider)
+  req <- req_url_path_append(req, "/messages/batches")
+
+  requests <- map(seq_along(turns), function(i) {
+    params <- chat_body(
+      provider,
+      stream = FALSE,
+      turns = turns[[i]],
+      type = type
+    )
+    list(
+      custom_id = paste0("chat-", i),
+      params = body
+    )
+  })
+  req <- req_body_json(req, list(requests = requests))
+
+  resp <- req_perform(req)
+  resp_body_json(resp)
+}
+
+# https://docs.anthropic.com/en/api/retrieving-message-batches
+method(batch_poll, ProviderClaude) <- function(provider, batch) {
+  req <- provider_request(provider)
+  req <- req_url_path_append(req, "/messages/batches/", batch$id)
+  resp <- req_perform(req)
+  body <- resp_body_json(resp)
+
+  list(
+    done = body$processing_status == "ended",
+    status = list(
+      processing = body$request_counts$processing,
+      succeeded = body$request_counts$completed,
+      failed = body$request_counts$errored +
+        body$request_counts$cancelled +
+        body$request_counts$expired
+    ),
+    body = body
+  )
+}
+
+# https://docs.anthropic.com/en/api/retrieving-message-batch-results
+method(batch_retrieve, ProviderClaude) <- function(provider, batch) {
+  req <- provider_request(provider)
+  req <- req_url_path(batch$results_url)
+  req <- req_progress(req, "down")
+
+  path <- withr::local_tempfile()
+  req <- req_perform(req, path = path)
+
+  #
+  jsonlite::stream_in(path, function(page) {
+    # Parse json a page at a time
+  })
+
+  # Re-align to match inputs
 }
 
 # Claude -> ellmer --------------------------------------------------------------

R/provider.R

@@ -28,12 +28,27 @@ Provider <- new_class(
 
 # Create a request------------------------------------
 
+base_request <- new_generic("base_request", "provider",
+  function(provider) {
+    S7_dispatch()
+  }
+)
+
 chat_request <- new_generic("chat_request", "provider",
   function(provider, stream = TRUE, turns = list(), tools = list(), type = NULL) {
     S7_dispatch()
   }
 )
 
+chat_body <- new_generic(
+  "chat_body",
+  "provider",
+  function(provider, stream = TRUE, turns = list(), tools = list(), type = NULL) {
+    S7_dispatch()
+  }
+)
+
+
 chat_resp_stream <- new_generic("chat_resp_stream", "provider",
   function(provider, resp) {
     S7_dispatch()
@@ -75,3 +90,29 @@ method(as_json, list(Provider, class_list)) <- function(provider, x) {
 method(as_json, list(Provider, ContentJson)) <- function(provider, x) {
   as_json(provider, ContentText("<structured data/>"))
 }
+
+# Batch API ---------------------------------------------------------------
+
+batch_submit <- new_generic(
+  "batch_submit",
+  "provider",
+  function(provider, turns, type = NULL) {
+    S7_dispatch()
+  }
+)
+
+batch_poll <- new_generic(
+  "batch_poll",
+  "provider",
+  function(provider, batch) {
+    S7_dispatch()
+  }
+)
+
+batch_retrieve <- new_generic(
+  "batch_retrieve",
+  "provider",
+  function(provider, batch) {
+    S7_dispatch()
+  }
+)