如何实现Python版Azure Function API流式响应，供Angular UI经.NET Core API调用

一、改造Python Azure Function，支持Azure OpenAI流式输出 #

现有代码为同步返回完整结果，需调整为流式响应模式：

• 开启Azure OpenAI流式参数：在chat.completions.create方法中添加stream=True
• 切换Azure Function响应方式：使用流式生成器逐块返回数据，采用Server-Sent Events（SSE）格式适配后续中间层与UI处理
• 处理流式片段：遍历OpenAI返回的每一块数据，按SSE规范输出内容片段

修改后的Python代码：

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
        )
        
        # 开启流式返回
        stream = 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,
            stream=True  # 关键参数：启用流式输出
        )

        # 定义流式响应生成器
        def stream_generator():
            for chunk in stream:
                if chunk.choices[0].delta.content is not None:
                    # 以SSE格式返回每个内容片段
                    yield f"data: {json.dumps({'content': chunk.choices[0].delta.content})}\n\n"
            # 发送流结束标记
            yield "data: [DONE]\n\n"

        # 返回流式HttpResponse，设置正确的Content-Type
        return func.HttpResponse(
            stream_generator(),
            mimetype="text/event-stream",
            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
        )

二、.NET Core API中间层适配流式响应 #

中间层需作为代理转发流式请求，保持连接实时性：

• 创建API接口：接收Angular的请求参数，转发至Azure Function
• 配置流式传输：使用HttpClient的流式请求能力，接收Azure Function返回的SSE流并原样转发给前端
• 设置响应头：指定text/event-stream类型，禁用缓存，保持长连接

示例.NET Core控制器代码片段：

[ApiController]
[Route("api/chat")]
public class ChatController : ControllerBase
{
    private readonly HttpClient _httpClient;

    public ChatController(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    [HttpPost]
    public async Task StreamChatResponse([FromBody] ChatRequest request)
    {
        // 设置SSE响应头
        Response.ContentType = "text/event-stream";
        Response.Headers.CacheControl = "no-cache";
        Response.Headers.Connection = "keep-alive";

        // 构造转发请求到Azure Function
        var azureFunctionRequest = new HttpRequestMessage(HttpMethod.Post, "你的Azure Function部署URL")
        {
            Content = JsonContent.Create(request)
        };

        using var response = await _httpClient.SendAsync(azureFunctionRequest, HttpCompletionOption.ResponseHeadersRead);
        response.EnsureSuccessStatusCode();

        // 流式读取响应并转发给前端
        using var stream = await response.Content.ReadAsStreamAsync();
        await stream.CopyToAsync(Response.Body);
        await Response.Body.FlushAsync();
    }

    // 请求参数模型
    public class ChatRequest
    {
        public string deployment_id { get; set; }
        public string api_base { get; set; }
        public List<Message> messages { get; set; }
    }

    public class Message
    {
        public string role { get; set; }
        public string content { get; set; }
    }
}

三、Angular UI消费流式响应 #

Angular需实时接收流式数据并更新UI：

• 使用HttpClient订阅流式响应：处理POST请求的流式返回，逐块解析SSE格式内容
• 实时更新UI：每收到一个内容片段，追加到对话显示区域
• 处理流结束标记：收到[DONE]标记后完成会话

示例Angular组件代码片段：

import { Component } from '@angular/core';
import { HttpClient } from '@angular/common/http';

@Component({
  selector: 'app-chat',
  template: `
    <div class="chat-box" [innerHTML]="chatContent"></div>
    <input type="text" [(ngModel)]="userInput" (keydown.enter)="sendMessage()">
    <button (click)="sendMessage()">发送</button>
  `
})
export class ChatComponent {
  chatContent = '';
  userInput = '';

  constructor(private httpClient: HttpClient) {}

  sendMessage() {
    if (!this.userInput.trim()) return;

    const requestBody = {
      deployment_id: '你的模型部署ID',
      api_base: '你的Azure OpenAI端点',
      messages: [{ role: 'user', content: this.userInput }]
    };

    // 订阅流式响应
    this.httpClient.post('api/chat', requestBody, { responseType: 'text', reportProgress: true })
      .subscribe({
        next: (chunk) => {
          // 解析SSE格式片段
          const lines = chunk.split('\n');
          lines.forEach(line => {
            if (line.startsWith('data: ')) {
              const data = line.slice(6);
              if (data === '[DONE]') return;
              const content = JSON.parse(data).content;
              this.chatContent += content;
            }
          });
        },
        error: (err) => console.error('流式请求失败:', err),
        complete: () => console.log('会话结束')
      });

    this.userInput = '';
  }
}

内容的提问来源于stack exchange，提问作者Rahul