用FastAPI搭建AI Agent API：从零到生产部署完整教程（2026）

📘 AI教程 2026-07-19 💬 🔥 Trending

🩺 摘要

FastAPI（GitHub 101K stars，截至2026年7月）是Python社区增长最快的Web框架之一。它的...

📝 详情

FastAPI（GitHub 101K stars，截至2026年7月）是Python社区增长最快的Web框架之一。它的异步原生支持、自动生成OpenAPI文档、Pydantic验证，让它天然适合AI Agent场景。

为什么FastAPI特别适合AI Agent API？

三个原因：

异步是第一公民 — AI Agent的核心操作（LLM调用、向量检索、外部API）全是I/O密集型，async/await能最大化吞吐
自动文档 — Swagger UI + ReDoc，前端和其他Agent调你的API不需要额外沟通
Pydantic v2 — 请求/响应模型验证，配合OpenAI的结构化输出，类型安全

这不是又一篇FastAPI入门教程。我们从项目结构开始，写一个真正能用的AI Agent API。

第一步：项目结构

agent-api/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI应用入口
│   ├── config.py        # 配置管理
│   ├── models/          # Pydantic模型
│   │   ├── __init__.py
│   │   ├── request.py   # 请求模型
│   │   └── response.py  # 响应模型
│   ├── agents/          # Agent逻辑
│   │   ├── __init__.py
│   │   ├── base.py      # 基础Agent类
│   │   └── tools.py     # 工具定义
│   ├── memory/          # 记忆系统
│   │   ├── __init__.py
│   │   └── store.py     # 向量记忆存储
│   └── router/          # 路由
│       ├── __init__.py
│       ├── chat.py      # 聊天路由
│       └── agent.py     # Agent路由
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

这个结构的好处：Agent逻辑和API路由分离，方便单独测试；记忆系统独立模块，后期换数据库只需要改memory/目录。

第二步：核心Agent逻辑

这是一个支持流式输出和工具调用的Agent：

# app/agents/base.py
import json
from typing import AsyncGenerator, Optional
from openai import AsyncOpenAI
from app.agents.tools import get_weather, search_wiki, search_web

class AIAgent:
    def __init__(self, system_prompt: str = None, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.system_prompt = system_prompt or "你是一个有用的AI助手"
        self.model = model
        self.tools = [
            {"type": "function", "function": {
                "name": "get_weather",
                "description": "查询指定城市的天气",
                "parameters": {"type": "object", "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                }, "required": ["city"]}
            }},
            {"type": "function", "function": {
                "name": "search_wiki",
                "description": "搜索维基百科获取信息",
                "parameters": {"type": "object", "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                }, "required": ["query"]}
            }}
        ]

    async def chat_stream(
        self,
        messages: list,
        temperature: float = 0.7
    ) -> AsyncGenerator[str, None]:
        """支持函数调用的流式对话"""
        kwargs = {
            "model": self.model,
            "messages": [{"role": "system", "content": self.system_prompt}] + messages,
            "tools": self.tools,
            "stream": True,
            "temperature": temperature,
        }

        while True:
            response = await self.client.chat.completions.create(**kwargs)

            # 收集流式回复
            collected_content = ""
            tool_calls = []

            async for chunk in response:
                delta = chunk.choices[0].delta if chunk.choices else None
                if delta is None:
                    continue

                if delta.content:
                    collected_content += delta.content
                    yield delta.content

                if delta.tool_calls:
                    for tc in delta.tool_calls:
                        while len(tool_calls) <= tc.index:
                            tool_calls.append({
                                "id": "", "type": "function",
                                "function": {"name": "", "arguments": ""}
                            })
                        if tc.id: tool_calls[tc.index]["id"] = tc.id
                        if tc.function.name: tool_calls[tc.index]["function"]["name"] = tc.function.name
                        if tc.function.arguments: tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

            # 如果没有工具调用，返回文本
            if not tool_calls:
                break

            # 执行工具调用
            yield "\n\n**正在调用工具...**\n\n"
            messages.append({"role": "assistant", "content": None, "tool_calls": tool_calls})

            for tc in tool_calls:
                fn_name = tc["function"]["name"]
                fn_args = json.loads(tc["function"]["arguments"])

                yield f"> 调用 `{fn_name}`，参数：`{json.dumps(fn_args, ensure_ascii=False)}`\n\n"

                # 执行对应函数
                tool_map = {
                    "get_weather": get_weather,
                    "search_wiki": search_wiki,
                }
                result = tool_map[fn_name](**fn_args)

                yield f"> 返回：`{result[:200]}{'...' if len(result)>200 else ''}`\n\n"

                messages.append({
                    "role": "tool",
                    "tool_call_id": tc["id"],
                    "content": json.dumps(result, ensure_ascii=False)
                })

            # 带工具结果的下一轮推理
            kwargs["messages"] = [{"role": "system", "content": self.system_prompt}] + messages

核心设计思路： while True循环让Agent可以无限次调用工具，直到它认为信息足够返回最终答案。每次工具调用结果追加到messages里作为上下文。

第三步：API路由

# app/router/agent.py
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from app.agents.base import AIAgent

router = APIRouter(prefix="/api/v1/agent", tags=["agent"])
agent = AIAgent(system_prompt="你是一个信息查询助手。回答要简洁，不确定时说不知道。")

class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"  # 支持多会话
    temperature: float = 0.7

class ChatResponse(BaseModel):
    response: str

@router.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """非流式对话接口"""
    messages = [{"role": "user", "content": request.message}]
    result = ""
    async for chunk in agent.chat_stream(messages, request.temperature):
        result += chunk
    return ChatResponse(response=result)

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """流式对话接口 - 适合前端实时展示"""
    messages = [{"role": "user", "content": request.message}]

    async def generate():
        async for chunk in agent.chat_stream(messages, request.temperature):
            yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

流式接口用SSE（Server-Sent Events）传回前端。前端用EventSource或fetch处理流式响应，用户能看到Agent生成内容的过程。

第四步：工具函数实现

# app/agents/tools.py
import httpx
from typing import Any

async def get_weather(city: str) -> dict[str, Any]:
    """查询天气（用异步HTTP）"""
    async with httpx.AsyncClient() as client:
        # 用免费天气API
        resp = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        data = resp.json()
        current = data["current_condition"][0]
        return {
            "city": city,
            "temperature": f"{current['temp_C']}°C",
            "humidity": f"{current['humidity']}%",
            "description": current["weatherDesc"][0]["value"]
        }

async def search_web(query: str) -> str:
    """搜索网页"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.duckduckgo.com/?q={query}&format=json",
            timeout=10.0
        )
        data = resp.json()
        return data.get("AbstractText", "未找到相关信息")

注意： 工具函数也必须是async的，否则会阻塞事件循环。

第五步：生产部署

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 用gunicorn + uvicorn workers
CMD ["gunicorn", "app.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

# docker-compose.yml
version: '3.8'
services:
  agent-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

几个生产要点： 1. gunicorn + uvicorn workers — 比单纯用uvicorn稳定，worker挂了自动重启 2. Redis做状态缓存 — 存会话历史，Agent重启后能继续对话 3. 环境变量管理 — API Key不在代码里写死，用.env或secrets管理 4. 健康检查 — 加个/health端点给负载均衡器用

@app.get("/health")
async def health():
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat()}

实际运行效果

启动后访问 http://localhost:8000/docs，Swagger UI自动出现。前端可以直接调 POST /api/v1/agent/chat/stream 拿SSE流。

测试调用：

curl -X POST "http://localhost:8000/api/v1/agent/chat" \
  -H "Content-Type: application/json" \
  -d '{"message": "北京今天天气怎么样？适合出门吗？"}'

Agent会：调用get_weather工具获取北京天气 → 根据天气数据给出出门建议。整套流程全自动，不需要写死任何业务逻辑。

总结

用FastAPI搭AI Agent API的最大好处是：类型安全、异步原生、自动文档、容易扩展。

这套代码的核心抽象——Agent类管理推理循环 + 工具函数异步执行 + SSE流式传输 + Pydantic验证——足够支撑从单文件脚本到生产服务的跨越。

下一步你可以： - 加上PostgreSQL存用户会话 - 集成向量数据库（Chroma/Pinecone）做长期记忆 - 加上速率限制（slowapi）防滥用 - 接上n8n/Dify做可视化编排

💡 建议收藏本文。 文中的项目结构可以直接作为AI Agent API的脚手架，下次不用从零开始。

📤 转发给写Python的朋友。 FastAPI + AI Agent是2026年最实用的后端技能组合之一。

上篇：Dify vs Flowise深度对比下篇：AI Agent记忆系统——向量数据库原理解析