如何设置来自 Azure Function API 的响应流，该 API 旨在利用 Azure OpenAI

Question

如何设置来自 Azure Function API 的响应流，该 API 旨在利用 Python 语言利用 Azure OpenAI，供 Angular UI 应用程序使用？此 Angular UI 应用程序通过 .NET Core API 中介与 Azure Function API 进行通信。需要采取哪些步骤才能有效地实现此流媒体功能？

import azure.functions as func
from azure.identity import DefaultAzureCredential
from openai import AzureOpenAI
import json

def main(req: func.HttpRequest, context: func.Context):
    try:

        data = req.get_json();
        model_name=data['deployment_id']
        api_base1=data['api_base']

        default_credential = DefaultAzureCredential()
        token = default_credential.get_token("https://cognitiveservices.azure.com/.default")

        client = AzureOpenAI(
                            api_version="2023-07-01-preview",
                            azure_endpoint=api_base1,
                            api_key=token.token
                            )
        
        completion = client.chat.completions.create(
              model=model_name,
              messages = data['messages'],
              temperature=0.05,
              max_tokens=500,
              top_p= 0.95,
              frequency_penalty= 0,
              presence_penalty= 0,
              stop=None
              )
        response = completion.choices[0].message.content;

        response_data=json.dumps({"message": response})
        return func.HttpResponse(
        response_data,
        mimetype="application/json",
        status_code=200
        )
       
    except Exception as e:
        error_data= json.dumps({"error_message": str(e)})
        return func.HttpResponse(
        error_data,
        mimetype="application/json",
        status_code=500
        )

Answer 1

使用此参考，使用 Python 和服务器发送事件 (SSE) 实时传输来自 OpenAI API 的响应。

安装必要的模块（

requests

、

json

和

sseclient-py

）。如果尚未安装这些库，请使用 pip 安装。使用您的 OpenAI API 密钥定义

API_KEY

变量。然后，实现

performRequestWithStreaming()

函数，向 OpenAI 的完成端点发送 POST 请求，指定 AI 模型所需的参数和提示。使用

sseclient

处理流响应并打印文本。

代码：

import azure.functions as func
from azure.identity import DefaultAzureCredential
from openai import AzureOpenAI
import json

def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        data = req.get_json()
        model_name = data['deployment_id']
        api_base1 = data['api_base']

        default_credential = DefaultAzureCredential()
        token = default_credential.get_token("https://cognitiveservices.azure.com/.default")

        client = AzureOpenAI(
            api_version="2023-07-01-preview",
            azure_endpoint=api_base1,
            api_key=token.token
        )

        completion = client.chat.completions.create(
            model=model_name,
            messages=data['messages'],
            temperature=0.05,
            max_tokens=500,
            top_p=0.95,
            frequency_penalty=0,
            presence_penalty=0,
            stop=None
        )

        headers = {
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
        response = func.HttpResponse(status_code=200, headers=headers)

        for choice in completion.choices:
            response_data = json.dumps({"message": choice.message.content})
            response.add_data(f"data: {response_data}\n\n")

        return response

    except Exception as e:
        error_data = json.dumps({"error_message": str(e)})
        return func.HttpResponse(
            error_data,
            mimetype="application/json",
            status_code=500
        )

• 有关更多详细信息，请参阅so了解StreamingResponse不使用生成器功能进行流式传输。

另一种方法是在 ASP.NET Core API 中通过

IAsyncEnumerable

使用此 link 和 git。