5分钟落地：Ollama-Python与FastAPI深度整合，构建智能问答系统

痛点引入

在企业级AI应用开发中，开发者常面临三大核心挑战：响应延迟导致用户体验下降、数据隐私泄露风险、云服务成本持续攀升。传统云API方案平均响应时间超过800ms，且每万次调用成本高达20美元。更棘手的是，金融、医疗等敏感领域的数据合规要求使得云端处理方案面临合规性风险。如何在本地环境实现高效、安全且低成本的AI能力集成，成为企业数字化转型的关键瓶颈。

技术选型对比

方案	响应速度	数据安全性	长期成本	部署复杂度	离线可用性
云API服务	500-1500ms	低（数据出境）	高（按调用计费）	低	不支持
自建GPU集群	100-300ms	高	极高（硬件+维护）	极高	支持
Ollama本地部署	50-150ms	极高（数据本地化）	中（一次性硬件投入）	低	完全支持
边缘计算节点	200-500ms	中（需边缘节点管理）	中高	中	部分支持

注意事项：Ollama方案特别适合中小团队，在普通服务器（8核16G配置）即可运行7B参数模型，平衡性能与成本。

环境搭建

1. 部署Ollama服务

# 安装Ollama（Linux/Unix系统）
curl -fsSL https://ollama.com/install.sh | sh

# 启动服务并拉取基础模型（约3.8GB）
ollama serve &
ollama pull mistral:7b-instruct

2. 创建FastAPI项目

# 克隆项目仓库
git clone https://gitcode.com/GitHub_Trending/ol/ollama-python
cd ollama-python/examples

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows使用: venv\Scripts\activate

# 安装依赖
pip install fastapi uvicorn pydantic ollama python-multipart

注意事项：建议使用Python 3.9+版本，低版本可能导致类型注解兼容性问题。

核心模块实现

1. 封装Ollama客户端服务

创建ollama_service.py实现服务封装：

from ollama import Client, AsyncClient
from pydantic import BaseModel
from typing import List, Optional, Dict

class Message(BaseModel):
    role: str  # "user" 或 "assistant"
    content: str

class OllamaClient:
    def __init__(self, host: str = "http://localhost:11434"):
        self.host = host
        self.client = Client(host=host)
        
    def get_available_models(self) -> List[str]:
        """获取所有可用模型列表"""
        models = self.client.list()
        return [model["name"] for model in models["models"]]
        
    def chat_completion(
        self, 
        model: str, 
        messages: List[Message],
        temperature: float = 0.7,
        max_tokens: Optional[int] = None
    ) -> str:
        """同步聊天接口"""
        try:
            response = self.client.chat(
                model=model,
                messages=[msg.dict() for msg in messages],
                options={"temperature": temperature, "max_tokens": max_tokens}
            )
            return response["message"]["content"]
        except Exception as e:
            return f"服务错误: {str(e)}"

# 异步版本实现
class AsyncOllamaClient(OllamaClient):
    async def chat_completion_async(self, model: str, messages: List[Message], **kwargs) -> str:
        async with AsyncClient(host=self.host) as client:
            response = await client.chat(
                model=model,
                messages=[msg.dict() for msg in messages],** kwargs
            )
            return response["message"]["content"]

2. 实现FastAPI接口

创建main.py定义API端点：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from ollama_service import OllamaClient, AsyncOllamaClient, Message

app = FastAPI(title="智能问答API服务")
ollama_sync = OllamaClient()
ollama_async = AsyncOllamaClient()

class ChatRequest(BaseModel):
    model: str = "mistral:7b-instruct"
    messages: List[Message]
    temperature: float = 0.7
    max_tokens: Optional[int] = None

@app.get("/models", summary="获取可用模型列表")
def list_models():
    return {"models": ollama_sync.get_available_models()}

@app.post("/chat", summary="同步聊天接口")
def chat(request: ChatRequest):
    try:
        response = ollama_sync.chat_completion(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens
        )
        return {"response": response}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"服务错误: {str(e)}")

@app.post("/chat/async", summary="异步聊天接口")
async def chat_async(request: ChatRequest):
    try:
        response = await ollama_async.chat_completion_async(
            model=request.model,
            messages=request.messages,** request.dict(exclude={"model", "messages"})
        )
        return {"response": response}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"服务错误: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

注意事项：生产环境需添加API密钥认证和请求限流，可使用fastapi.security和slowapi实现。

功能验证

1. 启动服务

# 启动FastAPI服务
python main.py

# 另一个终端验证模型列表
curl http://localhost:8000/models

2. 测试聊天接口

使用curl发送测试请求：

curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "mistral:7b-instruct",
    "messages": [{"role": "user", "content": "请解释什么是RESTful API"}],
    "temperature": 0.5
  }'

预期响应：

{
  "response": "RESTful API是一种软件架构风格，基于HTTP协议设计，通过标准的HTTP方法（GET、POST、PUT、DELETE等）对资源进行操作..."
}

3. 性能测试

使用Apache Bench进行并发测试：

ab -n 100 -c 10 "http://localhost:8000/chat -d '{\"messages\":[{\"role\":\"user\",\"content\":\"hello\"}]}'"

注意事项：首次请求会有模型加载延迟（约3-5秒），后续请求响应时间应稳定在100-300ms。

扩展方案

1. 多模型负载均衡

实现思路：创建模型池管理类，根据模型负载自动分配请求，代码示例：

class ModelLoadBalancer:
    def __init__(self, models: List[str]):
        self.models = models
        self.load_counter = {model: 0 for model in models}
        
    def get_next_model(self) -> str:
        """选择当前负载最低的模型"""
        return min(self.load_counter, key=self.load_counter.get)
        
    def update_load(self, model: str, increment: int = 1):
        self.load_counter[model] += increment
        # 定期重置计数器（可结合定时任务实现）

2. 对话历史持久化

实现思路：使用SQLite或Redis存储对话历史，添加用户认证中间件：

# 安装依赖
pip install sqlalchemy python-jose[cryptography]

# 核心实现
from sqlalchemy import create_engine, Column, String, Text, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from pydantic import BaseModel

class Conversation(BaseModel):
    user_id: str
    messages: List[Message] = []
    created_at: datetime = datetime.now()

3. 流式响应优化

实现思路：使用FastAPI的StreamingResponse实现打字机效果：

from fastapi.responses import StreamingResponse
import asyncio

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for chunk in ollama_async.stream_chat_completion(
            model=request.model, messages=request.messages
        ):
            yield f"data: {chunk}\n\n"
            await asyncio.sleep(0.05)  # 控制输出速度
    return StreamingResponse(generate(), media_type="text/event-stream")

资源导航

• 核心客户端源码：ollama/_client.py - 包含同步/异步客户端实现
• 示例代码库：examples/ - 包含20+种使用场景的完整示例
• 类型定义文档：ollama/_types.py - 所有API请求/响应类型定义
• 工具函数集：ollama/_utils.py - 提供模型管理、数据转换等实用工具
• 单元测试：tests/ - 包含客户端功能验证和类型检查测试用例

通过以上资源，开发者可以快速扩展功能，实现从简单问答到企业级智能应用的全流程落地。