三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

FastAPI与ollama大模型异步流式响应实战

发布时间：2026/8/6 6:34:01 三亩地 编程学习日记
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.6

3. 基础实现与首次翻车

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"]

这个版本的问题立即显现：

  1. 每个请求都会阻塞事件循环
  2. 无法处理并发请求
  3. 响应时间不可控

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 chunk

5. 前端对话窗口实现

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 优化用户体验

实际使用中发现几个需要改进的点：

  1. 添加消息发送禁用状态
  2. 实现打字机效果
  3. 处理网络中断情况
  4. 添加消息历史持久化

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 生产环境配置

实际部署时需要关注：

  1. ollama服务的启动参数
  2. FastAPI的worker数量配置
  3. 反向代理的超时设置
  4. 日志和监控集成

7.2 水平扩展方案

当单机无法满足需求时，可以考虑：

  1. ollama多实例负载均衡
  2. 模型并行计算
  3. 请求队列管理
  4. 结果缓存策略

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 异步编程的注意事项

  1. 避免阻塞操作：任何同步IO操作都会破坏事件循环
  2. 合理设置超时：特别是对于大模型这种响应时间不确定的服务
  3. 资源清理：确保所有异步资源都正确关闭
  4. 错误传播：异步栈中的错误处理需要特别小心

9.2 ollama集成技巧

  1. 模型预热：首次加载模型可能很慢，可以预先发送简单请求
  2. 内存管理：ollama默认会保留最近使用的模型，注意系统内存使用
  3. 版本控制：明确指定模型版本，避免自动更新导致的不兼容
  4. 本地缓存：对于常见问题，可以在应用层实现缓存机制

10. 性能监控与调优

10.1 关键指标监控

  1. 请求响应时间分布
  2. 错误率和重试次数
  3. 模型推理时间
  4. 系统资源使用率

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 response

11. 安全考量

  1. 输入验证：防止Prompt注入攻击
  2. 速率限制：防止滥用导致服务不可用
  3. 敏感信息过滤：模型输出可能包含训练数据中的敏感信息
  4. 认证授权：至少实现基本的API密钥验证

12. 扩展功能思路

  1. 对话历史持久化：使用Redis或数据库存储对话上下文
  2. 多模态支持：结合ollama的视觉模型能力
  3. 插件系统：允许动态加载不同的功能模块
  4. 性能分析仪表盘：实时监控模型使用情况

13. 国内环境特别优化

  1. 模型下载加速：配置国内镜像源
  2. 备用服务部署：考虑在多个区域部署ollama实例
  3. 离线包准备：预先下载模型并打包分发
  4. 网络连接优化：调整TCP参数适应国内网络环境

14. 常见问题解决方案

  1. ollama服务无响应

    • 检查服务是否正常运行：ollama serve
    • 验证端口11434是否可访问
    • 查看日志中的错误信息

  2. 流式响应中断

    • 检查客户端是否过早关闭连接
    • 增加网络超时设置
    • 实现断点续传机制

  3. 模型加载失败

    • 确保磁盘空间充足
    • 验证模型文件完整性
    • 尝试重新拉取模型

  4. 内存不足错误

    • 限制并发请求数量
    • 使用较小规模的模型
    • 增加交换空间或物理内存

15. 进阶话题探索

  1. 自定义模型集成：如何加载自己训练的模型
  2. 性能基准测试：不同硬件配置下的表现对比
  3. 混合部署方案：结合云端和本地模型
  4. 自动扩展策略：基于负载的动态资源分配

这个项目从最初的简单设想到最终稳定可用的服务，经历了多次迭代和优化。最大的收获是深入理解了Python异步编程在实际项目中的应用要点，以及如何平衡用户体验与系统性能。对于想要尝试类似项目的开发者，我的建议是从最简单的版本开始，逐步添加功能，并在每个阶段进行充分的测试和性能评估。

编程学习 学习日记 实战经验
← 返回列表