Guide — Streaming SSE

Principe

POST /spaces/:id/query supporte deux modes selon l'en-tête Accept :

Accept	Mode
application/json	Réponse complète
text/event-stream	Streaming SSE

Événements

Événement	Contenu
status	`{"stage":"retrieving\
sources	{"sources":[...]} — sources avant génération (affichage anticipé)
delta	{"text":"..."} — incrément de texte
done	Réponse finale complète + sources filtrées + usage
error	{"error":{"code":"...","message":"..."}}
ping	{} — keepalive toutes les 15 s

Ruby (net/http)

require "net/http"
require "json"
require "uri"

def stream_query(space_id:, question:, api_key:)
  uri = URI("https://pyxaura.com/api/v1/spaces/#{space_id}/query")
  accumulated = ""

  Net::HTTP.start(uri.host, uri.port, use_ssl: true) do |http|
    req = Net::HTTP::Post.new(uri)
    req["Authorization"] = "Bearer #{api_key}"
    req["Content-Type"]  = "application/json"
    req["Accept"]        = "text/event-stream"
    req.body = { question: question, options: { persist: false } }.to_json

    http.request(req) do |response|
      buffer = ""
      response.read_body do |chunk|
        buffer += chunk
        while (match = buffer.match(/^event: (\w+)\ndata: (.+)\n\n/))
          event, data_str = match[1], match[2]
          buffer = buffer[match.end(0)..]
          data = JSON.parse(data_str) rescue {}

          case event
          when "delta"   then accumulated += data["text"].to_s; print data["text"]
          when "done"    then puts "\n\n---\nSources : #{data["sources"]&.map { |s| s["title"] }.join(", ")}"
          when "error"   then raise "Erreur API : #{data.dig("error", "message")}"
          end
        end
      end
    end
  end
  accumulated
end

answer = stream_query(
  space_id: "spc_9f2",
  question: "Quel est le délai de retour ?",
  api_key:  ENV["PYXIA_API_KEY"]
)

JavaScript (fetch + ReadableStream)

async function streamQuery({ spaceId, question, apiKey }) {
  const response = await fetch(`/api/v1/spaces/${spaceId}/query`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${apiKey}`,
      "Content-Type": "application/json",
      Accept: "text/event-stream",
    },
    body: JSON.stringify({ question, options: { persist: false } }),
  });

  if (!response.ok) throw new Error(`HTTP ${response.status}`);

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

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const parts = buffer.split("\n\n");
    buffer = parts.pop();

    for (const part of parts) {
      const lines    = part.split("\n");
      const evLine   = lines.find(l => l.startsWith("event:"));
      const dataLine = lines.find(l => l.startsWith("data:"));
      if (!dataLine) continue;

      const event = evLine?.replace("event:", "").trim() ?? "message";
      const data  = JSON.parse(dataLine.replace("data:", "").trim());

      if (event === "delta")  { answer += data.text; process.stdout.write(data.text); }
      if (event === "done")   console.log("\n\nSources :", data.sources?.map(s => s.title));
      if (event === "error")  throw new Error(data.error?.message);
    }
  }
  return answer;
}

Notes importantes

• EventSource natif ne supporte pas les en-têtes custom → utiliser fetch + ReadableStream (voir exemples ci-dessus).
• Reconnexion : Last-Event-ID non supporté en v1. En cas de coupure, relancer la requête entière.
• Proxy/load balancer : ajouter X-Accel-Buffering: no (Nginx) ou équivalent pour désactiver le buffering.
• Timeout : le ping toutes les 15 s maintient la connexion. Configurez votre proxy pour tolérer les connexions longues (ex. proxy_read_timeout 120s dans Nginx).