streaming-sentiment-typechat.mts

import assert from "assert";
import dotenv from "dotenv";
import fetch from "node-fetch";
import { TypeChatLanguageModel, createJsonTranslator, processRequests, success } from "typechat";

dotenv.config();

const modelName = process.env.OPENAI_MODEL;
const apiKey = process.env.OPENAI_API_KEY;

assert(modelName);
assert(apiKey);

const model: TypeChatLanguageModel = {
    complete: createStreamingCompleter(s => process.stdout.write(s))
};

export interface SentimentResponse {
    sentiment: "positive" | "negative" | "neutral";
}
const schema = `export interface SentimentResponse {
    sentiment: "positive" | "negative" | "neutral";
}`;

const translator = createJsonTranslator(model, schema, "SentimentResponse");
processRequests("> ", undefined, async (req) => {
    const result = await translator.translate(req);
    
    console.log("\n------\n");
    if (result.success) {
        console.log(result.data);
    }
    else {
        console.log("Uh oh!");
        process.exit(-1);
    }
});

function createStreamingCompleter(onContent: (content: string) => void) {
    return async function complete(prompt: string) {
        const response = await fetch("https://api.openai.com/v1/chat/completions", {
            method: "POST",
            headers: {
                Authorization: `Bearer ${apiKey}`,
                "Content-Type": "application/json",
            },
            body: JSON.stringify({
                model: modelName,
                stream: true,
                temperature: 0,
                n: 1,
                messages: [
                    { role: "user", content: prompt }
                ],
            })
        });

        assert(response.body);

        let result = "";
        for await (const data of parseDataEvents(response.body)) {
            const deltaContent = JSON.parse(data).choices[0].delta.content;
            if (deltaContent !== undefined) {
                onContent(deltaContent);
            }
            result += deltaContent;
        }
        return success(result);
    }
}

/**
 * @param {AsyncIterable<Buffer | string>} stream 
 */
async function* parseDataEvents(stream) {
    let tmp = ""
    for await (let part of stream) {
        part = part.toString();
        let start = 0;
        let end = 0;
        while ((end = part.indexOf("\n\n", start)) >= 0) {
            tmp += part.slice(start, end);
            if (tmp.startsWith("data: ")) {
                tmp = tmp.slice("data: ".length);
                if (tmp === "[DONE]") {
                    break;
                }
                yield tmp;
            }
            tmp = "";
            start = end + 2
        }
    }
}