I’m trying to create a chat bot using open ai and streaming its responses.

I have an API written with nestjs. Here is how I communicate with open ai:

public async test(): Promise<Observable<string | null>> {
  return new Observable((subscriber) => {
    this._openAi.chat.completions
      .create({
        messages: [
          { role: "system", content: "You are a helpful assistant." },
        ],
        model: "gpt-4-turbo-2024-04-09",
        temperature: 1,
        stream: true,
      })
      .then(async (response) => {
        for await (const chunk of response) {
          if (chunk.choices[0].finish_reason === "stop") {
            subscriber.complete();
            return;
          }
          const content = chunk.choices[0].delta.content;
          console.log("sending content", content);
          subscriber.next(content);
        }
      });
  });
}

And the controller:

@Sse()
public async test(): Promise<Observable<string | null>> {
  return this._service.test();
}

This works fine, when I call the endpoint via the browser, I can see the list a message and it stops when done.

Now, I tried to consume it via an Angular application. Here is the service:

@Injectable({ providedIn: "root" })
export class AppService {
    private _eventSource: EventSource | null = null;

    constructor(private _ngZone: NgZone) {}

    getEventSource(url: string): EventSource {
        return new EventSource(url);
    }

    connectToServerSentEvents(): Observable<MessageEvent> {
        this._eventSource = this.getEventSource("http://localhost:3003/chat");

        return new Observable((subscriber: Subscriber<MessageEvent>) => {
            if (!this._eventSource) {
                return;
            }

            this._eventSource.onerror = (error) => {
                this._ngZone.run(() => subscriber.error(error));
            };

            this._eventSource.onmessage = (event) => {
                this._ngZone.run(() => subscriber.next(event));
            };
        });
    }

    close(): void {
        if (!this._eventSource) {
            return;
        }

        this._eventSource.close();
        this._eventSource = null;
    }
}

The subscribe:

export class AppComponent implements OnDestroy {
    private readonly _eventSourceSubscription: SubscriptionLike;

    constructor(private _service: AppService) {
        this._eventSourceSubscription = this._service.connectToServerSentEvents()
      .subscribe((data) => console.log("data", data));
    }

    public ngOnDestroy(): void {
        this._eventSourceSubscription.unsubscribe();
        this._service.close();
    }
}

When I load the page, the connection opens, and I can see the data coming in. However, two issues here:

1. The call keeps looping. Meaning, once it has reached the end of the stream, it calls the endpoint again. How can I stop that?
2. I don’t see any option to send post data, like the user input, the context, history etc. How can I send data to the endpoint?