스트림 소비하기

스트림 소비하기 (Consuming Streams)

CrewAI 스트림을 구독해 프레임이 도착하는 대로 출력하거나 라우팅하고 싶을 때 이 가이드를 쓰세요. stream_events()로 스트림을 열고 llm, tools, flow 같은 채널별 투영(projection)을 순회하는 기본 패턴을 정리했어요. 스트림을 끝까지 소비한 다음에 stream.result를 읽어야 한다는 점이 핵심입니다.

출처: 공식문서

본문

기본 패턴은 이렇게 생겼어요.

stream = flow.stream_events(inputs={"topic": "AI agents"})

with stream:
    for frame in stream:
        ...

result = stream.result

stream.result를 읽기 전에 반드시 스트림을 먼저 소비해야 합니다.

LLM 출력 출력하기 (Print LLM Output)

LLM 호출로 생성된 텍스트만 원한다면 llm 투영을 구독하고 frame.content를 출력하면 돼요.

stream = flow.stream_events(inputs={"topic": "AI agents"})

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

print()
result = stream.result

출력 가능한 텍스트가 없는 프레임은 frame.content가 빈 문자열이므로, 이렇게 조건을 달아도 안전해요.

with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.content:
            print(frame.content, end="", flush=True)

result = stream.result

도구 활동 출력하기 (Print Tool Activity)

도구 이벤트는 tools 채널로 도착합니다. frame.type으로 시작·완료·오류를 구분할 수 있어요.

with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.content:
            print(frame.content, end="", flush=True)

        if frame.channel == "tools" and frame.type == "tool_usage_started":
            print(f"\nTool started: {frame.event.get('tool_name')}")

        if frame.channel == "tools" and frame.type == "tool_usage_finished":
            print(f"\nTool finished: {frame.event.get('tool_name')}")

result = stream.result

frame.event는 원본 이벤트의 구조화된 페이로드입니다. 도구 이름, 인자, 메시지 role, 런타임 식별자 같은 메타데이터를 여기서 가져와요.

플로우 진행 상황 보기 (Watch Flow Progress)

플로우 라이프사이클과 메서드 실행 프레임은 flow 채널로 도착합니다.

with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
    for frame in stream.flow:
        print(frame.type, frame.namespace)

result = stream.result

토큰 단위 출력 대신 진행 로그가 필요할 때 유용해요.

선택한 채널 인터리브하기 (Interleave Selected Channels)

일부 채널만 골라 상대적 순서를 유지하고 싶다면 interleave()를 써요.

with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
    for frame in stream.interleave(["llm", "tools"]):
        if frame.channel == "llm":
            print(frame.content, end="", flush=True)
        elif frame.type == "tool_usage_started":
            print(f"\nTool: {frame.event.get('tool_name')}")

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("Explain streaming in one sentence.")

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

print()
result = stream.result

대화형 턴 스트리밍하기 (Stream a Conversational Turn)

대화형 플로우(Conversational Flows)는 사용자 메시지 한 건에 대해 stream_turn()을 제공합니다.

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

with stream:
    for frame in stream.interleave(["llm", "messages"]):
        if frame.channel == "llm":
            print(frame.content, end="", flush=True)
        elif frame.channel == "messages":
            print(f"\n{frame.event.get('role')}: {frame.event.get('content')}")

reply = stream.result

비동기 소비자 (Async Consumers)

비동기 스트림도 같은 채널 투영을 씁니다.

stream = flow.astream(inputs={"topic": "AI agents"})

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

result = stream.result

정리 (Cleanup)

가능하면 스트림을 컨텍스트 매니저로 쓰는 게 좋아요. 클라이언트가 끊기거나 일찍 소비를 멈추면 스트림을 닫아야 합니다.

stream = flow.stream_events(inputs={"topic": "AI agents"})

try:
    for frame in stream.events:
        print(frame.content, end="", flush=True)
finally:
    if not stream.is_exhausted:
        stream.close()

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

더 알아보기