从一个真实demo项目出发讲清楚 AI 聊天应用里最关键的流式通信模式。前言当你你在聊天页面输入【各科平均分是多少请用柱状图展示】点击发送。几秒后 AI 像打字机一样逐字吐出答案最后还附上一张 ECharts 图表。这个过程看似简单背后却是一段精心设计的旅程浏览器 → FastAPI → Dify AI Agent → FastAPI → 浏览器该篇将带你一步步拆解其中的关键节点帮助你建立【流式通信】的完整流程。一、三个角色角色技术栈职责前端HTML JS ECharts收集输入、打字机效果、渲染图表后端Python FastAPI httpx代理请求、安全转发流数据DifyDify NLP2SQL Agent理解自然语言、查库、生成回答为什么不直接从前端请求 DifyAPI Key 写在前端 JS 里任何人打开开发者工具都能看到。正确做法是前端 → 后端 → Dify。就像去餐厅点菜你告诉服务员服务员有钥匙进厨房。二、一次请求的完整旅程1. 前端发送请求用户点击发送后前端发起 POST 请求因为是在【提交】问题而非【获取】页面const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: 各科平均分是多少, conversation_id: }), });2. 后端接收并返回流式响应FastAPI 用 Pydantic 定义请求结构然后返回StreamingResponseclass ChatRequest(BaseModel): question: str conversation_id: str app.post(/api/chat) async def chat(req: ChatRequest): return StreamingResponse( ask_dify_stream(req.question, req.conversation_id), media_typetext/event-stream, )核心决策返回 StreamingResponse而非JSONResponsemedia_type设为text/event-stream——告诉浏览器数据会一段一段来。3. 后端请求 Dify后端从.env读取 Dify 配置组装请求headers {Authorization: fBearer {DIFY_API_TOKEN}} payload { inputs: {}, query: question, response_mode: streaming, # 关键要求 Dify 流式返回 conversation_id: conversation_id, user: web-user, }然后用httpx流式请求async with httpx.AsyncClient(timeoutNone) as client: async with client.stream(POST, DIFY_CHAT_URL, headersheaders, jsonpayload) as resp: ...timeoutNone避免 AI 生成较慢时过早断连。4. Dify 处理并流式返回Dify 的 NLP2SQL Agent 收到问题后NLP 理解意图SQL 转化为数据库查询。然后边生成边以 SSE 格式返回data: {event:message,answer:各科平均分如下} data: {event:message,answer:语文 86数学 91英语 88。} data: {event:message_end,conversation_id:abc-123}5. 后端原样转发后端逐行读取 Dify 返回遇到data:开头就转发——不改内容纯粹的传声筒async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n\nyield 而非return——一点一点「让渡」数据。6. 前端流式读取 —— 核心差异这是全文最重要的部分。前端拿到response后没有这样做// 普通做法等全部返回后一次性解析 const data await response.json();而是这样做了// 流式做法自己动手一块一块读 const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const events buffer.split(\n\n); buffer events.pop(); // 最后一个可能不完整留到下次拼接 for (const item of events) { const line item.split(\n).find(r r.startsWith(data:)); if (!line) continue; const data JSON.parse(line.slice(5)); // 去掉 data: 前缀 if (data.event message || data.event agent_message) { fullAnswer data.answer || ; aiMessageText.textContent fullAnswer; // 实时更新页面 } } }两种写法的本质区别response.json()response.body.getReader()获取方式等服务器全部返回完返回一点读一点体验等待后一次性出现打字机效果内存全部加载流式处理场景普通 API、配置AI 聊天、实时日志底层内部 read→JSON.parse直接操作 ReadableStream比喻response.json()是厨师做完一整桌菜再端出来getReader()是做好一道端一道。7. 流结束后渲染 Markdown 与图表这里没有导入其他库来做markdown的渲染而是采用了替换的方式流式接收期间前端用textContent纯文本显示避免 Markdown 不完整导致渲染错乱。结束后做三件事渲染 Markdown## 标题、**加粗**等转为 HTML。提取并渲染 EChartsconst regex /(?:json|echarts)?\s*([\s\S]*?)/g; const json JSON.parse(match[1]); if (json.xAxis || json.series) { // 用 ECharts 渲染 const chart echarts.init(chartDiv); chart.setOption(json); }三、为什么这样设计1. 安全性API Key 只存于服务端.env浏览器代码中找不到任何密钥。2. 体验非流式下用户要等十几秒才看到回复流式模式下第一秒就有反馈。3. 解耦前端只知道/api/chat和 SSE 协议。后端换 AI 服务商前端无需改动。四、动手实践最小示例后端FastAPIimport asyncio, json, uvicorn from fastapi import FastAPI from fastapi.responses import StreamingResponse, FileResponse from starlette.staticfiles import StaticFiles app FastAPI( title流式输出测试, version1.0.0 ) async def generate(): for word in [你好, , 这是, 流式, 响应, 。]: yield fdata: {json.dumps({event: message, answer: word})}\n\n await asyncio.sleep(0.3) app.get(/) async def root(): return FileResponse(index.html) app.post(/api/chat) async def chat(): return StreamingResponse(generate(), media_typetext/event-stream)前端流式读取(async () { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: 你好 }), }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const events buffer.split(\n\n); buffer events.pop(); for (const item of events) { const line item.split(\n).find(r r.startsWith(data:)); if (line) console.log(JSON.parse(line.slice(5)).answer); } } })();五、常见问题速查现象原因解决发送失败后端未启动访问/api/healthhas_token: false.env未配置检查根目录.envDify 401Key 错误重新复制密钥有文字无图表JSON 缺少xAxis/series检查代码块结构显示了图表代码JSON 格式非法确保纯 JSON六、总结这个项目的核心架构可以用四句话概括前端负责交互 —— 收集输入、打字机效果、渲染图表 后端负责安全转发 —— 持有密钥、代理请求、原样转发 SSE Dify 负责 AI —— NLP2SQL 理解意图、生成回答 前端负责最终展示 —— Markdown 转换、JSON 提取、ECharts 渲染而贯穿全链路的那条金线就是流式通信——从 Dify 的response_mode: streaming到后端的httpx.stream()yield再到前端的response.body.getReader()三个环节环环相扣共同构成了打字机般的流畅体验。当从const data await response.json()走向const reader response.body.getReader()的那一刻完成的不仅是一次 API 调用的升级更是从「请求-响应」思维到「流式通信」思维的跃迁。理解了这一点你就掌握了现代 AI 聊天应用最核心的通信模式。
HTML+fastAPI+Dify|打通前后端至智能体的路
从一个真实demo项目出发讲清楚 AI 聊天应用里最关键的流式通信模式。前言当你你在聊天页面输入【各科平均分是多少请用柱状图展示】点击发送。几秒后 AI 像打字机一样逐字吐出答案最后还附上一张 ECharts 图表。这个过程看似简单背后却是一段精心设计的旅程浏览器 → FastAPI → Dify AI Agent → FastAPI → 浏览器该篇将带你一步步拆解其中的关键节点帮助你建立【流式通信】的完整流程。一、三个角色角色技术栈职责前端HTML JS ECharts收集输入、打字机效果、渲染图表后端Python FastAPI httpx代理请求、安全转发流数据DifyDify NLP2SQL Agent理解自然语言、查库、生成回答为什么不直接从前端请求 DifyAPI Key 写在前端 JS 里任何人打开开发者工具都能看到。正确做法是前端 → 后端 → Dify。就像去餐厅点菜你告诉服务员服务员有钥匙进厨房。二、一次请求的完整旅程1. 前端发送请求用户点击发送后前端发起 POST 请求因为是在【提交】问题而非【获取】页面const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: 各科平均分是多少, conversation_id: }), });2. 后端接收并返回流式响应FastAPI 用 Pydantic 定义请求结构然后返回StreamingResponseclass ChatRequest(BaseModel): question: str conversation_id: str app.post(/api/chat) async def chat(req: ChatRequest): return StreamingResponse( ask_dify_stream(req.question, req.conversation_id), media_typetext/event-stream, )核心决策返回 StreamingResponse而非JSONResponsemedia_type设为text/event-stream——告诉浏览器数据会一段一段来。3. 后端请求 Dify后端从.env读取 Dify 配置组装请求headers {Authorization: fBearer {DIFY_API_TOKEN}} payload { inputs: {}, query: question, response_mode: streaming, # 关键要求 Dify 流式返回 conversation_id: conversation_id, user: web-user, }然后用httpx流式请求async with httpx.AsyncClient(timeoutNone) as client: async with client.stream(POST, DIFY_CHAT_URL, headersheaders, jsonpayload) as resp: ...timeoutNone避免 AI 生成较慢时过早断连。4. Dify 处理并流式返回Dify 的 NLP2SQL Agent 收到问题后NLP 理解意图SQL 转化为数据库查询。然后边生成边以 SSE 格式返回data: {event:message,answer:各科平均分如下} data: {event:message,answer:语文 86数学 91英语 88。} data: {event:message_end,conversation_id:abc-123}5. 后端原样转发后端逐行读取 Dify 返回遇到data:开头就转发——不改内容纯粹的传声筒async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n\nyield 而非return——一点一点「让渡」数据。6. 前端流式读取 —— 核心差异这是全文最重要的部分。前端拿到response后没有这样做// 普通做法等全部返回后一次性解析 const data await response.json();而是这样做了// 流式做法自己动手一块一块读 const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const events buffer.split(\n\n); buffer events.pop(); // 最后一个可能不完整留到下次拼接 for (const item of events) { const line item.split(\n).find(r r.startsWith(data:)); if (!line) continue; const data JSON.parse(line.slice(5)); // 去掉 data: 前缀 if (data.event message || data.event agent_message) { fullAnswer data.answer || ; aiMessageText.textContent fullAnswer; // 实时更新页面 } } }两种写法的本质区别response.json()response.body.getReader()获取方式等服务器全部返回完返回一点读一点体验等待后一次性出现打字机效果内存全部加载流式处理场景普通 API、配置AI 聊天、实时日志底层内部 read→JSON.parse直接操作 ReadableStream比喻response.json()是厨师做完一整桌菜再端出来getReader()是做好一道端一道。7. 流结束后渲染 Markdown 与图表这里没有导入其他库来做markdown的渲染而是采用了替换的方式流式接收期间前端用textContent纯文本显示避免 Markdown 不完整导致渲染错乱。结束后做三件事渲染 Markdown## 标题、**加粗**等转为 HTML。提取并渲染 EChartsconst regex /(?:json|echarts)?\s*([\s\S]*?)/g; const json JSON.parse(match[1]); if (json.xAxis || json.series) { // 用 ECharts 渲染 const chart echarts.init(chartDiv); chart.setOption(json); }三、为什么这样设计1. 安全性API Key 只存于服务端.env浏览器代码中找不到任何密钥。2. 体验非流式下用户要等十几秒才看到回复流式模式下第一秒就有反馈。3. 解耦前端只知道/api/chat和 SSE 协议。后端换 AI 服务商前端无需改动。四、动手实践最小示例后端FastAPIimport asyncio, json, uvicorn from fastapi import FastAPI from fastapi.responses import StreamingResponse, FileResponse from starlette.staticfiles import StaticFiles app FastAPI( title流式输出测试, version1.0.0 ) async def generate(): for word in [你好, , 这是, 流式, 响应, 。]: yield fdata: {json.dumps({event: message, answer: word})}\n\n await asyncio.sleep(0.3) app.get(/) async def root(): return FileResponse(index.html) app.post(/api/chat) async def chat(): return StreamingResponse(generate(), media_typetext/event-stream)前端流式读取(async () { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: 你好 }), }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const events buffer.split(\n\n); buffer events.pop(); for (const item of events) { const line item.split(\n).find(r r.startsWith(data:)); if (line) console.log(JSON.parse(line.slice(5)).answer); } } })();五、常见问题速查现象原因解决发送失败后端未启动访问/api/healthhas_token: false.env未配置检查根目录.envDify 401Key 错误重新复制密钥有文字无图表JSON 缺少xAxis/series检查代码块结构显示了图表代码JSON 格式非法确保纯 JSON六、总结这个项目的核心架构可以用四句话概括前端负责交互 —— 收集输入、打字机效果、渲染图表 后端负责安全转发 —— 持有密钥、代理请求、原样转发 SSE Dify 负责 AI —— NLP2SQL 理解意图、生成回答 前端负责最终展示 —— Markdown 转换、JSON 提取、ECharts 渲染而贯穿全链路的那条金线就是流式通信——从 Dify 的response_mode: streaming到后端的httpx.stream()yield再到前端的response.body.getReader()三个环节环环相扣共同构成了打字机般的流畅体验。当从const data await response.json()走向const reader response.body.getReader()的那一刻完成的不仅是一次 API 调用的升级更是从「请求-响应」思维到「流式通信」思维的跃迁。理解了这一点你就掌握了现代 AI 聊天应用最核心的通信模式。
