把Agent从实验笔记本搬进生产环境,核心工作就是给它套一层网络接口。FastAPI作为现代Python Web框架,原生支持异步、类型注解和交互式文档,非常适合承接这类任务。它不仅能把Agent的推理函数暴露成HTTP端点,还能借助Pydantic模型帮我们过滤脏输入,降低下游处理逻辑的崩溃风险。

服务骨架与路由设计
封装Agent的第一步是确定对外暴露哪些能力。多数Agent至少有「单次问答」和「多轮会话」两种模式,我们可以分别映射到/chat和/session/chat两个路由。在FastAPI里,用APIRouter拆分模块能让项目结构更清晰,也方便后续加鉴权中间件。路由函数本身应当尽量轻量,只做参数校验和任务分发,真正耗时的模型推理交给单独的类或异步函数处理。
下面示例展示了一个最小可运行骨架。注意我们在路由层用Depends注入了配置对象,这样Agent实例可以在应用启动时构建一次,避免每次请求都重新加载权重。这种单例模式在LLM类Agent上尤其重要,因为模型体积大、加载慢。
from fastapi import FastAPI, APIRouter, Depends
from pydantic import BaseModel
class ChatReq(BaseModel):
prompt: str
max_tokens: int = 512
app = FastAPI(title="Agent Service")
router = APIRouter()
def get_agent():
# 假设Agent在全局只加载一次
from my_agent import Agent
return Agent.load("config.yaml")
@router.post("/chat")
async def chat(req: ChatReq, agent=Depends(get_agent)):
result = await agent.run(req.prompt, req.max_tokens)
return {"answer": result}
app.include_router(router)
上述写法把Agent生命周期和请求处理解耦。如果直接把Agent.load写进路由函数,每来一个请求都重新读盘加载,接口延迟会从毫秒级飙到秒级甚至分钟级。另外,用Pydantic的BaseModel定义请求体,FastAPI会自动生成JSON Schema,前端联调时直接看/docs页面就能知道该传什么字段。
异步与阻塞调用的坑
很多开发者把Agent的推理包在一个普通函数里,然后在async路由里用await调用,结果发现并发一高服务就卡死。原因是如果那个函数内部是同步CPU密集或IO密集操作,且没有真正释放事件循环,FastAPI的异步优势完全发挥不出来。对于不支持原生协程的旧模型库,应当用run_in_threadpool或asyncio.to_thread把阻塞调用抛到线程池。
如果Agent支持流式输出,比如逐字返回生成内容,可以用StreamingResponse配合异步生成器,让用户体验更顺滑,也避免服务端攒完所有token才返回带来的超时问题。下面代码演示了线程池包裹同步Agent的用法,以及错误时如何统一返回结构。
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
from starlette.concurrency import run_in_threadpool
@router.post("/chat/stream")
async def chat_stream(req: ChatReq, agent=Depends(get_agent)):
try:
def blocking_gen():
for piece in agent.run_stream(req.prompt):
yield piece
return StreamingResponse(
(chunk async for chunk in run_in_threadpool(blocking_gen)),
media_type="text/plain"
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
需要强调的是,run_in_threadpool适合IO等待型阻塞;若是纯CPU计算且用到多进程模型,应考虑ProcessPoolExecutor或者把Agent推理独立成gRPC微服务。否则Python GIL会让多线程加速失效,仍然堵在单核上。这也是Agent部署里最容易低估的性能盲区。
鉴权与依赖注入实践
线上Agent服务绝不能裸奔。常见做法是在Depends链里加一个verify_token函数,从请求头取Authorization做校验,失败就抛401。FastAPI的依赖树可以全局挂载,也可以按路由选择性挂载,灵活性很高。对于多租户场景,还能在依赖里解析出tenant_id,传给Agent做隔离或计费。
除了鉴权,依赖注入还能统一管理日志、限流和追踪。例如用SlowAPI或自写中间件统计每个Agent调用的耗时,超过阈值就告警。下面示例展示了一个简单的令牌校验依赖,以及如何在路由中组合多个依赖。
from fastapi import Header, HTTPException, Depends
async def verify_token(authorization: str = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="missing token")
token = authorization.split(" ", 1)[1]
if token != "ipipp-secret":
raise HTTPException(status_code=403, detail="invalid token")
return token
@router.post("/secure/chat", dependencies=[Depends(verify_token)])
async def secure_chat(req: ChatReq, agent=Depends(get_agent)):
return {"answer": await agent.run(req.prompt)}
把鉴权写成依赖而不是在每个函数里复制粘贴,既减少出错也方便后续换鉴权方案,比如从静态令牌升级到JWT或OAuth2密码流。FastAPI自带OAuth2PasswordBearer等工具类,配合Pydantic甚至可以一键生成登录接口。当Agent服务要嵌入企业系统,这种可插拔的依赖设计会显著降低改造成本。
整体来看,用FastAPI封装Agent并不是简单套个@app.post就结束。从路由拆分、异步阻塞处理到鉴权注入,每一层都直接影响服务的稳定性和扩展能力。把重逻辑移出主事件循环、用依赖统一管理横切关注点,才能让原型真正变成可运维的部署服务。