스트리밍 출력
스트리밍 출력
이 항목에서는 Cortex Code Agent SDK에서 실시간 응답을 스트리밍하는 방법을 설명해요.
출처: Streaming output
본문
기본적으로 SDK는 모델이 각 응답 생성을 끝낸 후 완전한 AssistantMessage 객체를 생성해요. 텍스트와 사고 블록이 생성될 때 증분 업데이트를 받으려면 includePartialMessages(TypeScript) 또는 include_partial_messages(Python)를 true로 설정해 부분 메시지 스트리밍을 활성화해요.
부분 메시지가 활성화되면 Cortex Code는 부분 텍스트와 사고 콘텐츠에 대한 StreamEvent 객체를 내보내요. 완전한 도구 호출은 여전히 AssistantMessage 객체로 도착하고, 도구 결과는 여전히 UserMessage 객체로 도착해요.
스트리밍 출력 활성화하기
활성화되면 SDK는 평소의 AssistantMessage, UserMessage, ResultMessage 객체 외에 부분 스트리밍 이벤트를 포함하는 StreamEvent 메시지를 생성해요. 코드는 다음을 해야 해요.
StreamEvent를 다른 유형과 구분하기 위해 각 메시지의type을 확인.StreamEvent의 경우event필드를 추출하고 그type을 확인.delta.type이text_delta인content_block_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())
StreamEvent 참조
부분 메시지가 활성화되면 원시 스트리밍 이벤트를 객체에 감싼 채로 받아요.
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
event 필드는 Cortex Code가 내보낸 원시 부분 스트리밍 이벤트를 포함해요. 일반적인 이벤트 유형:
| 이벤트 유형 | 설명 |
|---|---|
content_block_start |
새 텍스트 또는 사고 블록의 시작 |
content_block_delta |
증분 텍스트 또는 사고 업데이트 |
content_block_stop |
현재 텍스트 또는 사고 블록의 끝 |
메시지 흐름
부분 메시지가 활성화되면 일반적으로 다음 순서로 메시지를 받아요.
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
부분 메시지가 활성화되지 않았어도 같은 완전한 assistant, user, result 메시지를 받지만 StreamEvent는 받지 않아요. 세션에 따라 SDK는 초기화, 상태, 백그라운드 태스크 알림 같은 시스템 이벤트도 내보낼 수 있어요.
텍스트 응답 스트리밍하기
텍스트가 생성될 때 표시하려면 delta.type이 text_delta인 content_block_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())
스트리밍 UI 구축하기
다음 예시는 지역 버퍼에 스트리밍된 텍스트를 누적하고 새 text_delta가 도착할 때마다 현재 응답을 다시 렌더링해요. 실제 애플리케이션에서는 render 함수를 프레임워크의 상태 업데이트 로직으로 바꿔요.
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())
알려진 제한 사항
| 기능 | 스트리밍에 미치는 영향 |
|---|---|
| 구조화된 출력 | JSON 결과는 ResultMessage.structured_output에만 나타나며 스트리밍 델타로는 나타나지 않아요. |
법적 고지
Model and Service Pass-Through Terms에 제공된 모델을 사용하도록 Cortex Code를 구성하는 경우, 해당 모델 사용은 해당 페이지의 모델에 대한 약관도 추가로 적용돼요.
입력 및 출력의 데이터 분류는 다음 표에 명시되어 있어요.
| 입력 데이터 분류 | 출력 데이터 분류 | 지정 |
|---|---|---|
| 사용 데이터 | 고객 데이터 | Covered AI Features [1] |
[1] AI 약관 및 허용 가능한 사용 정책에서 사용되는 정의된 용어를 나타내요.
추가 정보는 Snowflake AI 및 ML을 참고하세요.