FastAPI与ollama大模型异步流式响应实战
1. 项目背景与问题起源
上周我在尝试用FastAPI对接ollama大模型时,遇到了asyncio的深坑。原本以为简单的接口调用,结果在实现流式响应(StreamingResponse)时遭遇了各种异步编程的陷阱。这个案例特别适合分享给正在尝试将大模型能力集成到Web服务中的开发者们。
ollama作为当前热门的本地大模型运行工具,确实让开发者能够快速在本地环境运行各类开源模型。但当我试图将其与FastAPI这个异步Web框架结合时,发现官方文档中的简单示例远远不能满足实际生产需求。特别是在处理长时间运行的模型推理请求时,传统的同步调用方式会导致服务完全阻塞。
2. 技术栈选型分析
2.1 为什么选择FastAPI+ollama组合
FastAPI的异步特性理论上非常适合大模型服务:
- 原生支持ASGI标准
- 内置对WebSocket和SSE的支持
- 自动生成的API文档
- 出色的性能基准测试数据
ollama的优势在于:
- 简化了本地大模型的部署流程
- 提供统一的REST API接口
- 支持模型的热加载和版本管理
- 活跃的社区和持续更新
2.2 关键依赖版本
在实际开发中,版本兼容性至关重要:
fastapi==0.109.1 ollama==0.1.27 httpx==0.27.0 python-multipart==0.0.63. 基础实现与首次翻车
3.1 初始同步版本实现
我最开始的实现是这样的:
from fastapi import FastAPI import ollama app = FastAPI() @app.post("/chat") def chat(prompt: str): response = ollama.chat(model="llama3", messages=[{"role": "user", "content": prompt}]) return response["message"]["content"]这个版本的问题立即显现:
- 每个请求都会阻塞事件循环
- 无法处理并发请求
- 响应时间不可控
3.2 第一次异步改造
意识到问题后,我尝试了异步改造:
@app.post("/chat") async def chat(prompt: str): response = await ollama.chat(model="llama3", messages=[{"role": "user", "content": prompt}]) return response["message"]["content"]结果发现ollama的Python客户端并不原生支持异步!这就是第一个大坑。
4. 深入异步编程解决方案
4.1 使用httpx实现异步HTTP客户端
解决方案是绕过官方客户端,直接使用httpx调用ollama的HTTP接口:
import httpx async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post( "http://localhost:11434/api/chat", json={ "model": "llama3", "messages": [{"role": "user", "content": prompt}] } )4.2 处理流式响应
真正的挑战在于实现流式输出。大模型的响应往往需要较长时间,用户希望看到逐步输出的结果,而不是等待全部生成完毕。
from fastapi.responses import StreamingResponse @app.post("/stream-chat") async def stream_chat(prompt: str): async with httpx.AsyncClient() as client: async with client.stream( "POST", "http://localhost:11434/api/chat", json={"model": "llama3", "messages": [{"role": "user", "content": prompt}]}, timeout=None ) as response: async for chunk in response.aiter_bytes(): yield chunk5. 前端对话窗口实现
5.1 基本HTML/JS实现
配合后端流式接口,前端实现也很关键:
<div id="chat-container"> <div id="chat-history"></div> <input type="text" id="user-input"> <button onclick="sendMessage()">发送</button> </div> <script> async function sendMessage() { const input = document.getElementById('user-input').value; const response = await fetch('/stream-chat', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({prompt: input}) }); const reader = response.body.getReader(); while(true) { const {done, value} = await reader.read(); if(done) break; const text = new TextDecoder().decode(value); document.getElementById('chat-history').innerHTML += text; } } </script>5.2 优化用户体验
实际使用中发现几个需要改进的点:
- 添加消息发送禁用状态
- 实现打字机效果
- 处理网络中断情况
- 添加消息历史持久化
6. 性能优化与错误处理
6.1 连接池管理
频繁创建HTTP连接会导致性能问题,正确的做法是:
from contextlib import asynccontextmanager from fastapi import FastAPI import httpx client = None @asynccontextmanager async def lifespan(app: FastAPI): global client client = httpx.AsyncClient(timeout=60.0) yield await client.aclose() app = FastAPI(lifespan=lifespan)6.2 超时与重试机制
大模型响应不可预测,必须添加合理的超时和重试:
from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def safe_chat_request(prompt: str): try: async with client.stream( "POST", "http://localhost:11434/api/chat", json={"model": "llama3", "messages": [{"role": "user", "content": prompt}]}, timeout=30.0 ) as response: async for chunk in response.aiter_bytes(): yield chunk except httpx.ReadTimeout: yield b"模型响应超时,请重试或简化问题"7. 部署与扩展考量
7.1 生产环境配置
实际部署时需要关注:
- ollama服务的启动参数
- FastAPI的worker数量配置
- 反向代理的超时设置
- 日志和监控集成
7.2 水平扩展方案
当单机无法满足需求时,可以考虑:
- ollama多实例负载均衡
- 模型并行计算
- 请求队列管理
- 结果缓存策略
8. 完整代码示例
8.1 后端完整实现
from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from contextlib import asynccontextmanager import httpx from tenacity import retry, stop_after_attempt, wait_exponential import logging client = None @asynccontextmanager async def lifespan(app: FastAPI): global client client = httpx.AsyncClient(timeout=60.0) yield await client.aclose() app = FastAPI(lifespan=lifespan) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def generate_stream(prompt: str): try: async with client.stream( "POST", "http://localhost:11434/api/chat", json={ "model": "llama3", "messages": [{"role": "user", "content": prompt}], "stream": True }, timeout=30.0 ) as response: async for chunk in response.aiter_bytes(): yield chunk except Exception as e: logging.error(f"请求失败: {str(e)}") yield b"服务暂时不可用,请稍后重试" @app.post("/api/chat") async def chat_endpoint(request: Request): data = await request.json() return StreamingResponse( generate_stream(data["prompt"]), media_type="application/octet-stream" )8.2 前端优化版本
<!DOCTYPE html> <html> <head> <title>Ollama Chat</title> <style> #chat-container { max-width: 800px; margin: 0 auto; } #chat-history { height: 500px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; margin-bottom: 10px; white-space: pre-wrap; } #user-input { width: 80%; padding: 8px; } button { padding: 8px 15px; } .typing { color: #666; font-style: italic; } </style> </head> <body> <div id="chat-container"> <div id="chat-history"></div> <input type="text" id="user-input" placeholder="输入你的问题..."> <button id="send-button" onclick="sendMessage()">发送</button> </div> <script> let isProcessing = false; const chatHistory = document.getElementById('chat-history'); const userInput = document.getElementById('user-input'); const sendButton = document.getElementById('send-button'); function updateUIState() { userInput.disabled = isProcessing; sendButton.disabled = isProcessing; sendButton.textContent = isProcessing ? '处理中...' : '发送'; } function appendMessage(role, content) { const messageDiv = document.createElement('div'); messageDiv.innerHTML = `<strong>${role}:</strong> ${content}`; chatHistory.appendChild(messageDiv); chatHistory.scrollTop = chatHistory.scrollHeight; } async function sendMessage() { if (isProcessing || !userInput.value.trim()) return; isProcessing = true; updateUIState(); const prompt = userInput.value; userInput.value = ''; appendMessage('你', prompt); const typingIndicator = document.createElement('div'); typingIndicator.className = 'typing'; typingIndicator.textContent = '模型正在思考...'; chatHistory.appendChild(typingIndicator); chatHistory.scrollTop = chatHistory.scrollHeight; try { const response = await fetch('/api/chat', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({prompt: prompt}) }); chatHistory.removeChild(typingIndicator); const messageDiv = document.createElement('div'); messageDiv.innerHTML = '<strong>AI:</strong> '; chatHistory.appendChild(messageDiv); const reader = response.body.getReader(); while(true) { const {done, value} = await reader.read(); if(done) break; const text = new TextDecoder().decode(value); messageDiv.innerHTML += text; chatHistory.scrollTop = chatHistory.scrollHeight; } } catch (error) { chatHistory.removeChild(typingIndicator); appendMessage('系统', `请求失败: ${error.message}`); } finally { isProcessing = false; updateUIState(); } } userInput.addEventListener('keypress', (e) => { if (e.key === 'Enter') sendMessage(); }); </script> </body> </html>9. 经验教训与最佳实践
9.1 异步编程的注意事项
- 避免阻塞操作:任何同步IO操作都会破坏事件循环
- 合理设置超时:特别是对于大模型这种响应时间不确定的服务
- 资源清理:确保所有异步资源都正确关闭
- 错误传播:异步栈中的错误处理需要特别小心
9.2 ollama集成技巧
- 模型预热:首次加载模型可能很慢,可以预先发送简单请求
- 内存管理:ollama默认会保留最近使用的模型,注意系统内存使用
- 版本控制:明确指定模型版本,避免自动更新导致的不兼容
- 本地缓存:对于常见问题,可以在应用层实现缓存机制
10. 性能监控与调优
10.1 关键指标监控
- 请求响应时间分布
- 错误率和重试次数
- 模型推理时间
- 系统资源使用率
10.2 实用调试技巧
# 在FastAPI中添加中间件记录请求时间 @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response11. 安全考量
- 输入验证:防止Prompt注入攻击
- 速率限制:防止滥用导致服务不可用
- 敏感信息过滤:模型输出可能包含训练数据中的敏感信息
- 认证授权:至少实现基本的API密钥验证
12. 扩展功能思路
- 对话历史持久化:使用Redis或数据库存储对话上下文
- 多模态支持:结合ollama的视觉模型能力
- 插件系统:允许动态加载不同的功能模块
- 性能分析仪表盘:实时监控模型使用情况
13. 国内环境特别优化
- 模型下载加速:配置国内镜像源
- 备用服务部署:考虑在多个区域部署ollama实例
- 离线包准备:预先下载模型并打包分发
- 网络连接优化:调整TCP参数适应国内网络环境
14. 常见问题解决方案
ollama服务无响应
- 检查服务是否正常运行:
ollama serve - 验证端口11434是否可访问
- 查看日志中的错误信息
- 检查服务是否正常运行:
流式响应中断
- 检查客户端是否过早关闭连接
- 增加网络超时设置
- 实现断点续传机制
模型加载失败
- 确保磁盘空间充足
- 验证模型文件完整性
- 尝试重新拉取模型
内存不足错误
- 限制并发请求数量
- 使用较小规模的模型
- 增加交换空间或物理内存
15. 进阶话题探索
- 自定义模型集成:如何加载自己训练的模型
- 性能基准测试:不同硬件配置下的表现对比
- 混合部署方案:结合云端和本地模型
- 自动扩展策略:基于负载的动态资源分配
这个项目从最初的简单设想到最终稳定可用的服务,经历了多次迭代和优化。最大的收获是深入理解了Python异步编程在实际项目中的应用要点,以及如何平衡用户体验与系统性能。对于想要尝试类似项目的开发者,我的建议是从最简单的版本开始,逐步添加功能,并在每个阶段进行充分的测试和性能评估。