Function createOpenAICompatibleStream

apps/sim/providers/utils.ts:1338–1383

(
  stream: AsyncIterable<ChatCompletionChunk>,
  providerName: string,
  onComplete?: (content: string, usage: CompletionUsage) => void
)

 * @returns A ReadableStream that can be used for streaming responses
 */
export function createOpenAICompatibleStream(
  stream: AsyncIterable<ChatCompletionChunk>,
  providerName: string,
  onComplete?: (content: string, usage: CompletionUsage) => void
): ReadableStream<Uint8Array> {
  const streamLogger = createLogger(`${providerName}Utils`)
  let fullContent = ''
  let promptTokens = 0
  let completionTokens = 0
  let totalTokens = 0

  return new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          if (chunk.usage) {
            promptTokens = chunk.usage.prompt_tokens ?? 0
            completionTokens = chunk.usage.completion_tokens ?? 0
            totalTokens = chunk.usage.total_tokens ?? 0
          }

          const content = chunk.choices?.[0]?.delta?.content || ''
          if (content) {
            fullContent += content
            controller.enqueue(new TextEncoder().encode(content))
          }
        }

        if (onComplete) {
          if (promptTokens === 0 && completionTokens === 0) {
            streamLogger.warn(`${providerName} stream completed without usage data`)
          }
          onComplete(fullContent, {
            prompt_tokens: promptTokens,
            completion_tokens: completionTokens,
            total_tokens: totalTokens || promptTokens + completionTokens,
          })
        }

        controller.close()
      } catch (error) {
        controller.error(error)
      }
    },
  })
}

/**
 * Checks if a forced tool was used in an OpenAI-compatible response and updates tracking.