Class: SwarmSDK::Agent::LLMInstrumentationMiddleware

Inherits:
Faraday::Middleware
  • Object
show all
Defined in:
lib/swarm_sdk/agent/llm_instrumentation_middleware.rb

Overview

Faraday middleware for capturing LLM API requests and responses

This middleware intercepts HTTP calls to LLM providers and emits structured events via LogStream for logging and monitoring.

Events emitted:

  • llm_api_request: Before sending request to LLM API
  • llm_api_response: After receiving response from LLM API

The middleware is injected at runtime into the provider's Faraday connection stack (see Agent::Chat#inject_llm_instrumentation).

Instance Method Summary collapse

Constructor Details

#initialize(app, on_request:, on_response:, provider_name:) ⇒ LLMInstrumentationMiddleware

Initialize middleware

Parameters:

  • app (Faraday::Connection) —

    Faraday app

  • on_request (Proc) —

    Callback for request events

  • on_response (Proc) —

    Callback for response events

  • provider_name (String) —

    Provider name for logging



23
24
25
26
27
28
 
# File 'lib/swarm_sdk/agent/llm_instrumentation_middleware.rb', line 23

def initialize(app, on_request:, on_response:, provider_name:)
  super(app)
  @on_request = on_request
  @on_response = on_response
  @provider_name = provider_name
end

Instance Method Details

#call(env) ⇒ Faraday::Response

Intercept HTTP call

Parameters:

  • env (Faraday::Env) —

    Request environment

Returns:

  • (Faraday::Response) —

    HTTP response



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/swarm_sdk/agent/llm_instrumentation_middleware.rb', line 34

def call(env)
  start_time = Time.now
  accumulated_raw_chunks = []

  # Emit request event
  emit_request_event(env, start_time)

  # Wrap existing on_data to capture raw SSE chunks for streaming
  if env.request&.on_data
    original_on_data = env.request.on_data
    env.request.on_data = proc do |chunk, bytes, response_env|
      # Capture raw chunk BEFORE RubyLLM processes it
      accumulated_raw_chunks << chunk
      # Call original handler (RubyLLM's stream processing)
      original_on_data.call(chunk, bytes, response_env)
    end
  end

  # Execute request
  @app.call(env).on_complete do |response_env|
    end_time = Time.now

    # Determine if this was a streaming request based on whether chunks were accumulated
    # This is more reliable than parsing response content
    is_streaming = accumulated_raw_chunks.any?

    # For streaming: use accumulated raw SSE chunks
    # For non-streaming: use response body
    raw_body = is_streaming ? accumulated_raw_chunks.join : response_env.body

    # Store SSE body in Fiber-local for citation extraction
    # This allows append_citations_to_content to access the full SSE body
    # even though response.body is empty for streaming responses
    Fiber[:last_sse_body] = raw_body if is_streaming

    # Emit response event
    timing = { start_time: start_time, end_time: end_time, duration: end_time - start_time }
    emit_response_event(response_env, timing, raw_body, is_streaming)
  end
end