Assistant API 流式输出快速入门

效果展示

相比一次性输出完整内容，流式输出具有：

• 更自然的对话节奏：模拟真实对话的输出节奏，让交互过程更加流畅自然。

• 更精细的控制机制：支持实时内容监测，发现异常时可终止后续输出。

• 更可靠的连接保障：内容逐步返回，显著降低了长对话和复杂计算场景下的连接超时风险。

快速配置

如需启用流式输出，您只需使用Run.create(stream=True)运行 Assistant，使用 event 观察运行状态，通过 data获取相应的流式数据。

# 非流式输出 
from dashscope import Assistants, Threads, Runs, Messages

# 创建Assistant
assistant = Assistants.create(
        model='qwen-plus',
        name='示例Assistant',
        instructions='You are a helpful assistant'
    )

# 创建对话线程
thread = Threads.create(assistant_id=assistant.id,
                        messages=[{
                            'role': 'user',
                            'content': '你好！'
                        }])

# 执行对话并等待响应
run = Runs.create(thread.id,
                  assistant_id=assistant.id)
Runs.wait(run.id,
          thread_id=thread.id)

# 获取并打印结果
message = Messages.list(thread.id).data[0].content[0].text.value
print(message)

# 流式输出
from dashscope import Assistants, Threads, Runs

# 创建Assistant
assistant = Assistants.create(
        model='qwen-plus',
        name='示例Assistant',
        instructions='You are a helpful assistant'
    )

# 创建对话线程
thread = Threads.create(assistant_id=assistant.id,
                        messages=[{
                            'role': 'user',
                            'content': '你好！'
                        }])

# 执行流式对话
run = Runs.create(thread.id,
                  assistant_id=assistant.id,
                  stream=True)  # 启用流式输出的关键配置

# 实时输出结果
for event, data in run:
    if event == 'thread.message.delta':
        print(data.delta.content.text.value, end='', flush=True)

详细配置

在 Assistant API 中，为了呈现流式输出效果，需要完成两个步骤：

1. 流式运行 Assistant

2. 接收流式数据

如果将流式输出比作“水流”，那么运行 Assistant 就像是“打开水龙头”，而接收数据就像是“接收水流”。

流式运行 Assistant

就像打开水龙头，您只需在创建运行时，将stream参数设置为True，即可流式运行 Assistant。

run = Runs.create(thread_id=thread.id,
                  assistant_id=assistant.id,
                  stream=True)  # 只需配置stream=True，即可流式运行 Assistant

接收流式数据

在流式运行中，Assistant 产生的数据流主要分为两类：

• 状态信息流：用于追踪运行状态和进度

• 对话消息流：包含实际的输出内容

就像接住源源不断的水流，您需要构建一个循环来处理数据，并设置一些规则来区分不同类型的数据。作为开发者，您通常只需关注对话消息流，除非需要实现更复杂的状态管理。详细的事件和事件数据列表，请参考Assistant API 流式输出 开发参考。

for event, data in run_iterator:
    if event == 'thread.run.created':  # 【状态信息流】如果运行被创建，那么展示运行的当前状态
        print(data.status)  
    if event == 'thread.message.delta':  # 【对话消息流】如果产生消息片段，那么立即展示这个片段
        print(data.delta.content.text.value, end='', flush=True)

对话消息流

在 Assistant API 中，Assistant 会产生两类对话消息流：

• 文本消息流：在 Assistant 中，由大模型生成的文本消息流。

  for event, data in run_iterator:
      if event == 'thread.message.delta':  # 【对话消息流】消息增量对象，表示有新生成的文本消息片段
          print(data.delta.content.text.value, end='', flush=True)  # 输出这个片段

• 工具消息流：在 Assistant 中，由工具调用返回的工具消息流。以代码解释器为例：

  for event, data in run_iterator:
      if event == 'thread.run.step.delta':  # 【对话消息流】运行步骤增量对象，表示有新的工具调用
          tool_call = data.delta.step_details.tool_calls[0]
          if getattr(tool_call, 'type', '') == 'code_interpreter':  # 以代码解释器为例
              print(getattr(tool_call.code_interpreter, 'arguments', ''), end='', flush=True)  # 输出将要执行的代码
              print(getattr(tool_call.code_interpreter, 'output', ''), end='', flush=True)  # 输出代码执行结果

  代码解释器，夸克搜索，文生图和计算器支持流式输出。其他 Assistant API 工具不支持流式输出。详情请参阅运行步骤增量对象。

至此，您已经了解了构建流式输出的基本方法。接下来您将通过一个案例，构建带有工具调用的流式输出。

示例：代码教学助手（流式输出）

在这个简单的案例中，Assistant 配置了代码解释器工具，实现了文本消息、代码编写和代码执行的流式输出效果。如果您希望了解更多代码解释器的配置方法，请参考代码解释器-功能特性。

from dashscope import Assistants, Threads, Runs


def main():
    # 创建支持代码解释器的Assistant
    # tools列表中添加code_interpreter使Assistant能够执行Python代码
    assistant = Assistants.create(
        model='qwen-plus',
        name='简单助手',
        instructions='you are a helpful assistant',
        tools=[{
            'type': 'code_interpreter',
        }]
    )

    # 创建对话线程，请求绘制函数图像
    # 这里要求Assistant思考如何绘图，它会使用code_interpreter来完成任务
    thread = Threads.create(
        messages=[{
            'role': 'user',
            'content': '你好啊，帮我画一幅y=x**2的图像，请你先想想怎么画'
        }])

    # 启动流式对话，设置stream=True来获取实时响应
    run_iterator = Runs.create(
        thread.id, 
        assistant_id=assistant.id,
        stream=True
    )

    try:
        # 处理不同类型的流式响应事件
        for event, data in run_iterator:
            # 当一个步骤完成时打印换行
            if event == 'thread.run.step.completed':
                print("\n")
            # 处理Assistant的文本消息
            if event == 'thread.message.delta':
                print(data.delta.content.text.value, end='', flush=True)
            # 处理代码解释器的输出
            if event == 'thread.run.step.delta':
                tool_call = data.delta.step_details.tool_calls[0]
                if getattr(tool_call, 'type', '') == 'code_interpreter':
                    # 打印代码和执行结果
                    print(getattr(tool_call.code_interpreter, 'arguments', ''), end='', flush=True)
                    print(getattr(tool_call.code_interpreter, 'output', ''), end='', flush=True)
    
    # 异常处理：支持用户中断和错误处理            
    except KeyboardInterrupt:
        run_iterator.close()
        print("\n输出被用户中断")
    except Exception as e:
        print(f"\n遇到错误了：{str(e)}")
        run_iterator.close()


if __name__ == '__main__':
    main()

常见问题

在配置流式输出时，如果您遇到代码执行错误，可参阅错误码排查错误类型。