30分钟上手：用ollama-python+FastAPI构建高性能AI后端服务

你是否还在为AI模型集成到后端服务而烦恼？本文将带你使用ollama-python和FastAPI，快速搭建一个支持流式响应的AI后端服务，无需复杂配置，30分钟即可上线。## 准备工作### 环境要求- Python 3.8+- Ollama服务已安装并运行：[Ollama下载](https://ollama.com/download)- 安装依赖包：```bashpip in...

陶真蔷Scott

922人浏览 · 2025-09-07 13:04:53

陶真蔷Scott · 2025-09-07 13:04:53 发布

30分钟上手：用ollama-python+FastAPI构建高性能AI后端服务

【免费下载链接】ollama-python 项目地址: https://gitcode.com/GitHub_Trending/ol/ollama-python

你是否还在为AI模型集成到后端服务而烦恼？本文将带你使用ollama-python和FastAPI，快速搭建一个支持流式响应的AI后端服务，无需复杂配置，30分钟即可上线。

准备工作

环境要求

Python 3.8+
Ollama服务已安装并运行：Ollama下载
安装依赖包：

pip install ollama fastapi uvicorn

项目结构

GitHub_Trending/ol/ollama-python/
├── main.py              # FastAPI服务主文件
├── requirements.txt     # 项目依赖
├── examples/            # 官方示例代码
│   ├── async-chat.py    # 异步聊天示例
│   └── chat-stream.py   # 流式响应示例
└── ollama/              # 核心库代码
    ├── _client.py       # 客户端实现
    └── _types.py        # 类型定义

快速开始：构建基础AI服务

1. 创建FastAPI应用

新建main.py文件，实现一个简单的AI对话接口：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from ollama import AsyncClient, ChatResponse

app = FastAPI(title="Ollama AI服务")
client = AsyncClient()  # 使用异步客户端提高性能

class ChatRequest(BaseModel):
    model: str = "gemma3"  # 默认模型
    message: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat(
            model=request.model,
            messages=[{"role": "user", "content": request.message}]
        )
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

2. 运行服务

uvicorn main:app --reload

访问 http://localhost:8000/docs 即可测试API。

高级功能：流式响应实现

为提升用户体验，我们来实现流式响应功能，让AI回答像打字一样实时显示。

1. 添加流式接口

修改main.py，添加流式响应端点：

from fastapi.responses import StreamingResponse
import asyncio

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        async for part in client.chat(
            model=request.model,
            messages=[{"role": "user", "content": request.message}],
            stream=True  # 启用流式响应
        ):
            content = part["message"]["content"]
            if content:
                yield f"data: {content}\n\n"
            await asyncio.sleep(0.01)  # 控制流速度
        
        yield "data: [DONE]\n\n"  # 结束标记

    return StreamingResponse(event_generator(), media_type="text/event-stream")

2. 测试流式响应

使用curl测试：

curl -X POST "http://localhost:8000/chat/stream" \
  -H "Content-Type: application/json" \
  -d '{"model":"gemma3","message":"为什么天空是蓝色的？"}'

你将看到AI响应内容被逐段返回。

生产环境优化

1. 连接池配置

优化 Ollama 客户端连接池，提高并发处理能力：

from httpx import AsyncHTTPTransport

transport = AsyncHTTPTransport(
    limits=httpx.Limits(max_connections=100),
    timeout=30.0
)
client = AsyncClient(transport=transport)

2. 模型管理接口

添加模型管理接口，方便监控和管理模型：

@app.get("/models")
async def list_models():
    return await client.list()

@app.get("/models/{model}")
async def show_model(model: str):
    return await client.show(model)

完整代码示例

异步聊天示例

官方异步聊天实现：examples/async-chat.py

import asyncio
from ollama import AsyncClient

async def main():
    messages = [{"role": "user", "content": "Why is the sky blue?"}]
    response = await AsyncClient().chat('gemma3', messages=messages)
    print(response['message']['content'])

asyncio.run(main())

流式聊天示例

官方流式响应实现：examples/chat-stream.py

from ollama import chat

messages = [{"role": "user", "content": "Why is the sky blue?"}]
for part in chat('gemma3', messages=messages, stream=True):
    print(part['message']['content'], end='', flush=True)