Class: Openlayer::Integrations::GoogleConversationalSearchTracer

Inherits:
Object
  • Object
show all
Defined in:
lib/openlayer/integrations/google_conversational_search_tracer.rb

Overview

Tracer for Google Cloud DiscoveryEngine ConversationalSearchService

This class provides integration with Google's ConversationalSearchService to automatically trace answer_query calls and send them to the Openlayer platform.

Examples:

Basic usage

require 'openlayer/integrations/google_conversational_search_tracer'
require 'google/cloud/discovery_engine/v1'

google_client = Google::Cloud::DiscoveryEngine::V1::ConversationalSearchService::Client.new
openlayer = Openlayer::Client.new(api_key: ENV['OPENLAYER_API_KEY'])

Openlayer::Integrations::GoogleConversationalSearchTracer.trace_client(
  google_client,
  openlayer_client: openlayer,
  inference_pipeline_id: 'your-pipeline-id',
  additional_columns: { environment: 'production' }
)

# Now all answer_query calls are automatically traced! Pass
# additional_columns on an individual call to attach data (like your
# own trace ID) to just that row; it takes precedence over the
# static defaults above on a key conflict.
response = google_client.answer_query(
  serving_config: "projects/.../servingConfigs/default",
  query: { text: "What is the meaning of life?" },
  additional_columns: { trace_id: "abc-123" }
)

Constant Summary collapse

RESERVED_ROW_KEYS =

Row keys computed by this tracer. Any key in a caller-supplied additional_columns hash matching one of these is dropped, so custom data can never overwrite core trace fields.

[:query, :answer, :latency_ms, :timestamp, :metadata, :steps, :context, :session_id, :user_id].freeze

Class Method Summary collapse

Class Method Details

.send_trace(args:, kwargs:, response:, start_time:, end_time:, openlayer_client:, inference_pipeline_id:, session_id: nil, user_id: nil, additional_columns: {}, call_additional_columns: {}) ⇒ void

This method returns an undefined value.

Send trace data to Openlayer platform

Parameters:

  • args (Array) —

    Original method positional arguments

  • kwargs (Hash) —

    Original method keyword arguments

  • response (Google::Cloud::DiscoveryEngine::V1::AnswerQueryResponse) —

    The API response

  • start_time (Time) —

    Request start time

  • end_time (Time) —

    Request end time

  • openlayer_client (Openlayer::Client) —

    Openlayer client instance

  • inference_pipeline_id (String) —

    Pipeline ID

  • session_id (String, nil) (defaults to: nil) —

    Optional session ID (takes precedence over auto-extracted)

  • user_id (String, nil) (defaults to: nil) —

    Optional user ID

  • additional_columns (Hash, nil) (defaults to: {}) —

    Optional static column values (see trace_client)

  • call_additional_columns (Hash, nil) (defaults to: {}) —

    Optional per-call column values; takes precedence over additional_columns



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/openlayer/integrations/google_conversational_search_tracer.rb', line 121

def self.send_trace(args:, kwargs:, response:, start_time:, end_time:, openlayer_client:, inference_pipeline_id:, session_id: nil, user_id: nil, additional_columns: {}, call_additional_columns: {})
  # Calculate latency
  latency_ms = ((end_time - start_time) * 1000).round(2)

  # Extract query from request
  query_text = extract_query(args, kwargs)

  # Extract answer and metadata from response
  answer_data = extract_answer_data(response)

  # Extract additional metadata
  metadata = extract_metadata(args, kwargs, response, latency_ms)

  # Rough estimate of prompt and completion tokens
  prompt_tokens = (query_text.length / 4.0).ceil
  completion_tokens = (answer_data[:answer_text].length / 4.0).ceil

  # Extract grounding information from metadata for step root level
  citations = metadata.delete(:citations)
  references = metadata.delete(:references)
  related_questions = metadata.delete(:relatedQuestions)

  # Extract context from references (array of content strings)
  context = if references && references.is_a?(Array)
    references.map { |ref| ref[:content] }.compact
  else
    nil
  end

  # Extract nested steps (Google's execution steps)
  answer = response.respond_to?(:answer) ? response.answer : nil
  nested_steps = answer ? extract_steps(answer, query_text) : nil

  # Build step object
  step = {
    name: "Conversational Search answer_query",
    type: "chat_completion",
    provider: "Google",
    startTime: start_time.to_i,
    endTime: end_time.to_i,
    latency: latency_ms,
    metadata: metadata,
    inputs: {
      prompt: [
        {role: "user", content: query_text}
      ]
    },
    output: answer_data[:answer_text],
    promptTokens: prompt_tokens,
    completionTokens: completion_tokens,
    tokens: prompt_tokens + completion_tokens,
    model: "google-discovery-engine"
  }

  # Add grounding information at step root level
  step[:citations] = citations if citations
  step[:references] = references if references
  step[:relatedQuestions] = related_questions if related_questions

  # Add nested steps (Google's execution steps as RetrieverSteps)
  step[:steps] = nested_steps if nested_steps && !nested_steps.empty?

  # Build trace data in Openlayer format
  trace_data = {
    config: {
      inputVariableNames: ["query"],
      outputColumnName: "answer",
      latencyColumnName: "latency_ms",
      timestampColumnName: "timestamp"
    },
    rows: [
      {
        query: query_text,
        answer: answer_data[:answer_text],
        latency_ms: latency_ms,
        timestamp: start_time.to_i,
        metadata: metadata,
        steps: [step]
      }
    ]
  }

  # Add context column if available
  if context && !context.empty?
    trace_data[:rows][0][:context] = context
    trace_data[:config][:contextColumnName] = "context"
  end

  # Determine which session to use (kwarg takes precedence over auto-extracted)
  final_session = session_id || metadata[:session]
  if final_session
    trace_data[:rows][0][:session_id] = final_session
    trace_data[:config][:sessionIdColumnName] = "session_id"
  end

  # Determine which user_id to use (kwarg takes precedence over auto-extracted)
  user_pseudo_id = extract_user_pseudo_id(response)
  final_user_id = user_id || user_pseudo_id
  if final_user_id
    trace_data[:rows][0][:user_id] = final_user_id
    trace_data[:config][:userIdColumnName] = "user_id"
  end

  # Merge additional columns (per-call values take precedence over
  # static defaults; keys colliding with reserved row columns are
  # dropped so custom data can never corrupt core trace fields)
  extra_columns = resolve_additional_columns(additional_columns, call_additional_columns)
  trace_data[:rows][0].merge!(extra_columns) unless extra_columns.empty?

  # Send to Openlayer
  openlayer_client
    .inference_pipelines
    .data
    .stream(
      inference_pipeline_id,
      **trace_data
    )
end

.trace_client(client, openlayer_client:, inference_pipeline_id:, session_id: nil, user_id: nil, additional_columns: {}) ⇒ void

This method returns an undefined value.

Enable tracing on a Google ConversationalSearchService client

Parameters:

  • client (Google::Cloud::DiscoveryEngine::V1::ConversationalSearchService::Client) —

    The Google client instance to trace

  • openlayer_client (Openlayer::Client) —

    The Openlayer client instance for sending traces

  • inference_pipeline_id (String) —

    The Openlayer inference pipeline ID to send traces to

  • session_id (String, nil) (defaults to: nil) —

    Optional session ID to use for all traces. Takes precedence over auto-extracted sessions.

  • user_id (String, nil) (defaults to: nil) —

    Optional user ID to use for all traces.

  • additional_columns (Hash, nil) (defaults to: {}) —

    Optional static column values merged into every trace sent through this client (e.g. { environment: 'production' }). A value passed to an individual answer_query call takes precedence over these on a key conflict. Keys colliding with a reserved row column (query, answer, latency_ms, timestamp, metadata, steps, context, session_id, user_id) are dropped.



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/openlayer/integrations/google_conversational_search_tracer.rb', line 60

def self.trace_client(client, openlayer_client:, inference_pipeline_id:, session_id: nil, user_id: nil, additional_columns: {})
  # Store original method reference
  original_answer_query = client.method(:answer_query)

  # Define traced wrapper method
  client.define_singleton_method(:answer_query) do |*args, **kwargs, &block|
    # Capture start time
    start_time = Time.now

    # Extract per-call additional columns before forwarding to the
    # real client; Google's client never sees this key
    call_additional_columns = kwargs.delete(:additional_columns)

    # Execute the original method
    response = original_answer_query.call(*args, **kwargs, &block)

    # Capture end time
    end_time = Time.now

    # Send trace to Openlayer (with error handling)
    begin
      GoogleConversationalSearchTracer.send_trace(
        args: args,
        kwargs: kwargs,
        response: response,
        start_time: start_time,
        end_time: end_time,
        openlayer_client: openlayer_client,
        inference_pipeline_id: inference_pipeline_id,
        session_id: session_id,
        user_id: user_id,
        additional_columns: additional_columns,
        call_additional_columns: call_additional_columns
      )
    rescue StandardError => e
      # Never break the user's application due to tracing errors
      GoogleConversationalSearchTracer.warn_if_debug("[Openlayer] Failed to send trace: #{e.message}")
      GoogleConversationalSearchTracer.warn_if_debug("[Openlayer] #{e.backtrace.first(3).join("\n")}") if e.backtrace
    end

    # Always return the original response
    response
  end

  nil
end

.warn_if_debug(message) ⇒ void

This method returns an undefined value.

Log warning message if debug mode is enabled

Parameters:

  • message (String) —

    Warning message



709
710
711
 
# File 'lib/openlayer/integrations/google_conversational_search_tracer.rb', line 709

def self.warn_if_debug(message)
  warn(message) if ENV["OPENLAYER_DEBUG"]
end