从零开始一步步实现一个 AI 聊天服务。两种输出方式直接输出等 AI 生成完整回答后一次性返回⚡流式输出逐字实时推送打字机效果前端用原生fetch实现前端不依赖任何框架纯原生 HTML JavaScript好理解、易上手。 目录准备工作注册阿里云百炼 获取 API Key项目结构一览安装依赖 配置环境变量后端实现详解4.1 方式一直接输出/chat4.2 方式二流式输出/chat/stream前端实现详解5.1 直接输出页面5.2 流式输出页面原生 fetch ReadableStream运行 体验两种方式的对比总结常见问题 排错1. 准备工作注册阿里云百炼 获取 API Key1.1 什么是阿里云百炼阿里云百炼 是阿里云推出的大模型服务平台提供通义千问Qwen系列模型的 API 调用能力。为什么选阿里云百炼新用户有免费额度百万 Token足够学习和测试OpenAI 兼容接口和 ChatGPT API 调用方式几乎一样学习成本低国内访问稳定不需要代理按量付费用多少花多少1.2 注册步骤约 5 分钟text复制第一步打开浏览器访问 https://bailian.console.aliyun.com 第二步用阿里云账号登录没有的话用支付宝/淘宝账号注册一个 第三步进入控制台后左侧菜单找到「模型广场」 第四步在模型列表中找到「通义千问-Plus」或「通义千问-Turbo」 第五步点击模型 → 查看详情 → 开通服务新用户免费1.3 获取 API Keytext复制第一步控制台右上角点击头像 → 「API-KEY 管理」 第二步点击「创建 API-KEY」 第三步复制生成的 Key只显示一次务必保存好拿到 API Key 后创建一个.env文件保存bash复制# 在项目目录 fastapi-ai-chat/ 下创建 .env 文件 DASHSCOPE_API_KEYsk-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 jinja23.2 配置 API Key在fastapi-ai-chat/目录下创建.env文件env复制DASHSCOPE_API_KEYsk-你的阿里云百炼APIKey4. 后端实现详解后端用FastAPI框架定义两个端点分别对应两种输出方式。核心架构图text复制┌──────────┐ POST /chat ┌──────────────┐ streamFalse ┌──────────────┐ │ │ ──────────────────────▶ │ │ ─────────────────▶ │ │ │ 浏览器 │ │ FastAPI 后端 │ │ 阿里云百炼 │ │ │ ◀────────────────────── │ │ ◀───────────────── │ (通义千问) │ └──────────┘ JSON { reply } └──────────────┘ 完整回复 └──────────────┘ ┌──────────┐ POST /chat/stream ┌──────────────┐ streamTrue ┌──────────────┐ │ │ ──────────────────────▶ │ │ ─────────────────▶ │ │ │ 浏览器 │ │ FastAPI 后端 │ │ 阿里云百炼 │ │ │ ◀── SSE 逐字推送 ────── │ │ ◀── SSE 逐块接收 ── │ (通义千问) │ └──────────┘ └──────────────┘ └──────────────┘SSE 是什么Server-Sent Events服务器推送事件让服务器可以持续向浏览器推送数据而不需要浏览器反复请求。非常适合 AI 流式输出的场景。4.1 方式一直接输出/chatpython复制app.post(/chat) async def chat(request: Request): 等待 AI 完整回复后一次性返回。 body await request.json() user_message body.get(message, ) # 调用阿里云 DashScope APIstreamFalse headers {Authorization: fBearer {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, headersheaders, jsonpayload, ) data response.json() ai_reply data[choices][0][message][content] return {reply: ai_reply}流程接收前端发来的{message: 你好}调用阿里云 API设置streamFalse阿里云生成完整回复后返回后端把回复打包成{reply: 你好有什么...}返回前端优点代码简单一看就懂。缺点用户需要等待如果回复很长会等比较久。4.2 方式二流式输出/chat/streampython复制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: fBearer {DASHSCOPE_API_KEY}} payload { model: qwen-plus, messages: [...], stream: True, # ← 关键开启流式 } async with httpx.AsyncClient() as client: async with client.stream(POST, url, headersheaders, jsonpayload) as resp: async for line in resp.aiter_lines(): if line.startswith(data: ): data_str line[6:] if data_str [DONE]: yield fdata: {json.dumps({done: True})}\n\n break chunk json.loads(data_str) content chunk[choices][0][delta].get(content, ) if content: yield fdata: {json.dumps({content: content})}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)流程接收前端请求设置streamTrue调用阿里云阿里云不再等全部生成完而是每生成一段就发送一段SSE 格式后端用async for line in resp.aiter_lines()逐行读取每读到一段内容立刻用yield推送给前端前端收到一段就显示一段 →打字机效果关键函数对比特性直接输出流式输出API 调用client.post()client.stream()stream参数FalseTrue返回方式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} │ donetrue → 结束6. 运行 体验bash复制# 1. 进入项目目录 cd fastapi-ai-chat # 2. 确认 .env 文件中有你的 API Key # DASHSCOPE_API_KEYsk-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 xxxbash复制# 确保安装了所有依赖 pip install fastapi uvicorn httpx python-dotenv jinja2Q2API 返回 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万 TokenQ5如何部署到服务器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使用 ReadableStreamMDNServer-Sent Events恭喜你已经学会了如何用 FastAPI 阿里云百炼实现 AI 对话包括直接输出和流式输出两种方式。核心要点直接输出 streamFalse 普通fetch流式输出 streamTrueresponse.body.getReader()流式前端只需要 3 步fetch → getReader → while(read)
