Class: LittleGhost::Providers::Bedrock::StreamNormalizer

Inherits:

Object

• Object
• LittleGhost::Providers::Bedrock::StreamNormalizer

Defined in:

lib/little_ghost/providers/bedrock.rb

Overview

:nodoc:

Instance Method Summary

• #consume(event) ⇒ Object
• #finish ⇒ Object
• #initialize(model:) ⇒ StreamNormalizer constructor

  A new instance of StreamNormalizer.

Constructor Details

#initialize(model:) ⇒ StreamNormalizer

Returns a new instance of StreamNormalizer.

# File 'lib/little_ghost/providers/bedrock.rb', line 444

def initialize(model:)
  @model = model
  @message_id = nil
  @text = +""
  @reasoning_blocks = {}
  @tool_calls = {}
  @usage = Usage.new
  @stop_reason = nil
  @finished = false
end

Instance Method Details

#consume(event) ⇒ Object

# File 'lib/little_ghost/providers/bedrock.rb', line 455

def consume(event)
  if event["event_type"]
    type = event["event_type"].to_s
    payload = event.except("event_type")
  else
    type, payload = event.first
  end
  case type
  when "message_start"
    @message_id = payload["role"]
    [StreamEvent.build(:message_start, id: nil, model: @model)]
  when "content_block_start"
    content_start(payload)
  when "content_block_delta"
    content_delta(payload)
  when "content_block_stop"
    content_stop(payload)
  when "message_stop"
    @stop_reason = normalize_stop(payload["stop_reason"])
    @terminal = true
    []
  when "metadata"
    metadata(payload)
  when *TRANSIENT_STREAM_ERRORS, "validation_exception"
    message = payload.is_a?(Hash) ? payload["message"].to_s : ""
    message = "Bedrock returned #{type}" if message.empty?
    raise StreamError.new(message, event_type: type)
  else
    []
  end
end

#finish ⇒ Object

# File 'lib/little_ghost/providers/bedrock.rb', line 487

def finish
  return [] if @finished
  raise ProtocolError, "Bedrock stream ended before message_stop" unless @terminal

  @finished = true
  if @stop_reason == :malformed_tool_use
    raise MalformedToolCallError, "Bedrock returned malformed_tool_use"
  end

  blocks = []
  @reasoning_blocks.sort.each do |_index, reasoning|
    if !reasoning[:redacted_content].empty?
      blocks << Content::Reasoning.new(redacted_content: reasoning[:redacted_content])
    elsif !reasoning[:text].empty?
      signature = reasoning[:signature]
      blocks << Content::Reasoning.new(
        text: reasoning[:text],
        signature: signature.empty? ? nil : signature
      )
    end
  end
  blocks << Content::Text.new(text: @text) unless @text.empty?
  @tool_calls.sort.each do |_index, tool|
    input = tool[:arguments].empty? ? {} : JSON.parse(tool[:arguments])
    blocks << Content::ToolUse.new(id: tool[:id], name: tool[:name], input:)
  end
  response = ModelResponse.new(
    message: Message.new(role: :assistant, content: blocks),
    stop_reason: @stop_reason || (@tool_calls.empty? ? :end_turn : :tool_use),
    usage: @usage,
    metadata: {model: @model}
  )
  [StreamEvent.build(:message_stop, response:)]
rescue JSON::ParserError, ArgumentError => error
  raise MalformedToolCallError, "Bedrock returned an invalid tool call: #{error.message}"
end