provider-claude.R

#' @include provider.R
#' @include content.R
#' @include turns.R
#' @include tools-def.R
NULL

#' Chat with an Anthropic Claude model
#'
#' @description
#' [Anthropic](https://www.anthropic.com) provides a number of chat based
#' models under the [Claude](https://www.anthropic.com/claude) moniker.
#' Note that a Claude Pro membership does not give you the ability to call
#' models via the API; instead, you will need to sign up (and pay for) a
#' [developer account](https://console.anthropic.com/)
#'
#' ## Authentication
#'
#' To authenticate, we recommend saving your
#' [API key](https://console.anthropic.com/account/keys) to
#' the `ANTHROPIC_API_KEY` env var in your `.Renviron`
#' (which you can easily edit by calling `usethis::edit_r_environ()`).
#'
#' @inheritParams chat_openai
#' @inherit chat_openai return
#' @param api_key The API key to use for authentication. You generally should
#'   not supply this directly, but instead set the `ANTHROPIC_API_KEY` environment
#'   variable.
#' @param max_tokens Maximum number of tokens to generate before stopping.
#' @family chatbots
#' @export
#' @examplesIf has_credentials("claude")
#' chat <- chat_claude()
#' chat$chat("Tell me three jokes about statisticians")
chat_claude <- function(system_prompt = NULL,
                            turns = NULL,
                            max_tokens = 4096,
                            model = NULL,
                            api_args = list(),
                            base_url = "https://api.anthropic.com/v1",
                            api_key = anthropic_key(),
                            echo = NULL) {
  turns <- normalize_turns(turns, system_prompt)
  echo <- check_echo(echo)

  model <- model %||% "claude-3-5-sonnet-latest"

  provider <- ProviderClaude(
    model = model,
    max_tokens = max_tokens,
    extra_args = api_args,
    base_url = base_url,
    api_key = api_key
  )

  Chat$new(provider = provider, turns = turns, echo = echo)
}

ProviderClaude <- new_class(
  "ProviderClaude",
  parent = Provider,
  properties = list(
    api_key = prop_string(),
    model = prop_string(),
    max_tokens = prop_number_whole(min = 1)
  )
)

anthropic_key <- function() {
  key_get("ANTHROPIC_API_KEY")
}
anthropic_key_exists <- function() {
  key_exists("ANTHROPIC_API_KEY")
}

method(base_request, ProviderClaude) <- function(provider) {
  req <- request(provider@base_url)
  # <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>
  req <- req_headers_redacted(req, `x-api-key` = provider@api_key)
  # <https://docs.anthropic.com/en/api/rate-limits>
  req <- req_retry(
    req,
    # <https://docs.anthropic.com/en/api/errors#http-errors>
    is_transient = function(resp) resp_status(resp) %in% c(429, 503, 529),
    max_tries = 2
  )
  req <- ellmer_req_timeout(req, stream)

  # <https://docs.anthropic.com/en/api/errors>
  req <- req_error(req, body = function(resp) {
    if (resp_content_type(resp) == "application/json") {
      json <- resp_body_json(resp)
      paste0(json$error$message, " [", json$error$type, "]")
    }
  })

  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 {
    system <- NULL
  }

  messages <- compact(as_json(provider, turns))

  if (!is.null(type)) {
    tool_def <- ToolDef(
      fun = function(...) {},
      name = "_structured_tool_call",
      description = "Extract structured data",
      arguments = type_object(data = type)
    )
    tools[[tool_def@name]] <- tool_def
    tool_choice <- list(type = "tool", name = tool_def@name)
    stream <- FALSE
  } else {
    tool_choice <- NULL
  }
  tools <- as_json(provider, unname(tools))

  body <- compact(list2(
    model = provider@model,
    system = system,
    messages = messages,
    stream = stream,
    max_tokens = provider@max_tokens,
    tools = tools,
    tool_choice = tool_choice,
  ))
  modify_list(body, provider@extra_args)
}

# Batch chat -------------------------------------------------------------------

method(has_batch_support, ProviderClaude) <- function(provider) {
  TRUE
}

# https://docs.anthropic.com/en/api/creating-message-batches
method(batch_submit, ProviderClaude) <- function(provider, turns, type = NULL) {
  req <- base_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 = params
    )
  })
  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 <- base_request(provider)
  req <- req_url_path_append(req, "/messages/batches/", batch$id)
  resp <- req_perform(req)
  resp_body_json(resp)
}

method(batch_info, ProviderClaude) <- function(provider, batch) {
  counts <- batch$request_counts

  list(
    working = batch$processing_status != "ended",
    counts = list(
      processing = counts$processing,
      succeeded = counts$succeeded,
      failed = counts$errored + counts$canceled + counts$expired
    )
  )
}

# https://docs.anthropic.com/en/api/retrieving-message-batch-results
method(batch_retrieve, ProviderClaude) <- function(provider, batch) {
  req <- base_request(provider)
  req <- req_url(req, batch$results_url)
  req <- req_progress(req, "down")

  path <- withr::local_tempfile()
  req <- req_perform(req, path = path)

  lines <- readLines(path, warn = FALSE)
  json <- lapply(lines, jsonlite::fromJSON, simplifyVector = FALSE)

  ids <- as.numeric(gsub("chat-", "", map_chr(json, "[[", "custom_id")))
  results <- lapply(json, "[[", "result")
  results[order(ids)]
}

method(batch_result_ok, ProviderClaude) <- function(provider, result) {
  result$type == "succeeded"
}

method(batch_result_turn, ProviderClaude) <- function(provider, result, has_type = FALSE) {
  value_turn(provider, result$message, has_type = has_type)
}

# Claude -> ellmer --------------------------------------------------------------

method(stream_parse, ProviderClaude) <- function(provider, event) {
  if (is.null(event)) {
    cli::cli_abort("Connection closed unexpectedly")
  }

  data <- jsonlite::parse_json(event$data)
  if (identical(data$type, "message_stop")) {
    return(NULL)
  }

  data
}
method(stream_text, ProviderClaude) <- function(provider, event) {
  if (event$type == "content_block_delta") {
    event$delta$text
  }
}
method(stream_merge_chunks, ProviderClaude) <- function(provider, result, chunk) {
  if (chunk$type == "ping") {
    # nothing to do
  } else if (chunk$type == "message_start") {
    result <- chunk$message
  } else if (chunk$type == "content_block_start") {
    result$content[[chunk$index + 1L]] <- chunk$content_block
  } else if (chunk$type == "content_block_delta") {
    if (chunk$delta$type == "text_delta") {
      paste(result$content[[chunk$index + 1L]]$text) <- chunk$delta$text
    } else if (chunk$delta$type == "input_json_delta") {
      if (chunk$delta$partial_json != "") {
        # See issue #228 about partial_json sometimes being ""
        paste(result$content[[chunk$index + 1L]]$input) <- chunk$delta$partial_json
      }
    } else {
      cli::cli_inform(c("!" = "Unknown delta type {.str {chunk$delta$type}}."))
    }
  } else if (chunk$type == "content_block_stop") {
    # nothing to do
  } else if (chunk$type == "message_delta") {
    result$stop_reason <- chunk$delta$stop_reason
    result$stop_sequence <- chunk$delta$stop_sequence
    result$usage$output_tokens <- chunk$usage$output_tokens
  } else if (chunk$type == "error") {
    if (chunk$error$type == "overloaded_error") {
      # https://docs.anthropic.com/en/api/messages-streaming#error-events
      # TODO: track number of retries
      wait <- backoff_default(1)
      Sys.sleep(wait)
    } else {
      cli::cli_abort("{chunk$error$message}")
    }
  } else {
    cli::cli_inform(c("!" = "Unknown chunk type {.str {chunk$type}}."))
  }
  result
}

method(value_turn, ProviderClaude) <- function(provider, result, has_type = FALSE) {
  contents <- lapply(result$content, function(content) {
    if (content$type == "text") {
      ContentText(content$text)
    } else if (content$type == "tool_use") {
      if (has_type) {
        ContentJson(content$input$data)
      } else {
        if (is_string(content$input)) {
          content$input <- jsonlite::parse_json(content$input)
        }
        ContentToolRequest(content$id, content$name, content$input)
      }
    } else {
      cli::cli_abort(
        "Unknown content type {.str {content$type}}.",
        .internal = TRUE
      )
    }
  })

  tokens <- c(result$usage$input_tokens, result$usage$output_tokens)
  tokens_log("Claude", tokens)

  Turn(result$role, contents, json = result, tokens = tokens)
}

# ellmer -> Claude --------------------------------------------------------------

method(as_json, list(ProviderClaude, Turn)) <- function(provider, x) {
  if (x@role == "system") {
    # claude passes system prompt as separate arg
    NULL
  } else if (x@role %in% c("user", "assistant")) {
    list(role = x@role, content = as_json(provider, x@contents))
  } else {
    cli::cli_abort("Unknown role {turn@role}", .internal = TRUE)
  }
}

method(as_json, list(ProviderClaude, ContentText)) <- function(provider, x) {
  list(type = "text", text = x@text)
}

method(as_json, list(ProviderClaude, ContentPDF)) <- function(provider, x) {
  list(
    type = "document",
    source = list(
      type = "base64",
      media_type = x@type,
      data = x@data
    )
  )
}

method(as_json, list(ProviderClaude, ContentImageRemote)) <- function(provider, x) {
  cli::cli_abort("Claude doesn't support remote images")
}

method(as_json, list(ProviderClaude, ContentImageInline)) <- function(provider, x) {
  list(
    type = "image",
    source = list(
      type = "base64",
      media_type = x@type,
      data = x@data
    )
  )
}

# https://docs.anthropic.com/en/docs/build-with-claude/tool-use#handling-tool-use-and-tool-result-content-blocks
method(as_json, list(ProviderClaude, ContentToolRequest)) <- function(provider, x) {
  list(
    type = "tool_use",
    id = x@id,
    name = x@name,
    input = x@arguments
  )
}

# https://docs.anthropic.com/en/docs/build-with-claude/tool-use#handling-tool-use-and-tool-result-content-blocks
method(as_json, list(ProviderClaude, ContentToolResult)) <- function(provider, x) {
  list(
    type = "tool_result",
    tool_use_id = x@id,
    content = tool_string(x),
    is_error = tool_errored(x)
  )
}

method(as_json, list(ProviderClaude, ToolDef)) <- function(provider, x) {
  list(
    name = x@name,
    description = x@description,
    input_schema = compact(as_json(provider, x@arguments))
  )
}



# Helpers ----------------------------------------------------------------

# From httr2
backoff_default <- function(i) {
  round(min(stats::runif(1, min = 1, max = 2^i), 60), 1)
}