MyApp/myapp/src/app/tab3/class-rxjs-chat-completion.ts

import { Observable, from, of } from 'rxjs';
import { switchMap, map, catchError, finalize } from 'rxjs/operators';

export interface TestRxjsChatMessage {
    role: string;
    content: string;
}

export class TestRxjsChatCompletion {
    messageList: Array<TestRxjsChatMessage>;

    constructor(messageList: Array<TestRxjsChatMessage>) {
        this.messageList = messageList;
    }

    createCompletionByStream(options?:{
        model?:string
    }): Observable<{ content: string, cumulativeContent: string, done: boolean }> {
        const token = localStorage.getItem("token");
        const bodyJson = {
            "token": `Bearer ${token}`,
            "messages": this.messageList,
            "model": options?.model || "fmode-3.6-16k",
            "temperature": 0.5,
            "presence_penalty": 0,
            "frequency_penalty": 0,
            "top_p": 1,
            "stream": true
        };

        return from(fetch("https://test.fmode.cn/api/apig/aigc/gpt/v1/chat/completions", {
            "headers": {
                "accept": "text/event-stream",
                "sec-fetch-dest": "empty",
                "sec-fetch-mode": "cors",
                "sec-fetch-site": "same-site"
            },
            "referrer": "https://ai.fmode.cn/",
            "referrerPolicy": "strict-origin-when-cross-origin",
            "body": JSON.stringify(bodyJson),
            "method": "POST",
            "mode": "cors",
            "credentials": "omit"
        })).pipe(
            switchMap(response => {
                const reader = response.body?.getReader();
                if (!reader) {
                    throw new Error("Failed to get the response reader.");
                }
                const decoder = new TextDecoder();
                let buffer = "";
                let messageAiReply = "";
                let messageIndex = this.messageList.length;

                return new Observable<{ content: string, cumulativeContent: string, done: boolean }>(observer => {
                    const read = () => {
                        reader.read().then(({ done, value }) => {
                            if (done) {
                                observer.complete();
                                return;
                            }

                            buffer += decoder.decode(value);
                            let messages = buffer.split("\n");

                            for (let i = 0; i < messages.length - 1; i++) {
                                let message = messages[i];
                                let dataText = message.replace("data: ", "");

                                if (dataText.startsWith("{")) {
                                    try {
                                        let dataJson = JSON.parse(dataText);
                                        let content = dataJson?.choices?.[0]?.delta?.content || "";
                                        messageAiReply += content;
                                        this.messageList[messageIndex] = {
                                            role: "assistant",
                                            content: messageAiReply
                                        };
                                        observer.next({ content, cumulativeContent: messageAiReply, done: false });
                                    } catch (err) { }
                                }

                                if (dataText.startsWith("[")) {
                                    this.messageList[messageIndex] = {
                                        role: "assistant",
                                        content: messageAiReply
                                    };
                                    observer.next({ content: "", cumulativeContent: messageAiReply, done: true });
                                    messageAiReply = "";
                                }

                                buffer = buffer.slice(message.length + 1);
                            }

                            read();
                        }).catch(err => observer.error(err));
                    };

                    read();
                });
            }),
            catchError(err => {
                console.error(err);
                return of({ content: "", cumulativeContent: "", done: true });
            }),
            finalize(() => {
                console.log("Stream completed");
            })
        );
    }
}