本文摘要智能体此前只能在命令行里交互外部程序拿不到回复能力被锁在本地进程。新增的 api.py 用 FastAPI 把它包成 HTTP 接口请求体自动校验、回复包成 JSON 返回。一、环境与前提上一篇第五篇完成多轮对话的状态管理与记忆维护本篇解决把agent-demo/里的Agent与run_agent暴露成 HTTP API 的问题。调用约定沿用前五篇端点内调用run_agent传入一条用户输入字符串拿到回复字符串main.py一行不改。若你的run_agent需要显式接收Agent实例在调用处多传一个参数即可写法见第四节末尾。前置条件环境配置与_truncate_messages的截断策略见前几篇此处不重复展开Python 3.11、openai 1.x、tiktoken、python-dotenv已安装agent-demo/目录下main.py提供Agent类__init__、_count_tokens、_truncate_messages、run_agent、image_to_base64、main.env中已配置OPENAI_API_KEY。本篇只新增api.py不动main.py命令行入口继续可用服务出问题时用python main.py对照一次就能区分是接口层的问题还是Agent本身的问题。本篇新增两个依赖具体版本以 pip 实际安装到的为准未确认版本依赖在本篇的作用fastapi定义路由、用 Pydantic 校验请求体、用Depends做依赖注入uvicorn运行 FastAPI 生成的 ASGI 应用步骤 1安装依赖并确认导入链完好目的装上 Web 框架与 ASGI 服务器同时确认main.py的既有名称仍可导入不被新增依赖影响。操作在agent-demo/目录下依次执行pipinstallfastapi uvicorn python-cimport fastapi, uvicorn; print(ok)python-cfrom main import Agent, run_agent, image_to_base64, main; print(ok)预期输出后两条命令各回显一行ok实际输出未实测两条命令成功时都只回显ok缺包时抛ModuleNotFoundError名称不齐时抛ImportError。若导入main的那条卡住不返回说明main.py顶层直接调用了main()给它补上if __name__ __main__:守卫即可未实测。二、关键步骤一次请求的完整链路是curl把 JSON 发到/chat→uvicorn把请求交给 FastAPI → FastAPI 用ChatRequest校验并构造请求对象 →chat函数调用run_agent→ 返回值被ChatResponse序列化成 JSON。下面按这条链路落地。步骤 2新建agent-demo/api.py目的用 Pydantic 定义请求与响应模型暴露一个POST /chat端点去调用run_agentmain.py保持不动。操作新建文件agent-demo/api.py写入以下完整内容fromfastapiimportFastAPIfrompydanticimportBaseModelfrommainimportrun_agentclassChatRequest(BaseModel):message:strclassChatResponse(BaseModel):reply:strappFastAPI()app.post(/chat,response_modelChatResponse)defchat(request:ChatRequest)-ChatResponse:replyrun_agent(request.message)returnChatResponse(replyreply)四点说明路由函数chat用同步def声明run_agent同样是同步函数FastAPI 会把它放进线程池执行不会阻塞事件循环依据FastAPI 官方文档 Defining Asynchronous and Synchronous Functions。ChatRequest的message: str让 FastAPI 自动生成请求体校验字段缺失或类型不符会直接返回校验错误不用手写判断。response_modelChatResponse决定响应体只含reply字段同时把该结构写进 OpenAPI 文档/docs里能看到请求与响应的字段。app FastAPI()这一行的实例名必须是appuvicorn api:app靠它定位应用对象写成别的名字会启动失败见第三节。这个文件不构造Agent实例、也不碰self.messagesapi.py只做协议转换状态问题留给第三节。操作校验导入仍在agent-demo/目录下执行python-cimport api; print(type(api.app).__name__)预期输出FastAPI实际输出未实测成功时仅回显FastAPImain.py导入失败时会先抛出它的原始异常。步骤 3启动服务目的把应用跑在本地端口上供下一步调用。操作在agent-demo/目录下执行--reload会在文件改动后自动重启仅用于开发按CtrlC停止uvicorn api:app--reload预期输出启动行的具体格式随版本略有差异未确认版本INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit) INFO: Application startup complete.实际输出未实测进程保持前台运行控制台出现服务地址与Application startup complete.。启动成功后浏览器打开[/docs](/docs)能看到 FastAPI 自动生成的交互文档/chat端点应出现在列表里。默认只绑定127.0.0.1仅本机可访问要让局域网内其他机器调用启动命令需加--host 0.0.0.0未实测按 uvicorn 参数语义。步骤 4发送一次正常请求目的确认端点真的把用户输入交给run_agent并把回复包成 JSON 返回。操作另开一个终端服务保持运行执行curl-XPOST http://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{message: 用一句话介绍你自己}Content-Type: application/json不能省FastAPI 按 JSON 解析请求体缺这个请求头会直接返回校验错误。预期输出{reply:Agent 的回复文本}实际输出未实测响应体是含reply字段的 JSON 对象reply的文本由模型生成每次调用内容不同。步骤 5验证请求体校验目的确认字段写错时服务端直接拒绝校验不用自己写。操作curl-i-XPOST http://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{msg: 字段名写错了}预期输出状态行为HTTP/1.1 422 Unprocessable Entity响应体形如{detail:[{loc:[body,message],msg:Field required,type:missing}]}实际输出未实测返回校验失败响应detail数组元素包含loc、msg、type字段形态依据 FastAPI 官方文档。三、失败处理失败 1uvicorn 找不到应用对象报错原文AttributeError: module api has no attribute app模块名写错时报错原文Error loading ASGI app. Could not import module api.原因uvicorn 模块名:应用对象名要求模块里确实存在该名称的 ASGI 应用。常见触发方式有两种FastAPI 实例没有命名为app例如写成application FastAPI()或者在agent-demo/目录下写成uvicorn main:app模块名与app所在的文件不一致。修复把实例名改回app启动命令的模块名与文件名对齐。操作在agent-demo/目录下执行uvicorn api:app--reload预期输出与步骤 3 相同的启动行。实际输出未实测修复后启动行与步骤 3 一致。失败 2共享Agent实例导致对话状态串扰症状并发请求时一个请求写入的对话历史出现在另一个请求的上下文里历史累积到超限时_truncate_messages第五篇会静默截断用户侧表现为 Agent「忘记」早期对话。整个过程没有异常抛出。原因第五篇在Agent里用self.messages维护对话历史。若为了让 API「记住上下文」把Agent实例化成模块级变量并让所有请求共用历史就会跨请求累积。Depends本身每次请求都会调用依赖函数但函数返回的是同一个模块级对象注入的仍是同一实例依据FastAPI 官方文档 Dependencies。复现新建文件agent-demo/api_shared.py写入以下完整内容fromfastapiimportDepends,FastAPIfrompydanticimportBaseModelfrommainimportAgent,run_agent appFastAPI()shared_agentAgent()defget_agent()-Agent:returnshared_agentclassChatRequest(BaseModel):message:strclassChatResponse(BaseModel):reply:strapp.post(/chat,response_modelChatResponse)defchat(request:ChatRequest,agent:AgentDepends(get_agent))-ChatResponse:replyrun_agent(request.message)agent.messages.append({role:user,content:request.message})agent.messages.append({role:assistant,content:reply})returnChatResponse(replyreply)app.get(/history)defhistory(agent:AgentDepends(get_agent))-dict:return{messages:agent.messages}操作用uvicorn api_shared:app --reload启动另开终端连续发两次请求再看历史接口curl-XPOST http://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{message: 用户甲的第一条}curl-XPOST http://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{message: 用户乙的第一条}curlhttp://127.0.0.1:8000/history预期输出/history返回的messages里混着两次请求写入的全部消息用户甲的内容出现在用户乙的会话记录中。实际输出未实测串扰直接体现在/history的messages中若生成回复时读取同一实例的self.messages他人对话还会进入回复正文此处机制为推测。修复让每个请求拿到独立实例只改get_agent这一个函数defget_agent()-Agent:returnAgent()操作重启服务后再执行上一条curl [/history](/history)。预期输出messages只包含本次请求写入的消息。实际输出未实测修复后/history只含本次请求写入的消息不再出现其他请求的内容。代价要说清修复后跨请求不再有对话连续性要维持会话需要按会话标识隔离历史并引入额外存储超出本篇范围。失败 3执行目录不对导致导入失败报错原文在agent-demo/之外执行python -c import apiModuleNotFoundError: No module named main原因api.py用from main import run_agent导入同目录模块Python 只在当前工作目录与sys.path中查找main执行位置不在agent-demo/时就找不到mainuvicorn 侧则表现为找不到api模块。修复先切到agent-demo/再执行任何命令。操作cdagent-demo python-cimport api; print(ok)预期输出ok实际输出未实测成功时仅回显ok。四、替代方案与取舍两个决定点各有两种做法。先看Agent实例的管理方式维度方案 A单例模块级共享方案 B每请求新建本篇默认适用条件本地开发、单用户调试多用户服务、面向外部程序调用代价对话状态跨请求累积不同用户互相污染每次请求重新初始化Agenttiktoken编码等固定开销重复支付边界并发一超过单用户就出现串扰不能用于多用户没有跨请求对话连续性要维持会话需额外存储与会话管理超出本篇范围单例只有在一个人调试时才成立一旦有两个调用方它的边界就到了。每请求新建把状态彻底隔离代价是放弃跨请求记忆这也是第二节api.py不持有任何Agent实例的原因。再看路由处理函数的声明方式维度方案 A同步def本篇默认方案 B异步async defasyncio.to_thread适用条件run_agent是同步函数直接调用最省事需要更高并发愿意多写一层异步包装代价并发能力受 FastAPI 默认线程池大小限制多一次线程调度代码与调试都更复杂边界低流量、内部工具够用在async def里直接调用同步run_agent会阻塞事件循环延迟反而更高方案 B 的完整实现如下独立文件agent-demo/api_async.pyimportasynciofromfastapiimportFastAPIfrompydanticimportBaseModelfrommainimportrun_agentclassChatRequest(BaseModel):message:strclassChatResponse(BaseModel):reply:strappFastAPI()app.post(/chat,response_modelChatResponse)asyncdefchat(request:ChatRequest)-ChatResponse:replyawaitasyncio.to_thread(run_agent,request.message)returnChatResponse(replyreply)操作在agent-demo/目录下用uvicorn api_async:app --reload启动再执行步骤 4 的那条curl。预期输出与步骤 4 相同的 JSON 结构。实际输出未实测响应体结构与步骤 4 一致。若你的run_agent需要显式接收Agent实例把Depends(get_agent)每请求新建加进路由参数同步版写成run_agent(agent, request.message)异步版写成await asyncio.to_thread(run_agent, agent, request.message)。以下情况不该用本篇方案需要跨请求的多轮记忆/chat每次请求都是新实例别用它做必须记住上文的产品先补会话存储再谈接口。直接暴露公网本篇没有鉴权与限流不应作为公网服务运行。生产部署--reload是开发期的自动重载多进程部署与运维不在本篇范围。下一步要解决的问题是在这个 API 服务上增加流式输出让用户逐步看到 Agent 的推理与回复过程。参考资料n8n-io/n8nSignificant-Gravitas/AutoGPThuggingface/transformers