背景

阿里云百炼（原 DashScope）提供了兼容 OpenAI 接口协议的大模型服务，支持通义千问、DeepSeek 等主流模型。本文以实际项目代码为例，介绍如何在 Django 后端封装一个同步流式调用的 AI 客户端工具类。

---

一、准备工作

1. 获取 API Key

前往 阿里云百炼控制台 创建 API Key，建议配置到环境变量，避免硬编码泄露：

export DASHSCOPE_API_KEY="sk-your-api-key"

2. 安装依赖

仅需标准库 requests，无需安装额外 SDK：

pip install requests

3. Django 配置

在 settings.py 中添加模型配置：

# settings.py
CLOUD_AI_SETTINGS = {
    'model_name': 'qwen-plus',  # 可替换为 qwen-max、qwen-turbo 等
}

---

二、接口说明

阿里云百炼兼容 OpenAI 的 /chat/completions 接口，核心参数：

参数	说明
base_url	https://dashscope.aliyuncs.com/compatible-mode/v1
Authorization	Bearer YOUR_API_KEY
stream	true 开启流式输出
stream_options	{"include_usage": true} 返回 Token 用量

---

三、封装 AIClient 工具类

# utils/ai_client.py
import requests
import os
import json
import logging
from typing import Generator
from django.conf import settings

logger = logging.getLogger(__name__)


class AIClient:
    """阿里云百炼大模型调用客户端（同步流式版本）"""

    def __init__(self):
        self.api_key = os.getenv("DASHSCOPE_API_KEY")
        self.base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
        self.model = settings.CLOUD_AI_SETTINGS['model_name']
        self.timeout = 120

    def analyze_stream(self, prompt: str, timeout: int = None) -> Generator[str, None, None]:
        """
        流式调用大模型，逐块 yield 返回文本片段
        
        Args:
            prompt: 用户输入的提示词
            timeout: 超时秒数，默认 120s
            
        Yields:
            str: 每次返回的文本片段
        """
        if timeout is None:
            timeout = self.timeout

        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json',
        }

        data = {
            'model': self.model,
            'messages': [
                {'role': 'system', 'content': '你是一个专业的学术助手文献泡泡。'},
                {'role': 'user', 'content': prompt},
            ],
            'stream': True,
            'stream_options': {'include_usage': True},
        }

        try:
            with requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=data,
                stream=True,
                timeout=timeout,
            ) as response:

                if response.status_code != 200:
                    raise Exception(f"调用失败: {response.status_code} - {response.text}")

                for line in response.iter_lines():
                    if not line:
                        continue
                    line = line.decode('utf-8').strip()

                    # SSE 格式：每行以 "data: " 开头
                    if not line.startswith('data: '):
                        continue
                    line = line[6:]  # 去掉 "data: " 前缀

                    if line == '[DONE]':
                        break

                    try:
                        chunk = json.loads(line)
                        content = (
                            chunk.get('choices', [{}])[0]
                            .get('delta', {})
                            .get('content', '')
                        )
                        if content:
                            yield content
                    except json.JSONDecodeError:
                        continue

        except requests.Timeout:
            raise Exception(f"请求超时（{timeout}s）")
        except Exception as e:
            logger.error(f"AI 调用异常: {e}")
            raise

---

四、在 Django View 中使用

方式一：StreamingHttpResponse（推荐）

适合前端通过 EventSource 或 fetch 接收实时流：

# views.py
from django.http import StreamingHttpResponse
from .utils.ai_client import AIClient


def ai_stream_view(request):
    prompt = request.GET.get('prompt', '')

    def event_stream():
        client = AIClient()
        for chunk in client.analyze_stream(prompt):
            # SSE 格式
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingHttpResponse(
        event_stream(),
        content_type='text/event-stream',
    )

方式二：普通接口（拼接全文后返回）

def ai_full_view(request):
    from django.http import JsonResponse
    prompt = request.GET.get('prompt', '')
    client = AIClient()
    result = ''.join(client.analyze_stream(prompt))
    return JsonResponse({'result': result})

---

五、前端接收流式数据（示例）

const source = new EventSource(`/api/ai/stream/?prompt=介绍一下科研 AI Agent 文献泡泡`);

source.onmessage = (e) => {
    if (e.data === '[DONE]') {
        source.close();
        return;
    }
    document.getElementById('output').innerText += e.data;
};

---

六、关键设计说明

为什么用 requests 而不是 openai SDK？

• 项目已有 requests 依赖，无需引入新包
• 对 SSE 流式数据的解析逻辑完全可控
• 阿里云兼容 OpenAI 接口协议，HTTP 直调即可

流式解析流程：

响应体（SSE 格式）
    ↓ iter_lines() 逐行读取
    ↓ 过滤 "data: " 前缀
    ↓ JSON 解析 → 取 choices[0].delta.content
    ↓ yield 文本片段

---

七、支持的模型（部分）

系列	模型名称
通义千问 Max	qwen-max、qwen3-max
通义千问 Plus	qwen-plus、qwen3-plus
通义千问 Turbo	qwen-turbo
DeepSeek	deepseek-v3、deepseek-r1

完整列表见 阿里云百炼模型列表。

---

总结

整体流程非常简洁：配置 API Key → 封装工具类 → Django View 调用 → 前端接收。得益于阿里云兼容 OpenAI 协议，切换模型只需改一个 model_name 配置，几乎零改造成本。