Method handleStreamResponse

packages/core/src/pipes/pipes.ts:180–247

(
		options: RunOptionsStream,
		response: RunResponseStream,
	)

	}

	private async handleStreamResponse(
		options: RunOptionsStream,
		response: RunResponseStream,
	): Promise<RunResponseStream> {
		const endpoint = '/v1/pipes/run';
		const stream = this.isStreamRequested(options);
		const body = {...options, stream};

		const [streamForToolCall, streamForReturn] = response.stream.tee();
		const tools = await getToolsFromStream(streamForToolCall);

		if (tools.length) {
			let messages = options.messages || [];

			let currentResponse: RunResponseStream = {
				stream: streamForReturn,
				threadId: response.threadId,
				rawResponse: response.rawResponse,
			};

			let callCount = 0;

			while (callCount < this.maxCalls) {
				const [streamForToolCall, streamForReturn] =
					currentResponse.stream.tee();

				const tools = await getToolsFromStream(streamForToolCall);

				if (tools.length === 0) {
					return {
						stream: streamForReturn,
						threadId: currentResponse.threadId,
						rawResponse: response.rawResponse,
					};
				}

				const toolResults = await this.runTools(tools);

				const responseMessage = {
					role: 'assistant',
					content: null,
					tool_calls: tools,
				} as Message;

				messages = this.getMessagesToSend(
					messages,
					responseMessage,
					toolResults,
				);

				currentResponse = await this.createRequest<RunResponseStream>(
					endpoint,
					{
						...body,
						messages,
						threadId: currentResponse.threadId,
					},
				);