Rechercher la documentation ...Rechercher

CtrlK

Guide d'Intégration du Streaming Chat Completions

Vue d'ensemble

Ce guide explique comment intégrer correctement l'API de streaming chat completions de Devana dans votre application. Il couvre la gestion du streaming SSE (Server-Sent Events), le parsing des tokens, la gestion des balises spéciales (<think>, [tool:]), et les bonnes pratiques d'implémentation.

Architecture du streaming

Le système de streaming repose sur le protocole SSE (Server-Sent Events) qui permet au serveur d'envoyer des mises à jour en temps réel au client. Chaque "chunk" (morceau) reçu contient un token qui peut être :

Frontend Streaming - devana.ai/docs

data: {"choices":[{"delta":{"content":"token"}}],"conversation_id":"conv-123"}

data: [DONE]

POST /v1/chat/completions

{
  "Content-Type": "application/json",
  "Authorization": "Bearer YOUR_TOKEN"
}

{
  "stream": true,                          // Active le streaming
  "messages": [
    { "role": "user", "content": "..." }   // Vos messages
  ],
  "files": ["file-id-1", "file-id-2"],    // IDs de fichiers (optionnel)
  "clientModel": "model-name",             // Modèle spécifique (optionnel)
  "conversation_id": "conv-123",           // ID de conversation (optionnel)
  "model": "agent-id"                      // ID de l'agent
}

const response = await fetch("/v1/chat/completions", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${token}`,
  },
  body: JSON.stringify({
    stream: true,
    messages: [{ role: "user", content: message }],
    conversation_id: chatId,
    model: agentId,
  }),
  signal: abortController.signal, // Pour permettre l'annulation
});

const reader = response.body.getReader();
let buffer = "";
let lastFailedChunk = "";

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

  // Ajouter les nouvelles données au buffer
  buffer += new TextDecoder().decode(value);

  let splitIndex: number;

  // Traiter tous les messages complets dans le buffer
  while ((splitIndex = buffer.indexOf("\n\n")) !== -1) {
    const messageWithLastFail = buffer.slice(0, splitIndex).trim();
    buffer = buffer.slice(splitIndex + 2);

    // Ignorer les messages non-data
    if (!messageWithLastFail.startsWith("data:")) continue;

    // Fin du flux
    if (messageWithLastFail === "data: [DONE]") continue;

    try {
      // Extraire et parser le JSON
      const jsonString = messageWithLastFail.replace(/^data:\s*/, "");
      const jsonData = JSON.parse(jsonString);
      lastFailedChunk = "";

      // Extraire le token
      const token = jsonData?.choices?.[0]?.delta?.content;
      if (token) {
        processStreamChunk(token); // Traiter le token
      }

      // Capturer le conversation_id pour les nouvelles conversations
      if (jsonData.conversation_id) {
        conversationId = jsonData.conversation_id;
      }
    } catch (e) {
      // Si le parsing échoue, remettre dans le buffer
      // (peut arriver si le message est incomplet)
      if (messageWithLastFail === lastFailedChunk) continue;
      lastFailedChunk = messageWithLastFail;
      buffer = messageWithLastFail + "\n\n" + buffer;
      break;
    }
  }
}

<think>
Je dois d'abord vérifier les données...
Ensuite calculer le résultat...
</think>
Voici la réponse finale

function extractThinkAndMainContent(message: string) {
  let thinkContent = "";
  let mainContent = "";
  let currentPos = 0;
  let hasThinkBlock = false;

  while (currentPos < message.length) {
    const thinkStart = message.indexOf("<think>", currentPos);

    if (thinkStart === -1) {
      // Plus de balises <think>, ajouter le reste au contenu principal
      mainContent += message.substring(currentPos);
      break;
    }

    hasThinkBlock = true;
    // Ajouter le contenu avant <think> au contenu principal
    mainContent += message.substring(currentPos, thinkStart);

    const thinkEnd = message.indexOf("</think>", thinkStart);

    if (thinkEnd === -1) {
      // Bloc <think> ouvert (streaming en cours)
      const content = message.substring(thinkStart + 7);
      if (thinkContent && content) thinkContent += "\n\n---\n\n";
      thinkContent += content;
      break;
    } else {
      // Bloc <think> fermé
      const content = message.substring(thinkStart + 7, thinkEnd);
      if (thinkContent && content) thinkContent += "\n\n---\n\n";
      thinkContent += content;
      currentPos = thinkEnd + 8;
    }
  }

  // Nettoyage
  thinkContent = thinkContent.trim();
  mainContent = mainContent.replace(/<\/think>/g, "").trim();

  return { thinkContent, mainContent, hasThinkBlock };
}

// Dans votre composant de message
const { thinkContent, mainContent, hasThinkBlock } = extractThinkAndMainContent(message);

return (
  <>
    {hasThinkBlock && (
      <ThinkPanel
        content={thinkContent}
        isStreaming={isStreaming}
      />
    )}
    <MarkdownContent content={mainContent} />
  </>
);

// Démarrage d'un outil (toujours en JSON)
[tool:start:{"name":"web_search","args":{"query":"météo Paris"}}]

// Fin d'un outil - Format 1 : Simple (juste le nom)
[tool:end:web_search]

// Fin d'un outil - Format 2 : JSON structuré
[tool:end:{"name":"web_search"}]

interface ToolInfo {
  id: string;
  name: string;
  args: Record<string, unknown>;
  startTime: number;
  endTime?: number;
}

function processStreamChunk(token: string) {
  // Détecter les balises tool
  if (token.startsWith("[tool:")) {
    const toolMatch = token.match(/\[tool:(start|end):(.+)\]/);

    if (toolMatch) {
      const [, action, dataString] = toolMatch;

      try {
        let toolData;

        // Essayer de parser en JSON d'abord
        try {
          toolData = JSON.parse(dataString);
        } catch {
          // Si ce n'est pas du JSON, c'est probablement juste le nom (format simple)
          // Format : [tool:end:ToolName]
          toolData = { name: dataString };
        }

        if (action === "start") {
          // Créer un ID unique pour tracer l'outil
          const uniqueId = `${toolData.name}-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;

          const toolInfo: ToolInfo = {
            id: uniqueId,
            name: toolData.name,
            args: toolData.args || {},
            startTime: Date.now(),
          };

          // Ajouter aux outils en cours
          setRunnedTools((prev) => [...prev, toolInfo]);
          setActiveTool(toolInfo);
        } else if (action === "end") {
          // Marquer l'outil comme terminé
          setActiveTool((prev) => {
            if (prev?.name === toolData.name) {
              setRunnedTools((tools) =>
                tools.map((t) =>
                  t.id === prev.id ? { ...t, endTime: Date.now() } : t,
                ),
              );
              return null;
            }
            return prev;
          });
        }
      } catch (e) {
        console.error("Error parsing tool data", e);
      }
    }

    // Important : ne pas ajouter ce token au contenu du message
    return;
  }

  // Token normal, l'ajouter au message
  setStreamedResponse((prev) => prev + token);
}

interface ToolExecutionPanelProps {
  runnedTools: ToolInfo[];
  activeTool: ToolInfo | null;
  isStreaming: boolean;
}

function ToolExecutionPanel({ runnedTools, activeTool, isStreaming }: ToolExecutionPanelProps) {
  return (
    <div className="tool-panel">
      <h4>
        {isStreaming ? "Exécution des outils" : "Outils exécutés"}
        <Badge count={runnedTools.length} />
      </h4>

      {runnedTools.map((tool, index) => (
        <ToolStep
          key={tool.id}
          toolInfo={tool}
          status={activeTool?.id === tool.id ? "running" : "completed"}
          index={index}
        />
      ))}

      {/* Afficher l'outil actif s'il n'est pas déjà dans la liste */}
      {activeTool && !runnedTools.some(t => t.id === activeTool.id) && (
        <ToolStep
          toolInfo={activeTool}
          status="running"
          index={runnedTools.length}
        />
      )}
    </div>
  );
}

[tool:confirm:{"messageId":"msg-123","toolName":"delete_file","args":{...}}]

const TOOL_CONFIRM_KEY = "[tool:confirm:";

// Dans le rendu des messages
if (message.startsWith(TOOL_CONFIRM_KEY)) {
  return (
    <ToolConfirmDialog
      message={message}
      onConfirm={async (confirmed) => {
        // Envoyer la réponse de confirmation
        await sendMessage(
          `[tool:confirm:${JSON.stringify({
            messageId: message.id,
            confirm: confirmed
          })}]`
        );
      }}
    />
  );
}

interface Message {
  id: string;
  role: "USER" | "ASSISTANT" | "SYSTEM";
  message: string;
  status: "complete" | "streaming" | "temporary" | "error";
  fiability?: "GOOD" | "BAD" | "DEFAULT";
  comment?: string;
  files?: File[];
  sources?: Source[];
  runnedTools?: ToolInfo[];
  activeTool?: ToolInfo | null;
}

function useChat() {
  const [isStreaming, setIsStreaming] = useState(false);
  const [streamedResponse, setStreamedResponse] = useState("");
  const [temporaryQuestion, setTemporaryQuestion] = useState("");
  const [waitingForResponse, setWaitingForResponse] = useState(false);
  const [runnedTools, setRunnedTools] = useState<ToolInfo[]>([]);
  const [activeTool, setActiveTool] = useState<ToolInfo | null>(null);

  // Messages combinés : historique + messages temporaires
  const messages = useMemo(() => {
    const msgs: Message[] = [...historyMessages];

    // Ajouter la question temporaire
    if (temporaryQuestion) {
      msgs.push({
        id: "temp-user",
        role: "USER",
        message: temporaryQuestion,
        status: "temporary",
      });
    }

    // Ajouter la réponse en cours de streaming
    if (
      streamedResponse ||
      isStreaming ||
      activeTool ||
      runnedTools.length > 0
    ) {
      // Vérifier qu'on ne duplique pas avec l'historique
      const lastHistoryMsg = msgs[msgs.length - 1];
      const cleanStream = streamedResponse
        .replace(/<think>[\s\S]*?<\/think>/, "")
        .trim();
      const cleanHistory = lastHistoryMsg?.message
        ?.replace(/<think>[\s\S]*?<\/think>/, "")
        ?.trim();

      const isDuplicate =
        lastHistoryMsg?.role === "ASSISTANT" &&
        !isStreaming &&
        (lastHistoryMsg?.message === streamedResponse ||
          cleanHistory === cleanStream ||
          (cleanStream.length > 10 && cleanHistory?.includes(cleanStream)));

      if (!isDuplicate) {
        msgs.push({
          id: "streaming-assistant",
          role: "ASSISTANT",
          message: streamedResponse,
          status: isStreaming ? "streaming" : "complete",
          runnedTools,
          activeTool,
        });
      }
    }

    return msgs;
  }, [
    historyMessages,
    temporaryQuestion,
    streamedResponse,
    isStreaming,
    runnedTools,
    activeTool,
  ]);

  return { messages /* ... */ };
}

// Comparer le contenu sans les balises <think>
const cleanStream = streamedResponse
  .replace(/<think>[\s\S]*?<\/think>/, "")
  .trim();
const cleanHistory = lastHistoryMsg?.message
  ?.replace(/<think>[\s\S]*?<\/think>/, "")
  ?.trim();

const isDuplicate =
  lastHistoryMsg?.role === "ASSISTANT" &&
  !isStreaming &&
  (lastHistoryMsg?.message === streamedResponse ||
    cleanHistory === cleanStream ||
    (cleanStream.length > 10 && cleanHistory?.includes(cleanStream)));

const abortControllerRef = useRef<AbortController | null>(null);

function sendMessage(message: string) {
  // Créer un nouveau controller pour cette requête
  abortControllerRef.current = new AbortController();

  try {
    const response = await fetch(url, {
      // ...
      signal: abortControllerRef.current.signal,
    });
    // ...
  } catch (error) {
    if (error.name !== "AbortError") {
      // Erreur réelle, afficher un message
      toast.error("Impossible d'envoyer le message");
    }
    // Si c'est AbortError, c'est une annulation volontaire
  } finally {
    abortControllerRef.current = null;
  }
}

function stopRequest() {
  if (abortControllerRef.current) {
    abortControllerRef.current.abort();
  }
  setIsStreaming(false);
  setWaitingForResponse(false);
  setTemporaryQuestion("");
}

async function sendMessage(message: string) {
  try {
    const response = await fetch(url, {
      /* ... */
    });

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

    if (!response.body) {
      throw new Error("No response body");
    }

    // Traiter le flux...
  } catch (error) {
    if (error.name === "AbortError") {
      // Annulation volontaire, ne rien faire
      return;
    }

    // Erreur réelle
    console.error("Stream error:", error);
    toast.error("Erreur lors de l'envoi du message");

    // Réinitialiser les états
    setIsStreaming(false);
    setWaitingForResponse(false);
    setTemporaryQuestion("");
  }
}

// Créer un timeout pour la requête
const timeoutId = setTimeout(() => {
  abortControllerRef.current?.abort();
  toast.error("La requête a pris trop de temps");
}, 60000); // 60 secondes

try {
  // ... traitement du streaming
} finally {
  clearTimeout(timeoutId);
}

import { useCallback, useEffect, useMemo, useRef, useState } from "react";

interface ToolInfo {
  id: string;
  name: string;
  args: Record<string, unknown>;
  startTime: number;
  endTime?: number;
}

interface Message {
  id: string;
  role: "USER" | "ASSISTANT" | "SYSTEM";
  message: string;
  status: "complete" | "streaming" | "temporary" | "error";
  runnedTools?: ToolInfo[];
  activeTool?: ToolInfo | null;
}

export function useChat({ chatId, agentId, userToken }) {
  const [isStreaming, setIsStreaming] = useState(false);
  const [streamedResponse, setStreamedResponse] = useState("");
  const [temporaryQuestion, setTemporaryQuestion] = useState("");
  const [waitingForResponse, setWaitingForResponse] = useState(false);
  const [runnedTools, setRunnedTools] = useState<ToolInfo[]>([]);
  const [activeTool, setActiveTool] = useState<ToolInfo | null>(null);
  const abortControllerRef = useRef<AbortController | null>(null);

  // Charger l'historique des messages (GraphQL, REST, etc.)
  const historyMessages = useMemo(() => {
    // Votre logique de chargement
    return [];
  }, [chatId]);

  // Messages combinés
  const messages = useMemo(() => {
    const msgs: Message[] = [...historyMessages];

    if (temporaryQuestion) {
      msgs.push({
        id: "temp-user",
        role: "USER",
        message: temporaryQuestion,
        status: "temporary",
      });
    }

    if (
      streamedResponse ||
      isStreaming ||
      activeTool ||
      runnedTools.length > 0
    ) {
      msgs.push({
        id: "streaming-assistant",
        role: "ASSISTANT",
        message: streamedResponse,
        status: isStreaming ? "streaming" : "complete",
        runnedTools,
        activeTool,
      });
    }

    return msgs;
  }, [
    historyMessages,
    temporaryQuestion,
    streamedResponse,
    isStreaming,
    runnedTools,
    activeTool,
  ]);

  // Traiter un chunk du stream
  const processStreamChunk = useCallback((token: string) => {
    // Gestion des outils
    if (token.startsWith("[tool:")) {
      const toolMatch = token.match(/\[tool:(start|end):(.+)\]/);
      if (toolMatch) {
        const [, action, dataString] = toolMatch;
        try {
          let toolData;

          // Essayer de parser en JSON d'abord
          try {
            toolData = JSON.parse(dataString);
          } catch {
            // Si ce n'est pas du JSON, c'est le format simple : [tool:end:ToolName]
            toolData = { name: dataString };
          }

          if (action === "start") {
            const uniqueId = `${toolData.name}-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
            const toolInfo: ToolInfo = {
              id: uniqueId,
              name: toolData.name,
              args: toolData.args || {},
              startTime: Date.now(),
            };
            setRunnedTools((prev) => [...prev, toolInfo]);
            setActiveTool(toolInfo);
          } else if (action === "end") {
            setActiveTool((prev) => {
              if (prev?.name === toolData.name) {
                setRunnedTools((tools) =>
                  tools.map((t) =>
                    t.id === prev.id ? { ...t, endTime: Date.now() } : t,
                  ),
                );
                return null;
              }
              return prev;
            });
          }
        } catch (e) {
          console.error("Error parsing tool data", e);
        }
      }
      return;
    }

    // Token normal
    setStreamedResponse((prev) => prev + token);
  }, []);

  // Envoyer un message
  const sendMessage = useCallback(
    async (message: string, files: string[] = []) => {
      if (!message) return;

      setTemporaryQuestion(message);
      setWaitingForResponse(true);
      setIsStreaming(true);
      setStreamedResponse("");
      setRunnedTools([]);
      setActiveTool(null);

      const url = "/v1/chat/completions";
      abortControllerRef.current = new AbortController();

      try {
        const response = await fetch(url, {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            Authorization: `Bearer ${userToken}`,
          },
          body: JSON.stringify({
            stream: true,
            messages: [{ role: "user", content: message }],
            files,
            conversation_id: chatId,
            model: agentId,
          }),
          signal: abortControllerRef.current.signal,
        });

        if (!response.ok || !response.body) {
          throw new Error("Request failed");
        }

        setWaitingForResponse(false);

        const reader = response.body.getReader();
        let buffer = "";
        let lastFailedChunk = "";
        let finalChatId = chatId;

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

          buffer += new TextDecoder().decode(value);
          let splitIndex: number;

          while ((splitIndex = buffer.indexOf("\n\n")) !== -1) {
            const messageWithLastFail = buffer.slice(0, splitIndex).trim();
            buffer = buffer.slice(splitIndex + 2);

            if (!messageWithLastFail.startsWith("data:")) continue;
            if (messageWithLastFail === "data: [DONE]") continue;

            try {
              const jsonString = messageWithLastFail.replace(/^data:\s*/, "");
              const jsonData = JSON.parse(jsonString);
              lastFailedChunk = "";

              const token = jsonData?.choices?.[0]?.delta?.content;
              if (token) processStreamChunk(token);

              if (
                jsonData.conversation_id &&
                (!chatId || chatId !== jsonData.conversation_id)
              ) {
                finalChatId = jsonData.conversation_id;
              }
            } catch (e) {
              if (messageWithLastFail === lastFailedChunk) continue;
              lastFailedChunk = messageWithLastFail;
              buffer = messageWithLastFail + "\n\n" + buffer;
              break;
            }
          }
        }

        // Stream terminé
        setIsStreaming(false);
        setTemporaryQuestion("");

        // Rafraîchir l'historique
        if (finalChatId) {
          // Votre logique pour rafraîchir avec le nouvel ID
        }
      } catch (error) {
        if (error.name !== "AbortError") {
          console.error("Stream error:", error);
          // Afficher une erreur à l'utilisateur
        }
        setIsStreaming(false);
        setWaitingForResponse(false);
      } finally {
        abortControllerRef.current = null;
      }
    },
    [chatId, agentId, userToken, processStreamChunk],
  );

  // Arrêter le streaming
  const stopRequest = useCallback(() => {
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
    }
    setIsStreaming(false);
    setWaitingForResponse(false);
    setTemporaryQuestion("");
  }, []);

  return {
    messages,
    sendMessage,
    stopRequest,
    isStreaming,
    waitingForResponse,
  };
}

useEffect(() => {
  if (isStreaming && isAttachedToBottom) {
    requestAnimationFrame(() => {
      scrollContainerRef.current?.scrollTo({
        top: scrollContainerRef.current.scrollHeight,
        behavior: "auto",
      });
    });
  }
}, [messages, isStreaming, isAttachedToBottom]);

<button
  onClick={stopRequest}
  disabled={!isStreaming}
>
  {isStreaming ? "Arrêter la génération" : "Génération terminée"}
</button>

useEffect(() => {
  // Reset quand la conversation change
  setStreamedResponse("");
  setIsStreaming(false);
  setTemporaryQuestion("");
  setWaitingForResponse(false);
  setRunnedTools([]);
  setActiveTool(null);
}, [chatId]);

<div role="log" aria-live="polite" aria-atomic="false">
  {messages.map(msg => (
    <div key={msg.id} aria-label={`Message de ${msg.role}`}>
      {msg.message}
    </div>
  ))}
</div>

function escapeHtml(text: string): string {
  return text
    .replace(/&/g, "&amp;")
    .replace(/</g, "&lt;")
    .replace(/>/g, "&gt;")
    .replace(/"/g, "&quot;")
    .replace(/'/g, "&#039;");
}

// Appliquer uniquement en dehors des blocs de code
function addEscapeOutsideCodeBlocks(input: string): string {
  const segments = input.split(/(```[\s\S]*?```|`[^`]*`)/);
  return segments
    .map((segment, i) => {
      if (i % 2 === 0) {
        // Échapper uniquement en dehors des blocs de code
        return segment.replace(/</g, "&lt;").replace(/>/g, "&gt;");
      }
      return segment;
    })
    .join("");
}

// Test du parsing SSE
describe("SSE Parsing", () => {
  it("should parse complete messages", () => {
    const chunk = 'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n';
    const result = parseSSEChunk(chunk);
    expect(result.token).toBe("Hello");
  });

  it("should handle incomplete messages", () => {
    const chunk1 = 'data: {"choices":[{"delta":';
    const chunk2 = '{"content":"Hello"}}]}\n\n';
    const buffer = chunk1 + chunk2;
    const result = parseSSEChunk(buffer);
    expect(result.token).toBe("Hello");
  });
});

// Test de l'extraction des balises think
describe("Think Tag Extraction", () => {
  it("should extract think content", () => {
    const message = "<think>Thinking...</think>Response";
    const { thinkContent, mainContent } = extractThinkAndMainContent(message);
    expect(thinkContent).toBe("Thinking...");
    expect(mainContent).toBe("Response");
  });

  it("should handle unclosed think tags", () => {
    const message = "<think>Thinking...";
    const { thinkContent, mainContent } = extractThinkAndMainContent(message);
    expect(thinkContent).toBe("Thinking...");
    expect(mainContent).toBe("");
  });
});

// Test des balises tool
describe("Tool Tags", () => {
  it("should parse tool start", () => {
    const token = '[tool:start:{"name":"search","args":{"q":"test"}}]';
    const result = parseToolTag(token);
    expect(result.action).toBe("start");
    expect(result.name).toBe("search");
    expect(result.args).toEqual({ q: "test" });
  });
});

Guide d'Intégration du Streaming Chat Completions

Vue d'ensemble

Table des matières

Architecture du streaming

Guide d'Intégration du Streaming Chat Completions

Vue d'ensemble

Table des matières

Architecture du streaming

Format des messages SSE

Configuration de la requête

Endpoint

Headers requis

Corps de la requête

Exemple avec Fetch API

Parsing du flux SSE

Algorithme de parsing

Points clés du parsing

Gestion des balises spéciales

1. Balises <think> - Réflexions internes

Format

Extraction

Affichage UI

2. Balises [tool:] - Exécution d'outils

Format

Parsing et gestion

Affichage UI des outils

3. Balises [tool:confirm:] - Confirmation utilisateur

Format

Gestion

Gestion de l'état des messages

Structure des messages

États du flux de conversation

Gestion des états

Prévention des duplications

Gestion des erreurs

Annulation de requête

Gestion des erreurs réseau

Timeout

Exemple d'implémentation complète

Bonnes pratiques

1. Performance

2. UX

3. Gestion d'état

4. Accessibilité

5. Sécurité

6. Tests

Résumé

Support

1. Balises `<think>` - Réflexions internes

2. Balises `[tool:]` - Exécution d'outils

3. Balises `[tool:confirm:]` - Confirmation utilisateur