Skip to main content
Assistant API (Deprecated)

Learn about the streaming output feature of the Assistant API (being unpublished)

A saída em streaming permite recuperar o status de execução de um assistente em tempo real. Esse recurso exibe o conteúdo gerado pelo Large Language Model (LLM) aos usuários palavra por palavra.

A Assistant API está sendo descontinuada. Como alternativa, migre para a Responses API, que oferece ferramentas integradas e gerenciamento de contexto em múltiplos turnos.
A saída em streaming oferece os seguintes benefícios:
  • Fluxo de conversa mais natural: Simula padrões de comunicação humana para uma interação mais fluida e orgânica.
  • Controle mais granular: Monitore o status de execução do assistente em tempo real para exibir o progresso da geração ou interromper o processo em caso de erro.
  • Conexão mais confiável: O retorno da resposta em fragmentos evita longos tempos de espera no lado do cliente.

Primeiros passos

Para ativar a saída em streaming, execute o assistente com Run.create(stream=True). Certifique-se de ter configurado a chave da API como variável de ambiente e instalado o DashScope SDK.
Saída sem streaming
# Non-streaming output
from dashscope import Assistants, Threads, Runs, Messages

# Create an assistant
assistant = Assistants.create(
        model='qwen-plus',  # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models
        name='Sample Assistant',
        instructions='You are a helpful assistant'
    )

# Create a conversation thread
thread = Threads.create(assistant_id=assistant.id,
                        messages=[{
                            'role': 'user',
                            'content': 'Hello!'
                        }])

# Run the conversation and wait for the response
run = Runs.create(thread.id,
                  assistant_id=assistant.id)
Runs.wait(run.id,
          thread_id=thread.id)

# Get and print the result
message = Messages.list(thread.id).data[0].content[0].text.value
print(message)
Saída em streaming
# Streaming output
from dashscope import Assistants, Threads, Runs

# Create an assistant
assistant = Assistants.create(
        model='qwen-plus',  # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models
        name='Sample Assistant',
        instructions='You are a helpful assistant'
    )

# Create a conversation thread
thread = Threads.create(assistant_id=assistant.id,
                        messages=[{
                            'role': 'user',
                            'content': 'Hello!'
                        }])

# Run the streaming conversation
run = Runs.create(thread.id,
                  assistant_id=assistant.id,
                  stream=True)  # The key configuration to enable streaming output

# Output the result in real time
for event, data in run:
    if event == 'thread.message.delta':
        print(data.delta.content.text.value, end='', flush=True)

Configuração

Na Assistant API, a renderização da saída em streaming envolve duas etapas:
  1. Execute o assistente no modo de streaming.
  2. Receba os dados do stream.
Pense na saída em streaming como um "fluxo de água". Executar o assistente equivale a "abrir a torneira", enquanto receber os dados corresponde a "coletar a água".

Executar o assistente no modo de streaming

Para executar o assistente no modo de streaming, defina o parâmetro stream como True ao criar uma execução. 
run = Runs.create(thread_id=thread.id,
                  assistant_id=assistant.id,
                  stream=True)  # Simply set stream=True to run the assistant in stream mode

Receber os dados do stream

Em uma execução com streaming, o assistente produz dois tipos principais de fluxos de dados:
  • Fluxo de informações de status: Rastreia o status e o progresso da execução.
  • Fluxo de mensagens da conversa: Contém o conteúdo real da saída.
Para capturar esse fluxo contínuo de dados, crie um loop de processamento e estabeleça regras para distinguir os diferentes tipos de dados. Geralmente, acompanhe apenas o fluxo de mensagens da conversa, a menos que precise implementar um gerenciamento de estado mais complexo. Para obter uma lista detalhada de eventos e dados de eventos, consulte a referência de desenvolvimento da Saída em streaming da Assistant API. 
for event, data in run_iterator:
    if event == 'thread.run.created':  # [Status information stream] If a run is created, display its current status
        print(data.status)
    if event == 'thread.message.delta':  # [Conversation message stream] If a message chunk is generated, display it immediately
        print(data.delta.content.text.value, end='', flush=True)

Fluxo de mensagens da conversa

Na Assistant API, o assistente produz dois tipos de fluxos de mensagens de conversa:
  • Fluxo de mensagens de texto: Mensagens de texto geradas pelo LLM no assistente.
for event, data in run_iterator:
    if event == 'thread.message.delta':  # [Conversation message stream] A message delta object, which indicates a newly generated text message chunk.
        print(data.delta.content.text.value, end='', flush=True)  # Output this chunk.
  • Fluxo de mensagens de ferramenta: Mensagens retornadas a partir de chamadas de ferramenta no assistente. O exemplo abaixo utiliza o interpretador de código:
for event, data in run_iterator:
    if event == 'thread.run.step.delta':  # [Conversation message stream] A run step delta object, which indicates a new tool call.
        tool_call = data.delta.step_details.tool_calls[0]
        if getattr(tool_call, 'type', '') == 'code_interpreter':  # Take the code interpreter as an example.
            print(getattr(tool_call.code_interpreter, 'arguments', ''), end='', flush=True)  # Output the code to be executed.
            print(getattr(tool_call.code_interpreter, 'output', ''), end='', flush=True)  # Output the code execution result.
O interpretador de código, a busca Quark, a conversão de texto em imagem e a calculadora suportam saída em streaming. Outras ferramentas da Assistant API não oferecem suporte a esse recurso. Para mais informações, consulte Objeto delta da etapa de execução. Agora você conhece o método básico para ativar a saída em streaming. O exemplo a seguir demonstra como construir uma saída em streaming com chamadas de ferramenta.

Exemplo: Assistente de tutoria de código (saída em streaming)

Neste exemplo simples, o assistente usa a ferramenta de interpretador de código. Isso permite a saída em streaming para mensagens de texto, geração de código e execução de código. Para mais detalhes sobre como configurar o interpretador de código, consulte Interpretador de código - Recursos. Certifique-se de ter configurado a chave da API como variável de ambiente e instalado o DashScope SDK. 

from dashscope import Assistants, Threads, Runs
import time

class CodeTutorStream:
    """A demo class for the code tutoring assistant with streaming output."""

    def __init__(self):
        self.assistant = None
        self.thread = None

    def create_assistant(self):
        """Create a tutoring assistant configured with the code interpreter."""
        self.assistant = Assistants.create(
            model='qwen-plus',
            name="Python Tutoring Assistant",
            instructions="You are a patient code tutoring assistant. Please use streaming output to explain the code step by step.",
            tools=[{'type': 'code_interpreter'}]
        )
        print("Tutoring assistant initialized (code interpreter enabled)\n")

    def start_lesson(self, question):
        """Create a tutoring thread and start the streaming conversation."""
        self.thread = Threads.create(messages=[{
            'role': 'user',
            'content': question
        }])

        print("Student's question:", question)
        print("\nAssistant is thinking...\n")

        # Delay for 1 second to simulate processing time
        time.sleep(1)

        # Start the stream run
        run_stream = Runs.create(
            self.thread.id,
            assistant_id=self.assistant.id,
            stream=True
        )
        return run_stream

    def process_stream(self, stream):
        message_flag = False
        tool_call_flag = False
        """Process the streaming output and simulate a tutoring scenario."""
        try:
            for event, data in stream:
                # Process the text explanation
                if event == 'thread.message.delta':
                    if not message_flag:
                        print(f"\nStep-by-step explanation:")
                        message_flag = True
                    text_chunk = data.delta.content.text.value
                    self.simulate_typing(text_chunk)

                # Process the code demonstration
                if event == 'thread.run.step.delta':
                    if not tool_call_flag:
                        print(f"\nCode demonstration:")
                        tool_call_flag = True
                    tool_call = data.delta.step_details.tool_calls[0]
                    if getattr(tool_call, 'type', '') == 'code_interpreter':
                        code = getattr(tool_call.code_interpreter, 'arguments', '')
                        output = getattr(tool_call.code_interpreter, 'output', '')

                        self.simulate_typing(code, speed=0.03)
                        self.simulate_typing(output)

        except KeyboardInterrupt:
            stream.close()
            print("\nTutoring interrupted")

    @staticmethod
    def simulate_typing(text, speed=0.03):
        """Simulate a typewriter effect for the output."""
        for char in text:
            print(char, end='', flush=True)
            time.sleep(speed)

if __name__ == "__main__":
    tutor = CodeTutorStream()

    # Create the tutoring assistant
    tutor.create_assistant()

    # Set the tutoring question
    question = """Please explain in detail how to plot a sine function graph in Python:
1. Explain the mathematical principles step by step.
2. Demonstrate the usage of numpy and matplotlib.
3. Show the final visualization."""

    # Start the streaming tutorial
    stream = tutor.start_lesson(question)

    # Process the real-time output
    tutor.process_stream(stream)

Perguntas frequentes

Se ocorrerem erros de execução de código ao configurar a saída em streaming, consulte Mensagens de erro para solucionar o problema.

Segunda edição

Interromper a saída em streaming

Referências

Primeiros passos
Desenvolvimento de aplicações
Prompt
Application Data
Base de Conhecimento (RAG)
Plugin
Application Calling
Permissões
Assistant API (Deprecated)
Suporte