【问题记录】Python中用yield实现打字机效果

news2024/11/17 19:41:55

用yield实现打字机效果

前言
示例
注意点

前言

在人工智能（AI）领域，yield 函数在处理大数据集和实现异步编程时非常重要。
异步编程是一种常见的编程模式，它可以提高程序的执行效率。yield 可以用于实现协程，使得程序在等待 I/O 操作（如网络请求）的过程中，可以切换到其他任务，从而提高程序的并发性能。同时，使用异步编程可以实现聊天数据的流式传输，从而达到实时聊天的需求。

示例

这段代码展示了如何通过调用百度接口实现消息的流式传输以及消息的重新打包封装。
首先通过chat函数获取远端大模型输出，然后通过yield函数异步传输给下一级函数msg_adapt来进行消息的重新封装，经过重新封装以后的消息再通过yield函数传输到前端用于实现打字机的效果。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Generator

import qianfan
import os

from .base_llm_connector import BaseLLMConnector
from .. import config
from ..llm.chat_response import MessageResponse


class ErnieConnector(BaseLLMConnector):
    """
    百度文心内容生成连接器
    """
    _model = 'ERNIE-Speed-128K'

    def __init__(self):
        super().__init__()
        # 鉴权参数初始化
        # https://cloud.baidu.com/doc/WENXINWORKSHOP/s/3lmokh7n6
        os.environ["QIANFAN_ACCESS_KEY"] = config.get('llm').get('ernie').get('access_key')
        os.environ["QIANFAN_SECRET_KEY"] = config.get('llm').get('ernie').get('secret_key')

    # 开始聊天
    def chat(self, message: str, embedding: dict):
        chat_comp = qianfan.ChatCompletion()
        messages = [
            {
                "role": "user",
                "content": message
            }
        ]
        # https://cloud.baidu.com/doc/WENXINWORKSHOP/s/xlmokikxe
        resp = chat_comp.do(
            model=self._model,
            messages=messages,
            system=self.gen_prompt(embedding),
            stream=True
        )
        for r in resp:
            if r.code == 200:
                yield r.body

    def msg_adapt(self, responses) -> Generator[MessageResponse, None, None]:
        """数据类型转化方法

        Args:
            responses: 大模型返回内容

        Returns:
            返回数据列表

        """
        seq = 0
        for response in responses:
            msg_response = MessageResponse(type="message", content="", seq=0, is_last=False)
            msg_response['type'] = "message"
            msg_response['seq'] = seq
            msg_response['content'] = response['result']
            msg_response['is_last'] = response['is_end']
            # 直接返回对象，防止出现变量共同读写的问题
            yield msg_response
            seq += 1