Function streamPipelineChat

ui/frontend/src/shared/api/streaming.ts:22–93

(
  payload: StreamPipelineChatPayload,
)

}

export async function* streamPipelineChat(
  payload: StreamPipelineChatPayload,
): AsyncGenerator<PipelineChatEvent> {
  const endpoint = buildApiUrl(`/api/pipelines/${encodeURIComponent(payload.pipelineName)}/chat`);
  const dynamicParams: Record<string, unknown> = {
    memory: { user_id: payload.userId },
  };
  if (payload.collectionName) {
    dynamicParams.collection_name = payload.collectionName;
  }

  const response = await fetch(endpoint, {
    method: "POST",
    credentials: "include",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      question: payload.question,
      history: payload.history,
      is_demo: true,
      session_id: payload.sessionId,
      chat_session_id: payload.chatSessionId,
      dynamic_params: dynamicParams,
    }),
    signal: payload.signal,
  });

  if (!response.ok) {
    let apiMessage = response.statusText;
    try {
      const errorPayload = (await response.json()) as { error?: string; details?: string };
      apiMessage = errorPayload.error ?? errorPayload.details ?? apiMessage;
    } catch {
      // noop
    }
    throw new ApiHttpError(response.status, { error: apiMessage }, "Pipeline chat request failed");
  }

  if (!response.body) {
    throw new Error("Pipeline chat stream body is empty");
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const blocks = buffer.split("\n\n");
    buffer = blocks.pop() ?? "";

    for (const block of blocks) {
      const line = block
        .split("\n")
        .map((segment) => segment.trim())