Contrato de Streaming do Runtime

Visão geral

O CrewAI expõe um contrato de streaming baseado em frames para runtimes que precisam de mais do que chunks de texto simples. O contrato emite objetos StreamFrame ordenados para eventos de ciclo de vida de Flow, tokens de LLM diretos, atividade de ferramentas, mensagens de conversa e eventos personalizados. Use esta API ao criar uma UI, ponte de serviço, aplicativo de terminal ou runtime de implantação que precise de um fluxo estável de eventos estruturados enquanto um Flow, turno de chat ou chamada direta de LLM está em execução.

StreamFrame

Todo frame tem o mesmo envelope:

from crewai.types.streaming import StreamFrame

frame.id           # id único do frame
frame.seq          # ordem local da execução, quando disponível
frame.type         # tipo do evento de origem, como "flow_started"
frame.channel      # "llm", "flow", "tools", "messages", "lifecycle" ou "custom"
frame.namespace    # namespace de origem/runtime
frame.timestamp    # timestamp do evento
frame.parent_id    # id do evento pai, quando disponível
frame.previous_id  # id do evento anterior, quando disponível
frame.data         # payload do evento
frame.event        # alias para frame.data
frame.content      # texto imprimível para frames de token, caso contrário ""

O campo channel é a forma mais rápida de rotear frames em consumidores:

Canal	Contém
`llm`	Tokens e chunks de raciocínio de eventos de streaming de LLM
`flow`	Ciclo de vida do Flow, execução de métodos, roteamento e eventos de pausa/retomada
`tools`	Eventos de uso de ferramentas
`messages`	Eventos do transcript da conversa
`lifecycle`	Eventos de ciclo de vida do runtime que não pertencem a outro canal
`custom`	Eventos que não mapeiam para um canal integrado

frame.type preserva o tipo do evento de origem, para que consumidores possam tratar eventos específicos dentro de um canal.

Transmitir um Flow

Defina stream=True em um Flow para fazer kickoff() retornar uma sessão de stream:

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

Você deve consumir o stream antes de ler stream.result. Acessar o resultado cedo demais gera um RuntimeError, para que consumidores não tratem uma execução parcial como concluída. Você também pode chamar flow.stream_events(...) diretamente quando quiser streaming para uma única invocação sem definir stream=True na instância do Flow.

Filtrar por canal

StreamSession expõe projeções por canal que preservam a ordem global dos frames dentro do canal selecionado:

stream = flow.stream_events()

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

result = stream.result

As projeções disponíveis são:

Projeção	Frames
`stream.events`	Todos os frames
`stream.llm`	Frames de LLM
`stream.messages`	Frames de mensagens de conversa
`stream.flow`	Frames de Flow
`stream.tools`	Frames de ferramentas
`stream.interleave([...])`	Um conjunto selecionado de canais

Use stream.interleave(["flow", "llm", "messages"]) quando um consumidor quiser apenas alguns canais, mas ainda precisar da ordem relativa entre eles.

Streaming assíncrono

Use astream() para consumidores assíncronos:

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

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

result = stream.result

A sessão assíncrona tem as mesmas projeções da sessão síncrona.

Transmitir uma chamada direta de LLM

llm.call(...) ainda retorna o resultado final montado. Use llm.stream_events(...) quando quiser iterar pelos chunks conforme eles chegam, mantendo o payload estruturado do evento:

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(...) ativa temporariamente o streaming para a chamada encapsulada e restaura a configuração anterior de stream do LLM depois. As integrações de provedores continuam emitindo os eventos de stream de LLM subjacentes; esse helper fornece uma API de iterador comum sobre esses eventos para todos os provedores de LLM.

Turnos conversacionais

Flows conversacionais podem transmitir um turno de usuário com stream_turn():

from crewai import Flow
from crewai.experimental.conversational 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

Durante stream_turn(), o caminho de resposta conversacional integrado ativa o streaming de tokens de LLM para esse turno e restaura a configuração anterior de stream do LLM depois. Handlers de rota personalizados que criam seus próprios agentes ou instâncias de LLM devem configurar esses LLMs para streaming se precisarem de saída em nível de token.

Limpeza

Use a sessão como gerenciador de contexto quando possível. Se um cliente se desconectar antes de o stream ser esgotado, feche a sessão explicitamente:

stream = flow.stream_events()

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

Para streams assíncronos, use await stream.aclose().

Streaming de chunks legado

O streaming de Crew com stream=True ainda retorna a API orientada a chunks CrewStreamingOutput descrita em Streaming da Execução de Crew. Chamadas diretas llm.call(...) ainda retornam o resultado final do LLM. O contrato de frames é destinado a runtimes que precisam de um envelope de evento estável em Flows, chamadas diretas de LLM, turnos conversacionais, ferramentas e mensagens.

Começando

Guias

Conceitos-Chave

Integração MCP

Ferramentas

Observabilidade

Aprenda

Telemetria

Contrato de Streaming do Runtime

Visão geral

StreamFrame

Transmitir um Flow

Filtrar por canal

Streaming assíncrono

Transmitir uma chamada direta de LLM

Turnos conversacionais

Limpeza

Streaming de chunks legado

​Visão geral

​StreamFrame

​Transmitir um Flow

​Filtrar por canal

​Streaming assíncrono

​Transmitir uma chamada direta de LLM

​Turnos conversacionais

​Limpeza

​Streaming de chunks legado

Visão geral

StreamFrame

Transmitir um Flow

Filtrar por canal

Streaming assíncrono

Transmitir uma chamada direta de LLM

Turnos conversacionais

Limpeza

Streaming de chunks legado