FastAPI构建AI后端服务

# FastAPI构建AI后端服务

## 为什么选择FastAPI？

FastAPI是现代化的Python Web框架，以性能高、开发快著称。它内置支持异步、自动生成API文档，特别适合构建AI服务。

## 安装

```bash
pip install fastapi uvicorn
```

## 基础示例

```python
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Query(BaseModel):
    text: str
    max_tokens: int = 100

@app.post("/generate")
async def generate(query: Query):
    # 调用AI模型
    result = call_llm(query.text, query.max_tokens)
    return {"response": result}

# 运行: uvicorn main:app --reload
```

## 流式响应

```python
from fastapi.responses import StreamingResponse
import asyncio

async def generate_stream(prompt: str):
    for chunk in call_llm_stream(prompt):
        yield f"data: {chunk}

"
        await asyncio.sleep(0.01)

@app.post("/stream")
async def stream_generate(query: Query):
    return StreamingResponse(
        generate_stream(query.text),
        media_type="text/event-stream"
    )
```

## 完整的AI服务示例

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
from typing import List, Optional

app = FastAPI(title="AI服务API")
client = OpenAI()

class ChatMessage(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    model: str = "gpt-3.5-turbo"
    temperature: Optional[float] = 0.7

class ChatResponse(BaseModel):
    response: str
    tokens_used: int

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = client.chat.completions.create(
            model=request.model,
            messages=[m.dict() for m in request.messages],
            temperature=request.temperature
        )
        return ChatResponse(
            response=response.choices[0].message.content,
            tokens_used=response.usage.total_tokens
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "healthy"}
```

## 文件上传与处理

```python
from fastapi import File, UploadFile

@app.post("/analyze-document")
async def analyze_document(file: UploadFile = File(...)):
    content = await file.read()
    # 处理文档
    result = analyze_with_ai(content)
    return {"analysis": result}
```

## 认证与鉴权

```python
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if credentials.credentials != "valid_token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return credentials.credentials

@app.post("/protected")
async def protected_endpoint(token: str = Depends(verify_token)):
    return {"message": "访问成功"}
```

## 负载均衡与限流

```python
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/limited")
@limiter.limit("10/minute")
async def limited_endpoint(request: Request):
    return {"message": "success"}
```

## Docker部署

```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```

## 性能优化

1. **使用异步**: 充分利用async/await
2. **连接池**: 复用数据库和API连接
3. **缓存**: 缓存频繁请求的结果
4. **模型预加载**: 启动时加载模型

## 总结

FastAPI是构建AI服务的理想选择，它的异步支持和自动文档功能大大提高了开发效率。
评论 (0)
