API服务化用FastAPI把Agent封装成RESTful接口前面几十篇做的Agent都是在本地脚本里跑自己用没问题。但真要给别人用或者接到产品系统里你总不能让别人也开个终端跑Python。把Agent封装成一个HTTP接口别人发个请求过来Agent处理完返回结果这才是生产环境该有的样子。今天这篇就用FastAPI把Agent包成一个RESTful服务。从请求响应模型到会话管理到流式输出一步步搭出来最后给一套能直接跑的完整代码。为什么选FastAPIPython写Web框架不少Django太重Flask太裸FastAPI卡在中间。它天生支持异步写Agent这种IO密集的场景正合适。自带请求参数校验用Pydantic定义模型请求体的字段类型、必填可选全帮你管好。还有自动文档服务跑起来访问/docs就能看到接口文档和在线调试页面省得手写。安装就一行pip install fastapi uvicorn。fastapi是框架本身uvicorn是跑它的ASGI服务器。服务化架构怎么设计Agent变成服务以后要处理几个问题。请求来了怎么传给Agent。用户发一个JSON过来你得解析成Agent能理解的输入格式。Agent的输出也得转成JSON返回给用户。这中间需要请求模型和响应模型做转换。会话怎么保持。Agent是有记忆的用户说第二句话的时候Agent得记得第一句说了什么。HTTP是无状态的每个请求独立你得自己管会话状态。最简单的做法是用session_id关联一组对话历史存在内存里。慢请求怎么处理。Agent调大模型慢的时候一个请求十几秒用户盯着转圈圈很焦虑。流式响应能解决这个问题Agent生成一点就往外吐一点用户看到内容一个字一个字蹦出来体验好很多。并发怎么扛。多个用户同时请求Agent处理要异步不能一个请求阻塞了其他全等着。FastAPI的异步路由天然支持这个但你的Agent底层也得是异步的才行。请求响应模型设计用Pydantic定义模型请求体和响应体都有明确的字段定义。请求模型至少要有用户输入的消息内容可选带上session_id表示是哪个会话。响应模型要有Agent的回复内容加上session_id方便客户端关联再带个元数据字段放token用量之类的信息。下面是完整代码包含一个简单的Agent封装、会话管理、普通接口和流式接口。importuuidimportasynciofromtypingimportOptional,AsyncGeneratorfromfastapiimportFastAPI,HTTPException,DependsfrompydanticimportBaseModel,Fieldfromcontextlibimportasynccontextmanager# ---------- 会话管理 ----------classSessionManager:管理用户会话每个session_id对应一组对话历史def__init__(self):self.sessions{}# session_id - 消息历史列表defcreate_session(self)-str:创建新会话返回session_idsession_idstr(uuid.uuid4())# 生成唯一会话IDself.sessions[session_id][]# 初始化空的消息历史returnsession_iddefget_history(self,session_id:str)-list:获取某个会话的消息历史ifsession_idnotinself.sessions:raiseHTTPException(status_code404,detailf会话{session_id}不存在)returnself.sessions[session_id]defadd_message(self,session_id:str,role:str,content:str):往会话历史里加一条消息ifsession_idnotinself.sessions:self.sessions[session_id][]self.sessions[session_id].append({role:role,content:content})defdelete_session(self,session_id:str):删除某个会话ifsession_idinself.sessions:delself.sessions[session_id]# 全局会话管理器整个应用共用一个session_managerSessionManager()# ---------- Agent封装 ----------classSimpleAgent:一个简单的Agent封装实际项目里换成你的Agent实现def__init__(self):self.name助手Agent# Agent名字asyncdefchat(self,message:str,history:list)-str:异步聊天接口接收消息和历史返回回复# 这里用模拟回复代替真实LLM调用# 实际项目里换成 await llm.ainvoke(message, history)awaitasyncio.sleep(0.5)# 模拟网络延迟replyf收到你的消息:{message}。ifhistory:replyf我记得你之前说了{len(history)}句话。returnreplyasyncdefstream_chat(self,message:str,history:list)-AsyncGenerator[str,None]:流式聊天接口逐字返回回复内容replyawaitself.chat(message,history)# 把回复拆成一个字一个字往外吐forcharinreply:awaitasyncio.sleep(0.02)# 模拟生成延迟yieldchar# 每次yield一个字符# 全局Agent实例agentSimpleAgent()# ---------- Pydantic模型 ----------classChatRequest(BaseModel):聊天请求模型定义了客户端发过来的JSON结构message:strField(# 用户的消息内容必填...,min_length1,max_length2000,description用户输入的消息)session_id:Optional[str]Field(# 会话ID可选不传就新建会话None,description会话ID首次对话不传)classChatResponse(BaseModel):聊天响应模型定义了返回给客户端的JSON结构reply:strField(...,descriptionAgent的回复内容)session_id:strField(...,description会话ID后续对话要带上)message_count:intField(# 当前会话的消息总数...,description当前会话累计消息数)classSessionResponse(BaseModel):创建会话的响应模型session_id:strField(...,description新创建的会话ID)# ---------- FastAPI应用 ----------asynccontextmanagerasyncdeflifespan(app:FastAPI):应用生命周期管理启动和关闭时执行print(Agent服务启动)# 启动时打印日志yield# 应用运行期间print(Agent服务关闭)# 关闭时打印日志appFastAPI(titleAgent API服务,description把Agent封装成RESTful接口,version1.0.0,lifespanlifespan,)# ---------- 接口定义 ----------app.post(/sessions,response_modelSessionResponse)asyncdefcreate_session():创建新会话返回session_idsession_idsession_manager.create_session()returnSessionResponse(session_idsession_id)app.delete(/sessions/{session_id})asyncdefdelete_session(session_id:str):删除指定会话session_manager.delete_session(session_id)return{status:已删除}app.post(/chat,response_modelChatResponse)asyncdefchat(request:ChatRequest):普通聊天接口等Agent处理完一次性返回# 如果没传session_id自动创建新会话session_idrequest.session_idifsession_idisNone:session_idsession_manager.create_session()# 获取会话历史historysession_manager.get_history(session_id)# 记录用户消息session_manager.add_message(session_id,user,request.message)# 调用Agent获取回复replyawaitagent.chat(request.message,history)# 记录Agent回复session_manager.add_message(session_id,assistant,reply)# 返回响应returnChatResponse(replyreply,session_idsession_id,message_countlen(session_manager.get_history(session_id)),)app.post(/chat/stream)asyncdefchat_stream(request:ChatRequest):流式聊天接口逐字返回Agent的回复fromfastapi.responsesimportStreamingResponse# 同样处理会话逻辑session_idrequest.session_idifsession_idisNone:session_idsession_manager.create_session()historysession_manager.get_history(session_id)session_manager.add_message(session_id,user,request.message)asyncdefgenerate():生成器函数逐字yield回复内容full_replyasyncforcharinagent.stream_chat(request.message,history):full_replychar# 拼接完整回复yieldchar# 往客户端吐一个字符# 流结束后把完整回复存入历史session_manager.add_message(session_id,assistant,full_reply)# 返回流式响应媒体类型设为纯文本returnStreamingResponse(generate(),media_typetext/plain,headers{X-Session-Id:session_id}# 通过header返回session_id)app.get(/sessions/{session_id}/history)asyncdefget_history(session_id:str):获取某个会话的完整历史记录historysession_manager.get_history(session_id)return{session_id:session_id,history:history}# ---------- 启动服务 ----------if__name____main__:importuvicorn# host设0.0.0.0允许外部访问port按需改# reloadTrue开发时自动重载生产环境关掉uvicorn.run(app,host0.0.0.0,port8000)效果验证服务跑起来以后有几种方式验证。最简单的是直接访问http://localhost:8000/docsFastAPI自动生成的文档页面。在里面找到POST /chat接口点Try it out填一段message点Execute能看到Agent的回复JSON。session_id留空第一次会自动创建后续请求把返回的session_id填进去就能保持对话上下文。用curl测普通接口发一个POST请求。curl-XPOST http://localhost:8000/chat\-HContent-Type: application/json\-d{message: 你好}返回的JSON里有reply、session_id、message_count三个字段。拿着返回的session_id再发一条message_count会变成2说明会话历史在累加。测流式接口用curl加-N参数能看到内容一个字一个字返回。curl-N-XPOST http://localhost:8000/chat/stream\-HContent-Type: application/json\-d{message: 讲个故事}判断成功的标准普通接口返回200状态码和完整的JSON。流式接口返回200内容逐步输出不卡顿。连续发多条消息message_count递增说明会话状态保持正常。常见报错422 Validation Error说明请求体格式不对检查message字段有没有传、长度有没有超限。404说明session_id不对检查是不是拼错了或者会话已经被删了。踩坑记录第一个坑流式接口里会话历史没存上。我一开始在generate函数外面调了add_message存用户消息Agent回复在generate函数里流式输出但完整回复的存储放在了generate函数外面。结果流式响应返回以后generate函数还没跑完外面的代码先执行了存了一个空字符串。后来把完整回复的存储挪到generate函数内部流结束后再存问题才解决。异步生成器的执行顺序跟同步代码不一样跟数据存储相关的操作一定要放在生成器内部。第二个坑并发请求串会话。一开始会话管理器用字典存没加锁。两个请求同时操作同一个session_id一个在读历史一个在写偶发数据错乱。测试的时候单线程测不出来上了并发压测才暴露。后来给会话操作加了asyncio.Lock同一个session_id的操作串行化不同session_id之间不阻塞。如果你的会话量大锁的粒度可以细到每个session_id一把锁别全局一把锁把并发全卡死了。第三个坑生产环境内存涨。会话历史存在内存里用户多了或者对话长了内存一直涨。一开始没做清理跑了一天内存就爆了。后来加了两个策略一是限制每个会话的历史长度超过20条就裁掉最早的只保留最近20条。二是设个过期时间24小时没活动的会话自动清理。生产环境建议把会话存Redis别放内存里重启就丢了。部署注意事项开发阶段用uvicorn的reload模式很方便改了代码自动重启。上生产别用reload性能差。用uvicorn或者gunicorn跑多个worker进程扛并发。模型API的key别硬编码在代码里用环境变量或者配置文件管理。FastAPI可以用Settings管理配置从环境变量读。跨域问题别忘了。前端调你的接口域名不一样浏览器会拦。FastAPI加个CORSMiddleware把允许的域名配上就行。开发阶段可以设allow_origins为星号放开所有域名生产环境收敛到具体域名。日志要打好。每个请求记session_id、消息内容、处理耗时、token用量。出了问题能根据session_id追溯完整对话过程。FastAPI可以加中间件统一记录请求日志不用每个接口里手写。延伸与判断这个服务化架构可以扩展的方向很多。把上一篇文章的多Agent项目团队封装成接口用户发一个需求三个Agent跑完返回代码和测试报告。接口设计上把/chat换成/project请求体里带上需求描述响应体里返回多阶段结果。认证授权也得加。生产环境的接口不能裸奔加个API key或者JWT认证。FastAPI用Depends做依赖注入写个认证依赖挂在路由上就行。如果有长任务比如多Agent跑一遍要几分钟同步等不了。可以改成任务队列模式用户发请求返回一个task_id后台慢慢跑用户拿task_id轮询结果或者通过WebSocket推送进度。结尾把Agent封装成API服务核心就四件事请求响应模型定义好会话状态管好慢请求用流式响应兜住并发和资源该限制的限制。这套东西搭完你的Agent就能被任何系统调用了从本地脚本变成了真正的在线服务。AI Agent企业级实战系列到这里就收尾了从单个Agent的搭建到多Agent协作再到服务化部署整条路走通了。