diff --git a/lib/ruby_llm/protocols/converse/chat.rb b/lib/ruby_llm/protocols/converse/chat.rb index 08629c4b5..af1f9d2f5 100644 --- a/lib/ruby_llm/protocols/converse/chat.rb +++ b/lib/ruby_llm/protocols/converse/chat.rb @@ -395,15 +395,35 @@ def budget_reasoning_config(thinking) # When the turn had more than one reasoning block (common under adaptive/interleaved # thinking during tool use), thinking.blocks holds the exact original blocks and must # be replayed verbatim instead of reconstructed from the merged text/signature. + # + # Blocks captured by the InvokeAnthropic protocol are in the Anthropic Messages shape + # (type: thinking / redacted_thinking); those are translated to reasoningContent so a + # chat that moves from InvokeModel back to Converse can still replay its reasoning. + # reasoningContent blocks pass through untouched. def format_thinking_blocks(thinking) return nil unless thinking - return thinking.blocks if thinking.blocks + return thinking.blocks.map { |block| converse_thinking_block(block) } if thinking.blocks block = format_single_thinking_block(thinking) block ? [block] : nil end + def converse_thinking_block(block) + return block unless block.is_a?(Hash) + + case (block['type'] || block[:type]).to_s + when 'thinking' + text = { text: block['thinking'] || block[:thinking] || '', + signature: block['signature'] || block[:signature] } + { reasoningContent: { reasoningText: text.compact } } + when 'redacted_thinking' + { reasoningContent: { redactedContent: block['data'] || block[:data] } } + else + block + end + end + # Lossy fallback for messages persisted without raw thinking blocks. `signature` # here is always a real reasoningText signature (streaming captures it from # reasoningContent signature deltas; redacted content is only ever captured into diff --git a/lib/ruby_llm/protocols/converse/event_stream.rb b/lib/ruby_llm/protocols/converse/event_stream.rb new file mode 100644 index 000000000..4a9e5bc5c --- /dev/null +++ b/lib/ruby_llm/protocols/converse/event_stream.rb @@ -0,0 +1,149 @@ +# frozen_string_literal: true + +require 'base64' +require 'faraday' +require 'json' + +module RubyLLM + module Protocols + class Converse + # AWS Event Stream plumbing shared by every bedrock-runtime streaming endpoint: ConverseStream + # (Converse::Streaming) and InvokeModelWithResponseStream (InvokeAnthropic::Streaming). Both + # frame each event as a binary eventstream message whose JSON payload may wrap the real + # event as base64 `bytes`, and both report mid-stream failures as `:exception-type` + # messages. What each endpoint puts INSIDE the payload differs, so event interpretation + # stays with the including protocol. + module EventStream + ErrorResponse = Struct.new(:body, :status) + + # Bedrock stream exception type => HTTP status, so ErrorMiddleware raises the same error + # class a synchronous call with that failure would. + EXCEPTION_STATUSES = { + 'throttlingException' => 429, + 'validationException' => 400, + 'accessDeniedException' => 401, + 'unrecognizedClientException' => 401, + 'serviceUnavailableException' => 503 + }.freeze + + private + + # POSTs a SigV4-signed eventstream request and hands every raw body chunk of a 200 + # response to the block. Non-200 chunks go through handle_failed_stream. + def post_event_stream(url, payload, additional_headers = {}, &on_chunk) + body = JSON.generate(payload) + + @connection.post(url, payload) do |req| + req.headers.merge!(@provider.sign_headers('POST', url, body)) + req.headers.merge!(additional_headers) unless additional_headers.empty? + req.headers['Accept'] = 'application/vnd.amazon.eventstream' + + if Faraday::VERSION.start_with?('1') + req.options[:on_data] = proc { |chunk, _size| on_chunk.call(chunk) } + else + req.options.on_data = proc do |chunk, _bytes, env| + if env&.status == 200 + on_chunk.call(chunk) + else + handle_failed_stream(chunk, env) + end + end + end + end + end + + def event_stream_decoder + require 'aws-eventstream' + Aws::EventStream::Decoder.new + rescue LoadError + raise Error, + 'The aws-eventstream gem is required for Bedrock streaming. ' \ + 'Please add it to your Gemfile: gem "aws-eventstream"' + end + + def handle_failed_stream(chunk, env) + data = JSON.parse(chunk) + error_response = env.merge(body: data) + ErrorMiddleware.parse_error(provider: self, response: error_response) + rescue JSON::ParserError + RubyLLM.logger.debug { "Failed Bedrock stream error chunk: #{chunk}" } + end + + def handle_non_eventstream_error_chunk(raw_chunk) + text = raw_chunk.to_s + + if text.start_with?('event: error') + payload = text.lines.find { |line| line.start_with?('data:') }&.delete_prefix('data:')&.strip + raise_streaming_chunk_error(payload) if payload + return + end + + return unless text.lstrip.start_with?('{') && text.include?('"error"') + + raise_streaming_chunk_error(text) + end + + def raise_streaming_chunk_error(payload) + parsed = JSON.parse(payload) + message = parsed.dig('error', 'message') || parsed['message'] || 'Bedrock streaming error' + response = ErrorResponse.new({ 'message' => message }, 500) + ErrorMiddleware.parse_error(provider: self, response: response) + rescue JSON::ParserError + nil + end + + # Every complete eventstream message in raw_chunk (the decoder buffers partial ones + # across calls). + def decode_event_messages(decoder, raw_chunk) + messages = [] + message, eof = decoder.decode_chunk(raw_chunk) + + while message + messages << message + break if eof + + message, eof = decoder.decode_chunk + end + + messages + end + + def event_type_header(message) + headers = message.headers + header = headers[':event-type'] || headers[':exception-type'] + value = header.respond_to?(:value) ? header.value : header + value.is_a?(String) && !value.empty? ? value : nil + rescue StandardError + nil + end + + def exception_type_header(message) + header = message.headers[':exception-type'] + value = header.respond_to?(:value) ? header.value : header + value.is_a?(String) && !value.empty? ? value : nil + rescue StandardError + nil + end + + def decode_event_payload(payload) + outer = JSON.parse(payload) + + if outer['bytes'].is_a?(String) + JSON.parse(Base64.decode64(outer['bytes'])) + else + outer + end + rescue JSON::ParserError => e + RubyLLM.logger.debug { "Failed to decode Bedrock stream event payload: #{e.message}" } + nil + end + + def raise_event_stream_exception(exception_type, message) + status = EXCEPTION_STATUSES.fetch(exception_type, 500) + response = ErrorResponse.new({ 'message' => message || exception_type }, status) + ErrorMiddleware.parse_error(provider: self, response: response) + end + end + end + end +end diff --git a/lib/ruby_llm/protocols/converse/streaming.rb b/lib/ruby_llm/protocols/converse/streaming.rb index 431b1d7e4..83868832d 100644 --- a/lib/ruby_llm/protocols/converse/streaming.rb +++ b/lib/ruby_llm/protocols/converse/streaming.rb @@ -1,7 +1,5 @@ # frozen_string_literal: true -require 'base64' -require 'faraday' require 'json' module RubyLLM @@ -9,7 +7,9 @@ module Protocols class Converse # Streaming implementation for Bedrock ConverseStream (AWS Event Stream). module Streaming - ErrorResponse = Struct.new(:body, :status) + include EventStream + + ErrorResponse = EventStream::ErrorResponse private @@ -21,26 +21,9 @@ def stream_response(payload, additional_headers = {}, &block) accumulator = StreamAccumulator.new decoder = event_stream_decoder thinking_state = {} - body = JSON.generate(payload) - - response = @connection.post(stream_url, payload) do |req| - req.headers.merge!(@provider.sign_headers('POST', stream_url, body)) - req.headers.merge!(additional_headers) unless additional_headers.empty? - req.headers['Accept'] = 'application/vnd.amazon.eventstream' - - if Faraday::VERSION.start_with?('1') - req.options[:on_data] = proc do |chunk, _size| - parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block) - end - else - req.options.on_data = proc do |chunk, _bytes, env| - if env&.status == 200 - parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block) - else - handle_failed_stream(chunk, env) - end - end - end + + response = post_event_stream(stream_url, payload, additional_headers) do |chunk| + parse_stream_chunk(decoder, chunk, accumulator, thinking_state, &block) end message = accumulator.to_message(response) @@ -48,23 +31,6 @@ def stream_response(payload, additional_headers = {}, &block) message end - def event_stream_decoder - require 'aws-eventstream' - Aws::EventStream::Decoder.new - rescue LoadError - raise Error, - 'The aws-eventstream gem is required for Bedrock streaming. ' \ - 'Please add it to your Gemfile: gem "aws-eventstream"' - end - - def handle_failed_stream(chunk, env) - data = JSON.parse(chunk) - error_response = env.merge(body: data) - ErrorMiddleware.parse_error(provider: self, response: error_response) - rescue JSON::ParserError - RubyLLM.logger.debug { "Failed Bedrock stream error chunk: #{chunk}" } - end - def parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state) handle_non_eventstream_error_chunk(raw_chunk) @@ -77,48 +43,15 @@ def parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state) end end - def handle_non_eventstream_error_chunk(raw_chunk) - text = raw_chunk.to_s - - if text.start_with?('event: error') - payload = text.lines.find { |line| line.start_with?('data:') }&.delete_prefix('data:')&.strip - raise_streaming_chunk_error(payload) if payload - return - end - - return unless text.lstrip.start_with?('{') && text.include?('"error"') - - raise_streaming_chunk_error(text) - end - - def raise_streaming_chunk_error(payload) - parsed = JSON.parse(payload) - message = parsed.dig('error', 'message') || parsed['message'] || 'Bedrock streaming error' - response = ErrorResponse.new({ 'message' => message }, 500) - ErrorMiddleware.parse_error(provider: self, response: response) - rescue JSON::ParserError - nil - end - def decode_events(decoder, raw_chunk) - events = [] - message, eof = decoder.decode_chunk(raw_chunk) - - while message + decode_event_messages(decoder, raw_chunk).filter_map do |message| event = decode_event_payload(message.payload.read) - event = nest_event_under_type(event, message) if event - if event && RubyLLM.config.log_stream_debug - RubyLLM.logger.debug do - "Bedrock stream event keys: #{event.keys}" - end - end - events << event if event - break if eof - - message, eof = decoder.decode_chunk - end + next unless event - events + event = nest_event_under_type(event, message) + RubyLLM.logger.debug { "Bedrock stream event keys: #{event.keys}" } if RubyLLM.config.log_stream_debug + event + end end # A ConverseStream wire message carries its event type in the eventstream @@ -137,28 +70,6 @@ def nest_event_under_type(event, message) { type => event } end - def event_type_header(message) - headers = message.headers - header = headers[':event-type'] || headers[':exception-type'] - value = header.respond_to?(:value) ? header.value : header - value.is_a?(String) && !value.empty? ? value : nil - rescue StandardError - nil - end - - def decode_event_payload(payload) - outer = JSON.parse(payload) - - if outer['bytes'].is_a?(String) - JSON.parse(Base64.decode64(outer['bytes'])) - else - outer - end - rescue JSON::ParserError => e - RubyLLM.logger.debug { "Failed to decode Bedrock stream event payload: #{e.message}" } - nil - end - def build_chunk(event, thinking_state = {}) raise_stream_error(event) if stream_error_event?(event) @@ -317,18 +228,7 @@ def raise_stream_error(event) end key = event.keys.find { |candidate| candidate.end_with?('Exception') } - payload = event[key] - message = payload['message'] || key - status = case key - when 'throttlingException' then 429 - when 'validationException' then 400 - when 'accessDeniedException', 'unrecognizedClientException' then 401 - when 'serviceUnavailableException' then 503 - else 500 - end - - response = ErrorResponse.new({ 'message' => message }, status) - ErrorMiddleware.parse_error(provider: self, response: response) + raise_event_stream_exception(key, event[key]['message']) end def extract_content_delta(event) diff --git a/lib/ruby_llm/protocols/invoke_anthropic.rb b/lib/ruby_llm/protocols/invoke_anthropic.rb new file mode 100644 index 000000000..ffcf68034 --- /dev/null +++ b/lib/ruby_llm/protocols/invoke_anthropic.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +module RubyLLM + module Protocols + # AWS Bedrock InvokeModel / InvokeModelWithResponseStream for Anthropic Claude models, + # speaking the Anthropic Messages request/response format natively instead of Bedrock's + # Converse translation layer. Reaches Messages-API features Converse doesn't expose + # (anthropic_beta headers-in-body, context_management, cache_control placement) without + # leaving bedrock-runtime or SigV4. + # + # Differences from the first-party Anthropic protocol, all on the envelope: + # - POST /model/{id}/invoke (and /invoke-with-response-stream), SigV4-signed by the provider. + # - The body carries anthropic_version: "bedrock-2023-05-31" and never `model` or `stream` + # (the model is in the path; streaming is chosen by endpoint). + # - The stream is an AWS eventstream whose `chunk` events wrap each Anthropic SSE event + # as base64 `bytes` (see Converse::EventStream), not an SSE body. + # + # Never the Bedrock default: selected per request with `protocol: :invoke_anthropic` + # (RubyLLM::Chat#with_protocol) or config.bedrock_protocol. + class InvokeAnthropic < Anthropic + include InvokeAnthropic::Chat + include InvokeAnthropic::Streaming + + private + + def sync_response(payload, additional_headers = {}) + body = JSON.generate(payload) + response = @connection.post(completion_url, payload) do |req| + req.headers.merge!(@provider.sign_headers('POST', completion_url, body)) + req.headers.merge!(additional_headers) unless additional_headers.empty? + end + parse_completion_response(response) + end + end + end +end diff --git a/lib/ruby_llm/protocols/invoke_anthropic/chat.rb b/lib/ruby_llm/protocols/invoke_anthropic/chat.rb new file mode 100644 index 000000000..9f49a4cd6 --- /dev/null +++ b/lib/ruby_llm/protocols/invoke_anthropic/chat.rb @@ -0,0 +1,333 @@ +# frozen_string_literal: true + +module RubyLLM + module Protocols + class InvokeAnthropic + # Request rendering and response parsing for Bedrock InvokeModel with Anthropic Messages + # bodies. Rendering reuses Anthropic::Chat's payload assembly (tools, tool_choice, system, + # temperature, thinking, output_config) and overrides message formatting so it matches + # Converse's replay guarantees: thinking blocks are always replayed verbatim, consecutive + # same-role rows are folded, and tool results are batched into one user turn. + module Chat + ANTHROPIC_VERSION = 'bedrock-2023-05-31' + DEFAULT_MAX_TOKENS = 4096 + + THINKING_TYPES = %w[thinking redacted_thinking].freeze + + def completion_url + "/model/#{escape_model_id(@model.id)}/invoke" + end + + def escape_model_id(model_id) + Converse::Chat.escape_model_id(model_id) + end + + def render(messages, **) + finalize_invoke_payload(super) + end + + # The one place the InvokeModel envelope is assembled. `model`/`stream` never belong in + # an InvokeModel body; anthropic_version must. Betas from params (`anthropic_beta:`) + # and from invoke_anthropic_betas are merged into a single de-duplicated list, so a + # later caller enabling a beta feature (e.g. context_management) adds the beta here + # and the feature's own block via params, without touching the rest of rendering. + def finalize_invoke_payload(payload) + payload = payload.dup + payload.delete(:model) + payload.delete(:stream) + betas = (Array(payload.delete(:anthropic_beta)) + invoke_anthropic_betas).map(&:to_s).uniq + + envelope = { anthropic_version: ANTHROPIC_VERSION } + envelope[:anthropic_beta] = betas unless betas.empty? + envelope.merge(payload) + end + + def invoke_anthropic_betas + [] + end + + def max_tokens_for(model) + model.max_tokens || DEFAULT_MAX_TOKENS + end + + # InvokeModel does not read the Anthropic Files API, so there is no provider file to + # reference; large attachments are sent inline. + def supports_provider_file_references? + false + end + + def build_system_content(system_messages) + system_messages.flat_map do |msg| + content = msg.content + content.is_a?(RubyLLM::Content::Raw) ? Array(content.value) : format_content(content) + end + end + + def build_base_payload(chat_messages, model, _stream, thinking, citations: false) + @render_citations = citations + payload = { + messages: format_messages(chat_messages), + max_tokens: max_tokens_for(model) + } + + add_thinking_fields(payload, thinking, model) + payload + end + + # Bedrock's model registry rarely carries Anthropic reasoning_options, so don't gate on + # them the way the first-party protocol does; Bedrock validates the request itself. + def build_thinking_payload(thinking, _model) + return nil unless thinking&.enabled? + + effort = resolve_effort(thinking) + return nil if effort == 'none' + + budget = resolve_budget(thinking) + return enabled_thinking_payload(budget) if budget + return adaptive_thinking_payload(effort) if effort + + raise ArgumentError, 'Anthropic adaptive thinking requires an effort' + end + + def format_messages(messages) + rendered = [] + tool_result_blocks = [] + + messages.each do |msg| + if msg.tool_result? + tool_result_blocks.concat(format_tool_result_blocks(msg)) + next + end + + unless tool_result_blocks.empty? + append_message(rendered, { role: 'user', content: tool_result_blocks }) + tool_result_blocks = [] + end + + message = format_non_tool_message(msg) + append_message(rendered, message) if message + end + + append_message(rendered, { role: 'user', content: tool_result_blocks }) unless tool_result_blocks.empty? + rendered + end + + # Anthropic requires strict user/assistant alternation, so a synthesized tool-result + # user turn immediately followed by a real user message is folded into one turn, as + # Converse::Chat#append_message does. + def append_message(rendered, message) + previous = rendered.last + + if previous && previous[:role] == message[:role] + previous[:content] = merge_content_blocks(previous[:content], message[:content]) + else + rendered << message + end + end + + # Thinking blocks first, then tool_result blocks, then everything else, each group in + # its original order: Anthropic requires tool_result blocks to lead a user turn and + # thinking blocks to lead an assistant turn. + def merge_content_blocks(existing_blocks, incoming_blocks) + combined = existing_blocks + incoming_blocks + thinking, rest = combined.partition { |block| thinking_block?(block) } + tool_results, others = rest.partition { |block| block_type(block) == 'tool_result' } + thinking + tool_results + others + end + + def format_non_tool_message(msg) + content = format_message_content(msg) + return nil if content.empty? + + { role: format_role(msg.role), content: content } + end + + def format_role(role) + role == :assistant ? 'assistant' : 'user' + end + + def format_message_content(msg) + if msg.content.is_a?(RubyLLM::Content::Raw) + blocks = Array(msg.content.value) + return blocks if msg.role == :assistant + + return blocks.reject { |block| thinking_block?(block) } + end + + blocks = [] + blocks.concat(format_thinking_blocks(msg.thinking) || []) if msg.role == :assistant + blocks.concat(format_message_body(msg.content)) + blocks.concat(format_tool_use_blocks(msg)) if msg.tool_call? + blocks + end + + def format_message_body(content) + return [] if content.nil? || (content.respond_to?(:empty?) && content.empty?) + + Array(format_content(content, citations: @render_citations)).reject do |block| + block.is_a?(Hash) && block[:type] == 'text' && block[:text].to_s.empty? + end + end + + def format_tool_use_blocks(msg) + msg.tool_calls.values.map do |tool_call| + { type: 'tool_use', id: tool_call.id, name: tool_call.name, input: tool_call.arguments || {} } + end + end + + # Anthropic rejects a request whose latest assistant turn doesn't replay every + # thinking/redacted_thinking block exactly as received, so blocks are replayed whenever + # present (not only when thinking is enabled on this request, unlike the first-party + # protocol). Blocks captured by the Converse protocol are translated to the Messages + # shape so a chat can be served by either endpoint. + def format_thinking_blocks(thinking) + return nil unless thinking + return thinking.blocks.filter_map { |block| anthropic_thinking_block(block) } if thinking.blocks + return nil unless thinking.text || thinking.signature + + [{ type: 'thinking', thinking: thinking.text || '', signature: thinking.signature }.compact] + end + + def anthropic_thinking_block(block) + return nil unless block.is_a?(Hash) + + reasoning = indifferent_fetch(block, :reasoningContent) + return block unless reasoning.is_a?(Hash) + + redacted = indifferent_fetch(reasoning, :redactedContent) + return { type: 'redacted_thinking', data: redacted } if redacted + + text_block = indifferent_fetch(reasoning, :reasoningText) || {} + { + type: 'thinking', + thinking: indifferent_fetch(text_block, :text) || '', + signature: indifferent_fetch(text_block, :signature) + }.compact + end + + def indifferent_fetch(hash, key) + hash[key.to_s] || hash[key] + end + + def format_tool_result_blocks(msg) + return Array(msg.content.value) if msg.content.is_a?(RubyLLM::Content::Raw) && raw_tool_result_turn?(msg) + + [format_tool_result_block(msg)] + end + + # A Raw tool-result whose blocks are already complete tool_result blocks (the shape the + # first-party protocol's format_tool_result passes through) is used as-is. + def raw_tool_result_turn?(msg) + blocks = Array(msg.content.value) + blocks.any? && blocks.all? { |block| block_type(block) == 'tool_result' } + end + + def format_tool_result_block(msg) + { + type: 'tool_result', + tool_use_id: msg.tool_call_id, + content: format_tool_result_content(msg.content) + } + end + + def format_tool_result_content(content) + return format_raw_tool_result_content(content.value) if content.is_a?(RubyLLM::Content::Raw) + return content.results.map { |r| Anthropic::Tools.search_result_block(r) } if search_results?(content) + return [{ type: 'text', text: '(no output)' }] if blank_content?(content) + + blocks = format_message_body(content) + blocks.empty? ? [{ type: 'text', text: '(no output)' }] : blocks + end + + def format_raw_tool_result_content(raw_value) + blocks = Array(raw_value).grep(Hash) + blocks.empty? ? [{ type: 'text', text: raw_value.to_s }] : blocks + end + + def search_results?(content) + content.is_a?(RubyLLM::SearchResults) + end + + def blank_content?(content) + content.nil? || (content.respond_to?(:empty?) && content.empty?) + end + + # Images and PDFs always travel base64-inline: InvokeModel has no URL fetch and no + # Files API. Everything else matches Anthropic::Media. + def format_content(content, citations: false) + return Array(content.value) if content.is_a?(RubyLLM::Content::Raw) + return [Anthropic::Media.format_text(content.to_json)] if content.is_a?(Hash) || content.is_a?(Array) + return [Anthropic::Media.format_text(content)] unless content.is_a?(RubyLLM::Content) + + parts = [] + parts << Anthropic::Media.format_text(content.text) if content.text + content.attachments.each { |attachment| parts << format_attachment(attachment, citations:) } + parts + end + + def format_attachment(attachment, citations:) + raise UnsupportedAttachmentError, attachment.mime_type if attachment.provider_file? + + case attachment.type + when :image then { type: 'image', source: base64_source(attachment) } + when :pdf then format_inline_pdf(attachment, citations:) + when :text + citations ? Anthropic::Media.format_text_document(attachment) : Anthropic::Media.format_text_file(attachment) + else + raise UnsupportedAttachmentError, attachment.mime_type + end + end + + def format_inline_pdf(pdf, citations:) + document = { type: 'document', source: base64_source(pdf) } + Anthropic::Media.enable_citations(document, pdf) if citations + document + end + + def base64_source(attachment) + { type: 'base64', media_type: attachment.mime_type, data: attachment.encoded } + end + + # Same as Anthropic::Chat#build_message plus thinking.blocks: the raw thinking / + # redacted_thinking blocks, which must be replayed verbatim on the next turn. + def build_message(data, content, citations, thinking, thinking_signature, tool_use_blocks, raw) # rubocop:disable Metrics/ParameterLists + usage = data['usage'] || {} + blocks = raw_thinking_blocks(data['content'] || []) + + Message.new( + role: :assistant, + content: content, + citations: citations, + thinking: Thinking.build(text: thinking, signature: thinking_signature, blocks: blocks), + tool_calls: Anthropic::Tools.parse_tool_calls(tool_use_blocks), + input_tokens: usage['input_tokens'], + output_tokens: usage['output_tokens'], + cached_tokens: extract_cached_tokens(data), + cache_creation_tokens: extract_cache_creation_tokens(data), + thinking_tokens: usage.dig('output_tokens_details', 'thinking_tokens'), + finish_reason: data['stop_reason'], + model_id: data['model'], + raw: raw + ) + end + + def raw_thinking_blocks(content_blocks) + blocks = content_blocks.select { |block| THINKING_TYPES.include?(block['type']) } + blocks.empty? ? nil : blocks + end + + def thinking_block?(block) + return false unless block.is_a?(Hash) + + THINKING_TYPES.include?(block_type(block)) || block.key?(:reasoningContent) || block.key?('reasoningContent') + end + + def block_type(block) + return nil unless block.is_a?(Hash) + + (block[:type] || block['type'])&.to_s + end + end + end + end +end diff --git a/lib/ruby_llm/protocols/invoke_anthropic/streaming.rb b/lib/ruby_llm/protocols/invoke_anthropic/streaming.rb new file mode 100644 index 000000000..acfc768c8 --- /dev/null +++ b/lib/ruby_llm/protocols/invoke_anthropic/streaming.rb @@ -0,0 +1,168 @@ +# frozen_string_literal: true + +require 'json' + +module RubyLLM + module Protocols + class InvokeAnthropic + # Streaming for Bedrock InvokeModelWithResponseStream. The wire format is an AWS + # eventstream (decoded by Converse::EventStream) whose `chunk` events each carry one + # Anthropic Messages streaming event as base64 `bytes`; the decoded event is then + # interpreted exactly as the first-party Anthropic SSE stream would be, plus raw thinking + # block capture for verbatim replay. + module Streaming + include Converse::EventStream + + private + + def stream_url + "/model/#{escape_model_id(@model.id)}/invoke-with-response-stream" + end + + def stream_response(payload, additional_headers = {}, &block) + accumulator = StreamAccumulator.new + decoder = event_stream_decoder + thinking_state = {} + + response = post_event_stream(stream_url, payload, additional_headers) do |raw_chunk| + parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state, &block) + end + + message = accumulator.to_message(response) + RubyLLM.logger.debug { "Stream completed: #{message.content}" } + message + end + + def parse_stream_chunk(decoder, raw_chunk, accumulator, thinking_state) + handle_non_eventstream_error_chunk(raw_chunk) + + decode_event_messages(decoder, raw_chunk).each do |message| + data = decode_invoke_event(message) + next unless data + + chunk = build_chunk(data, thinking_state) + accumulator.add(chunk) + yield chunk + end + end + + # Returns the Anthropic event carried by one eventstream message, or raises for a + # Bedrock exception message or an Anthropic `error` event. + def decode_invoke_event(message) + payload = message.payload.read + exception_type = exception_type_header(message) + if exception_type + parsed = decode_event_payload(payload) || {} + raise_event_stream_exception(exception_type, parsed['message'] || parsed['Message']) + end + + data = decode_event_payload(payload) + return nil unless data.is_a?(Hash) + + RubyLLM.logger.debug { "Bedrock invoke stream event: #{data['type']}" } if RubyLLM.config.log_stream_debug + raise_anthropic_stream_error(data) if data['type'] == 'error' + data + end + + def raise_anthropic_stream_error(data) + status = data.dig('error', 'type') == 'overloaded_error' ? 529 : 500 + message = data.dig('error', 'message') || 'Bedrock streaming error' + response = Converse::EventStream::ErrorResponse.new({ 'message' => message }, status) + ErrorMiddleware.parse_error(provider: self, response: response) + end + + def build_chunk(data, thinking_state = {}) + delta_type = data.dig('delta', 'type') + track_thinking_block(data, delta_type, thinking_state) + + Chunk.new( + role: :assistant, + model_id: extract_model_id(data), + content: extract_content_delta(data, delta_type), + citations: extract_citations_delta(data, delta_type), + thinking: Thinking.build( + text: extract_thinking_delta(data, delta_type), + signature: extract_signature_delta(data, delta_type), + blocks: extract_finalized_thinking_blocks(data, thinking_state) + ), + input_tokens: extract_input_tokens(data), + output_tokens: extract_output_tokens(data), + cached_tokens: extract_cached_tokens(data), + cache_creation_tokens: extract_cache_creation_tokens(data), + tool_calls: extract_tool_calls(data), + finish_reason: data.dig('delta', 'stop_reason') + ) + end + + # Thinking blocks arrive as content_block_start (type thinking / redacted_thinking, + # the latter carrying its opaque `data` whole), then thinking_delta / signature_delta + # events for the same index, then content_block_stop. State is keyed by index and + # finalized into the exact block shape the non-streaming response returns. + def track_thinking_block(data, delta_type, thinking_state) + index = data['index'] + return if index.nil? + + if data['type'] == 'content_block_start' + start_thinking_state(data['content_block'] || {}, index, thinking_state) + return + end + + state = thinking_state[index] + return unless state && data['type'] == 'content_block_delta' + + case delta_type + when 'thinking_delta' then state[:text] << data.dig('delta', 'thinking').to_s + when 'signature_delta' + state[:signature] = stream_presence(data.dig('delta', 'signature')) || state[:signature] + end + end + + def start_thinking_state(block, index, thinking_state) + case block['type'] + when 'thinking' + thinking_state[index] = { + text: +block['thinking'].to_s, signature: stream_presence(block['signature']), redacted: nil + } + when 'redacted_thinking' + thinking_state[index] = { text: +'', signature: nil, redacted: block['data'] } + end + end + + # A block finalizes on its own content_block_stop. As in Converse::Streaming, message_stop + # also finalizes any still-open block that is structurally complete (signed or + # redacted); a half-formed block is dropped because replaying it would be rejected. + def extract_finalized_thinking_blocks(data, thinking_state) + case data['type'] + when 'content_block_stop' + state = thinking_state.delete(data['index']) + block = state && finalize_thinking_block(state) + block ? [block] : nil + when 'message_stop' + finalize_remaining_thinking_blocks(thinking_state) + end + end + + def finalize_remaining_thinking_blocks(thinking_state) + blocks = thinking_state.keys.sort.filter_map do |index| + state = thinking_state.delete(index) + next unless state[:redacted] || state[:signature] + + finalize_thinking_block(state) + end + blocks.empty? ? nil : blocks + end + + def finalize_thinking_block(state) + return { 'type' => 'redacted_thinking', 'data' => state[:redacted] } if state[:redacted] + return nil if state[:text].empty? && state[:signature].nil? + + { 'type' => 'thinking', 'thinking' => state[:text], 'signature' => state[:signature] }.compact + end + + def stream_presence(value) + value.is_a?(String) && !value.empty? ? value : nil + end + end + end + end +end diff --git a/lib/ruby_llm/providers/bedrock.rb b/lib/ruby_llm/providers/bedrock.rb index 3b7aa9049..2e4985031 100644 --- a/lib/ruby_llm/providers/bedrock.rb +++ b/lib/ruby_llm/providers/bedrock.rb @@ -9,6 +9,9 @@ class Bedrock < Provider protocol :converse, Protocols::Converse, batches: Protocols::Converse::Batches protocol :mantle_responses, Protocols::MantleResponses + # Opt-in only (protocol: :invoke_anthropic / config.bedrock_protocol); protocol_for never + # picks it, so Converse stays the default for Claude. + protocol :invoke_anthropic, Protocols::InvokeAnthropic files Bedrock::Files # SigV4 requests to bedrock-mantle sign against this service namespace, not "bedrock" — @@ -50,7 +53,13 @@ def headers end def complete(messages, model:, params: {}, **rest, &) - params = mantle_only_model?(model) ? strip_converse_only_params(params) : normalize_params(params, model:) + params = if mantle_only_model?(model) + strip_converse_only_params(params) + elsif invoke_anthropic_request?(rest[:protocol]) + normalize_invoke_params(params) + else + normalize_params(params, model:) + end # Bare `super` forwards current bindings, so it picks up the reassigned `params` above. super end @@ -165,6 +174,24 @@ def normalize_params(params, model:) normalized end + def invoke_anthropic_request?(protocol) + (protocol || configured_protocol)&.to_sym == :invoke_anthropic + end + + # Callers build params for Converse, where Anthropic-native request fields (thinking, + # output_config, top_k, anthropic_beta, context_management, ...) ride inside + # additionalModelRequestFields. InvokeModel takes the Messages body directly, so those + # fields are lifted to the top level. `model` and `stream` are never valid in an + # InvokeModel body. + def normalize_invoke_params(params) + normalized = RubyLLM::Utils.deep_symbolize_keys(params || {}) + additional_fields = normalized.delete(:additionalModelRequestFields) || {} + normalized = RubyLLM::Utils.deep_merge(normalized, additional_fields) + normalized.delete(:model) + normalized.delete(:stream) + normalized + end + def model_supports_top_k?(model) Protocols::Converse.reasoning_embedded?(model) end diff --git a/spec/ruby_llm/protocols/invoke_anthropic_spec.rb b/spec/ruby_llm/protocols/invoke_anthropic_spec.rb new file mode 100644 index 000000000..023ff7fbe --- /dev/null +++ b/spec/ruby_llm/protocols/invoke_anthropic_spec.rb @@ -0,0 +1,358 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'aws-eventstream' + +RSpec.describe RubyLLM::Protocols::InvokeAnthropic do + let(:config) do + RubyLLM::Configuration.new.tap do |c| + c.bedrock_region = 'us-west-2' + c.bedrock_api_key = 'static-key' + c.bedrock_secret_key = 'static-secret' + end + end + let(:provider) { RubyLLM::Providers::Bedrock.new(config) } + let(:model_id) { 'arn:aws:bedrock:us-west-2:123456789012:application-inference-profile/abc123' } + let(:model) { model_info(model_id) } + let(:protocol) { described_class.new(provider, model) } + + def base + 'https://bedrock-runtime.us-west-2.amazonaws.com' + end + + def escaped_path + "/model/#{model_id.gsub('/', '%2F')}" + end + + def model_info(id, max_output_tokens: 8192) + RubyLLM::Model::Info.new( + id: id, name: id, provider: 'bedrock', family: 'claude', created_at: nil, + context_window: 200_000, max_output_tokens: max_output_tokens, + modalities: { input: [], output: [] }, capabilities: [], pricing: {}, metadata: {} + ) + end + + def user(text) + RubyLLM::Message.new(role: :user, content: text) + end + + def signed?(req) + req.headers['Authorization']&.include?('us-west-2/bedrock/aws4_request') + end + + def json_response(body) + { status: 200, body: JSON.generate(body), headers: { 'Content-Type' => 'application/json' } } + end + + def render(messages, **opts) + protocol.render(messages, tools: opts.fetch(:tools, {}), temperature: nil, params: opts.fetch(:params, {}), + thinking: opts[:thinking], stream: opts.fetch(:stream, false)) + end + + describe 'URLs' do + it 'targets /invoke and /invoke-with-response-stream with the model id as one escaped path segment' do + expect(protocol.completion_url).to eq("#{escaped_path}/invoke") + expect(protocol.send(:stream_url)).to eq("#{escaped_path}/invoke-with-response-stream") + end + end + + describe 'request body' do + it 'carries anthropic_version and max_tokens, and never model or stream' do + payload = render([user('hi')], stream: true) + + expect(payload[:anthropic_version]).to eq('bedrock-2023-05-31') + expect(payload).not_to have_key(:model) + expect(payload).not_to have_key(:stream) + expect(payload[:max_tokens]).to eq(8192) + expect(payload[:messages]).to eq([{ role: 'user', content: [{ type: 'text', text: 'hi' }] }]) + end + + it 'merges anthropic_beta from params into one de-duplicated list' do + payload = render([user('hi')], params: { anthropic_beta: %w[a b a] }) + + expect(payload[:anthropic_beta]).to eq(%w[a b]) + end + + it 'omits anthropic_beta when no betas are requested' do + expect(render([user('hi')])).not_to have_key(:anthropic_beta) + end + + it 'formats tools, tool_use and a folded tool_result + user turn' do + tool = Class.new(RubyLLM::Tool) do + def self.name = 'Weather' + description 'Get weather' + param :city, desc: 'City' + end.new + messages = [ + user('weather?'), + RubyLLM::Message.new(role: :assistant, content: nil, + tool_calls: { 't1' => RubyLLM::ToolCall.new(id: 't1', name: 'weather', + arguments: { 'city' => 'SF' }) }), + RubyLLM::Message.new(role: :tool, content: 'sunny', tool_call_id: 't1'), + user('thanks') + ] + + payload = render(messages, tools: { weather: tool }) + + expect(payload[:tools].first).to include(name: 'weather', description: 'Get weather') + expect(payload[:messages][1]).to eq( + role: 'assistant', + content: [{ type: 'tool_use', id: 't1', name: 'weather', input: { 'city' => 'SF' } }] + ) + expect(payload[:messages][2]).to eq( + role: 'user', + content: [ + { type: 'tool_result', tool_use_id: 't1', content: [{ type: 'text', text: 'sunny' }] }, + { type: 'text', text: 'thanks' } + ] + ) + expect(payload[:messages].size).to eq(3) + end + + it 'sends images base64-inline' do + content = RubyLLM::Content.new('look', [File.expand_path('../../fixtures/ruby.png', __dir__)]) + + block = render([user(content)])[:messages].first[:content].last + + expect(block[:type]).to eq('image') + expect(block[:source]).to include(type: 'base64', media_type: 'image/png') + expect(block[:source][:data]).not_to be_empty + end + + it 'replays Anthropic thinking blocks verbatim, before tool_use, even without thinking enabled' do + blocks = [{ 'type' => 'thinking', 'thinking' => 'hmm', 'signature' => 'sig' }, + { 'type' => 'redacted_thinking', 'data' => 'opaque' }] + assistant = RubyLLM::Message.new( + role: :assistant, content: 'ok', + thinking: RubyLLM::Thinking.build(text: 'hmm', signature: 'sig', blocks: blocks) + ) + + content = render([user('q'), assistant, user('next')])[:messages][1][:content] + + expect(content).to eq(blocks + [{ type: 'text', text: 'ok' }]) + end + + it 'translates Converse-captured reasoningContent blocks to the Messages shape' do + blocks = [{ 'reasoningContent' => { 'reasoningText' => { 'text' => 'hmm', 'signature' => 'sig' } } }, + { 'reasoningContent' => { 'redactedContent' => 'opaque' } }] + assistant = RubyLLM::Message.new(role: :assistant, content: 'ok', + thinking: RubyLLM::Thinking.build(blocks: blocks)) + + content = render([user('q'), assistant])[:messages][1][:content] + + expect(content.first(2)).to eq( + [{ type: 'thinking', thinking: 'hmm', signature: 'sig' }, { type: 'redacted_thinking', data: 'opaque' }] + ) + end + + it 'lifts nothing on its own: thinking/output_config arrive via params' do + payload = render([user('q')], params: { thinking: { type: 'adaptive' }, output_config: { effort: 'high' } }) + + expect(payload).to include(thinking: { type: 'adaptive' }, output_config: { effort: 'high' }) + end + end + + describe 'sync completion' do + it 'posts a SigV4-signed body and parses text, thinking blocks, tool calls and cache usage' do + response = { + 'id' => 'msg_1', 'type' => 'message', 'role' => 'assistant', 'model' => 'claude-sonnet-5', + 'content' => [ + { 'type' => 'thinking', 'thinking' => 'let me look', 'signature' => 'sig-1' }, + { 'type' => 'redacted_thinking', 'data' => 'enc' }, + { 'type' => 'text', 'text' => 'Checking.' }, + { 'type' => 'tool_use', 'id' => 'toolu_1', 'name' => 'weather', 'input' => { 'city' => 'SF' } } + ], + 'stop_reason' => 'tool_use', + 'usage' => { 'input_tokens' => 12, 'output_tokens' => 34, 'cache_read_input_tokens' => 1000, + 'cache_creation_input_tokens' => 200 } + } + stub = stub_request(:post, "#{base}#{escaped_path}/invoke") + .with do |req| + body = JSON.parse(req.body) + signed?(req) && body['anthropic_version'] == 'bedrock-2023-05-31' && !body.key?('model') + end + .to_return(json_response(response)) + + message = protocol.complete([user('weather?')], tools: {}, temperature: nil) + + expect(stub).to have_been_requested + expect(message.content).to eq('Checking.') + expect(message.thinking.text).to eq('let me look') + expect(message.thinking.signature).to eq('sig-1') + expect(message.thinking.blocks).to eq(response['content'].first(2)) + expect(message.tool_calls['toolu_1'].arguments).to eq('city' => 'SF') + expect(message.finish_reason).to eq('tool_use') + expect(message.input_tokens).to eq(12) + expect(message.output_tokens).to eq(34) + expect(message.cached_tokens).to eq(1000) + expect(message.cache_creation_tokens).to eq(200) + end + + it 'maps a Bedrock throttling response to RateLimitError' do + stub_request(:post, "#{base}#{escaped_path}/invoke") + .to_return(status: 429, body: JSON.generate('message' => 'Too many requests'), + headers: { 'Content-Type' => 'application/json' }) + + expect { protocol.complete([user('hi')], tools: {}, temperature: nil) } + .to raise_error(RubyLLM::RateLimitError, /Too many requests/) + end + end + + describe 'streaming' do + def event(payload) + Aws::EventStream::Message.new( + headers: { + ':event-type' => Aws::EventStream::HeaderValue.new(value: 'chunk', type: 'string'), + ':content-type' => Aws::EventStream::HeaderValue.new(value: 'application/json', type: 'string'), + ':message-type' => Aws::EventStream::HeaderValue.new(value: 'event', type: 'string') + }, + payload: StringIO.new(JSON.generate('bytes' => Base64.strict_encode64(JSON.generate(payload)))) + ) + end + + def exception(type, message) + Aws::EventStream::Message.new( + headers: { + ':exception-type' => Aws::EventStream::HeaderValue.new(value: type, type: 'string'), + ':content-type' => Aws::EventStream::HeaderValue.new(value: 'application/json', type: 'string'), + ':message-type' => Aws::EventStream::HeaderValue.new(value: 'exception', type: 'string') + }, + payload: StringIO.new(JSON.generate('message' => message)) + ) + end + + def wire(*messages) + encoder = Aws::EventStream::Encoder.new + messages.map { |message| encoder.encode(message) }.join + end + + def stub_stream(body) + stub_request(:post, "#{base}#{escaped_path}/invoke-with-response-stream") + .with { |req| signed?(req) && !JSON.parse(req.body).key?('stream') } + .to_return(status: 200, body: body, headers: { 'Content-Type' => 'application/vnd.amazon.eventstream' }) + end + + def stream(messages = [user('hi')]) + chunks = [] + message = protocol.complete(messages, tools: {}, temperature: nil) { |chunk| chunks << chunk } + [message, chunks] + end + + it 'decodes base64 chunk events into text, thinking blocks, tool calls and usage' do + stub_stream(wire( + event('type' => 'message_start', + 'message' => { 'model' => 'claude-sonnet-5', + 'usage' => { 'input_tokens' => 7, 'output_tokens' => 1, + 'cache_read_input_tokens' => 900, + 'cache_creation_input_tokens' => 100 } }), + event('type' => 'content_block_start', 'index' => 0, + 'content_block' => { 'type' => 'thinking', 'thinking' => '' }), + event('type' => 'content_block_delta', 'index' => 0, + 'delta' => { 'type' => 'thinking_delta', 'thinking' => 'plan ' }), + event('type' => 'content_block_delta', 'index' => 0, + 'delta' => { 'type' => 'thinking_delta', 'thinking' => 'it' }), + event('type' => 'content_block_delta', 'index' => 0, + 'delta' => { 'type' => 'signature_delta', 'signature' => 'sig-1' }), + event('type' => 'content_block_stop', 'index' => 0), + event('type' => 'content_block_start', 'index' => 1, + 'content_block' => { 'type' => 'redacted_thinking', 'data' => 'enc' }), + event('type' => 'content_block_stop', 'index' => 1), + event('type' => 'content_block_start', 'index' => 2, + 'content_block' => { 'type' => 'text', 'text' => '' }), + event('type' => 'content_block_delta', 'index' => 2, + 'delta' => { 'type' => 'text_delta', 'text' => 'Hel' }), + event('type' => 'content_block_delta', 'index' => 2, + 'delta' => { 'type' => 'text_delta', 'text' => 'lo' }), + event('type' => 'content_block_stop', 'index' => 2), + event('type' => 'content_block_start', 'index' => 3, + 'content_block' => { 'type' => 'tool_use', 'id' => 'toolu_1', 'name' => 'weather', + 'input' => {} }), + event('type' => 'content_block_delta', 'index' => 3, + 'delta' => { 'type' => 'input_json_delta', 'partial_json' => '{"city":' }), + event('type' => 'content_block_delta', 'index' => 3, + 'delta' => { 'type' => 'input_json_delta', 'partial_json' => '"SF"}' }), + event('type' => 'content_block_stop', 'index' => 3), + event('type' => 'message_delta', 'delta' => { 'stop_reason' => 'tool_use' }, + 'usage' => { 'output_tokens' => 42 }), + event('type' => 'message_stop', + 'amazon-bedrock-invocationMetrics' => { 'inputTokenCount' => 7 }) + )) + + message, chunks = stream + + expect(chunks).not_to be_empty + expect(message.content).to eq('Hello') + expect(message.model_id).to eq('claude-sonnet-5') + expect(message.thinking.text).to eq('plan it') + expect(message.thinking.signature).to eq('sig-1') + expect(message.thinking.blocks).to eq( + [{ 'type' => 'thinking', 'thinking' => 'plan it', 'signature' => 'sig-1' }, + { 'type' => 'redacted_thinking', 'data' => 'enc' }] + ) + expect(message.tool_calls['toolu_1'].name).to eq('weather') + expect(message.tool_calls['toolu_1'].arguments).to eq('city' => 'SF') + expect(message.finish_reason).to eq('tool_use') + expect(message.input_tokens).to eq(7) + expect(message.output_tokens).to eq(42) + expect(message.cached_tokens).to eq(900) + expect(message.cache_creation_tokens).to eq(100) + end + + it 'keeps a signed thinking block left open at message_stop and drops an unsigned one' do + stub_stream(wire( + event('type' => 'message_start', 'message' => { 'usage' => { 'input_tokens' => 1 } }), + event('type' => 'content_block_start', 'index' => 0, + 'content_block' => { 'type' => 'thinking', 'thinking' => '' }), + event('type' => 'content_block_delta', 'index' => 0, + 'delta' => { 'type' => 'signature_delta', 'signature' => 'sig-open' }), + event('type' => 'content_block_start', 'index' => 1, + 'content_block' => { 'type' => 'thinking', 'thinking' => '' }), + event('type' => 'content_block_delta', 'index' => 1, + 'delta' => { 'type' => 'thinking_delta', 'thinking' => 'half' }), + event('type' => 'message_stop') + )) + + message, = stream + + expect(message.thinking.blocks).to eq([{ 'type' => 'thinking', 'thinking' => '', 'signature' => 'sig-open' }]) + end + + it 'raises RateLimitError for a throttlingException event' do + stub_stream(wire( + event('type' => 'message_start', 'message' => { 'usage' => { 'input_tokens' => 1 } }), + exception('throttlingException', 'Rate exceeded') + )) + + expect { stream }.to raise_error(RubyLLM::RateLimitError, /Rate exceeded/) + end + + it 'raises BadRequestError for a validationException event' do + stub_stream(wire(exception('validationException', 'bad thinking block'))) + + expect { stream }.to raise_error(RubyLLM::BadRequestError, /bad thinking block/) + end + + it 'raises OverloadedError for an Anthropic overloaded_error event' do + stub_stream(wire(event('type' => 'error', + 'error' => { 'type' => 'overloaded_error', 'message' => 'Overloaded' }))) + + expect { stream }.to raise_error(RubyLLM::OverloadedError, /Overloaded/) + end + end + + describe 'Converse replay of InvokeModel-captured thinking' do + it 'translates Anthropic thinking blocks to reasoningContent' do + blocks = [{ 'type' => 'thinking', 'thinking' => 'hmm', 'signature' => 'sig' }, + { 'type' => 'redacted_thinking', 'data' => 'opaque' }] + assistant = RubyLLM::Message.new(role: :assistant, content: 'ok', + thinking: RubyLLM::Thinking.build(blocks: blocks)) + + content = RubyLLM::Protocols::Converse::Chat.format_messages([assistant]).first[:content] + + expect(content.first(2)).to eq( + [{ reasoningContent: { reasoningText: { text: 'hmm', signature: 'sig' } } }, + { reasoningContent: { redactedContent: 'opaque' } }] + ) + end + end +end diff --git a/spec/ruby_llm/providers/bedrock_spec.rb b/spec/ruby_llm/providers/bedrock_spec.rb index 31ed1d0c4..72b818baf 100644 --- a/spec/ruby_llm/providers/bedrock_spec.rb +++ b/spec/ruby_llm/providers/bedrock_spec.rb @@ -282,6 +282,41 @@ def model_double(id, metadata: {}) end end + describe '#complete invoke_anthropic params normalization' do + let(:provider) { described_class.new(bedrock_config(api_key: 'k', secret_key: 's')) } + let(:model) do + instance_double(RubyLLM::Model::Info, id: 'us.anthropic.claude-sonnet-5', max_tokens: 4096, metadata: {}, + provider: 'bedrock') + end + let(:protocol) { instance_double(RubyLLM::Protocols::InvokeAnthropic) } + + before do + allow(RubyLLM::Protocols::InvokeAnthropic).to receive(:new).and_return(protocol) + allow(protocol).to receive(:complete) + end + + it 'lifts additionalModelRequestFields to the top level and drops model/stream' do + params = { additionalModelRequestFields: { thinking: { type: 'adaptive' }, output_config: { effort: 'high' } }, + top_k: 5, model: 'x', stream: true } + + provider.complete([], model: model, tools: {}, temperature: nil, params: params, protocol: :invoke_anthropic) + + expect(protocol).to have_received(:complete).with( + [], hash_including(params: { thinking: { type: 'adaptive' }, output_config: { effort: 'high' }, top_k: 5 }), + any_args + ) + end + + it 'is selectable by config.bedrock_protocol too' do + config = bedrock_config(api_key: 'k', secret_key: 's').tap { |c| c.bedrock_protocol = :invoke_anthropic } + + described_class.new(config).complete([], model: model, tools: {}, temperature: nil, + params: { additionalModelRequestFields: { top_k: 3 } }) + + expect(protocol).to have_received(:complete).with([], hash_including(params: { top_k: 3 }), any_args) + end + end + describe '#protocol_for' do def build_bedrock described_class.new(bedrock_config(api_key: 'k', secret_key: 's'))