FastAPI + 阿里云百炼（通义千问）AI 单轮对话实战教程-尧图网络科技

从零开始，一步步实现一个 AI 聊天服务。
两种输出方式：
📨直接输出：等 AI 生成完整回答后一次性返回
⚡流式输出：逐字实时推送，打字机效果，前端用原生fetch实现
前端不依赖任何框架，纯原生 HTML + JavaScript，好理解、易上手。

📖 目录

准备工作：注册阿里云百炼 & 获取 API Key
项目结构一览
安装依赖 & 配置环境变量
后端实现详解
- 4.1 方式一：直接输出（/chat）
- 4.2 方式二：流式输出（/chat/stream）
前端实现详解
- 5.1 直接输出页面
- 5.2 流式输出页面（原生 fetch + ReadableStream）
运行 & 体验
两种方式的对比总结
常见问题 & 排错

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 官方文档
MDN：使用 ReadableStream
MDN：Server-Sent Events

🎉恭喜！你已经学会了如何用 FastAPI + 阿里云百炼实现 AI 对话，包括直接输出和流式输出两种方式。
核心要点：
直接输出 =stream=False+ 普通fetch
流式输出 =stream=True+response.body.getReader()
流式前端只需要 3 步：fetch → getReader → while(read)