初探：用 FastAPI 搭建你的第一个 AI Agent 接口

什么是 AI Agent 接口？想象一下，你有一个智能助手，它能理解你的问题，调用外部工具（比如搜索引擎、计算器），然后给出精准回答。这个智能助手的“大脑”就是 AI Agent。而 FastAPI，就像是一个“门卫”，它帮我们快速搭建一个接口，让用户可以通过网络访问这个 Agent。在本文中，我们将从零开始，用 FastAPI 搭建一个简单的 AI Agent 接口。这个 Agent 能接收用户问题，模拟调用计算工具，并返回结果。我们将循序渐进，从基础概念讲到实际代码。## 为什么选择 FastAPI？FastAPI 是一个现代、高性能的 Python Web 框架，专为构建 API 而设计。它的优势包括：- 速度极快：基于 Starlette 和 Pydantic，性能堪比 Node.js 和 Go。- 自动文档生成：自动生成 Swagger 和 ReDoc 文档。- 类型安全：使用 Python 类型提示，减少 bug。对于 AI Agent 接口，FastAPI 能让我们快速搭建一个稳定、可扩展的服务。## 第一步：安装 FastAPI 和 Uvicorn首先，我们需要安装 FastAPI 和 ASGI 服务器 Uvicorn。打开终端，运行以下命令：bashpip install fastapi uvicorn这将会安装 FastAPI 及其依赖。Uvicorn 是运行 FastAPI 应用的服务器。## 第二步：创建你的第一个 FastAPI 应用让我们从最基础的代码开始：一个简单的“Hello, AI Agent!”接口。创建一个文件 main.py，输入以下代码：python# 导入 FastAPI 模块from fastapi import FastAPI# 创建 FastAPI 应用实例app = FastAPI()# 定义根路径的 GET 请求@app.get("/")async def root(): """返回欢迎信息""" return {"message": "Hello, AI Agent!"}代码解释：- FastAPI() 创建一个应用对象。- @app.get("/") 是一个装饰器，表示当用户访问根路径时，执行下面的函数。- async def root() 定义了一个异步函数，返回一个字典，FastAPI 会自动转换为 JSON。运行这个应用：bashuvicorn main:app --reload打开浏览器，访问 http://127.0.0.1:8000，你会看到 {"message":"Hello, AI Agent!"}。访问 http://127.0.0.1:8000/docs，你会看到自动生成的交互式 API 文档。## 第三步：设计 AI Agent 的核心逻辑一个真正的 AI Agent 需要理解问题、调用工具、生成回答。现在，我们来模拟一个简单的 Agent，它能根据用户输入，调用一个“计算工具”来求和。我们创建一个新的文件 agent.py，包含 Agent 的核心逻辑：python# agent.pyfrom typing import Dict, Anyclass SimpleAgent: """ 一个简单的 AI Agent，模拟调用计算工具。 """ def __init__(self): self.tools = { "add": self.add_tool, "multiply": self.multiply_tool } def add_tool(self, a: float, b: float) -> float: """模拟加法工具""" return a + b def multiply_tool(self, a: float, b: float) -> float: """模拟乘法工具""" return a * b def process_query(self, query: str) -> Dict[str, Any]: """ 处理用户查询，识别工具和参数，并返回结果。 """ # 简单解析：假设用户输入格式为 "add 3 5" 或 "multiply 4 6" parts = query.lower().split() if len(parts) != 3: return {"status": "error", "message": "请使用格式：工具名 数字1 数字2，例如 add 3 5"} tool_name, arg1_str, arg2_str = parts if tool_name not in self.tools: return {"status": "error", "message": f"未知工具：{tool_name}。可用工具：add, multiply"} try: arg1 = float(arg1_str) arg2 = float(arg2_str) except ValueError: return {"status": "error", "message": "参数必须是数字"} result = self.tools[tool_name](arg1, arg2) return { "status": "success", "tool_used": tool_name, "arguments": [arg1, arg2], "result": result }代码解释：- SimpleAgent 类封装了 Agent 的逻辑。- tools 字典存储了可用的工具函数。- process_query 方法解析用户输入，调用相应工具，并返回结果。## 第四步：将 AI Agent 集成到 FastAPI 中现在，我们把 Agent 连接到 FastAPI 接口。修改 main.py，加入一个新的端点：python# main.py（更新版）from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom agent import SimpleAgent# 创建 FastAPI 应用app = FastAPI()# 创建 Agent 实例（全局单例）agent = SimpleAgent()# 定义请求体模型class QueryRequest(BaseModel): """用户查询的请求体""" query: str# 定义响应模型class QueryResponse(BaseModel): """接口响应模型""" status: str tool_used: str = "" # 可选字段 arguments: list = [] # 可选字段 result: float = None # 可选字段 message: str = "" # 错误信息@app.get("/")async def root(): """根路径，返回欢迎信息""" return {"message": "欢迎使用 AI Agent 接口！"}@app.post("/agent/query", response_model=QueryResponse)async def agent_query(request: QueryRequest): """ 接收用户查询，由 AI Agent 处理并返回结果。 """ # 调用 Agent 处理查询 result = agent.process_query(request.query) if result["status"] == "error": # 如果出错，返回 400 错误 raise HTTPException(status_code=400, detail=result["message"]) # 返回成功响应 return QueryResponse( status="success", tool_used=result["tool_used"], arguments=result["arguments"], result=result["result"] )代码解释：- QueryRequest 和 QueryResponse 是 Pydantic 模型，用于验证请求和响应的数据结构。- @app.post("/agent/query") 定义了一个 POST 端点，接收 JSON 请求体。- response_model=QueryResponse 确保返回的数据符合模型规范。- 如果 Agent 返回错误，我们使用 HTTPException 返回 400 状态码。## 第五步：运行并测试接口重新启动 Uvicorn（如果已运行，它会自动重载）：bashuvicorn main:app --reload现在，我们可以测试接口了。使用 curl 或 Postman 发送请求：bashcurl -X POST http://127.0.0.1:8000/agent/query \ -H "Content-Type: application/json" \ -d '{"query": "add 3 5"}'你会得到类似这样的响应：json{ "status": "success", "tool_used": "add", "arguments": [3.0, 5.0], "result": 8.0}如果发送错误查询：bashcurl -X POST http://127.0.0.1:8000/agent/query \ -H "Content-Type: application/json" \ -d '{"query": "hello"}'你会得到 400 错误，并附带提示信息。## 第六步：添加更高级的功能（可选）现在我们已经有了一个基础接口，可以进一步扩展。例如，添加日志记录、错误处理、或集成真实的 AI 模型（如 OpenAI API）。以下是一个简单示例，添加一个“记忆”功能，让 Agent 记住历史对话：python# agent.py（增强版）from typing import Dict, Any, Listclass AdvancedAgent: def __init__(self): self.tools = {"add": self.add_tool, "multiply": self.multiply_tool} self.history: List[Dict[str, Any]] = [] def add_tool(self, a: float, b: float) -> float: return a + b def multiply_tool(self, a: float, b: float) -> float: return a * b def process_query(self, query: str) -> Dict[str, Any]: parts = query.lower().split() if len(parts) != 3: return {"status": "error", "message": "格式错误"} tool_name, arg1_str, arg2_str = parts if tool_name not in self.tools: return {"status": "error", "message": "未知工具"} try: arg1 = float(arg1_str) arg2 = float(arg2_str) except ValueError: return {"status": "error", "message": "参数必须是数字"} result = self.tools[tool_name](arg1, arg2) # 记录历史 self.history.append({ "query": query, "result": result, "tool": tool_name }) return { "status": "success", "tool_used": tool_name, "arguments": [arg1, arg2], "result": result, "history_count": len(self.history) }在 main.py 中，你可以添加一个端点查看历史：python@app.get("/agent/history")async def get_history(): """返回 Agent 的历史记录""" return {"history": agent.history}## 总结通过本文，我们从零开始搭建了一个基于 FastAPI 的 AI Agent 接口。我们学到了：1. FastAPI 基础：安装、创建应用、定义端点。2. AI Agent 设计：如何封装工具和逻辑。3. 数据验证：使用 Pydantic 模型确保接口安全。4. 错误处理：返回合适的 HTTP 状态码。5. 扩展性：添加历史记录等高级功能。这个简单的例子只是起点。在实际项目中，你可以将 Agent 连接到大型语言模型（如 GPT-4）、数据库、或外部 API，构建一个功能强大的智能助手。FastAPI 的异步特性和类型提示，会让你的开发过程更加高效和可靠。现在，你已经掌握了基础，快去搭建属于你自己的 AI Agent 接口吧！