如何使用FastAPI
我们现在想将agent分享给朋友或者更多用户在线体验，需要解决2个问题：
- 接受朋友的数据（ta想对agent说什么话，问什么问题）
- 将数据返回给朋友的电脑（agent的回答）
为了解决这2个问题，我们需要实现一个后端，可以使用FastAPI来构建一个API接口，朋友可以通过这个接口向我们的agent发送消息，并接收agent的回答。
欢迎来到API开发的现代化利器——FastAPI！🚀💻
FastAPI是什么？
FastAPI是一个现代、快速（高性能）的Web框架，用于基于Python 3.6+构建API，它基于标准的Python类型提示，让你能够轻松构建出既快速又易用的API。就像它的名字一样，FastAPI不仅开发速度快，执行速度也快！
如果把传统Web框架比作老式轿车，那么FastAPI就是一辆配备了自动驾驶和智能辅助系统的跑车！🏎️
为什么在Agent开发中使用FastAPI？
当你开发AI Agent时，通常需要为其提供API接口以便与其他系统交互。FastAPI在这里闪耀光芒的原因有：
- 性能极佳：比肩NodeJS和Go的速度，是最快的Python框架之一
- 简单直观：编写少量代码即可创建功能丰富的API
- 自动生成文档：无需额外工作即可获得交互式API文档
- 基于标准：完全兼容OpenAPI和JSON Schema
- 类型安全：利用Python类型提示提供编辑器支持和自动验证
- 易于学习：如果你了解Python，很快就能上手
安装与设置
FastAPI基础知识
创建第一个API
我们先实现返还数据给朋友的功能，只需要创建一个名为main.py的文件，内容如下：
from fastapi import FastAPI
app = FastAPI()
@app.get("/return_json")
def read_root1():
return {"ai reply": "Hello, World!"}
@app.get("/return_text")
def read_root2():
return "Hello, World!"
@app.get("/return_html")
def read_root3():
from fastapi.responses import HTMLResponse
return HTMLResponse(content="<h1>Hello, World!</h1>")
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
然后在命令行运行：
现在，访问 http://127.0.0.1:8000/return_json 将看到 {"ai reply": "Hello, World!"} 的JSON响应。
访问 http://127.0.0.1:8000/docs 将看到自动生成的交互式API文档！这是FastAPI的一大亮点，无需额外工作即可获得专业的API文档🎉。也可以在交互式文档中直接发送请求，查看响应。
路径操作
有了以上基础，我们可以开始实现接受朋友的数据（ta想对agent说什么话，问什么问题），并返回agent的回答。
将数据传给后端分为大差不差的4种请求方式，分别是get、post、put、delete。但是最常用的就是get和post请求（其实在现代项目中，很多项目只用post请求一个）。
在FastAPI中，函数上方的装饰器如@app.get("/")被称为"路径操作装饰器"，它告诉FastAPI这个函数应该处理哪个路径和HTTP方法的请求：
@app.get("/") # GET请求，路径为"/"
@app.post("/items/") # POST请求，路径为"/items/"
@app.put("/items/1") # PUT请求，路径为"/items/1"
@app.delete("/items/1") # DELETE请求，路径为"/items/1"
每种请求方式的差异
- GET请求：用于获取资源，通常用于查询数据
- POST请求：用于创建资源，通常用于提交数据
- PUT请求：用于更新资源，通常用于修改数据
- DELETE请求：用于删除资源，通常用于删除数据
但是post请求就可以实现以上所有功能，要做什么操作（查询、创建、修改、删除），只需要在post请求中定义好一个字段就行。
传递参数
你可以在路径中定义参数，并在函数中接收它们：
完整代码
参数user_id将被自动解析为整数，如果无法解析（例如用户访问/users/abc），FastAPI会自动返回适当的错误响应。
我们可以使用交互式文档或者浏览器访问http://127.0.0.1:8000/users/123，将得到{"user_id": 246}的响应。
为什么可以通过浏览器可以直接访问我们写的后端
因为浏览器本身就是设计用来访问web服务器的，我们见到的网页，就是浏览器访问web服务器（同样是后端）返回的html文件。区别在于，我们写的后端是返回json数据，而网页是返回html文件，浏览器解析出来不一样。我们写的fastapi后端也一样可以返回html格式的文件（在上面的例子中，我们返回了
Hello, World!
）。
不在路径中的函数参数会被自动解释为查询参数：
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
完整代码
访问http://127.0.0.1:8000/items/?skip=20&limit=30将得到{"skip": 20, "limit": 30}的响应。
使用Pydantic模型定义请求体：
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
app = FastAPI()
@app.post("/items/")
def create_item(item: Item):
return item
完整代码
使用post请求需要实现定义好每个字段的类型，FastAPI将自动： - 读取请求体为JSON - 转换为对应的类型 - 验证数据 - 为你的编辑器提供补全和类型检查 - 在API文档中生成模型的文档
由于浏览器使用的get请求，所以无法使用post请求，需要使用curl、requests、httpx等工具来发送post请求。
(base) ➜ bi_tools git:(master) curl -X 'POST' \
'http://172.20.90.202:8003/items/' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"name": "test",
"description": "string",
"price": 10,
"tax": 1
}'
# {"name":"test","description":"string","price":10.0,"tax":1.0}%
🌟 小案例：简单TODO API
让我们创建一个简单的待办事项API：
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uuid
app = FastAPI(title="TODO API")
# 数据模型
class TodoCreate(BaseModel):
title: str
description: Optional[str] = None
completed: bool = False
class Todo(TodoCreate):
id: str
# 内存数据库
todos = {}
@app.post("/todos/", response_model=Todo)
def create_todo(todo: TodoCreate):
todo_id = str(uuid.uuid4())
todos[todo_id] = Todo(id=todo_id, **todo.dict())
return todos[todo_id]
@app.get("/todos/", response_model=List[Todo])
def read_todos(skip: int = 0, limit: int = 10):
return list(todos.values())[skip : skip + limit]
@app.get("/todos/{todo_id}", response_model=Todo)
def read_todo(todo_id: str):
if todo_id not in todos:
raise HTTPException(status_code=404, detail="Todo not found")
return todos[todo_id]
@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: str, todo: TodoCreate):
if todo_id not in todos:
raise HTTPException(status_code=404, detail="Todo not found")
todos[todo_id] = Todo(id=todo_id, **todo.dict())
return todos[todo_id]
@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: str):
if todo_id not in todos:
raise HTTPException(status_code=404, detail="Todo not found")
del todos[todo_id]
return {"message": "Todo deleted successfully"}
高级功能
1. 依赖注入系统
FastAPI提供了一个强大的依赖注入系统，可以用于： - 共享数据库连接 - 强制执行安全性、身份验证和授权 - 提取和验证数据
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: str = Header(...)):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
return x_token
@app.get("/items/", dependencies=[Depends(verify_token)])
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
2. 后台任务
需要执行不影响响应的长时间运行任务？使用后台任务：
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
3. WebSockets支持
FastAPI原生支持WebSockets，适用于实时通信需求：
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>聊天</title>
</head>
<body>
<h1>WebSocket 聊天</h1>
<input type="text" id="messageText" autocomplete="off"/>
<button onclick="sendMessage()">发送</button>
<ul id="messages"></ul>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages');
var message = document.createElement('li');
var content = document.createTextNode(event.data);
message.appendChild(content);
messages.appendChild(message);
};
function sendMessage() {
var input = document.getElementById("messageText");
ws.send(input.value);
input.value = '';
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"消息: {data}")
实际案例：AI Agent API
让我们创建一个更实际的例子，一个为AI Agent提供API接口的服务：
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import uuid
import time
from datetime import datetime
# 你的AI Agent模块导入（假设）
# from my_agent import AgentExecutor, Tool
app = FastAPI(
title="AI Agent API",
description="提供AI Agent功能的RESTful API",
version="1.0.0"
)
# 启用CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 在生产环境中应该限制源
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 数据模型
class AgentRequest(BaseModel):
query: str
context: Optional[Dict[str, Any]] = {}
tools: List[str] = []
verbose: bool = False
class AgentAction(BaseModel):
tool: str
tool_input: str
log: str
class AgentResponse(BaseModel):
query: str
answer: str
actions: List[AgentAction] = []
conversation_id: str
processing_time: float
timestamp: datetime
# 内存存储
conversations = {}
# 中间件记录请求时间
@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
# 依赖项：获取对话历史
def get_conversation(conversation_id: str = None):
if conversation_id and conversation_id in conversations:
return conversations[conversation_id]
new_id = str(uuid.uuid4())
conversations[new_id] = {
"id": new_id,
"history": [],
"created_at": datetime.now()
}
return conversations[new_id]
@app.post("/agent/query", response_model=AgentResponse)
async def query_agent(
request: AgentRequest,
conversation: dict = Depends(get_conversation)
):
start_time = time.time()
conversation_id = conversation["id"]
# 这里是与Agent交互的模拟逻辑
# 实际应用中，你会调用真实的Agent
actions = []
if "weather" in request.tools and "天气" in request.query:
actions.append(AgentAction(
tool="weather_api",
tool_input="北京",
log="检测到天气查询意图，调用天气API获取北京天气"
))
# 模拟Agent响应
answer = f"这是对'{request.query}'的回答。在实际应用中，这将由真实的AI Agent生成。"
# 记录到对话历史
response = AgentResponse(
query=request.query,
answer=answer,
actions=actions,
conversation_id=conversation_id,
processing_time=time.time() - start_time,
timestamp=datetime.now()
)
conversation["history"].append({
"query": request.query,
"response": answer,
"timestamp": datetime.now()
})
return response
@app.get("/conversations/{conversation_id}")
async def get_conversation_history(conversation_id: str):
if conversation_id not in conversations:
raise HTTPException(status_code=404, detail="Conversation not found")
return conversations[conversation_id]
@app.delete("/conversations/{conversation_id}")
async def delete_conversation(conversation_id: str):
if conversation_id not in conversations:
raise HTTPException(status_code=404, detail="Conversation not found")
del conversations[conversation_id]
return {"message": "Conversation deleted successfully"}
部署FastAPI应用
开发环境
在开发中运行FastAPI应用：
生产环境
在生产环境中，推荐使用Gunicorn作为ASGI服务器的管理器：
这会使用4个worker进程，适合多核CPU。
使用Docker部署
创建一个Dockerfile：
FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]
创建一个docker-compose.yml文件：
然后运行：
最佳实践
- 路径操作函数要短小精悍：业务逻辑应放在单独的函数或类中
- 使用Pydantic模型：为请求和响应定义清晰的数据模型
- 利用依赖注入：重用代码并保持路径操作函数简洁
- 异步在I/O密集型操作中使用：利用
async/await提高性能 - 分层架构：控制器（路径操作）-> 服务 -> 仓储，保持关注点分离
- 合理使用状态码：返回准确的HTTP状态码和信息
- 全面的错误处理：捕获异常并返回友好的错误信息
- 使用中间件：跨请求处理如日志记录、认证和性能监控
下一步
现在你已经了解了FastAPI的基础和高级功能，接下来可以：
- 学习更多关于Pydantic的高级验证功能
- 探索FastAPI的安全功能（OAuth2、JWT等）
- 集成数据库（使用SQLAlchemy、Tortoise-ORM等）
- 构建更复杂的多服务架构
- 为你的Agent添加API网关和限流功能
- 学习如何编写测试（FastAPI提供了测试客户端）
通过FastAPI，你可以为你的AI Agent构建一个高性能、易用且文档完善的API接口，让它能够无缝集成到各种应用场景中！🚀