update tool call function

sutigit · sutigit · commit 0d96a2984320 · 2025-05-26T16:42:30.000+03:00
diff --git a/src/server/routes/openai.ts b/src/server/routes/openai.ts
@@ -5,7 +5,7 @@ import { CourseChatRequest, RequestWithUser } from '../types'
 import { isError } from '../util/parser'
 import { calculateUsage, incrementUsage, checkUsage, checkCourseUsage, incrementCourseUsage } from '../services/chatInstances/usage'
 import { getCompletionEvents, streamCompletion } from '../util/azure/client'
-import { streamResponsesEvents, getResponsesEvents } from '../util/azure/clientV2'
+import { ResponsesClient } from '../util/azure/clientV2'
 import { getMessageContext, getModelContextLimit, getCourseModel, getAllowedModels } from '../util/util'
 import getEncoding from '../util/tiktoken'
 import logger from '../util/logger'
@@ -100,17 +100,14 @@ openaiRouter.post('/stream/:version?', upload.single('file'), async (r, res) =>
     return
   }
 
+  const responsesClient = new ResponsesClient(model)
+
   let events
   if (version === 'v2') {
-    events = await getResponsesEvents({
-      model: options.model,
-      input: options.messages,
-      stream: options.stream,
-    })
+    events = await responsesClient.createResponse({ input: options.messages })
   } else {
     events = await getCompletionEvents(options)
   }
-
   if (isError(events)) {
     res.status(424)
     return
@@ -120,7 +117,12 @@ openaiRouter.post('/stream/:version?', upload.single('file'), async (r, res) =>
 
   let completion
   if (version === 'v2') {
-    completion = await streamResponsesEvents(events, encoding, res)
+    completion = await responsesClient.handleResponse({
+      events,
+      prevMessages: options.messages,
+      encoding,
+      res,
+    })
   } else {
     completion = await streamCompletion(events, options, encoding, res)
   }
@@ -214,13 +216,11 @@ openaiRouter.post('/stream/:courseId/:version?', upload.single('file'), async (r
     return
   }
 
+  const responsesClient = new ResponsesClient(model)
+
   let events
   if (version === 'v2') {
-    events = await getResponsesEvents({
-      model: options.model,
-      input: options.messages,
-      stream: options.stream,
-    })
+    events = await responsesClient.createResponse({ input: options.messages })
   } else {
     events = await getCompletionEvents(options)
   }
@@ -234,7 +234,12 @@ openaiRouter.post('/stream/:courseId/:version?', upload.single('file'), async (r
 
   let completion
   if (version === 'v2') {
-    completion = await streamResponsesEvents(events, encoding, res)
+    completion = await responsesClient.handleResponse({
+      events,
+      prevMessages: options.messages,
+      encoding,
+      res,
+    })
   } else {
     completion = await streamCompletion(events, options, encoding, res)
   }
diff --git a/src/server/util/azure/clientV2.ts b/src/server/util/azure/clientV2.ts
@@ -1,5 +1,6 @@
 import { Tiktoken } from '@dqbd/tiktoken'
 import { Response } from 'express'
+import { isError } from '../../util/parser'
 
 import { AZURE_RESOURCE, AZURE_API_KEY } from '../config'
 import { validModels, inProduction } from '../../../config'
@@ -9,7 +10,14 @@ import { APIError } from '../../types'
 import { AzureOpenAI } from 'openai'
 // import { EventStream } from '@azure/openai'
 import { Stream } from 'openai/streaming'
-import { ResponseStreamEvent } from 'openai/resources/responses/responses'
+import {
+  FunctionTool,
+  ResponseInput,
+  ResponseInputItem,
+  ResponseStreamEvent,
+} from 'openai/resources/responses/responses'
+
+import { testTool } from './tools'
 
 const endpoint = `https://${AZURE_RESOURCE}.openai.azure.com/`
 
@@ -23,90 +31,139 @@ export const getAzureOpenAIClient = (deployment: string) =>
 
 const client = getAzureOpenAIClient(process.env.GPT_4O)
 
-/**
- * Mock stream for testing
- */
-// const getMockCompletionEvents: () => Promise<
-//   EventStream<ResponseStreamEvent>
-// > = async () => {
-//   const mockStream = new ReadableStream<ResponseStreamEvent>({
-//     start(controller) {
-//       for (let i = 0; i < 10; i += 1) {
-//         controller.enqueue({
-//           event: "response",
-//           data: ""
-//         })
-//       }
-//       controller.close()
-//     },
-//   }) as EventStream<ResponseStreamEvent>
-
-//   return mockStream
-// }
-
-export const getResponsesEvents = async ({
-  model,
-  input,
-  stream,
-}: any): Promise<
-  | Stream<ResponseStreamEvent>
-  // EventStream<ChatCompletionChunk>
-  | APIError
-  | any
-> => {
-  const deploymentId = validModels.find((m) => m.name === model)?.deployment
-
-  if (!deploymentId) throw new Error(`Invalid model: ${model}, not one of ${validModels.map((m) => m.name).join(', ')}`)
-
-  // Mocking disabled because it's difficult to mock a event stream for responses API.
-  // if (deploymentId === 'mock') return getMockCompletionEvents()
-
-  try {
-    const events = await client.responses.create({
-      model: deploymentId,
-      instructions: 'Olet avulias apuri.',
-      input,
-      stream,
-      tools: [],
-    })
+export class ResponsesClient {
+  model: string
+  instructions: string
+  tools: FunctionTool[]
 
-    return events
-  } catch (error: any) {
-    logger.error(error)
+  constructor(model: string, instructions?: string) {
+    const deploymentId = validModels.find((m) => m.name === model)?.deployment
 
-    return { error } as any as APIError
+    if (!deploymentId)
+      throw new Error(
+        `Invalid model: ${model}, not one of ${validModels.map((m) => m.name).join(', ')}`
+      )
+
+    this.model = deploymentId
+    this.instructions = instructions || 'Olet avulias apuri.'
+    this.tools = [testTool.definition]
   }
-}
 
-export const streamResponsesEvents = async (events: Stream<ResponseStreamEvent>, encoding: Tiktoken, res: Response) => {
-  let tokenCount = 0
-  const contents = []
-
-  for await (const event of events) {
-    switch (event.type) {
-      case 'response.output_text.delta':
-        if (!inProduction) logger.info(event.delta)
-
-        await new Promise((resolve) => {
-          if (
-            !res.write(event.delta, (err) => {
-              if (err) logger.error(`${event.delta} ${err}`)
-            })
-          ) {
-            logger.info(`${event.delta} res.write returned false, waiting for drain`)
-            res.once('drain', resolve)
-          } else {
-            process.nextTick(resolve)
-          }
-        })
-        contents.push(event.delta)
-        tokenCount += encoding.encode(event.delta).length ?? 0
-        break
+  async createResponse({
+    input,
+  }: {
+    input: ResponseInput
+  }): Promise<Stream<ResponseStreamEvent> | APIError> {
+    try {
+      return await client.responses.create({
+        model: this.model,
+        instructions: this.instructions,
+        input,
+        stream: true,
+        tools: this.tools,
+      })
+    } catch (error: any) {
+      logger.error(error)
+
+      return { error } as any as APIError
+    }
+  }
+
+  async handleResponse({
+    events,
+    prevMessages,
+    encoding,
+    res,
+  }: {
+    events: Stream<any>
+    prevMessages: ResponseInput
+    encoding: Tiktoken
+    res: Response
+  }) {
+    let tokenCount = 0
+    const contents = []
+
+    for await (const event of events) {
+      console.log('event type:', event.type)
+
+      switch (event.type) {
+        case 'response.output_text.delta':
+          await this.writeDelta(event.delta, res)
+
+          contents.push(event.delta)
+          tokenCount += encoding.encode(event.delta).length ?? 0
+          break
+
+        case 'response.function_call_arguments.done':
+          // WORK IN PROGRESS
+
+          // const augRetrieval = await this.callToolFunction(
+          //   event.arguments,
+          //   event.call_id
+          // )
+          // const newEvents = await this.createResponse({
+          //   input: [...prevMessages, augRetrieval],
+          // })
+
+          // if (isError(events)) {
+          //   throw new Error(`Error creating response from function call`)
+          // }
+
+          // await this.handleResponse({
+          //   events: newEvents as Stream<ResponseStreamEvent>,
+          //   prevMessages: [...prevMessages, augRetrieval],
+          //   encoding,
+          //   res,
+          // })
+          break
+      }
     }
+
+    return {
+      tokenCount,
+      response: contents.join(''),
+    }
+  }
+
+  private async writeDelta(text: string, res: Response) {
+    // if (!inProduction) logger.info(text)
+
+    await new Promise((resolve) => {
+      if (
+        !res.write(text, (err) => {
+          if (err) logger.error(`${text} ${err}`)
+        })
+      ) {
+        logger.info(`${text} res.write returned false, waiting for drain`)
+        res.once('drain', resolve)
+      } else {
+        process.nextTick(resolve)
+      }
+    })
   }
 
-  return {
-    tokenCount,
-    response: contents.join(''),
+  private async callToolFunction(
+    args: string,
+    callId: string
+  ): Promise<ResponseInputItem[]> {
+    const { query } = JSON.parse(args)
+    try {
+      const retrieval = await testTool.function(query)
+
+      return [
+        {
+          role: 'user',
+          content: retrieval.query,
+        },
+        {
+          type: 'function_call_output',
+          call_id: callId,
+          output: retrieval.result,
+        },
+      ]
+    } catch (error) {
+      logger.error('Error calling tool function:', error)
+      return null
+    }
   }
 }
diff --git a/src/server/util/azure/tools.ts b/src/server/util/azure/tools.ts
@@ -1,4 +1,11 @@
-export const testTool = {
+import { FunctionTool } from 'openai/resources/responses/responses'
+
+interface ToolObject {
+  definition: FunctionTool
+  function: (query: string) => Promise<any>
+}
+
+export const testTool: ToolObject = {
   definition: {
     type: 'function',
     name: 'test_knowledge_retrieval',
@@ -13,13 +20,18 @@ export const testTool = {
         },
       },
       required: ['query'],
+      additionalProperties: false,
     },
+    strict: true, // or true, depending on your requirements
   },
-  function: async (query: string) => {
+  function: async (
+    query: string
+  ): Promise<{ query: string; result: string }> => {
     // Simulate a tool function that returns a simple message
     return {
       query,
-      message: 'This is a test tool function',
+      result:
+        'This is a test result from the test tool. The secret is: Chili kastike',
     }
   },
 }
