스트리밍 런타임 계약

스트리밍 런타임 계약 (Streaming Runtime Contract)

UI, 서비스 브리지, 터미널 앱 또는 배포 런타임을 만들 때 "플레인 텍스트 청크 이상"의 안정적인 구조화 이벤트 스트림이 필요할 수 있어요. CrewAI는 이런 런타임을 위해 정렬된 StreamFrame 객체를 발행하는 프레임 기반 스트리밍 계약을 제공합니다. 플로우 라이프사이클, 직접 LLM 토큰, 도구 활동, 대화 메시지, 커스텀 이벤트가 모두 이 계약 위에서 흘러요.

출처: 공식문서

본문

플로우, 채팅 턴 또는 직접 LLM 호출이 실행되는 동안 구조화된 이벤트를 안정적으로 받아야 하는 런타임을 만들 때 이 API를 쓰세요.

StreamFrame

모든 프레임은 같은 봉투(envelope)를 가집니다.

from crewai.types.streaming import StreamFrame

frame.id           # unique frame id
frame.seq          # execution-local order, when available
frame.type         # source event type, such as "flow_started"
frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", or "custom"
frame.namespace    # source/runtime namespace
frame.timestamp    # event timestamp
frame.parent_id    # parent event id, when available
frame.previous_id  # previous event id, when available
frame.data         # event payload
frame.event        # alias for frame.data
frame.content      # printable text for token-like frames, otherwise ""

channel 필드는 소비자에서 프레임을 라우팅하는 가장 빠른 방법이에요.

Channel Contains
llm Token and thinking chunks from LLM streaming events
flow Flow lifecycle, method execution, routing, and pause/resume events
tools Tool usage events
messages Conversation transcript events
lifecycle Runtime lifecycle events that are not specific to another channel
custom Events that do not map to a built-in channel

frame.type은 원본 이벤트 타입을 보존하므로, 소비자가 채널 안에서 특정 이벤트를 처리할 수 있어요.

플로우 스트리밍하기 (Stream a Flow)

플로우에 stream=True를 설정하면 kickoff()가 스트림 세션을 반환합니다.

from crewai.flow import Flow, start

class ReportFlow(Flow):
    @start()
    def generate(self):
        return "done"

flow = ReportFlow(stream=True)
stream = flow.kickoff()

with stream:
    for chunk in stream:
        print(chunk.content, end="", flush=True)
        if chunk.type == "tool_usage_started":
            print(chunk.event["tool_name"])

result = stream.result

stream.result를 읽기 전에 반드시 스트림을 소비해야 해요. 일찍 접근하면 RuntimeError가 발생해, 부분 실행을 완료로 오인하지 않도록 방지합니다. 플로우 인스턴스에 stream=True를 설정하지 않고 한 번의 호출에 대해서만 스트리밍을 원한다면 flow.stream_events(...)를 직접 호출할 수도 있어요.

채널로 필터링하기 (Filter by Channel)

StreamSession은 선택한 채널 내에서 전역 프레임 순서를 보존하는 채널 투영을 제공합니다.

stream = flow.stream_events()

with stream:
    for frame in stream.llm:
        print(frame.content, end="", flush=True)

result = stream.result

사용 가능한 투영:

Projection Frames
stream.events All frames
stream.llm LLM frames
stream.messages Conversation message frames
stream.flow Flow frames
stream.tools Tool frames
stream.interleave([...]) A selected set of channels

일부 채널만 원하면서 상대적 순서는 유지하고 싶다면 stream.interleave(["flow", "llm", "messages"])를 써요.

비동기 스트리밍 (Async Streaming)

비동기 소비자는 astream()을 씁니다.

flow = ReportFlow()
stream = flow.astream()

async with stream:
    async for chunk in stream.events:
        print(chunk.channel, chunk.type, chunk.content)

result = stream.result

비동기 세션도 동기 세션과 같은 투영을 가집니다.

직접 LLM 호출 스트리밍하기 (Stream a Direct LLM Call)

llm.call(...)은 여전히 최종 합쳐진 결과를 반환합니다. 청크가 도착하는 대로 순회하면서 구조화된 이벤트 페이로드를 유지하려면 llm.stream_events(...)를 쓰세요.

from crewai import LLM

llm = LLM(model="gpt-4o-mini")
stream = llm.stream_events(
    messages=[
        {
            "role": "user",
            "content": "Explain CrewAI streaming in two short sentences.",
        }
    ]
)

with stream:
    for chunk in stream:
        print(chunk.content, end="", flush=True)

result = stream.result

llm.stream_events(...)는 감싼 호출 동안 임시로 스트리밍을 켜고, 이후 LLM의 이전 stream 설정을 복원합니다. 제공자 통합은 계속해서 기본 LLM 스트림 이벤트를 발행하며, 이 헬퍼는 모든 LLM 제공자에 대해 공통 이터레이터 API를 제공해요.

대화형 턴 (Conversational Turns)

대화형 플로우는 stream_turn()으로 사용자 턴 하나를 스트리밍할 수 있습니다.

from crewai import Flow
from crewai.flow import ConversationConfig, ConversationState

@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
class ChatFlow(Flow[ConversationState]):
    conversational = True

flow = ChatFlow()
stream = flow.stream_turn("What can you help me with?", session_id="session-1")

with stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
            print(frame.content, end="", flush=True)

reply = stream.result

stream_turn() 동안 내장 대화형 응답 경로가 해당 턴에 대해 LLM 토큰 스트리밍을 켜고, 이후 이전 stream 설정을 복원합니다. 자체 에이전트나 LLM 인스턴스를 만드는 커스텀 라우트 핸들러는 토큰 단위 출력이 필요하면 해당 LLM에 스트리밍을 직접 구성해야 해요.

정리 (Cleanup)

가능하면 세션을 컨텍스트 매니저로 씁니다. 클라이언트가 스트림이 소진되기 전에 끊기면 세션을 명시적으로 닫아야 해요.

stream = flow.stream_events()

try:
    for frame in stream.events:
        print(frame.type)
finally:
    if not stream.is_exhausted:
        stream.close()

비동기 스트림은 await stream.aclose()를 호출하세요.

레거시 청크 스트리밍 (Legacy Chunk Streaming)

stream=True를 설정한 크루 스트리밍은 Streaming Crew Execution에 설명된 청크 지향 CrewStreamingOutput API를 여전히 반환합니다. 직접 llm.call(...)도 여전히 최종 LLM 결과를 반환해요. 프레임 계약은 플로우, 직접 LLM 호출, 대화형 턴, 도구, 메시지 전반에 걸쳐 안정적인 이벤트 봉투가 필요한 런타임을 위한 것입니다.

더 알아보기