Saída de streaming¶

Este tópico descreve como transmitir respostas em tempo real do SDK do Cortex Code Agent.

Por padrão, o SDK produz objetos AssistantMessage completos depois que o modelo termina de gerar cada resposta. Para receber atualizações incrementais à medida que o texto e os blocos inteligentes são gerados, habilite o streaming de mensagens parciais definindo includePartialMessages (TypeScript) ou include_partial_messages (Python) como true.

Quando as mensagens parciais estão habilitadas, o Cortex Code emite objetos StreamEvent para texto e conteúdo de raciocínio parciais. Chamadas de ferramentas completas ainda chegam como objetos AssistantMessage, e resultados de ferramentas ainda chegam como objetos UserMessage.

Habilitar a saída de streaming¶

Quando habilitada, o SDK produz mensagens StreamEvent contendo eventos de streaming parciais, além do objetos habituais AssistantMessage, UserMessage e ResultMessage. Seu código precisa:

Verificar o tipo de cada mensagem para distinguir StreamEvent de outros tipos.
Para StreamEvent, extrair o campo event e verificar seu type.
Procurar eventos content_block_delta em que delta.type é 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())

Referência de StreamEvent¶

Quando as mensagens parciais estão habilitadas, você recebe eventos de streaming brutos encapsulados em um objeto:

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

O campo event contém o evento de streaming parcial bruto emitido pelo Cortex Code. Tipos de eventos comuns:


Tipo de evento	Descrição
`content_block_start`	Início de um novo bloco de texto ou raciocínio
`content_block_delta`	Atualização incremental de texto ou raciocínio
`content_block_stop`	Fim do bloco de texto ou raciocínio atual

Fluxo de mensagens¶

Com as mensagens parciais habilitadas, normalmente você recebe as mensagens na seguinte ordem:

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

Sem as mensagens parciais habilitadas, você ainda recebe as mesmas mensagens completas de assistente, usuário e resultado, mas não StreamEvent. Dependendo da sessão, o SDK também pode emitir eventos system, como notificações de inicialização, status e tarefas em segundo plano.

Transmitir respostas de texto¶

Para exibir o texto conforme ele é gerado, procure eventos content_block_delta em que delta.type é 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())

Criar UI de streaming¶

O exemplo a seguir acumula texto transmitido em um buffer local e renderiza a resposta atual sempre que um novo text_delta chega. Em um aplicativo real, substitua a função render pela lógica de atualização de estado da sua estrutura:

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

Limitações conhecidas¶


Recurso	Impacto no streaming
Saída estruturada	O resultado JSON aparece somente em `ResultMessage.structured_output`, não como deltas de streaming

Avisos legais¶

Quando sua configuração do Cortex Code usa um modelo que consta em Model and Service Pass-Through Terms, seu uso desse modelo está sujeito aos respectivos termos nessa página.

A classificação dos dados de entradas e saídas é definido na tabela a seguir.


Classificação de dados de entrada	Classificação de dados de saída	Designação
Usage Data	Dados de cliente	Recursos de AI incluídos [1]

Para obter informações adicionais, consulte AI e ML Snowflake.