流式输出是大语言模型的重要功能,可以提高模型输出效率、提升用户交互体验。

本文为小伙伴们介绍AgentScope中流式输出的用法。

1.AgentScope流式输出

1.1 流式输出的概念

流式输出是指大模型在生成文本或其他类型的数据时,生成时即刻发送,而不是等待所有数据生成完毕后一次性发送。

流式输出适用于生成较长文本或需要即时反馈的场景,可以减少等待时间,提高用户体验,并允许系统在数据生成过程中进行实时处理。

1.2.流式输出和非流式输出的应用场景

流式输出适合需要实时反馈和处理连续数据的应用,而非流式输出适合批量处理和结果完整性要求较高的场景。

在多智能体(Multi-Agent)系统开发过程中,可根据业务需求和系统设计选择合适的输出方式:

  • 流式输出

    • 实时交互应用:需要实时反馈的应用,如在线聊天机器人或实时监控系统。
    • 长任务处理:需要长时间处理的任务,流式输出可以让用户或系统在处理过程中逐步获取结果,而不是等待整个任务完成。
    • 数据流处理:处理连续的数据流时,如传感器数据或实时日志。
  • 非流式输出

    • 批量处理任务:不需要实时反馈的批量处理任务,如夜间批处理作业。
    • 结果完整性要求:需要确保所有输出数据的完整性和一致性时,非流式输出可以确保在所有数据处理完成后一次性输出结果。
    • 复杂决策过程:需要综合所有输入数据才能做出决策的场景中,非流式输出可以确保在所有数据被处理和分析后,再输出最终的决策结果。

1.3.AgentScope如何支持流式输出

AgentScope支持集成不同厂家的大模型服务,对各家模型的流式输出也做了统一封装:

  • 通过stream参数设置流式输出(参数值为True表示流式输出),这个参数会覆盖模型中的stream参数。

目前,AgentScope支持以下大模型 API 的流式输出:

  • OpenAI API:对话(Chat)
  • DashScope API:对话(Chat)
  • Gemini API:对话(Chat)
  • ZhipuAi API:对话(Chat)
  • Ollama API:对话(Chat)
  • LiteLLM API:对话(Chat)

2.AgentScope流式输出的配置和调用

2.1.流式输出的配置

AgentScope支持两种方式配置流式输出:

  • 在配置文件中
    • model_config模型配置中,将stream参数设置为True即表示启用流式输出模式:
1
2
3
4
5
6
model_config = {
    "config_name": "xxx",
    "model_type": "xxx",
    "stream": True,
    # ...
}
  • 在python模型调用时
    • 在python脚本中调用模型时,将stream参数设置为True,启用模型的流式输出模式:
1
2
3
4
5
6
7
8
9
class MyAgent(AgentBase):
    # ...
    def reply(self, x: Optional[Msg, Sequence[Msg]] = None) -> Msg:
        # ...
        response = self.model(
            prompt,
            stream=True,
        )
        # ...

2.2.流式输出的调用

(1)流式打印

  • 启用流式输出后,模型输出的stream字段是生成器,text 字段为 None。
  • 使用speak方法打印流式输出,参数是stream
  • 访问text 字段,生成器将被迭代以生成完整的文本并存入text 字段。
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def reply(self, x: Optional[Msg, Sequence[Msg]] = None) -> Msg:
    # response.steam 为 生成器,response.text 为 None
    response = self.model(prompt)
    
    # 在 terminal 和 AgentScope Studio 中流式打印文本
    self.speak(response.stream)
    
    # 生成器被迭代时,产生的文本将自动被存储在 response.text 中,因此用户可以直接使用 response.text 处理响应文本
    msg = Msg(self.name, content=response.text, role="assistant")
    self.memory.add(msg)
    return msg

(2)自定义流式输出

  • AgentScope提供了log_stream_msg函数,用户使用此函数自行处理生成器类型的消息,比如上例中的response.steam:
    • stream 字段中的生成器将生成一个布尔值和字符串的二元组。布尔值表示当前是否是最后一段文本,字符串是到目前为止的响应文本。
1
2
3
4
5
6
7
8
9
# ...
elif isinstance(content, GeneratorType):
    # 流式消息必须共享相同的 id 才能在 AgentScope Studio 中显示,因此这里通过同一条消息切换 content 字段来实现
    msg = Msg(name=self.name, content="", role="assistant")
    for last, text_chunk in content:
        msg.content = text_chunk
        log_stream_msg(msg, last=last)
else:
# ...

3.总结

本文要点如下:

  • 流式输出是大模型在生成数据时即时响应的一种方法。

  • 流式输出、非流式输出有各自的应用场景:

    • 流式输出适合需要实时反馈和处理连续数据的应用,而非流式输出适合批量处理和结果完整性要求较高的场景。
  • AgentScope支持多种模型的流式输出,包括OpenAI、DashScope、Gemini、ZhipuAi、Ollama、LiteLLM。

  • AgentScope流式输出的配置:

    • model_configstream参数。
    • 脚本中的模型调用参数stream
  • AgentScope流式输出的调用:

    • 模型输出的stream字段是个生成器。
    • 使用speak方法直接流式打印。
    • 使用log_stream_msg方法自定义流式打印。

参考:https://arxiv.org/pdf/2402.14034