Skip to content

stage6-fastapi-tutorial.md

Latest commit

 

History

History
223 lines (152 loc) · 8.12 KB

stage6-fastapi-tutorial.md

File metadata and controls

223 lines (152 loc) · 8.12 KB
Raw

第 6 阶段：FastAPI 教程

0. 学习目标

这一阶段的目标是：能用 Python 快速搭建 Agent 原型服务和 API 包装层。

学完后，你应该能：

  • 创建 app、定义 route、写 GET / POST。
  • 用路径参数、查询参数、请求体。
  • 用 Pydantic 定义请求和响应模型（复用第 4 阶段）。
  • 用依赖注入（Depends）和异常处理（exception_handler）。
  • 启动 uvicorn，并用自动生成的 OpenAPI 文档调试。
  • 给接口写不起真服务器的测试。

这一阶段把前几阶段串起来：Pydantic（第 4）做请求/响应模型，asyncio（第 5）做异步接口和批量并发。对你来说，它几乎就是"用 Python 写 Spring Boot Controller"。

1. Java/Spring 对照速查

Spring Boot FastAPI 说明
@RestController app = FastAPI() + 路由装饰器 应用/路由载体
@GetMapping / @PostMapping @app.get / @app.post 方法路由
@RequestBody DTO Pydantic 模型作函数参数 自动解析 + 校验
@PathVariable 路径参数 item_id: int 自动类型转换
@RequestParam 查询参数 q: str = Query(...) 可选/默认值
@Valid + Bean Validation Pydantic 校验（失败自动 422） 不用手写校验
@ControllerAdvice @app.exception_handler 全局异常转响应
构造注入 Bean Depends(...) 依赖注入
ResponseEntity return 模型 / JSONResponse 响应
Swagger / springdoc 内置 /docs、/openapi.json 零配置 OpenAPI
@SpringBootApplication 启动 uvicorn app:app ASGI 服务器

2. 最小可运行示例

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class ChatRequest(BaseModel):
    message: str


class ChatResponse(BaseModel):
    answer: str


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    return ChatResponse(answer=f"echo: {request.message}")

启动：

.\.venv\Scripts\python.exe -m uvicorn services.app:app --reload

打开 http://127.0.0.1:8000/docs 就能看到自动生成的交互式文档，直接点按钮调接口。

关键认知：把 Pydantic 模型作为函数参数，FastAPI 就自动从请求体解析 JSON、校验、转成对象；非法输入自动返回 422，连同详细错误位置——你不用写一行解析或校验代码。这就是 @RequestBody + @Valid 的合体。

3. 路径参数 / 查询参数

from fastapi import Query


@app.get("/echo/{item_id}")
def echo(item_id: int, q: str | None = Query(default=None, max_length=50)):
    return {"item_id": item_id, "q": q}
  • item_id: int —— 路径参数，自动转 int（传 abc 直接 422），≈ @PathVariable。
  • q: str | None = Query(...) —— 查询参数（/echo/7?q=hi），可选、带约束，≈ @RequestParam。

4. 任务 4：工具调用（路径参数选工具 + 异常处理）

from fastapi import HTTPException

TOOLS = {"weather": tool_weather, "search": tool_search}


@app.post("/tools/{tool_name}/invoke", response_model=ToolResponse)
def invoke_tool(tool_name: str, payload: ToolRequest) -> ToolResponse:
    tool = TOOLS.get(tool_name)
    if tool is None:
        raise HTTPException(status_code=404, detail=f"unknown tool: {tool_name}")
    return ToolResponse(tool=tool_name, result=tool(payload.arguments))

两种报错方式：

  • HTTPException(status_code, detail) —— FastAPI 标准报错，最常用（这里找不到工具返回 404）。
  • 自定义异常 + @app.exception_handler —— 把某类业务异常统一转成响应，≈ @ControllerAdvice：
class ToolError(Exception):
    def __init__(self, message): self.message = message

@app.exception_handler(ToolError)
async def handle_tool_error(request, exc):
    return JSONResponse(status_code=400, content={"error": exc.message})

这样工具内部 raise ToolError("...")，外层自动变成 400，无需每个 route 各写一遍 try/except。

5. 依赖注入（Depends）

from fastapi import Depends


def get_settings() -> dict:
    return {"model_name": "fake-llm", "max_concurrency": 10}


@app.post("/eval/run")
async def eval_run(request: EvalRequest, settings: dict = Depends(get_settings)):
    ...

Depends(get_settings) 让 FastAPI 在每次请求时调用 get_settings() 把结果注入进来，≈ Spring 构造注入一个 Bean。依赖可以是配置、数据库连接、当前用户、鉴权校验等。

测试时可替换（极其有用）：

app.dependency_overrides[get_settings] = lambda: {"max_concurrency": 1}

≈ Spring 测试里用 @MockBean 换掉真实依赖。

6. 任务 5：异步批量接口（复用第 5 阶段）

接口本身就是 async def，内部用 asyncio.gather + Semaphore 并发处理一批问题：

@app.post("/eval/run", response_model=EvalResponse)
async def eval_run(request: EvalRequest, settings: dict = Depends(get_settings)):
    limit = min(request.concurrency, settings["max_concurrency"])  # 服务端兜底上限
    sem = asyncio.Semaphore(limit)

    async def run_one(q):
        async with sem:
            try:
                return EvalItemResult(id=q, ok=True, answer=await fake_llm(q))
            except Exception as exc:
                return EvalItemResult(id=q, ok=False, error=str(exc))

    results = await asyncio.gather(*(run_one(q) for q in request.questions))
    return EvalResponse(total=len(results), ok=sum(r.ok for r in results), results=results)

注意"并发上限取请求值和服务端配置的较小者"——别让客户端随便要个 concurrency=99999 把你打爆。

7. Agent 服务常见接口

参考实现 services/app.py 提供了这套最小 Agent API：

接口 作用
GET /health 健康检查（K8s / 负载均衡探活）
POST /chat 单轮对话
POST /tools/{tool_name}/invoke 调用某个工具
POST /eval/run 批量评估一组问题

真实项目里还常见 POST /agent/invoke、POST /agent/stream（流式响应，对应 SSE）。

8. 测试：不起真服务器

fastapi.testclient.TestClient 在进程内直接打 ASGI app，不开端口、不走网络，又快又稳：

from fastapi.testclient import TestClient
client = TestClient(app)

def test_chat():
    resp = client.post("/chat", json={"message": "hi"})
    assert resp.status_code == 200
    assert resp.json() == {"answer": "echo: hi"}

≈ Spring 的 MockMvc / @WebMvcTest。完整测试见 tests/test_stage6_fastapi_app.py，覆盖：正常响应、422 校验、404、自定义异常转 400、依赖 override、OpenAPI schema。

9. 可运行验证

启动服务（需要时手动开，按 Ctrl+C 停）：

.\.venv\Scripts\python.exe -m uvicorn services.app:app --reload

然后浏览器开 http://127.0.0.1:8000/docs 交互式调接口。

运行测试（离线，不起服务器）：

.\.venv\Scripts\python.exe -m pytest tests\test_stage6_fastapi_app.py

10. 本阶段掌握标准

  • 能创建 app、定义 GET/POST route。
  • 能用路径参数、查询参数、Pydantic 请求/响应模型。
  • 能用 HTTPException 和 exception_handler 处理错误。
  • 能用 Depends 做依赖注入，并在测试里 override。
  • 能写一个 async def 接口并在里面并发处理。
  • 能用 TestClient 给接口写离线测试。
  • 能打开 /docs 调试并理解自动 OpenAPI。

11. 练习任务（自己动手）

参考 services/app.py 已实现 roadmap 第 9 节 5 个任务，可在其上扩展：

  1. 给所有接口加一个简单鉴权依赖：Depends 校验 header 里的 X-API-Key，缺失/错误返回 401。
  2. 加 POST /agent/stream，用 StreamingResponse 逐块吐字符串，模拟流式响应。
  3. 给 /eval/run 接真实并发上限和超时（结合第 5 阶段 wait_for），超时项标记失败。
  4. 加一个中间件（@app.middleware("http")）记录每个请求的耗时并打日志。
  5. 把请求/响应模型抽到独立的 services/models.py，用 model_json_schema() 打印 schema。