Streaming-Ausgabe¶

Dieses Thema beschreibt, wie Sie Echtzeitantworten vom Cortex Code Agent SDK streamen können.

Standardmäßig gibt das SDK vollständige AssistantMessage-Objekte aus, nachdem das Modell die Generierung jeder Antwort abgeschlossen hat. Um inkrementelle Aktualisierungen bei der Generierung von Text- und Denkblöcken zu erhalten, aktivieren Sie teilweises Nachrichten-Streaming, indem Sie includePartialMessages (TypeScript) oder include_partial_messages (Python) auf true einstellen.

Wenn Teilmeldungen aktiviert sind, gibt Cortex Code StreamEvent-Objekte für Teiltexte und Denkinhalte aus. Vollständige Toolaufrufe gehen weiterhin als AssistantMessage-Objekte ein, und Toolergebnisse kommen immer noch als UserMessage-Objekte an.

Aktivieren der Streaming-Ausgabe¶

Wenn diese Option aktiviert ist, gibt die SDK StreamEvent-Meldungen, die teilweise Streaming-Ereignisse enthalten, zusätzlich zu den üblichen AssistantMessage-, UserMessage- und ResultMessage-Objekten aus. Ihr Code muss:

Den Typ jeder Meldung überprüfen, um StreamEvent von anderen Typen zu unterscheiden.
Für StreamEvent das Feld event extrahieren und dessen type überprüfen.
Nach content_block_delta-Ereignissen suchen, bei denen delta.type gleich text_delta ist.

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())

StreamEvent-Referenz¶

Wenn Teilmeldungen aktiviert sind, erhalten Sie Rohdaten-Streaming-Ereignisse, die in ein Objekt eingeschlossen sind:

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

Das event-Feld enthält den von Cortex Code ausgegebenen Teil des Streaming-Ereignisses. Gängige Ereignistypen:


Ereignistyp	Beschreibung
`content_block_start`	Beginn eines neuen Text- oder Denkblocks
`content_block_delta`	Inkrementelle Text- oder Denk-Aktualisierung
`content_block_stop`	Ende des aktuellen Text- oder Denkblocks

Meldungsfluss¶

Wenn Teilmeldungen aktiviert sind, erhalten Sie normalerweise Meldungen in der folgenden Reihenfolge:

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

Ohne aktivierte Teilmeldungen erhalten Sie immer noch dieselben vollständigen Assistenten-, Benutzer- und Ergebnismeldungen, aber nicht StreamEvent. Abhängig von der Sitzung kann das SDK auch system-Ereignisse wie Initialisierungs-, Status- und Hintergrundaufgabenbenachrichtigungen ausgeben.

Streamen von Textantworten¶

Um Text sofort anzuzeigen, wenn er generiert wird, suchen Sie nach content_block_delta-Ereignissen, bei denen delta.type gleich text_delta ist:

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())

Erstellen einer Streaming-UI¶

Das folgende Beispiel sammelt gestreamten Text in einem lokalen Puffer und rendert die aktuelle Antwort jedes Mal neu, wenn ein neues text_delta ankommt. Ersetzen Sie in einer echten Anwendung die render-Funktion mit der Statusaktualisierungslogik des Frameworks:

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())

Bekannte Einschränkungen¶


Feature	Auswirkungen auf das Streaming
Strukturierte Ausgabe	Das JSON-Ergebnis wird nur in `ResultMessage.structured_output` angezeigt, nicht als Streaming-Delta

Rechtliche Hinweise¶

Wenn Ihre Cortex Code-Konfiguration ein Modell verwendet, das im Rahmen der Modell- und Service-Pass-Through-Bedingungen bereitgestellt wurde, unterliegt Ihre Nutzung dieses Modells zusätzlich den Bedingungen für dieses Modell auf dieser Seite.

Die Datenklassifizierung der Eingaben und Ausgaben ist in der folgenden Tabelle aufgeführt.


Klassifizierung von Eingabedaten	Klassifizierung von Ausgabedaten	Benennung
Usage Data	Kundendaten	Abgedeckte AI-Features [1]

Weitere Informationen dazu finden Sie unter KI und ML in Snowflake.