feat: add e2e test for official vs unofficial assistant apis

2023-11-14 07:16:20 -06:00
parent b855fc0548
commit 7d43b8fe38
8 changed files with 317 additions and 7 deletions
@@ -0,0 +1,213 @@
+import { createAIFunction } from '@dexaai/dexter/prompt'
+import { sha256 } from 'crypto-hash'
+import delay from 'delay'
+import 'dotenv/config'
+import OpenAI from 'openai'
+import { oraPromise } from 'ora'
+import pMap from 'p-map'
+import plur from 'plur'
+import { z } from 'zod'
+
+import type { Run } from '~/lib/db'
+
+async function main() {
+  const defaultBaseUrl = 'https://api.openai.com/v1'
+  const baseUrl = process.env.OPENAI_API_BASE_URL ?? defaultBaseUrl
+  const isOfficalAPI = baseUrl === defaultBaseUrl
+  const testId =
+    process.env.TEST_ID ??
+    `test_${(await sha256(Date.now().toString())).slice(0, 24)}`
+  const metadata = { testId, isOfficalAPI }
+  const cleanupTest = !process.env.NO_TEST_CLEANUP
+
+  console.log('baseUrl', baseUrl)
+  console.log('testId', testId)
+  console.log()
+
+  const openai = new OpenAI({
+    baseURL: baseUrl
+  })
+
+  const getWeather = createAIFunction(
+    {
+      name: 'get_weather',
+      description: 'Gets the weather for a given location',
+      argsSchema: z.object({
+        location: z
+          .string()
+          .describe('The city and state e.g. San Francisco, CA'),
+        unit: z
+          .enum(['c', 'f'])
+          .optional()
+          .default('f')
+          .describe('The unit of temperature to use')
+      })
+    },
+    // Fake weather API implementation which returns a random temperature
+    // after a short delay
+    async function getWeather(args) {
+      await delay(500)
+
+      return {
+        location: args.location,
+        unit: args.unit,
+        temperature: (Math.random() * 100) | 0
+      }
+    }
+  )
+
+  const assistant = await openai.beta.assistants.create({
+    model: 'gpt-4-1106-preview',
+    instructions: 'You are a helpful assistant.',
+    metadata,
+    tools: [
+      {
+        type: 'function',
+        function: getWeather.spec
+      }
+    ]
+  })
+  console.log('created assistant', assistant)
+
+  let thread = await openai.beta.threads.create({
+    metadata,
+    messages: [
+      {
+        role: 'user',
+        content: 'What is the weather in San Francisco today?',
+        metadata
+      }
+    ]
+  })
+  console.log('created thread', thread)
+
+  let listMessages = await openai.beta.threads.messages.list(thread.id)
+  console.log('messages', listMessages.data)
+
+  let run = await openai.beta.threads.runs.create(thread.id, {
+    assistant_id: assistant.id,
+    metadata,
+    instructions: assistant.instructions,
+    model: assistant.model,
+    tools: assistant.tools
+  })
+  console.log('created run', run)
+
+  let listRunSteps = await openai.beta.threads.runs.steps.list(
+    thread.id,
+    run.id
+  )
+  console.log('runSteps', listRunSteps.data)
+
+  async function waitForRunStatus(
+    status: Run['status'],
+    { intervalMs = 500 }: { intervalMs?: number } = {}
+  ) {
+    return oraPromise(async () => {
+      while (run.status !== status) {
+        await delay(intervalMs)
+        run = await openai.beta.threads.runs.retrieve(thread.id, run.id)
+      }
+    }, `waiting for run "${run.id}" to have status "${status}"...`)
+  }
+
+  await waitForRunStatus('requires_action')
+  console.log('run', run)
+
+  listRunSteps = await openai.beta.threads.runs.steps.list(thread.id, run.id)
+  console.log('runSteps', listRunSteps.data)
+
+  if (run.status !== 'requires_action') {
+    throw new Error(
+      `run "${run.id}" status expected to be "requires_action"; found "${run.status}"`
+    )
+  }
+
+  if (!run.required_action) {
+    throw new Error(
+      `run "${run.id}" expected to have "required_action"; none found`
+    )
+  }
+
+  if (run.required_action.type !== 'submit_tool_outputs') {
+    throw new Error(
+      `run "${run.id}" expected to have "required_action.type" of "submit_tool_outputs; found "${run.required_action.type}"`
+    )
+  }
+
+  if (!run.required_action.submit_tool_outputs?.tool_calls?.length) {
+    throw new Error(
+      `run "${run.id}" expected to have non-empty "required_action.submit_tool_outputs"`
+    )
+  }
+
+  // Resolve tool calls
+  const toolCalls = run.required_action.submit_tool_outputs.tool_calls
+
+  const toolOutputs = await oraPromise(
+    pMap(
+      toolCalls,
+      async (toolCall) => {
+        if (toolCall.type !== 'function') {
+          throw new Error(
+            `run "${run.id}" invalid submit_tool_outputs tool_call type "${toolCall.type}"`
+          )
+        }
+
+        if (!toolCall.function) {
+          throw new Error(
+            `run "${run.id}" invalid submit_tool_outputs tool_call function"`
+          )
+        }
+
+        if (toolCall.function.name !== getWeather.spec.name) {
+          throw new Error(
+            `run "${run.id}" invalid submit_tool_outputs tool_call function name "${toolCall.function.name}"`
+          )
+        }
+
+        const toolCallResult = await getWeather(toolCall.function.arguments)
+        return {
+          output: JSON.stringify(toolCallResult),
+          tool_call_id: toolCall.id
+        }
+      },
+      { concurrency: 4 }
+    ),
+    `run "${run.id}" resolving ${toolCalls.length} tool ${plur(
+      'call',
+      toolCalls.length
+    )}`
+  )
+
+  console.log(`submitting tool outputs for run "${run.id}"`, toolOutputs)
+  run = await openai.beta.threads.runs.submitToolOutputs(thread.id, run.id, {
+    tool_outputs: toolOutputs
+  })
+  console.log('run', run)
+
+  listRunSteps = await openai.beta.threads.runs.steps.list(thread.id, run.id)
+  console.log('runSteps', listRunSteps.data)
+
+  await waitForRunStatus('completed')
+  console.log('run', run)
+
+  listRunSteps = await openai.beta.threads.runs.steps.list(thread.id, run.id)
+  console.log('runSteps', listRunSteps.data)
+
+  thread = await openai.beta.threads.retrieve(thread.id)
+  console.log('thread', thread)
+
+  listMessages = await openai.beta.threads.messages.list(thread.id)
+  console.log('messages', listMessages.data)
+
+  if (cleanupTest) {
+    // TODO: no way to delete messages or runs or run steps
+    // maybe deleting the thread implicitly causes a cascade of deletes?
+    // TODO: test this assumption
+    await openai.beta.threads.del(thread.id)
+    await openai.beta.assistants.del(assistant.id)
+  }
+}
+
+main()
@@ -50,6 +50,7 @@
    "@types/http-errors": "^2.0.4",
    "@types/node": "^20.9.0",
    "del-cli": "^5.1.0",
+    "delay": "^6.0.0",
    "dotenv": "^16.3.1",
    "dotenv-cli": "^7.3.0",
    "eslint": "^8.53.0",
@@ -60,6 +61,8 @@
    "openai": "^4.17.3",
    "openapi-jsonschema-parameters": "^12.1.3",
    "openapi-types": "^12.1.3",
+    "ora": "^7.0.1",
+    "plur": "^5.1.0",
    "prettier": "^2.8.8",
    "prisma": "^5.5.2",
    "prisma-json-types-generator": "^3.0.3",
@@ -73,6 +73,9 @@ devDependencies:
  del-cli:
    specifier: ^5.1.0
    version: 5.1.0
+  delay:
+    specifier: ^6.0.0
+    version: 6.0.0
  dotenv:
    specifier: ^16.3.1
    version: 16.3.1
@@ -103,6 +106,12 @@ devDependencies:
  openapi-types:
    specifier: ^12.1.3
    version: 12.1.3
+  ora:
+    specifier: ^7.0.1
+    version: 7.0.1
+  plur:
+    specifier: ^5.1.0
+    version: 5.1.0
  prettier:
    specifier: ^2.8.8
    version: 2.8.8
@@ -2789,6 +2798,14 @@ packages:
      readable-stream: 3.6.2
    dev: true

+  /bl@5.1.0:
+    resolution: {integrity: sha512-tv1ZJHLfTDnXE6tMHv73YgSJaWR2AFuPwMntBe7XL/GBFHnT0CLnsHMogfk5+GzCDC5ZWarSCYaIGATZt9dNsQ==}
+    dependencies:
+      buffer: 6.0.3
+      inherits: 2.0.4
+      readable-stream: 3.6.2
+    dev: true
+
  /bowser@2.11.0:
    resolution: {integrity: sha512-AlcaJBi/pqqJBIQ8U9Mcpc9i8Aqxn88Skv5d+xBX006BY5u8N3mGLHa5Lgppa7L/HfwgwLgZ6NYs+Ag6uUmJRA==}
    dev: false
@@ -3456,6 +3473,11 @@ packages:
      slash: 4.0.0
    dev: true

+  /delay@6.0.0:
+    resolution: {integrity: sha512-2NJozoOHQ4NuZuVIr5CWd0iiLVIRSDepakaovIN+9eIDHEhdCAEvSy2cuf1DCrPPQLvHmbqTHODlhHg8UCy4zw==}
+    engines: {node: '>=16'}
+    dev: true
+
  /delayed-stream@1.0.0:
    resolution: {integrity: sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==}
    engines: {node: '>=0.4.0'}
@@ -3560,6 +3582,10 @@ packages:
    engines: {node: '>=0.10.0'}
    dev: true

+  /emoji-regex@10.3.0:
+    resolution: {integrity: sha512-QpLs9D9v9kArv4lfDEgg1X/gN5XLnf/A6l9cs8SPZLRZR3ZkY9+kwIQTxm+fsSej5UMYGE8fdoaZVIBlqG0XTw==}
+    dev: true
+
  /emoji-regex@8.0.0:
    resolution: {integrity: sha512-MSjYzcWNOA0ewAHpz0MxpYFvwg6yjy1NG3xteoqz644VCo/RPgnr1/GGt+ic3iJTzQ8Eu3TdM14SawnVUmGE6A==}
    dev: true
@@ -4919,6 +4945,11 @@ packages:
      - supports-color
    dev: false

+  /irregular-plurals@3.5.0:
+    resolution: {integrity: sha512-1ANGLZ+Nkv1ptFb2pa8oG8Lem4krflKuX/gINiHJHjJUKaJHk/SXk5x6K3J+39/p0h1RQ2saROclJJ+QLvETCQ==}
+    engines: {node: '>=8'}
+    dev: true
+
  /is-arguments@1.1.1:
    resolution: {integrity: sha512-8Q7EARjzEnKpt/PCD7e1cgUS0a6X8u5tdSiMqXhojOdoV9TsMsiO+9VLC5vAmO8N7/GmXn7yjR8qnA6bVAEzfA==}
    engines: {node: '>= 0.4'}
@@ -6249,6 +6280,21 @@ packages:
      wcwidth: 1.0.1
    dev: true

+  /ora@7.0.1:
+    resolution: {integrity: sha512-0TUxTiFJWv+JnjWm4o9yvuskpEJLXTcng8MJuKd+SzAzp2o+OP3HWqNhB4OdJRt1Vsd9/mR0oyaEYlOnL7XIRw==}
+    engines: {node: '>=16'}
+    dependencies:
+      chalk: 5.3.0
+      cli-cursor: 4.0.0
+      cli-spinners: 2.9.1
+      is-interactive: 2.0.0
+      is-unicode-supported: 1.3.0
+      log-symbols: 5.1.0
+      stdin-discarder: 0.1.0
+      string-width: 6.1.0
+      strip-ansi: 7.1.0
+    dev: true
+
  /org-regex@1.0.0:
    resolution: {integrity: sha512-7bqkxkEJwzJQUAlyYniqEZ3Ilzjh0yoa62c7gL6Ijxj5bEpPL+8IE1Z0PFj0ywjjXQcdrwR51g9MIcLezR0hKQ==}
    engines: {node: '>=8'}
@@ -6550,6 +6596,13 @@ packages:
      pathe: 1.1.1
    dev: true

+  /plur@5.1.0:
+    resolution: {integrity: sha512-VP/72JeXqak2KiOzjgKtQen5y3IZHn+9GOuLDafPv0eXa47xq0At93XahYBs26MsifCQ4enGKwbjBTKgb9QJXg==}
+    engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0}
+    dependencies:
+      irregular-plurals: 3.5.0
+    dev: true
+
  /pluralize@8.0.0:
    resolution: {integrity: sha512-Nc3IT5yHzflTfbjgqWcCPpo7DaKy4FnpB0l/zCAW0Tc7jxAiuqSxHasntB3D7887LSrA93kDJ9IXovxJYxyLCA==}
    engines: {node: '>=4'}
@@ -7249,6 +7302,13 @@ packages:
    resolution: {integrity: sha512-f9aPhy8fYBuMN+sNfakZV18U39PbalgjXG3lLB9WkaYTxijru61wb57V9wxxNthXM5Sd88ETBWi29qLAsHO52Q==}
    dev: true

+  /stdin-discarder@0.1.0:
+    resolution: {integrity: sha512-xhV7w8S+bUwlPTb4bAOUQhv8/cSS5offJuX8GQGq32ONF0ZtDWKfkdomM3HMRA+LhX6um/FZ0COqlwsjD53LeQ==}
+    engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0}
+    dependencies:
+      bl: 5.1.0
+    dev: true
+
  /stop-iteration-iterator@1.0.0:
    resolution: {integrity: sha512-iCGQj+0l0HOdZ2AEeBADlsRC+vsnDsZsbdSiH1yNSjcfKM7fdpCMfqAL/dwF5BLiw/XhRft/Wax6zQbhq2BcjQ==}
    engines: {node: '>= 0.4'}
@@ -7296,6 +7356,15 @@ packages:
      strip-ansi: 7.1.0
    dev: true

+  /string-width@6.1.0:
+    resolution: {integrity: sha512-k01swCJAgQmuADB0YIc+7TuatfNvTBVOoaUWJjTB9R4VJzR5vNWzf5t42ESVZFPS8xTySF7CAdV4t/aaIm3UnQ==}
+    engines: {node: '>=16'}
+    dependencies:
+      eastasianwidth: 0.2.0
+      emoji-regex: 10.3.0
+      strip-ansi: 7.1.0
+    dev: true
+
  /string.prototype.matchall@4.0.10:
    resolution: {integrity: sha512-rGXbGmOEosIQi6Qva94HUjgPs9vKW+dkG7Y8Q5O2OYkWL6wFaTRZO8zM4mhP94uX55wgyrXzfS2aGtGzUL7EJQ==}
    dependencies:
@@ -1,4 +1,4 @@
-import { type ConnectionOptions } from 'bullmq'
+import { type ConnectionOptions, type DefaultJobOptions } from 'bullmq'

 export const env = process.env.NODE_ENV || 'development'
 export const isDev = env === 'development'
@@ -20,6 +20,10 @@ export namespace queue {
    password: process.env.REDIS_PASSWORD,
    username: process.env.REDIS_USERNAME ?? 'default'
  }
+  export const defaultJobOptions: DefaultJobOptions = {
+    removeOnComplete: true,
+    removeOnFail: 1000
+  }

  export const concurrency = isDev ? 1 : 16
  export const stalledInterval = runs.maxRunTime
@@ -5,9 +5,23 @@ import type { Run, RunStep } from './db'
 import type { JobData, JobResult } from './types'

 export const queue = new Queue<JobData, JobResult>(config.queue.name, {
-  connection: config.queue.redisConfig
+  connection: config.queue.redisConfig,
+  defaultJobOptions: config.queue.defaultJobOptions
 })

 export function getJobId(run: Run, runStep?: RunStep) {
  return `${run.id}${runStep?.id ? '-' + runStep.id : ''}`
 }
+
+const active = (await queue.getActive()).map((job) => job.asJSON())
+const completed = (await queue.getCompleted()).map((job) => job.asJSON())
+const failed = (await queue.getFailed()).map((job) => job.asJSON())
+const delayed = (await queue.getDelayed()).map((job) => job.asJSON())
+const jobs = (await queue.getJobs()).map((job) => job.asJSON())
+
+// console.log('jobs', jobs)
+console.log('active', active)
+console.log('completed', completed)
+console.log('failed', failed)
+console.log('delayed', delayed)
+console.log('jobs', jobs)
@@ -24,6 +24,9 @@ export const worker = new Worker<JobData, JobResult>(
    }

    const { runId } = job.data
+    console.log(
+      `Runner processing job "${job.id}" ${job.name} for run "${runId}"`
+    )
    let jobErrorResult: JobResult | undefined

    async function checkRunStatus(
@@ -31,7 +34,7 @@ export const worker = new Worker<JobData, JobResult>(
      { strict = true }: { strict?: boolean } = {}
    ) {
      if (!run) {
-        throw new Error(`Invalid run id "${runId}"`)
+        throw new Error(`Invalid run "${runId}"`)
      }

      if (run.status === 'cancelling') {
@@ -78,6 +81,7 @@ export const worker = new Worker<JobData, JobResult>(
        return jobErrorResult
      }

+      await job.updateProgress(50)
      return null
    }

@@ -47,13 +47,14 @@ app.openapi(routes.createThreadAndRun, async (c) => {
  })

  // Kick off async task
-  await queue.add(
+  const job = await queue.add(
    config.queue.threadRunJobName,
    { runId: run.id },
    {
      jobId: getJobId(run)
    }
  )
+  console.log('job', job.asJSON())

  return c.jsonT(utils.convertPrismaToOAI(run))
 })
@@ -84,13 +85,14 @@ app.openapi(routes.createRun, async (c) => {
  })

  // Kick off async task
-  await queue.add(
+  const job = await queue.add(
    config.queue.threadRunJobName,
    { runId: run.id },
    {
      jobId: getJobId(run)
    }
  )
+  console.log('job', job.asJSON())

  return c.jsonT(utils.convertPrismaToOAI(run))
 })
@@ -263,13 +265,14 @@ app.openapi(routes.submitToolOuputsToRun, async (c) => {
      })

      // Resume async task
-      await queue.add(
+      const job = await queue.add(
        config.queue.threadRunJobName,
        { runId: run.id },
        {
          jobId: getJobId(run, runStep)
        }
      )
+      console.log('job', job.asJSON())
      break
    }

@@ -1,5 +1,5 @@
 {
-  "include": ["src", "bin"],
+  "include": ["src", "bin", "e2e"],
  "exclude": ["node_modules", "dist", "openai-openapi"],
  "compilerOptions": {
    "target": "es2020",