Integrate Answer Relevancy Analysis into analysis workflow

This adds the BaseMetricjob and AnswerRelevancyJob. The AnswerRelevancyJobs
handles:
- making calls to the AnswerRelevancy class
- compiling the results
- creating a AnswerRelevancyAggregate record to store the mean score
- calling the AnswerRelevancyAggregate#create_run_from_result method to
  delegate run creation to the AutoEvaluationMetricRun model

The BaseMetricJob is used to store shard functionality for future metric jobs.

I've moved logic for saving the records into a concern so it can be reused
by other metrics.

Author: davidgisbey

app/jobs/answer_analysis/answer_relevancy_job.rb

@@ -0,0 +1,30 @@
+module AnswerAnalysis
+  class AnswerRelevancyJob < BaseMetricJob
+    def perform(answer_id)
+      return if warning_logged?(answer_id)
+
+      answer = Answer.includes(:question, :answer_relevancy_aggregate).find_by(id: answer_id)
+      return logger.warn(aggregate_exists_warn_message(answer.id)) if answer.answer_relevancy_aggregate.present?
+
+      results = []
+      NUMBER_OF_RUNS.times do
+        results << AutoEvaluation::AnswerRelevancy.call(
+          question_message: answer.rephrased_question || answer.question.message,
+          answer_message: answer.message,
+        )
+      end
+
+      begin
+        AnswerAnalysis::AnswerRelevancyAggregate.create_mean_aggregate_and_score_runs(answer, results)
+      rescue ActiveRecord::RecordNotUnique
+        logger.warn(aggregate_exists_warn_message(answer.id))
+      end
+    end
+
+  private
+
+    def aggregate_exists_warn_message(answer_id)
+      "Answer #{answer_id} has already been evaluated for relevancy"
+    end
+  end
+end

app/jobs/answer_analysis/base_metric_job.rb

@@ -0,0 +1,24 @@
+module AnswerAnalysis
+  class BaseMetricJob < ApplicationJob
+    NUMBER_OF_RUNS = 3
+    MAX_RETRIES = 5
+    retry_on Aws::Errors::ServiceError, wait: 1.minute, attempts: MAX_RETRIES
+
+  private
+
+    def warning_logged?(answer_id)
+      answer = Answer.find_by(id: answer_id)
+      unless answer
+        logger.warn("No answer found for #{answer_id}")
+        return true
+      end
+
+      unless answer.status_answered?
+        logger.info("Answer #{answer.id} is not eligible for auto-evaluation")
+        return true
+      end
+
+      false
+    end
+  end
+end

app/jobs/compose_answer_job.rb

@@ -14,6 +14,11 @@ def perform(question_id)
       logger.warn("Already an answer created for #{question_id}")
     end
 
-    AnswerTopicsJob.perform_later(answer.id) if answer.persisted?
+    if answer.persisted?
+      # TODO: Once we've added a few metrics we should move these to a single job that
+      # kicks off all analysis jobs.
+      AnswerTopicsJob.perform_later(answer.id)
+      AnswerAnalysis::AnswerRelevancyJob.perform_later(answer.id)
+    end
   end
 end

app/models/answer_analysis/answer_relevancy_aggregate.rb

@@ -1,5 +1,7 @@
 module AnswerAnalysis
   class AnswerRelevancyAggregate < ApplicationRecord
+    include AnalysisResultsCreatable
+
     belongs_to :answer
     has_many :runs, class_name: "AnswerAnalysis::AnswerRelevancyRun"
   end

app/models/concerns/analysis_results_creatable.rb

@@ -0,0 +1,26 @@
+module AnalysisResultsCreatable
+  extend ActiveSupport::Concern
+
+  included do
+    def self.create_mean_aggregate_and_score_runs(answer, results)
+      mean_score = results.sum(&:score) / results.size.to_f
+      aggregate = create!(answer:, mean_score:)
+
+      results.each do |result|
+        run = aggregate.runs.build(
+          score: result.score,
+          reason: result.reason,
+        )
+
+        result.llm_responses.stringify_keys.each do |name, llm_response|
+          run.assign_llm_response(name, llm_response)
+        end
+        result.metrics.stringify_keys.each do |name, metrics|
+          run.assign_metrics(name, metrics)
+        end
+      end
+
+      aggregate.save!
+    end
+  end
+end

spec/jobs/answer_analysis/answer_relevancy_job_spec.rb

@@ -0,0 +1,203 @@
+RSpec.describe AnswerAnalysis::AnswerRelevancyJob do
+  include ActiveJob::TestHelper
+
+  let(:answer) { create(:answer) }
+  let(:question) { answer.question }
+  let(:first_result) do
+    AutoEvaluation::AnswerRelevancy::Result.new(
+      score: 0.8,
+      reason: "The first reason.",
+      success: true,
+      llm_responses: {
+        "response_1" => { "content" => "LLM response content 1" },
+        "response_2" => { "content" => "LLM response content 2" },
+      },
+      metrics: {
+        "metric_1" => { "detail" => "Metric detail 1" },
+        "metric_2" => { "detail" => "Metric detail 2" },
+      },
+    )
+  end
+  let(:second_result) do
+    AutoEvaluation::AnswerRelevancy::Result.new(
+      score: 0.7,
+      reason: "The second reason.",
+      success: true,
+      llm_responses: {
+        "response_3" => { "content" => "LLM response content 3" },
+        "response_4" => { "content" => "LLM response content 4" },
+      },
+      metrics: {
+        "metric_3" => { "detail" => "Metric detail 3" },
+        "metric_4" => { "detail" => "Metric detail 4" },
+      },
+    )
+  end
+  let(:third_result) do
+    AutoEvaluation::AnswerRelevancy::Result.new(
+      score: 0.9,
+      reason: "The third reason.",
+      success: true,
+      llm_responses: {
+        "response_5" => { "content" => "LLM response content 5" },
+        "response_6" => { "content" => "LLM response content 6" },
+      },
+      metrics: {
+        "metric_5" => { "detail" => "Metric detail 5" },
+        "metric_6" => { "detail" => "Metric detail 6" },
+      },
+    )
+  end
+
+  before do
+    allow(AutoEvaluation::AnswerRelevancy)
+      .to receive(:call).and_return(first_result, second_result, third_result)
+    stub_const("AnswerAnalysis::BaseMetricJob::NUMBER_OF_RUNS", 3)
+  end
+
+  it_behaves_like "a job in queue", "default"
+
+  describe "#perform" do
+    it "calls AutoEvaluation::AnswerRelevancy the configured number of times with the correct arguments" do
+      described_class.new.perform(answer.id)
+
+      expect(AutoEvaluation::AnswerRelevancy)
+        .to have_received(:call)
+        .with(
+          question_message: question.message,
+          answer_message: answer.message,
+        )
+        .exactly(3).times
+    end
+
+    it "creates answer relevancy aggregate with the correct score" do
+      expect {
+        described_class.new.perform(answer.id)
+      }.to change(AnswerAnalysis::AnswerRelevancyAggregate, :count).by(1)
+      answer = Answer.includes(:answer_relevancy_aggregate)
+                     .find(AnswerAnalysis::AnswerRelevancyAggregate.last.answer_id)
+      expect(answer.answer_relevancy_aggregate.mean_score.round(2)).to eq(0.8)
+    end
+
+    it "creates answer relevancy runs for each result" do
+      expect {
+        described_class.new.perform(answer.id)
+      }.to change(AnswerAnalysis::AnswerRelevancyRun, :count).by(3)
+
+      answer = Answer.includes(answer_relevancy_aggregate: :runs)
+                     .find(AnswerAnalysis::AnswerRelevancyAggregate.last.answer_id)
+
+      [first_result, second_result, third_result].each_with_index do |result, index|
+        expect(answer.answer_relevancy_aggregate.runs.sort_by(&:created_at)[index])
+          .to have_attributes(
+            score: result.score.round(2),
+            reason: result.reason,
+            llm_responses: result.llm_responses,
+            metrics: result.metrics,
+          )
+      end
+    end
+
+    context "when the answer has a rephrased_question" do
+      let(:rephrased_question) { "This is a rephrased_question" }
+
+      it "passes the rephrased question to AutoEvaluation::AnswerRelevancy as the question_message" do
+        answer = create(:answer, rephrased_question: rephrased_question)
+
+        described_class.new.perform(answer.id)
+
+        expect(AutoEvaluation::AnswerRelevancy)
+          .to have_received(:call)
+          .with(
+            question_message: rephrased_question,
+            answer_message: answer.message,
+          )
+          .exactly(3).times
+      end
+    end
+
+    context "when aggegate data is persisted mid job" do
+      before do
+        allow(AnswerAnalysis::AnswerRelevancyAggregate)
+          .to receive(:create_mean_aggregate_and_score_runs)
+          .with(answer, anything)
+          .and_raise(ActiveRecord::RecordNotUnique)
+      end
+
+      it "logs a warning" do
+        expect(described_class.logger)
+          .to receive(:warn)
+          .with("Answer #{answer.id} has already been evaluated for relevancy")
+
+        described_class.new.perform(answer.id)
+      end
+    end
+
+    context "when the answer does not exist" do
+      let(:answer_id) { 999 }
+
+      it "logs a warning" do
+        expect(described_class.logger)
+          .to receive(:warn)
+          .with("No answer found for #{answer_id}")
+
+        described_class.new.perform(answer_id)
+      end
+
+      it "doesn't call AutoEvaluation::AnswerRelevancy" do
+        described_class.new.perform(answer_id)
+        expect(AutoEvaluation::AnswerRelevancy).not_to have_received(:call)
+      end
+    end
+
+    context "when answer relevancy has already been evaluated" do
+      let(:aggregate) { create(:answer_relevancy_aggregate) }
+      let(:answer) { aggregate.answer }
+
+      it "logs a warning" do
+        expect(described_class.logger)
+          .to receive(:warn)
+          .with("Answer #{answer.id} has already been evaluated for relevancy")
+
+        described_class.new.perform(answer.id)
+      end
+
+      it "doesn't call AutoEvaluation::AnswerRelevancy" do
+        described_class.new.perform(answer.id)
+        expect(AutoEvaluation::AnswerRelevancy).not_to have_received(:call)
+      end
+    end
+
+    context "when the AnswerRelevancy metric raises an Aws::Errors::ServiceError" do
+      it "retries the job the max number of times" do
+        allow(AutoEvaluation::AnswerRelevancy).to receive(:call)
+          .and_raise(Aws::Errors::ServiceError.new(nil, "error"))
+
+        (described_class::MAX_RETRIES - 1).times do
+          described_class.perform_later(answer.id)
+          expect { perform_enqueued_jobs }.not_to raise_error
+        end
+
+        described_class.perform_later(answer.id)
+        expect { perform_enqueued_jobs }.to raise_error(Aws::Errors::ServiceError)
+      end
+    end
+
+    context "when the answer is not eligible for auto-evaluation" do
+      let(:answer) { create(:answer, status: Answer.statuses.except(:answered).keys.sample) }
+
+      it "logs an info message" do
+        expect(described_class.logger)
+          .to receive(:info)
+          .with("Answer #{answer.id} is not eligible for auto-evaluation")
+
+        described_class.new.perform(answer.id)
+      end
+
+      it "does not call AutoEvaluation::AnswerRelevancy" do
+        expect(AutoEvaluation::AnswerRelevancy).not_to receive(:call)
+        described_class.new.perform(answer.id)
+      end
+    end
+  end
+end

spec/jobs/compose_answer_job_spec.rb

@@ -6,6 +6,7 @@
   before do
     allow(AnswerComposition::Composer).to receive(:call).and_return(returned_answer)
     allow(AnswerTopicsJob).to receive(:perform_later)
+    allow(AnswerAnalysis::AnswerRelevancyJob).to receive(:perform_later)
   end
 
   it_behaves_like "a job in queue", "answer"
@@ -22,6 +23,11 @@
       expect(AnswerTopicsJob).to have_received(:perform_later).with(returned_answer.id)
     end
 
+    it "calls the AnswerAnalysis::AnswerRelevancyJob with the answer_id" do
+      described_class.new.perform(question.id)
+      expect(AnswerAnalysis::AnswerRelevancyJob).to have_received(:perform_later).with(returned_answer.id)
+    end
+
     context "when the question has already been answered" do
       let(:question) { create(:question, :with_answer) }
 

spec/models/answer_analysis/answer_relevancy_aggregate_spec.rb

@@ -0,0 +1,6 @@
+RSpec.describe AnswerAnalysis::AnswerRelevancyAggregate do
+  include_examples "analysis results creatable",
+                   :answer_relevancy_aggregate,
+                   AnswerAnalysis::AnswerRelevancyRun,
+                   AutoEvaluation::AnswerRelevancy::Result
+end

spec/requests/api/v1/conversation_flow_spec.rb

@@ -74,6 +74,7 @@ def when_i_create_a_conversation
       )
     end
     allow(AnswerTopicsJob).to receive(:perform_later)
+    allow(AnswerAnalysis::AnswerRelevancyJob).to receive(:perform_later)
 
     post api_v1_create_conversation_path,
          params: { user_question: "What is the capital of France?" },