Trigger live evaluations on every interaction (#1878)

Author: neoxelox

packages/core/src/events/events.d.ts

@@ -32,6 +32,7 @@ export type Events =
   | 'workspaceCreated'
   | 'projectCreated'
   | 'documentLogCreated'
+  | 'documentLogInteracted'
   | 'documentSuggestionCreated'
   | 'documentSuggestionApplied'
   | 'documentSuggestionDiscarded'
@@ -171,6 +172,11 @@ export type DocumentLogCreatedEvent = LatitudeEventGeneric<
   Pick<DocumentLog, 'id'> & { workspaceId: number }
 >
 
+export type DocumentLogInteractedEvent = LatitudeEventGeneric<
+  'documentLogInteracted',
+  Pick<DocumentLog, 'id'> & { workspaceId: number }
+>
+
 export type DocumentSuggestionCreatedEvent = LatitudeEventGeneric<
   'documentSuggestionCreated',
   {
@@ -734,6 +740,7 @@ export type LatitudeEvent =
   | WorkspaceCreatedEvent
   | ProjectCreatedEvent
   | DocumentLogCreatedEvent
+  | DocumentLogInteractedEvent
   | DocumentSuggestionCreatedEvent
   | DocumentSuggestionAppliedEvent
   | DocumentSuggestionDiscardedEvent
@@ -802,6 +809,7 @@ export interface IEventsHandlers {
   workspaceCreated: EventHandler<WorkspaceCreatedEvent>[]
   projectCreated: EventHandler<ProjectCreatedEvent>[]
   documentLogCreated: EventHandler<DocumentLogCreatedEvent>[]
+  documentLogInteracted: EventHandler<DocumentLogInteractedEvent>[]
   documentSuggestionCreated: EventHandler<DocumentSuggestionCreatedEvent>[]
   documentSuggestionApplied: EventHandler<DocumentSuggestionAppliedEvent>[]
   documentSuggestionDiscarded: EventHandler<DocumentSuggestionDiscardedEvent>[]

packages/core/src/events/handlers/evaluateLiveLog.ts

@@ -1,6 +1,6 @@
 import { LogSources } from '../../constants'
-import { findWorkspaceFromDocumentLog } from '../../data-access/workspaces'
 import { findLastProviderLogFromDocumentLogUuid } from '../../data-access/providerLogs'
+import { findWorkspaceFromDocumentLog } from '../../data-access/workspaces'
 import { runEvaluationV2JobKey } from '../../jobs/job-definitions'
 import { queues } from '../../jobs/queues'
 import { NotFoundError } from '../../lib/errors'
@@ -10,7 +10,7 @@ import {
   EvaluationsV2Repository,
 } from '../../repositories'
 import { getEvaluationMetricSpecification } from '../../services/evaluationsV2/specifications'
-import { DocumentLogCreatedEvent } from '../events'
+import { DocumentLogCreatedEvent, DocumentLogInteractedEvent } from '../events'
 
 const LIVE_EVALUABLE_LOG_SOURCES = Object.values(LogSources).filter(
   (source) => source !== 'evaluation' && source !== 'experiment',
@@ -19,7 +19,7 @@ const LIVE_EVALUABLE_LOG_SOURCES = Object.values(LogSources).filter(
 export const evaluateLiveLogJob = async ({
   data: event,
 }: {
-  data: DocumentLogCreatedEvent
+  data: DocumentLogCreatedEvent | DocumentLogInteractedEvent
 }) => {
   const { id, workspaceId } = event.data
   const repo = new DocumentLogsRepository(workspaceId)

packages/core/src/events/handlers/index.ts

@@ -34,6 +34,7 @@ export const EventHandlers: IEventsHandlers = {
   datasetUploaded: [createDatasetRowsJob],
   documentCreated: [],
   documentLogCreated: [evaluateLiveLogJob, notifyToClientDocumentLogCreatedJob],
+  documentLogInteracted: [evaluateLiveLogJob],
   experimentVariantsCreated: [],
   documentSuggestionCreated: [
     notifyClientOfDocumentSuggestionCreated,

packages/core/src/services/documentLogs/addMessages/index.ts

@@ -1,11 +1,10 @@
-import { type Message } from '@latitude-data/constants/legacyCompiler'
-
 import { NotFoundError } from '@latitude-data/constants/errors'
 import { LatitudePromptConfig } from '@latitude-data/constants/latitudePromptSchema'
-import { buildConversation } from '../../../helpers'
+import { type Message } from '@latitude-data/constants/legacyCompiler'
 import { LogSources } from '../../../constants'
-import { type Workspace } from '../../../schema/models/types/Workspace'
 import { unsafelyFindProviderApiKey } from '../../../data-access/providerApiKeys'
+import { publisher } from '../../../events/publisher'
+import { buildConversation } from '../../../helpers'
 import { Result } from '../../../lib/Result'
 import { ToolHandler } from '../../../lib/streamManager/clientTools/handlers'
 import { DefaultStreamManager } from '../../../lib/streamManager/defaultStreamManager'
@@ -15,9 +14,11 @@ import {
   DocumentVersionsRepository,
   ProviderLogsRepository,
 } from '../../../repositories'
+import { type Workspace } from '../../../schema/models/types/Workspace'
 import { BACKGROUND, telemetry, TelemetryContext } from '../../../telemetry'
 import { getInputSchema, getOutputType } from '../../chains/ChainValidator'
 import { scanDocumentContent } from '../../documents'
+import { isErrorRetryable } from '../../evaluationsV2/run'
 import serializeProviderLog from '../../providerLogs/serialize'
 
 type AddMessagesArgs = {
@@ -48,7 +49,8 @@ export async function addMessages({
     documentLogUuid,
   })
   if (dataResult.error) return dataResult
-  const { document, commit, providerLog, globalConfig } = dataResult.unwrap()
+  const { document, commit, documentLog, providerLog, globalConfig } =
+    dataResult.unwrap()
 
   const $prompt = telemetry.prompt(context, {
     documentLogUuid,
@@ -105,9 +107,24 @@ export async function addMessages({
 
   start()
 
-  streamResult.error.then((error) =>
-    error ? $prompt.fail(error) : $prompt.end(),
-  )
+  streamResult.response.then(async (response) => {
+    const error = await streamResult.error
+    if (error) {
+      $prompt.fail(error)
+
+      if (isErrorRetryable(error)) return response
+    } else {
+      $prompt.end()
+    }
+
+    await publisher.publishLater({
+      type: 'documentLogInteracted',
+      data: {
+        id: documentLog.id,
+        workspaceId: workspace.id,
+      },
+    })
+  })
 
   return Result.ok(streamResult)
 }
@@ -150,5 +167,5 @@ async function retrieveData({
   if (metadataResult.error) return metadataResult
   const globalConfig = metadataResult.value.config as LatitudePromptConfig
 
-  return Result.ok({ commit, document, providerLog, globalConfig })
+  return Result.ok({ commit, document, documentLog, providerLog, globalConfig })
 }