中国中建设计网站成都网站建设 小兵cms

中国中建设计网站,成都网站建设 小兵cms,网站建设人员招聘要求,网站风险解除AI - 用 FastAPI 暴露你的第一个 Google ADK Agent1. 起点#xff1a;我们已经有了一个最简单的 Agent2. 目标架构#xff1a;把 Agent 变成一个 HTTP 服务3. 新增一个 api.py#xff1a;FastAPI 入口 ADK 运行时3.1 安装 FastAPI Uvicorn python-dotenv#xff08…AI - 用 FastAPI 暴露你的第一个 Google ADK Agent1. 起点我们已经有了一个最简单的 Agent2. 目标架构把 Agent 变成一个 HTTP 服务3. 新增一个 api.pyFastAPI 入口 ADK 运行时3.1 安装 FastAPI Uvicorn python-dotenv可选3.2 会话相关InMemorySessionService Runner4. 完整的 api.py 源码带 session 支持5. 逐段拆解一下关键技术点5.1 SessionService为什么要 user_id session_id5.2 Runner为什么不用直接 root_agent.run()5.3 FastAPIPydantic 模型 JSON Body6. 如何跑起来 测试6.1 启动服务6.2 创建会话6.3 在会话里聊天6.4 通过 Swagger UI 测试7. 小结在上一篇《 AI - 使用 Google ADK 创建你的第一个 AI Agent》我们已经用 Google ADKAgent Development Kit写了一个最简单的 Agent会用 Gemini 模型能调用一个工具函数回答“某个城市现在几点”。这一篇我们来做一件现实世界里更有用的事情把这个 Agent 用 FastAPI 暴露成 HTTP API支持多用户、多会话user_id session_id让前端、移动端、其他服务都可以通过 REST 调用它。过程中会顺带拆一下几个关键技术点FastAPI Uvicorn 是怎么跑起来的ADK 里的 SessionService / Runner 是干嘛的user_id session_id 在 ADK 中是怎么维护对话上下文的异步 run_async 和流式事件到底是什么1. 起点我们已经有了一个最简单的 Agent先快速回顾一下我们的 agent.py# my_agent/agent.pyfromgoogle.adk.agents.llm_agentimportAgent# 一个非常简单的工具函数返回某个城市的时间写死defget_current_time(city:str)-dict:Returns the current time in a specified city.return{status:success,city:city,time:10:30 AM}# 根 AgentADK 约定必须叫 root_agentroot_agentAgent(modelgemini-3-pro-preview,# 或其他 Gemini 模型nameroot_agent,descriptionTells the current time in a specified city.,instructionYou are a helpful assistant that tells the current time in cities. Use the get_current_time tool for this purpose.,tools[get_current_time],)项目结构大概是这样:my_agent/ agent.py # main agent code __init__.py .env # API keys or project IDs现在它可以用 adk run my_agent 或 adk web 跑起来但只能在本机 CLI / Dev UI 里玩对外没有 HTTP 接口。接下来我们就基于这个项目加一个 api.py用 FastAPI 暴露出来。2. 目标架构把 Agent 变成一个 HTTP 服务我们想要的是这样一套能力POST /session传入 user_id创建一个新的对话会话返回 session_idPOST /chat传入 user_id session_id messageAgent 在对应会话上继续对话并返回回复整体逻辑前端 / 其他服务 │ ├─ POST /session - 返回 session_id │ └─ POST /chat(user_id, session_id, message) │ ▼ FastAPI 路由 │ ▼ ADK Runner SessionService (负责会话 state) │ ▼ root_agent.run_async(...) │ ▼ Gemini tools其中FastAPIPython 高性能 Web 框架非常适合写 JSON API。UvicornASGI 服务器用来实际跑 FastAPI 应用。SessionServiceADK 中管理会话Session的组件提供创建 / 获取 / 列表 / 删除等能力。RunnerADK 的“执行引擎”负责拿到 Session、调用 Agent、处理生成的 Event 并写回 Session。3. 新增一个 api.pyFastAPI 入口 ADK 运行时我们计划把 api.py 放到 my_agent/ 包里所以最终目录变成这样my_agent/ ├─ agent.py # 定义 root_agent ├─ api.py # ✅ 新增FastAPI Runner Session └─ __init__.py3.1 安装 FastAPI Uvicorn python-dotenv可选在你的虚拟环境里安装pipinstallfastapi uvicorn[standard]python-dotenvFastAPI写 API 路由用 Pydantic 定义请求 / 响应模型Uvicorn跑 ASGI 服务器python-dotenv方便从 .env 自动加载 GOOGLE_API_KEY 等环境变量3.2 会话相关InMemorySessionService Runner在 ADK 里一个 Session 就代表一段连续的对话里面包含idsession_iduser_id对应用户events所有历史对话 / tool 调用等事件state当前会话的状态字典短期记忆Session 的生命周期由 SessionService 管理例如InMemorySessionService存在内存里应用重启就没了很适合本地开发 / demoDatabaseSessionService持久化到数据库适合生产环境VertexAiSessionService把 Session 存到 Vertex AI 的 Agent Engine 里而 Runner 则负责根据 user_id session_id 找到对应 Session把新一轮 new_message 加到 Session 的事件里调用 root_agent.run_async(…)接收流式 Event 并处理把新的事件和状态写回 Session4. 完整的 api.py 源码带 session 支持下面是一份可以直接用的 my_agent/api.py你可以一次性拷贝过去# my_agent/api.py 用 FastAPI 暴露基于 Google ADK 的第一个 Agent。 - /session : 创建会话返回 session_id - /chat : 带 user_id session_id 发消息支持多轮对话 importosimportuvicornfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromdotenvimportload_dotenvfromgoogle.adk.sessionsimportInMemorySessionServicefromgoogle.adk.runnersimportRunnerfromgoogle.genaiimporttypesasgenai_types# 导入我们在 agent.py 里定义好的 root_agentfrom.agentimportroot_agent# 环境变量 # 加载同级或上级目录下的 .env包含 GOOGLE_API_KEY 等load_dotenv()ifnotos.getenv(GOOGLE_API_KEY):# 不强制报错只打印一个友好提示print([WARN] GOOGLE_API_KEY 未设置请在 .env 或环境变量中配置否则模型调用会失败。)# ADKSessionService Runner # 应用名ADK 用它来区分不同应用的会话空间APP_NAMEmy_first_adk_agent# 1) 会话服务这里用内存版适合本地开发 / demosession_serviceInMemorySessionService()# 2) Runner负责从 Session 取历史、调用 Agent、写回事件runnerRunner(app_nameAPP_NAME,agentroot_agent,session_servicesession_service,)# FastAPI 初始化 appFastAPI(titleMy First ADK Agent API,description基于 Google ADK FastAPI 的简单对话 Agent支持 user_id session_id 会话。,)# 请求 / 响应模型 classCreateSessionRequest(BaseModel):创建会话请求只需要 user_id。user_id:strclassCreateSessionResponse(BaseModel):user_id:strsession_id:strclassChatRequest(BaseModel):对话请求必须带 user_id session_id message。user_id:strsession_id:strmessage:strclassChatResponse(BaseModel):user_id:strsession_id:strreply:str# 路由创建会话 app.post(/session,response_modelCreateSessionResponse)asyncdefcreate_session(req:CreateSessionRequest): 创建一个新的 Session - 输入: user_id - 输出: 自动生成的 session_id # 注意SessionService 在新版本 ADK 里是异步的需要 await。sessionawaitsession_service.create_session(app_nameAPP_NAME,user_idreq.user_id,state{},# 初始 state你也可以在这里塞一些默认值)returnCreateSessionResponse(user_idreq.user_id,session_idsession.id,)# 路由在指定会话里聊天 app.post(/chat,response_modelChatResponse)asyncdefchat(req:ChatRequest): 使用 user_id session_id message 调用 Runner 在对应的 Session 上继续对话并返回模型回复。 # 把用户输入包装成 google-genai 的 Content/Part 结构# Runner.run_async 需要的就是这个类型。user_contentgenai_types.Content(roleuser,parts[genai_types.Part.from_text(textreq.message)],)reply_chunks:list[str][]try:# Runner.run_async 是一个异步生成器会不断 yield Event。asyncforeventinrunner.run_async(user_idreq.user_id,session_idreq.session_id,new_messageuser_content,):# 把事件中的文本部分收集起来ifevent.contentandevent.content.parts:forpartinevent.content.parts:ifgetattr(part,text,None):reply_chunks.append(part.text)# is_final_response() 为 True 时说明这是最终回复可以结束本轮ifevent.is_final_response():breakexceptExceptionase:msgstr(e)# 一个常见错误是 session_id 不存在ifsessioninmsg.lower()andnot foundinmsg.lower():raiseHTTPException(status_code404,detail会话不存在请先调用 /session 创建会话。,)raiseHTTPException(status_code500,detailfAgent 调用失败:{msg},)reply_text.join(reply_chunks).strip()orAgent 没有返回任何文本内容returnChatResponse(user_idreq.user_id,session_idreq.session_id,replyreply_text,)# 本地启动可选 if__name____main__:# 在项目根目录运行python -m my_agent.apiuvicorn.run(my_agent.api:app,host0.0.0.0,port8001,reloadTrue)5. 逐段拆解一下关键技术点5.1 SessionService为什么要 user_id session_idADK 把一段连续对话抽象成一个 Session 对象由 SessionService 统一管理sessionawaitsession_service.create_session(app_nameAPP_NAME,user_idreq.user_id,state{})app_name区分是哪个应用的会话你完全可以在一个 Python 进程里跑多个 Agent。user_id谁的会话。session.idADK 生成的唯一 session_id。在后续每一轮对话里我们都要显式带上这两个维度asyncforeventinrunner.run_async(user_idreq.user_id,session_idreq.session_id,new_messageuser_content,):...这相当于告诉 Runner“请在 APP_NAME 这个应用里找到 user_id X、session_id Y 的那条会话然后在它的历史基础上继续处理这条新消息。”ADK 会帮你自动从 SessionService 读出该 Session 的 events 和 state把新消息加入事件链把 Agent 产生的新事件 / state 变化写回该 Session所以user_id 让你可以区分不同用户session_id 让你可以区分同一用户的多段对话多窗口 / 多任务如果以后要做“跨会话的长期记忆”可以再加上 MemoryService比如 InMemoryMemoryService / VertexAiMemoryBankService那就是另外一层了。5.2 Runner为什么不用直接 root_agent.run()理论上你也可以在 FastAPI 里直接调用awaitroot_agent.run(...)但官方更推荐用 Runner 来跑 Agent因为 Runner 额外做了很多底层活创建 / 获取 Session把新消息写入 Session 的事件链处理 Agent 生成的每个 Event包括工具调用状态变化state_delta产出的内容content把这些变化提交给 SessionService、MemoryService 等也就是说Runner Agent 的“执行引擎 会话协调器”对你来说Runner 有两个重要特性run_async(…) 是异步的适合 FastAPI 等 async 框架返回的是一个 异步生成器AsyncGenerator[Event]可以边生成边消费天然契合“流式输出”的场景在我们的 /chat 接口里就是用一个 async for 来消费这些 Eventasyncforeventinrunner.run_async(...):ifevent.contentandevent.content.parts:...ifevent.is_final_response():break现在我们只是把所有文本 part 拼起来做一个简单的“整段回复”。但你以后完全可以玩得更高级逐块推给前端做“打字机效果”根据不同类型的 Event 做不同 UI例如工具调用 / 代码执行5.3 FastAPIPydantic 模型 JSON BodyFastAPI 的一个特点是用 Pydantic 模型来声明请求体 / 响应体自动帮你做验证和文档生成。比如我们定义classChatRequest(BaseModel):user_id:strsession_id:strmessage:str然后在路由里写app.post(/chat,response_modelChatResponse)asyncdefchat(req:ChatRequest):...FastAPI 会自动把 JSON 请求体解析成 ChatRequest 对象在 /docs 里生成 Swagger UI 文档检查字段类型不符合直接 422 返回这一点和 ADK 本身的“结构化输出”output_schema非常契合——前端 / 调用方看的是 FastAPI 的 schemaAgent 内部看的是 ADK 自己的 state / schema中间用 Runner 做桥接。6. 如何跑起来 测试6.1 启动服务在项目根目录包含 my_agent/ 那层执行uvicorn my_agent.api:app --reload --port8001或直接运行 api.py 文件看到类似日志说明启动成功INFO: Uvicorn running on http://0.0.0.0:8001(Press CTRLC to quit)6.2 创建会话第一次聊天前先创建一个 Sessioncurl-X POSThttp://127.0.0.1:8000/session\-HContent-Type: application/json\-d{user_id: user_123}返回示例{user_id:user_123,session_id:c2a4a3d8-2a4e-4c2c-87a0-xxxxxx}前端要把这个 session_id 存起来例如存在 localStorage / cookie / 内存。6.3 在会话里聊天后续每一轮对话都用这个 user_id session_idcurl-X POSThttp://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{ user_id: user_123, session_id: c2a4a3d8-2a4e-4c2c-87a0-xxxxxx, message: 现在东京几点 }你会收到类似{user_id:user_123,session_id:c2a4a3d8-2a4e-4c2c-87a0-xxxxxx,reply:东京现在是 10:30 AM。}再发第二条{user_id:user_123,session_id:c2a4a3d8-2a4e-4c2c-87a0-xxxxxx,message:那和上海差多少}ADK 会自动在同一条 Session 里拼接上下文让 Agent 有“短期记忆”。{user_id:user_123,session_id:c2a4a3d8-2a4e-4c2c-87a0-xxxxxx,reply:东京和上海的当前时间都是上午10:30所以它们之间没有时差。}6.4 通过 Swagger UI 测试可以访问 http://localhost:8001/docs 通过 UI 来进行会话测试。7. 小结我们这篇文章做了几件实打实的事在已有的 root_agent 基础上新增 api.py用 FastAPI Uvicorn 暴露 HTTP API用 InMemorySessionService Runner 支持 user_id session_id 的对话会话拆解了 Session / Runner / FastAPI / GenAI Content 等关键技术点一句话总结就是用 ADK 写 Agent用 FastAPI 做壳用 SessionService Runner 管会话你就从“玩具 demo”迈进了“真正可被任何服务调用的 Agent 服务”。