Class: LanguageOperator::Agent::StreamingBody

Inherits:
Object
  • Object
show all
Defined in:
lib/language_operator/agent/web_server.rb

Overview

Streaming body for Server-Sent Events (SSE)

Implements the Rack streaming protocol for chat completion responses. Streams agent output as it's generated.

Defined Under Namespace

Classes: MockStream

Instance Method Summary collapse

Constructor Details

#initialize(agent, prompt, model_name) ⇒ StreamingBody

Returns a new instance of StreamingBody.



1167
1168
1169
1170
1171
1172
 
# File 'lib/language_operator/agent/web_server.rb', line 1167

def initialize(agent, prompt, model_name)
  @agent = agent
  @prompt = prompt
  @model_name = model_name
  @id = "chatcmpl-#{SecureRandom.hex(12)}"
end

Instance Method Details

#call(stream) ⇒ void

This method returns an undefined value.

Called by Rack to stream the response

Parameters:

  • The stream object



1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
 
# File 'lib/language_operator/agent/web_server.rb', line 1225

def call(stream)
  # Execute agent and stream response
  result = @agent.execute(@prompt)

  # Send the result as a single chunk (for simplicity)
  # In a real implementation, this could stream token-by-token
  chunk = {
    id: @id,
    object: 'chat.completion.chunk',
    created: Time.now.to_i,
    model: @model_name,
    choices: [
      {
        index: 0,
        delta: {
          role: 'assistant',
          content: result
        },
        finish_reason: nil
      }
    ]
  }

  stream.write("data: #{JSON.generate(chunk)}\n\n")

  # Send final chunk with finish_reason
  final_chunk = {
    id: @id,
    object: 'chat.completion.chunk',
    created: Time.now.to_i,
    model: @model_name,
    choices: [
      {
        index: 0,
        delta: {},
        finish_reason: 'stop'
      }
    ]
  }

  stream.write("data: #{JSON.generate(final_chunk)}\n\n")
  stream.write("data: [DONE]\n\n")
rescue StandardError => e
  error_chunk = {
    error: {
      message: e.message,
      type: 'server_error',
      code: nil
    }
  }
  stream.write("data: #{JSON.generate(error_chunk)}\n\n")
ensure
  stream.close
end

#each {|String| ... } ⇒ void

This method returns an undefined value.

Implement each for Rack::Test compatibility

Yields:

  • (String) —

    Each chunk of data



1178
1179
1180
1181
1182
1183
 
# File 'lib/language_operator/agent/web_server.rb', line 1178

def each
  buffer = StringIO.new
  stream = MockStream.new(buffer)
  call(stream)
  yield buffer.string
end