将Agent从本地脚本转变为线上服务，核心是把推理逻辑和HTTP层解耦。FastAPI的异步特性与Pydantic模型，能让Agent的入参校验、并发调度和接口文档自动化，降低部署成本。下面通过一个对话型Agent示例，逐步说明封装过程。
项目结构与Agent核心逻辑封装
在动手写接口前，先把Agent本身从Web框架中剥离出来。假设我们有一个基于大语言模型的问答Agent，它接收一个用户问题，内部调用模型并维护会话记忆。如果直接把模型调用写在路由函数里，后期很难测试和复用。推荐做法是单独建立一个agent_core.py模块，暴露一个类或者异步函数。
下面代码展示了一个最简的Agent封装。它不依赖任何Web框架，只负责接收文本并返回结果。这样在单元测试中可以脱离网络直接验证Agent行为，也为后面FastAPI的调用提供了清晰边界。
import asyncio
class SimpleAgent:
def __init__(self, model_name="demo-model"):
self.model_name = model_name
async def ask(self, question: str) -> str:
# 模拟模型推理耗时
await asyncio.sleep(0.5)
return f"[{self.model_name}] 回答：{question}"
# 单独使用时
async def main():
agent = SimpleAgent()
print(await agent.ask("什么是FastAPI"))
if __name__ == "__main__":
asyncio.run(main())
这种结构的好处是，当模型从本地推理换成远程API时，只需修改SimpleAgent内部，FastAPI层完全不用动。在工程实践中，Agent往往还要接检索、工具调用和记忆模块，把它们都收敛到核心类里，服务代码才能保持轻薄。
另外要注意，Agent的初始化如果有较重资源（如加载向量库、建立连接池），应在应用启动时完成，而不是每次请求都重建。FastAPI提供的生命周期事件正好适合做这件事，后面会提到。
用FastAPI暴露Agent为HTTP接口
有了独立的Agent核心，下一步就是写FastAPI应用。我们使用pydantic定义请求体和响应体，让框架自动完成参数校验和OpenAPI文档生成。对于问答场景，请求只需要一个问题字段，响应包含回答和会话标识即可。
下面的示例创建了/chat接口，在应用启动事件中实例化Agent，避免重复加载。路由函数通过Request或直接依赖注入拿到Agent对象，调用其异步ask方法。因为Agent本身是异步的，整个链路不会阻塞事件循环。
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from contextlib import asynccontextmanager
from agent_core import SimpleAgent
class ChatRequest(BaseModel):
question: str
session_id: str = "default"
class ChatResponse(BaseModel):
answer: str
session_id: str
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.agent = SimpleAgent()
yield
# 关闭时清理资源
app.state.agent = None
app = FastAPI(lifespan=lifespan)
async def get_agent(app: FastAPI = Depends()):
return app.state.agent
@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest, agent: SimpleAgent = Depends(get_agent)):
result = await agent.ask(req.question)
return ChatResponse(answer=result, session_id=req.session_id)
运行后访问/docs就能看到自动生成的Swagger界面，前端工程师可以直接调试。如果Agent支持流式输出，可改用StreamingResponse逐字返回，提升用户体验。对于同步阻塞型的旧模型，可以用run_in_executor丢到线程池，防止拖垮整个服务。
在真实部署中，还要考虑请求体大小限制和超时控制。FastAPI本身不限制请求体，但反向代理（如Nginx）通常会设上限，Agent上传文件时要提前规划。接口层应保持薄，只做协议转换，所有业务判断交给Agent核心。
鉴权、并发与部署上线要点
开放Agent服务后，第一步是加鉴权。最简单的是在依赖里校验请求头中的Token，不通过直接抛HTTPException。下面示例用了一个固定密钥，生产环境应改为读取数据库或调用鉴权中心。
from fastapi import Header, HTTPException
async def verify_token(x_token: str = Header(...)):
if x_token != "secret-key":
raise HTTPException(status_code=401, detail="无效令牌")
return x_token
@app.post("/chat", response_model=ChatResponse, dependencies=[Depends(verify_token)])
async def chat_secure(req: ChatRequest, agent: SimpleAgent = Depends(get_agent)):
result = await agent.ask(req.question)
return ChatResponse(answer=result, session_id=req.session_id)
并发方面，FastAPI基于ASGI，单进程就能处理大量连接。但Agent推理可能吃满CPU或显存，需要用信号量限制同时运行的Agent任务数，避免资源挤兑。可以用asyncio.Semaphore包裹ask调用，或在网关层做限流。
部署时推荐用uvicorn多进程配合gunicorn管理器，例如gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app。容器化时把模型权重和依赖打进镜像，用环境变量区分开发和生产配置。日志要记录每次请求的session_id和耗时，方便排查慢查询。只要核心逻辑与接口分离清晰，后续接入监控系统或灰度发布都会轻松很多。