Sortie en streaming¶

Cette rubrique décrit comment diffuser des réponses en temps réel à partir du SDK Agent Cortex Code.

Par défaut, leSDK rend des objets``AssistantMessage`` complets une fois que le modèle a fini de générer chaque réponse. Pour recevoir des mises à jour incrémentielles à mesure que du texte et des blocs de réflexion sont générés, activez le streaming partiel des messages en définissant includePartialMessages (TypeScript) ou include_partial_messages (Python) sur``true``.

Lorsque des messages partiels sont activés, Cortex Code émet des objets StreamEvent pour le texte partiel et le contenu de la réflexion. Les appels complets des outils arrivent toujours en tant qu’objets AssistantMessage et les résultats des outils arrivent toujours sous la forme d’objets UserMessage.

Activer la sortie en streaming¶

Lorsqu’il est activé, le SDK rend des messages StreamEvent contenant des événements de streaming partiels, en plus des objets AssistantMessage,``UserMessage``, et ResultMessage habituels. Votre code doit :

Vérifier le type de chaque message pour différencier StreamEvent des autres types.
Pour StreamEvent, extraire le champ event et vérifier son type.
Chercher des événements content_block_delta où``delta.type`` est text_delta.

import { query } from "cortex-code-agent-sdk";

for await (const message of query({
  prompt: "List the files in my project",
  options: {
    cwd: process.cwd(),
    includePartialMessages: true,
    allowedTools: ["Bash", "Read"],
  },
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta") {
      if (event.delta.type === "text_delta") {
        process.stdout.write(event.delta.text);
      }
    }
  }
}

import asyncio
from cortex_code_agent_sdk import query, CortexCodeAgentOptions
from cortex_code_agent_sdk.types import StreamEvent

async def stream_response():
    async for message in query(
        prompt="List the files in my project",
        options=CortexCodeAgentOptions(
            cwd=".",
            include_partial_messages=True,
            allowed_tools=["Bash", "Read"],
        ),
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)

asyncio.run(stream_response())

Référence StreamEvent¶

Lorsque les messages partiels sont activés, vous recevez des événements de streaming bruts encapsulés dans un objet :

interface SDKPartialAssistantMessage {
  type: "stream_event";
  event: Record<string, unknown>;  // Raw streaming event
  parent_tool_use_id: string | null;
  uuid: string;
  session_id: string;
}

@dataclass
class StreamEvent:
    uuid: str               # Unique identifier
    session_id: str          # Session identifier
    event: dict[str, Any]    # Raw streaming event
    parent_tool_use_id: str | None  # Parent tool ID if from a subagent

Le champ event contient l’événement de streaming partiel brut émis par Cortex Code. Types d’événements courants :


Type d’événement.	Description
`content_block_start`	Début d’un nouveau texte ou d’un bloc de réflexion
`content_block_delta`	Mise à jour du texte incrémentiel ou de la réflexion
`content_block_stop`	Fin du texte actuel ou du bloc de réflexion

Flux de messages¶

Lorsque les messages partiels sont activés, vous recevez généralement les messages dans l’ordre suivant :

SystemMessage -- session initialization
StreamEvent (content_block_start) -- text or thinking block
StreamEvent (content_block_delta) -- text_delta or thinking_delta chunks...
StreamEvent (content_block_stop)
AssistantMessage -- complete text/thinking block, or complete tool_use block
UserMessage -- complete tool_result block
... more assistant/user turns ...
ResultMessage -- final result

Si les messages partiels ne sont pas activés, vous recevez toujours les mêmes messages d’assistant, d’utilisateur et de résultat, mais pas StreamEvent. Selon la session, le SDK peut également émettre des événements system tels que l’initialisation, l’état et les notifications de tâches en arrière-plan.

Diffuser des réponses de texte¶

Pour afficher le texte tel qu’il est généré, recherchez des événements content_block_delta où delta.type est text_delta :

import { query } from "cortex-code-agent-sdk";

for await (const message of query({
  prompt: "Explain how databases work",
  options: { cwd: process.cwd(), includePartialMessages: true },
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
}
console.log(); // Final newline

import asyncio
from cortex_code_agent_sdk import query, CortexCodeAgentOptions
from cortex_code_agent_sdk.types import StreamEvent

async def stream_text():
    async for message in query(
        prompt="Explain how databases work",
        options=CortexCodeAgentOptions(cwd=".", include_partial_messages=True),
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)
    print()  # Final newline

asyncio.run(stream_text())

Créer une UI de streaming¶

L’exemple suivant accumule le texte ave streaming dans un tampon local et rend à nouveau la réponse actuelle chaque fois qu’un nouveau text_delta arrive. Dans une application réelle, remplacez la fonction render par la logique de mise à jour de l’état de votre framework :

import { query } from "cortex-code-agent-sdk";

let currentText = "";

function render(text: string) {
  console.clear();
  console.log("Assistant:\n");
  process.stdout.write(text);
}

for await (const message of query({
  prompt: "Explain how databases work",
  options: {
    cwd: process.cwd(),
    includePartialMessages: true,
  },
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      currentText += event.delta.text;
      render(currentText);
    }
  } else if (message.type === "result") {
    console.log("\n\n--- Complete ---");
  }
}

import asyncio
import sys
from cortex_code_agent_sdk import query, CortexCodeAgentOptions, ResultMessage
from cortex_code_agent_sdk.types import StreamEvent

def render(text: str) -> None:
    sys.stdout.write("\033[2J\033[H")
    sys.stdout.write("Assistant:\n\n")
    sys.stdout.write(text)
    sys.stdout.flush()

async def streaming_ui():
    current_text = ""

    async for message in query(
        prompt="Explain how databases work",
        options=CortexCodeAgentOptions(
            cwd=".",
            include_partial_messages=True,
        ),
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    current_text += delta.get("text", "")
                    render(current_text)
        elif isinstance(message, ResultMessage):
            print("\n\n--- Complete ---")

asyncio.run(streaming_ui())

Limitations connues¶


Fonctionnalité	Conséquences sur le streaming
Sortie structurée	Le résultat JSON n’apparaît que dans `ResultMessage.structured_output`, et non sous forme de deltas de streaming

Avis juridiques¶

Lorsque votre configuration de Cortex Code utilise un modèle fourni conformément aux Conditions de répercussion relatives aux modèles et aux services, votre utilisation de ce modèle est en outre soumise aux conditions de ce modèle sur cette page.

La classification des données d’entrées et de sorties est présentée dans la table suivante.


Classification des données d’entrée	Classification des données de sortie	Désignation
Usage Data	Données sur les clients	Fonctionnalités AI couvertes [1]

Pour plus d’informations, reportez-vous à Snowflake AI et ML.