Streaming Runtime Contract

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

Flow가 돌아가는 동안 그 실행을 실시간으로 받아 UI에 보여주거나 서비스로 중계해야 할 때가 있어요. CrewAI는 단순 텍스트 청크 이상을 원하는 런타임을 위해 프레임 기반 스트리밍 계약을 제공합니다. Flow 수명주기 이벤트, 직접 LLM 토큰, 도구 활동, 대화 메시지, 커스텀 이벤트를 순서 있는 StreamFrame 객체로 방출해요. UI·서비스 브리지·터미널 앱·배포 런타임을 만들 때 이 계약을 쓰면 구조화된 이벤트를 안정적으로 받을 수 있습니다.

출처: 공식문서

본문

StreamFrame

모든 프레임은 같은 엔벨로프를 가집니다.

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 필드는 컨슈머에서 프레임을 라우팅하는 가장 빠른 방법입니다.

frame.type은 소스 이벤트 타입을 보존하므로, 컨슈머가 채널 안에서 특정 이벤트를 처리할 수 있습니다.

Flow 스트리밍

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가 발생해서 컨슈머가 부분 실행을 완료로 오인하지 않게 해 줍니다. Flow 인스턴스에 stream=True를 설정하지 않고 단일 호출만 스트리밍하고 싶다면 flow.stream_events(...)를 직접 호출할 수도 있습니다.

채널로 필터링

StreamSession은 선택한 채널 내에서 전역 프레임 순서를 보존하는 채널 프로젝션을 노출합니다.

stream = flow.stream_events()

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

result = stream.result

사용 가능한 프로젝션:

일부 채널만 필요하지만 상대적 순서까지 유지해야 하는 컨슈머라면 stream.interleave(["flow", "llm", "messages"])를 씁니다.

비동기 스트리밍

async 컨슈머에는 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

async 세션도 sync 세션과 같은 프로젝션을 가집니다.

직접 LLM 호출 스트리밍

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 Flows는 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 토큰 스트리밍을 켜고 이후 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()

async 스트림은 await stream.aclose()를 사용합니다.

레거시 청크 스트리밍

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

더 알아보기

• CrewAI Flows — Flow 실행·스트리밍 기초
• Conversational Flows — 턴 스트리밍
• Streaming Runtime Contract 공식 문서