API服务化，用FastAPI把Agent封装成RESTful接口

API服务化，用FastAPI把Agent封装成RESTful接口

前面几十篇做的Agent都是在本地脚本里跑，自己用没问题。但真要给别人用，或者接到产品系统里，你总不能让别人也开个终端跑Python。把Agent封装成一个HTTP接口，别人发个请求过来，Agent处理完返回结果，这才是生产环境该有的样子。

今天这篇就用FastAPI把Agent包成一个RESTful服务。从请求响应模型到会话管理到流式输出，一步步搭出来，最后给一套能直接跑的完整代码。

为什么选FastAPI

Python写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_id"""session_id=str(uuid.uuid4())# 生成唯一会话IDself.sessions[session_id]=[]# 初始化空的消息历史returnsession_iddefget_history(self,session_id:str)->list:"""获取某个会话的消息历史"""ifsession_idnotinself.sessions:raiseHTTPException(status_code=404,detail=f"会话{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_manager=SessionManager()# ---------- 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)# 模拟网络延迟reply=f"收到你的消息:{message}。"ifhistory:reply+=f"我记得你之前说了{len(history)}句话。"returnreplyasyncdefstream_chat(self,message:str,history:list)->AsyncGenerator[str,None]:"""流式聊天接口，逐字返回回复内容"""reply=awaitself.chat(message,history)# 把回复拆成一个字一个字往外吐forcharinreply:awaitasyncio.sleep(0.02)# 模拟生成延迟yieldchar# 每次yield一个字符# 全局Agent实例agent=SimpleAgent()# ---------- Pydantic模型 ----------classChatRequest(BaseModel):"""聊天请求模型，定义了客户端发过来的JSON结构"""message:str=Field(# 用户的消息内容，必填...,min_length=1,max_length=2000,description="用户输入的消息")session_id:Optional[str]=Field(# 会话ID，可选，不传就新建会话None,description="会话ID，首次对话不传")classChatResponse(BaseModel):"""聊天响应模型，定义了返回给客户端的JSON结构"""reply:str=Field(...,description="Agent的回复内容")session_id:str=Field(...,description="会话ID，后续对话要带上")message_count:int=Field(# 当前会话的消息总数...,description="当前会话累计消息数")classSessionResponse(BaseModel):"""创建会话的响应模型"""session_id:str=Field(...,description="新创建的会话ID")# ---------- FastAPI应用 ----------@asynccontextmanagerasyncdeflifespan(app:FastAPI):"""应用生命周期管理，启动和关闭时执行"""print("Agent服务启动")# 启动时打印日志yield# 应用运行期间print("Agent服务关闭")# 关闭时打印日志app=FastAPI(title="Agent API服务",description="把Agent封装成RESTful接口",version="1.0.0",lifespan=lifespan,)# ---------- 接口定义 ----------@app.post("/sessions",response_model=SessionResponse)asyncdefcreate_session():"""创建新会话，返回session_id"""session_id=session_manager.create_session()returnSessionResponse(session_id=session_id)@app.delete("/sessions/{session_id}")asyncdefdelete_session(session_id:str):"""删除指定会话"""session_manager.delete_session(session_id)return{"status":"已删除"}@app.post("/chat",response_model=ChatResponse)asyncdefchat(request:ChatRequest):"""普通聊天接口，等Agent处理完一次性返回"""# 如果没传session_id，自动创建新会话session_id=request.session_idifsession_idisNone:session_id=session_manager.create_session()# 获取会话历史history=session_manager.get_history(session_id)# 记录用户消息session_manager.add_message(session_id,"user",request.message)# 调用Agent获取回复reply=awaitagent.chat(request.message,history)# 记录Agent回复session_manager.add_message(session_id,"assistant",reply)# 返回响应returnChatResponse(reply=reply,session_id=session_id,message_count=len(session_manager.get_history(session_id)),)@app.post("/chat/stream")asyncdefchat_stream(request:ChatRequest):"""流式聊天接口，逐字返回Agent的回复"""fromfastapi.responsesimportStreamingResponse# 同样处理会话逻辑session_id=request.session_idifsession_idisNone:session_id=session_manager.create_session()history=session_manager.get_history(session_id)session_manager.add_message(session_id,"user",request.message)asyncdefgenerate():"""生成器函数，逐字yield回复内容"""full_reply=""asyncforcharinagent.stream_chat(request.message,history):full_reply+=char# 拼接完整回复yieldchar# 往客户端吐一个字符# 流结束后把完整回复存入历史session_manager.add_message(session_id,"assistant",full_reply)# 返回流式响应，媒体类型设为纯文本returnStreamingResponse(generate(),media_type="text/plain",headers={"X-Session-Id":session_id}# 通过header返回session_id)@app.get("/sessions/{session_id}/history")asyncdefget_history(session_id:str):"""获取某个会话的完整历史记录"""history=session_manager.get_history(session_id)return{"session_id":session_id,"history":history}# ---------- 启动服务 ----------if__name__=="__main__":importuvicorn# host设0.0.0.0允许外部访问，port按需改# reload=True开发时自动重载，生产环境关掉uvicorn.run(app,host="0.0.0.0",port=8000)

效果验证

服务跑起来以后，有几种方式验证。

最简单的是直接访问http://localhost:8000/docs，FastAPI自动生成的文档页面。在里面找到POST /chat接口，点Try it out，填一段message，点Execute，能看到Agent的回复JSON。session_id留空第一次会自动创建，后续请求把返回的session_id填进去就能保持对话上下文。

用curl测普通接口，发一个POST请求。

curl-XPOST http://localhost:8000/chat\-H"Content-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\-H"Content-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协作再到服务化部署，整条路走通了。