FastAPI + 阿里云百炼（通义千问）AI 单轮对话实战教程
·
从零开始，一步步实现一个 AI 聊天服务。
两种输出方式：
- 📨 直接输出：等 AI 生成完整回答后一次性返回
- ⚡ 流式输出：逐字实时推送，打字机效果，前端用原生
fetch实现前端不依赖任何框架，纯原生 HTML + JavaScript，好理解、易上手。
📖 目录
1. 准备工作：注册阿里云百炼 & 获取 API Key
1.1 什么是阿里云百炼？
阿里云百炼 是阿里云推出的大模型服务平台，提供通义千问（Qwen）系列模型的 API 调用能力。
为什么选阿里云百炼？
- 🆓 新用户有免费额度（百万 Token），足够学习和测试
- 🔗 OpenAI 兼容接口，和 ChatGPT API 调用方式几乎一样，学习成本低
- 🇨🇳 国内访问稳定，不需要代理
- 💰 按量付费，用多少花多少
1.2 注册步骤（约 5 分钟）
text
复制
第一步：打开浏览器，访问 https://bailian.console.aliyun.com
第二步：用阿里云账号登录（没有的话用支付宝/淘宝账号注册一个）
第三步：进入控制台后，左侧菜单找到「模型广场」
第四步：在模型列表中找到「通义千问-Plus」或「通义千问-Turbo」
第五步：点击模型 → 查看详情 → 开通服务（新用户免费）
1.3 获取 API Key
text
复制
第一步：控制台右上角，点击头像 → 「API-KEY 管理」
第二步：点击「创建 API-KEY」
第三步：复制生成的 Key（只显示一次！务必保存好）
拿到 API Key 后，创建一个 .env 文件保存：
bash
复制
# 在项目目录 fastapi-ai-chat/ 下创建 .env 文件
DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
⚠️ 安全提醒：
.env文件不要提交到 Git！项目已包含.gitignore排除它。
2. 项目结构一览
text
复制
fastapi-ai-chat/
├── backend.py # FastAPI 后端（核心代码）
├── .env # 环境变量（你的 API Key）
├── requirements.txt # Python 依赖
└── templates/ # 前端页面
├── index.html # 首页（入口）
├── direct.html # 直接输出页面
└── stream.html # 流式输出页面
3. 安装依赖 & 配置环境变量
3.1 安装 Python 依赖
bash
复制
# 进入项目目录
cd fastapi-ai-chat
# 创建虚拟环境（推荐）
python -m venv venv
# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate
# 安装依赖
pip install fastapi uvicorn httpx python-dotenv jinja2
3.2 配置 API Key
在 fastapi-ai-chat/ 目录下创建 .env 文件：
env
复制
DASHSCOPE_API_KEY=sk-你的阿里云百炼APIKey
4. 后端实现详解
后端用 FastAPI 框架，定义两个端点，分别对应两种输出方式。
核心架构图
text
复制
┌──────────┐ POST /chat ┌──────────────┐ stream=False ┌──────────────┐
│ │ ──────────────────────▶ │ │ ─────────────────▶ │ │
│ 浏览器 │ │ FastAPI 后端 │ │ 阿里云百炼 │
│ │ ◀────────────────────── │ │ ◀───────────────── │ (通义千问) │
└──────────┘ JSON { reply } └──────────────┘ 完整回复 └──────────────┘
┌──────────┐ POST /chat/stream ┌──────────────┐ stream=True ┌──────────────┐
│ │ ──────────────────────▶ │ │ ─────────────────▶ │ │
│ 浏览器 │ │ FastAPI 后端 │ │ 阿里云百炼 │
│ │ ◀── SSE 逐字推送 ────── │ │ ◀── SSE 逐块接收 ── │ (通义千问) │
└──────────┘ └──────────────┘ └──────────────┘
SSE 是什么？ Server-Sent Events（服务器推送事件），让服务器可以持续向浏览器推送数据，而不需要浏览器反复请求。非常适合 AI 流式输出的场景。
4.1 方式一：直接输出（/chat）
python
复制
@app.post("/chat")
async def chat(request: Request):
"""等待 AI 完整回复后一次性返回。"""
body = await request.json()
user_message = body.get("message", "")
# 调用阿里云 DashScope API（stream=False）
headers = {"Authorization": f"Bearer {DASHSCOPE_API_KEY}"}
payload = {
"model": "qwen-plus",
"messages": [
{"role": "system", "content": "你是一个乐于助人的AI助手"},
{"role": "user", "content": user_message},
],
"stream": False, # ← 关键：不开启流式
}
async with httpx.AsyncClient() as client:
response = await client.post(
"https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
headers=headers,
json=payload,
)
data = response.json()
ai_reply = data["choices"][0]["message"]["content"]
return {"reply": ai_reply}
流程：
- 接收前端发来的
{"message": "你好"} - 调用阿里云 API，设置
stream=False - 阿里云生成完整回复后返回
- 后端把回复打包成
{"reply": "你好！有什么..."}返回前端
优点：代码简单，一看就懂。 缺点：用户需要等待，如果回复很长会等比较久。
4.2 方式二：流式输出（/chat/stream）
python
复制
@app.post("/chat/stream")
async def chat_stream(request: Request):
"""逐字推送 AI 回复到前端。"""
body = await request.json()
user_message = body.get("message", "")
async def event_generator():
"""异步生成器：逐块产出 SSE 数据"""
headers = {"Authorization": f"Bearer {DASHSCOPE_API_KEY}"}
payload = {
"model": "qwen-plus",
"messages": [...],
"stream": True, # ← 关键：开启流式
}
async with httpx.AsyncClient() as client:
async with client.stream("POST", url, headers=headers, json=payload) as resp:
async for line in resp.aiter_lines():
if line.startswith("data: "):
data_str = line[6:]
if data_str == "[DONE]":
yield f"data: {json.dumps({'done': True})}\n\n"
break
chunk = json.loads(data_str)
content = chunk["choices"][0]["delta"].get("content", "")
if content:
yield f"data: {json.dumps({'content': content})}\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
流程：
- 接收前端请求，设置
stream=True调用阿里云 - 阿里云不再等全部生成完，而是每生成一段就发送一段（SSE 格式）
- 后端用
async for line in resp.aiter_lines()逐行读取 - 每读到一段内容，立刻用
yield推送给前端 - 前端收到一段就显示一段 → 打字机效果
关键函数对比：
| 特性 | 直接输出 | 流式输出 |
|---|---|---|
| API 调用 | client.post() |
client.stream() |
stream 参数 |
False |
True |
| 返回方式 | return {"reply"} |
StreamingResponse(generator) |
| 前端接收 | 一次性 JSON | 逐块 SSE 数据 |
5. 前端实现详解
前端使用纯原生技术栈，不依赖 React、Vue 等框架。
5.1 直接输出页面
调用方式就是标准 fetch，和其他 API 请求完全一样：
javascript
复制
// 1. 发送请求
const response = await fetch('/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: '你好' }),
});
// 2. 解析 JSON
const data = await response.json();
// 3. 拿到回复
console.log(data.reply); // "你好！有什么可以帮助你的吗？"
就这么简单！ 和调用任何普通 REST API 没有任何区别。
5.2 流式输出页面（原生 fetch + ReadableStream）⭐
这是本教程的重点——不依赖 EventSource，不用任何库，纯原生 fetch 实现流式读取。
为什么不直接用 EventSource？因为 EventSource 只支持 GET 请求，而我们需要 POST 发送用户消息。
三步核心流程
text
复制
fetch('/chat/stream', { method: 'POST', body: ... })
│
▼
response.body.getReader() ← 获取流读取器
│
▼
while (true) {
const { done, value } = await reader.read()
if (done) break ← 流结束，退出
解码 value → 解析 SSE → 逐字显示
}
完整代码（带详细注释）
javascript
复制
async function sendMessageStream() {
const message = '你好，请介绍一下你自己';
// ═══════════════════════════════════════════
// 第 1 步：发起 POST 请求
// 和普通 fetch 完全一样，没有任何区别！
// ═══════════════════════════════════════════
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message }),
});
// ═══════════════════════════════════════════
// 第 2 步：获取流读取器
//
// response.body 是 ReadableStream 对象
// .getReader() 返回一个 reader
// reader.read() 每次读一块数据
//
// 对比传统方式：
// 传统：response.json() → 等全部数据到齐
// 流式：response.body.getReader() → 来一块读一块
// ═══════════════════════════════════════════
const reader = response.body.getReader();
const decoder = new TextDecoder(); // 二进制 → 字符串
let buffer = ''; // 缓冲区：处理不完整的 SSE 行
// ═══════════════════════════════════════════
// 第 3 步：循环读取
//
// reader.read() 返回 { done, value }
// - done: true → 流结束了
// - value: Uint8Array → 这次收到的二进制数据
// ═══════════════════════════════════════════
while (true) {
const { done, value } = await reader.read();
if (done) {
console.log('流结束，完整回复：', fullContent);
break;
}
// 解码二进制数据
buffer += decoder.decode(value, { stream: true });
// 按行分割（SSE 格式是 data: {...}\n\n）
const lines = buffer.split('\n');
buffer = lines.pop() || ''; // 最后一行可能不完整，留着下次用
for (const line of lines) {
if (!line.startsWith('data: ')) continue;
const data = JSON.parse(line.slice(6));
if (data.content) {
// ★ 收到一段文本 → 立刻显示！
document.getElementById('output').textContent += data.content;
}
if (data.done) {
console.log('生成完成！');
}
}
}
}
逐行解读
| 代码 | 作用 |
|---|---|
response.body.getReader() |
获取流的"水龙头"，可以控制开关 |
new TextDecoder() |
把二进制数据（Uint8Array）翻译成人类可读的文字 |
decoder.decode(value, {stream: true}) |
stream: true 告诉解码器"后面还有数据"，防止多字节字符（如中文）被截断 |
reader.read() |
读取下一块数据。返回 {done: false, value: Uint8Array} |
buffer 缓冲区 |
因为网络是分块到达的，一行 SSE 数据可能被切成两半，用 buffer 拼接完整行 |
数据流示意图
text
复制
阿里云 ──SSE──▶ FastAPI ──SSE──▶ 浏览器 fetch
│
data: {"content":"你"} │ reader.read()
data: {"content":"好"} │ ↓
data: {"content":"！"} │ "你" → 显示
data: {"content":"我"} │ "好" → 显示
data: {"content":"是"} │ "！" → 显示
... │ ...
data: {"done":true} │ done=true → 结束
6. 运行 & 体验
bash
复制
# 1. 进入项目目录
cd fastapi-ai-chat
# 2. 确认 .env 文件中有你的 API Key
# DASHSCOPE_API_KEY=sk-xxxxxxxx
# 3. 启动服务器
python backend.py
# 4. 打开浏览器访问
# 首页： http://localhost:8000
# 直接输出： http://localhost:8000/direct
# 流式输出： http://localhost:8000/stream
体验对比
- 先打开 直接输出页面 (
/direct)，输入一个问题 → 观察"等待 → 一次性出现" - 再打开 流式输出页面 (
/stream)，输入同样的问题 → 观察"逐字打出"的效果 - 对比两种方式的体验差异
7. 两种方式的对比总结
| 维度 | 直接输出 (POST /chat) | 流式输出 (POST /chat/stream) |
|---|---|---|
| 用户体验 | 需要等待，可能感觉"卡住了" | 逐字展示，像真人在打字 ⭐ |
| 后端实现 | 简单：await client.post() |
稍复杂：client.stream() + 生成器 |
| 前端实现 | 极简：标准 fetch + .json() |
需处理 ReadableStream，但也不难 |
| 适用场景 | 短回复、批量处理、API 集成 | 聊天 UI、长文生成、需要感知进度的场景 |
| 首字延迟 | 等于总生成时间 | 通常 < 1 秒 |
| 网络开销 | 较低（一次请求） | 稍高（持续连接） |
选择建议
- 如果你在做 后台批量任务、API 对 API 调用 → 直接输出就够了
- 如果你在做 聊天界面、面向用户的产品 → 一定要用流式输出
8. 常见问题 & 排错
Q1：启动报错 ModuleNotFoundError: No module named 'xxx'
bash
复制
# 确保安装了所有依赖
pip install fastapi uvicorn httpx python-dotenv jinja2
Q2：API 返回 401 Unauthorized
检查 .env 文件中的 DASHSCOPE_API_KEY 是否正确：
bash
复制
# 在项目目录运行
python -c "from dotenv import load_dotenv; import os; load_dotenv(); print(os.getenv('DASHSCOPE_API_KEY')[:10] + '...')"
Q3：流式输出不显示 / 卡住
- 检查浏览器控制台（F12）有没有报错
- 确认后端是否正常运行（终端有没有日志）
- 网络问题：阿里云 API 在国内访问通常没问题，如果超时检查网络
Q4：如何换成其他模型？
修改 backend.py 中的 MODEL 变量：
python
复制
# 更多选择：
MODEL = "qwen-turbo" # 更快、更便宜，适合简单任务
MODEL = "qwen-plus" # 均衡，推荐日常使用
MODEL = "qwen-max" # 最强，适合复杂推理
MODEL = "qwen-long" # 超长上下文（1000万 Token）
Q5：如何部署到服务器？
bash
复制
# 使用 uvicorn 启动（生产环境）
uvicorn backend:app --host 0.0.0.0 --port 8000 --workers 4
# 建议搭配 nginx 反向代理 + systemd 守护进程
Q6：免费额度用完了怎么办？
阿里云百炼按量付费，价格很便宜：
qwen-turbo：约 ¥0.3/百万 Token（输入），¥0.6/百万 Token（输出）qwen-plus：约 ¥0.8/百万 Token（输入），¥2/百万 Token（输出）
一次普通对话约消耗 500-2000 Token，成本几乎可以忽略不计。
📚 延伸学习
🎉 恭喜！你已经学会了如何用 FastAPI + 阿里云百炼实现 AI 对话，包括直接输出和流式输出两种方式。
核心要点：
- 直接输出 =
stream=False+ 普通fetch- 流式输出 =
stream=True+response.body.getReader()- 流式前端只需要 3 步：fetch → getReader → while(read)
AtomGit AI 社区提供模型库、数据集、Agent、Token等资源
更多推荐
3
0- 0
已为社区贡献1条内容
所有评论(0)