通过 API 将Deepseek相应流式内容输出到前端

要实现通过 API 将流式内容输出到前端，可以采用以下技能方案（以 Python 后端 + 前端 JavaScript 为例）：
方案一：使用 Server-Sent Events (SSE)

这是浏览器原生支持的流式传输方案，推荐首选

1. # Flask 示例
2. from flask import Response, stream_with_context
4. @app.route('/stream')
5. def stream_data():
6.     def generate():
7.         response = client.chat.completions.create(
8.             model="deepseek-chat",
9.             messages=messages,
10.             stream=True
11.         )
12.         
13.         for chunk in response:
14.             if chunk.choices:
15.                 content = chunk.choices[0].delta.content or ""
16.                 # SSE 格式要求 data: 前缀和双换行符
17.                 yield f"data: {json.dumps({'content': content})}\n\n"
18.    
19.     return Response(stream_with_context(generate()), mimetype='text/event-stream')

复制代码

1. // 前端 JavaScript
2. const eventSource = new EventSource('/stream');
4. eventSource.onmessage = (event) => {
5.     const data = JSON.parse(event.data);
6.     document.getElementById('output').innerHTML += data.content;
7. };
9. eventSource.onerror = (err) => {
10.     console.error('EventSource failed:', err);
11.     eventSource.close();
12. };

复制代码

方案二：使用流式 HTTP 相应（NDJSON）

更通用的流式传输方案，适合非浏览器客户端

1. # FastAPI 示例
2. from fastapi import APIRouter
3. from fastapi.responses import StreamingResponse
4. import json
6. @app.get("/stream")
7. async def stream_data():
8.     async def generate():
9.         response = client.chat.completions.create(
10.             model="deepseek-chat",
11.             messages=messages,
12.             stream=True
13.         )
14.         
15.         async for chunk in response:
16.             if chunk.choices:
17.                 content = chunk.choices[0].delta.content or ""
18.                 yield json.dumps({"content": content}) + "\n"  # NDJSON 格式
19.    
20.     return StreamingResponse(generate(), media_type='application/x-ndjson')

复制代码

1. // 前端 JavaScript 使用 Fetch API
2. async function streamData() {
3.     const response = await fetch('/stream');
4.     const reader = response.body.getReader();
5.     const decoder = new TextDecoder();
6.    
7.     while(true) {
8.         const { done, value } = await reader.read();
9.         if(done) break;
10.         
11.         const chunk = decoder.decode(value);
12.         const data = JSON.parse(chunk);
13.         document.getElementById('output').innerHTML += data.content;
14.     }
15. }

复制代码

关键设置说明

• 相应头设置：
  
  1. # Flask
  2. headers = {
  3.     'Cache-Control': 'no-cache',
  4.     'Connection': 'keep-alive'
  5. }
  7. # FastAPI 自动处理

  复制代码
• 数据格式选择：
  
  
  
  • SSE (text/event-stream)：浏览器原生支持，自动重连
    
  • NDJSON (application/x-ndjson)：更通用的流式 JSON 格式
    
  • 纯文本流：简单但结构化本领弱
    
  
  
• 前端处理建议：
  
  1. // 更健壮的读取方式（处理分块不完整情况）
  2. let buffer = '';
  4. async function processChunk(chunk) {
  5.     buffer += chunk;
  6.     while(buffer.includes('\n')) {
  7.         const lineEnd = buffer.indexOf('\n');
  8.         const line = buffer.slice(0, lineEnd);
  9.         buffer = buffer.slice(lineEnd + 1);
  10.         
  11.         try {
  12.             const data = JSON.parse(line);
  13.             // 处理数据...
  14.         } catch(e) {
  15.             console.error('解析错误:', e);
  16.         }
  17.     }
  18. }

  复制代码

完整工作流程示例（FastAPI + React）

后端：

1. # main.py
2. from fastapi import FastAPI
3. from fastapi.middleware.cors import CORSMiddleware
5. app = FastAPI()
7. app.add_middleware(
8.     CORSMiddleware,
9.     allow_origins=["*"],
10.     allow_methods=["*"],
11.     allow_headers=["*"],
12. )
14. @app.get("/chat")
15. async def chat_stream(prompt: str):
16.     async def generate():
17.         response = client.chat.completions.create(
18.             model="deepseek-chat",
19.             messages=[{"role": "user", "content": prompt}],
20.             stream=True
21.         )
22.         
23.         async for chunk in response:
24.             if content := chunk.choices[0].delta.content:
25.                 yield json.dumps({"content": content})
26.    
27.     return StreamingResponse(generate(), media_type="application/x-ndjson")

复制代码

前端 React 组件：

1. // ChatComponent.jsx
2. import { useState } from 'react';
4. export default function ChatComponent() {
5.     const [output, setOutput] = useState('');
6.    
7.     const startStream = async () => {
8.         const response = await fetch('http://api/chat?prompt=你好');
9.         const reader = response.body.getReader();
10.         const decoder = new TextDecoder();
11.         let buffer = '';
12.         
13.         while(true) {
14.             const { done, value } = await reader.read();
15.             if(done) break;
16.             
17.             buffer += decoder.decode(value);
18.             while(buffer.includes('}')) {
19.                 const endIndex = buffer.indexOf('}') + 1;
20.                 const chunk = buffer.slice(0, endIndex);
21.                 buffer = buffer.slice(endIndex);
22.                
23.                 try {
24.                     const data = JSON.parse(chunk);
25.                     setOutput(prev => prev + data.content);
26.                 } catch(e) {
27.                     console.error('解析错误:', e);
28.                 }
29.             }
30.         }
31.     };
32.    
33.     return (
34.         <div>
35.             <button onClick={startStream}>开始对话</button>
36.             <div id="output">{output}</div>
37.         </div>
38.     );
39. }

复制代码

注意事项

• 毗连管理：
  
  
  
  • 设置合理的超时时间（通常 30-60 秒）
    
  • 处理客户端提前断开毗连的情况
    
  1. # FastAPI 示例
  2. try:
  3.     async for chunk in response:
  4.         # ...处理数据
  5.         if await request.is_disconnected():
  6.             break
  7. finally:
  8.     await client.close()  # 清理资源

  复制代码
  
  
• 性能优化：
  
  
  
  • 使用异步框架（FastAPI 性能优于 Flask）
    
  • 启用相应压缩
    
  1. app = FastAPI()
  2. @app.middleware("http")
  3. async def add_compression(request, call_next):
  4.     response = await call_next(request)
  5.     response.headers["Content-Encoding"] = "gzip"
  6.     return response

  复制代码
  
  
• 安全考虑：
  
  
  
  • 限定最大并发毗连数
    
  • 实施速率限定
    
  1. from fastapi import Request
  2. from fastapi.middleware import Middleware
  3. from slowapi import Limiter
  4. from slowapi.util import get_remote_address
  6. limiter = Limiter(key_func=get_remote_address)
  7. app.state.limiter = limiter
  9. @app.get("/chat")
  10. @limiter.limit("10/minute")
  11. async def chat_stream(request: Request):
  12.     # ...

  复制代码
  
  
• 错误处理增强：
  
  1. async def generate():
  2.     try:
  3.         response = client.chat.completions.create(...)
  4.         async for chunk in response:
  5.             # 处理数据...
  6.     except Exception as e:
  7.         yield json.dumps({"error": str(e)})
  8.     finally:
  9.         await client.close()  # 确保释放资源

  复制代码

这些方案可根据具体需求组合使用，建议优先选择 SSE 方案（浏览器兼容性好），需要支持更复杂场景时可考虑 WebSocket，但后者实现成本较高。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。