尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Genkit Python Agent 后端实战:testapp 如何用 FastAPI 挂载一整套 Agent 服务

Genkit Python Agent 后端实战:testapp 如何用 FastAPI 挂载一整套 Agent 服务 Genkit Python Agent 后端实战testapp 如何用 FastAPI 挂载一整套 Agent 服务【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit本文围绕 Genkit Python 仓库中的py/samples/agents/testapp/示例展开它用一个 FastAPI 进程把同一目录下的所有 Agent 统一挂载到/api/name路径下会话状态存放在内存或磁盘文件中仅凭GEMINI_API_KEY即可在本地跑通部署时再平滑切换到 Firestore 会话存储。读完后你将掌握如何基于genkit_fastapi的serve_agent/serve_flow把实验性experimentalAgent 变成可被前端消费的 HTTP 后端以及 Agent 的快照getSnapshot与中止abort端点是如何在源码层面自动生成的。一、testapp 是什么一个「即插即用」的 Python Agent 后端testapp的定位写得很直白见 testapp/README.mdA FastAPI process with the agents in this folder mounted at/api/name. Sessions stay in memory or on disk so it runs onGEMINI_API_KEYtonight. Swap inFirestoreSessionStore()fromgenkit_google_cloud.expwhen you deploy.翻译成三个关键设计点单进程多 Agent一个 FastAPI 应用实例承载本目录下的全部 Agent路由路径与 Agent 名一一对应会话存储默认本地化会话快照存放在内存或磁盘文件里本地开发只需要GEMINI_API_KEY一个环境变量生产存储可替换正式部署时把会话存储换成genkit_google_cloud.exp中的FirestoreSessionStore()业务代码与前端路径均不需要改动。server.py 的模块 docstring 进一步解释了动机一次include_router就能拿到对话turn路由外加/getSnapshot和/abort两个端点配合prefix/api挂载后一个小团队可以直接把它当作 Python Agent 后端替换进现有系统而不必改动前端路径。它与 py/samples/agents 总目录 的关系是basic/目录下的编号文件每个演示一个独立概念从 01 开始逐个看而testapp/则是把这些 Agent 聚合成一个可对外提供服务的进程。Agents 目前仍是实验性 API需要从genkit.exp导入from genkit.exp import Genkit, InMemorySessionStore二、快速启动三条命令跑起来前置条件由 py/samples/agents/pyproject.toml 确认Python3.10依赖genkit、genkit-google-genai、genkit-middleware、genkit-fastapi、fastapi0.100.0、uvicorn[standard]0.24.0、httpx0.27.0、pydantic2.10.5环境变量GEMINI_API_KEY示例使用 Google AI 的 Gemini 模型。注意该项目的[tool.uv]配置中package false——注释明确说明这些是可直接运行的脚本不是可导入的库uv run只需要解析依赖无需构建安装cd py/samples/agents uv sync genkit start -- uv run testapp/server.py启动后有两个地址来自 testapp/README.mdDev UIhttp://localhost:4000Genkit 开发者界面可点选运行各个 flow、观察流式输出HTTP APIhttp://localhost:8080server.py 末尾以uvicorn.run(app, host127.0.0.1, port8080)启动。三、server.py 逐段解析挂载列表如何生成3.1 共享 Genkit 实例_ai.py所有 Agent 文件都从一个共享实例注册自己见 _ai.pyfrom genkit_google_genai import GoogleAI from genkit_middleware import Middleware from genkit.exp import Genkit DEFAULT_MODEL GoogleAI.gemini_model(gemini-flash-latest) LITE_MODEL GoogleAI.gemini_model(gemini-flash-lite-latest) # Middleware 插件提供 Artifacts() 与 ToolApproval() 行为 ai Genkit(plugins[GoogleAI(), Middleware()], modelDEFAULT_MODEL)要点GoogleAI()插件提供 Gemini 模型接入gemini-flash-latest为默认模型gemini-flash-lite-latest供轻量子步骤使用如任务拆解、安全检查Middleware()插件支撑Artifacts()工件与ToolApproval()工具审批行为workspace 和 banking 两个 Agent 依赖它由于导入 Agent 模块即在该ai上完成注册所以无论是 Dev UI 单文件运行还是 FastAPI 服务导入全部文件看到的都是同一份注册表。这也是为什么 server.py 顶部那一串from background_agent import ...导入语句本身就是挂载列表的一部分——注释说得很清楚列在这里正是让它们出现在 Dev UI 中的原因。3.2 Agent 挂载循环app FastAPI(titleGenkit Agents (Python)) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], expose_headers[X-Genkit-Stream-Id], ) for agent in ( weather_agent, weather_agent_stateless, file_store_agent, research_agent, task_agent, banking_agent, workspace_agent, background_agent, branching_agent, orchestrator_agent, trip_planner_agent, coding_agent, ): app.include_router(serve_agent(agent), prefix/api)从源码结构看这个循环覆盖了 12 个 Agent分别对应独立的示例文件Agent 模块演示主题由各文件 docstring 概括weather_agent.py入门 Agent一个工具 文件会话存储 流式多轮对话client_state_agent.py无状态无 server 端 store变体file_store_agent.py文件存储会话research_agent.py研究型 Agenttask_agent.py任务型 Agentbanking_agent.py依赖 Artifacts / ToolApproval 的银行场景workspace_agent.py工件Artifacts工作区background_agent.py后台 Agentbranching_agent.py分支时间旅行/分支快照orchestrator_agent.py多 Agent 编排委托给专家子 Agenttrip_planner_agent.py行程规划coding_agent.py沙箱编码助手Filesystem 中间件 写操作审批两个值得注意的实现细节CORS 中的expose_headers[X-Genkit-Stream-Id]server.py 注释说明前端与后端是不同源必须放行跨域并暴露流式响应头X-Genkit-Stream-Id——客户端靠它把分块chunks关联回同一次请求。路由与 Agent 名对齐每个 Agent 的路径来自它自己的名字因此/api/agentName与前端调用方约定一致。3.3 纯 flow 的挂载serve_flow与固定base_path除了 Agentserver 还挂载了两个普通 flow服务于编码 Agent 的文件浏览器 UIapp.include_router(serve_flow(list_workspace_files, base_path/workspace/files), prefix/api) app.include_router(serve_flow(read_workspace_file, base_path/workspace/file), prefix/api)注释解释了固定base_path的原因文件浏览器访问这两个路径固定 URL 可以保持稳定而不是跟随 flow 名变化。list_workspace_files与read_workspace_file定义在 coding_agent.py 中。四、源码层证据serve_agent如何自动生成 turn / getSnapshot / abort 三个端点serve_agent与serve_flow的实现在 genkit-fastapi 的 handler.py。从源码可以看到def serve_agent( agent: Agent[StateT], *, base_path: str | None None, context_dependency: Callable[..., Any] | None None, ) - APIRouter: resolved_base_path f/{agent.name} if base_path is None else base_path router APIRouter(tags[agent.name]) _mount_action(router, resolved_base_path, agent, context_dependencycontext_dependency) if agent.store is not None: # 注册 {agent.name}_snapshot 与 {agent.name}_abort 两个 Action # 挂载到 {base}/getSnapshot 与 {base}/abort可以确认的行为turn 路由Agent 本身作为 Action 挂载在/{agent.name}testapp 中经prefix/api后即为/api/name快照与中止是条件性的只有当agent.store is not None即 Agent 有 server 端会话存储时才会额外生成{agent.name}_snapshot和{agent.name}_abort两个 Action分别挂载到/api/name/getSnapshot与/api/name/abortcontext_dependency允许传入一个 FastAPI 依赖其解析值会作为 Action 上下文应用到 turn、getSnapshot 与 abort 三条路由上——这是复用现有Depends鉴权 / 资源注入机制的扩展点serve_flow同理base_path缺省为/{flow.name}tags取 flow 名最终同样由_mount_action完成路由注册。这与 server.py docstring 中一次 include_router 即获得 turn 路由外加 /getSnapshot 与 /abort的描述完全对应。4.1 为什么「有 store」才需要这两个端点以 weather_agent.py 为例weather_agent ai.define_agent( nameweatherAgent, systemYou are an assistant helping with weather information. Use the getWeather tool., tools[get_weather], storeFileSessionStore(./.snapshots), )文件中的注释解释了 store 的语义store 让 Agent 成为 server 托管式历史在磁盘上客户端只需持有 session id 即可续接对话无需把状态在线路上来回搬运。FileSessionStore来自genkit.exp.agent快照落到本地./.snapshots目录coding_agent 则用./.snapshots-coding。而 client_state_agent.py 提供的是无 store 的对照形态——这类 Agent 只暴露 turn 路由不产生 getSnapshot/abort 端点。五、会话存储的三档选择内存、文件、Firestore结合 py/samples/agents 总 README 与 testapp README 的说明当前仓库给出的存储演进路径是场景存储导入位置本地示例只需GEMINI_API_KEYInMemorySessionStore/FileSessionStoregenkit.exp/genkit.exp.agenttestapp 文件存储FileSessionStore(./.snapshots)genkit.exp.agent生产部署FirestoreSessionStore()genkit_google_cloud.exp切换方式是同一插槽替换ai.define_agent(..., store...)中把本地存储换成FirestoreSessionStore()即可路由、Dev UI、前端调用路径都不受影响。这也是 testapp README 中tonight 本地能跑、部署时再换承诺的技术基础。六、两个典型 Agent 的实现模式6.1 流式多轮weather_agent 的 flow 写法weather_agent.py 的test_weather_agent_streamflow 展示了标准的多轮驱动方式chat weather_agent.chat() turn chat.send_stream(text or What is the weather like in Paris?) async for chunk in turn: if chunk.text: ctx.send_chunk(chunk.text) await turn followup chat.send_stream(now say that in French) async for chunk in followup: ...同一个chat对象承载跨轮次历史后续轮次自动知道上下文每个chunk.tool_requests会被ctx.send_chunk(f[tool] {name})提示工具调用过程因此在 Dev UI 中可见。6.2 多 Agent 编排orchestrator_agent 的委托工具orchestrator_agent.py 展示了纯工具 对话的多 Agent 组合方式Agent 本身就是可以chat()的对象因此一个委托工具只需对子 Agent 跑一轮对话并把答案作为工具结果返回ai.tool(namedelegate_to_researcher, descriptionHand a research question to the researcher specialist.) async def delegate_to_researcher(input: Task) - str: return (await researcher.chat().send(input.task)).text编排者 Agent 的系统提示词要求它分析请求并委托研究用 delegate_to_researcher代码用 delegate_to_coder两者都需要时依次调用最后综合专家结果给出最终答案。文件注释将其概括为仅靠工具和对话实现的多 Agent 组合。6.3 受控写操作coding_agent 的中间件组合coding_agent.py 展示了Middleware插件的两个行为如何叠加coding_agent ai.define_agent( namecodingAgent, ... use[ ToolApproval(allowed_tools[list_files, read_file]), # 只读工具自动放行 Filesystem(root_dirstr(WORKSPACE_DIR), allow_write_accessTrue), # 沙箱文件工具 ], storeFileSessionStore(./.snapshots-coding), max_turns30, )顺序是关键ToolApproval必须在Filesystem之前这样才能在文件工具执行前拦截到写操作、暂停等待人工批准只读的list_files/read_file自动放行。配套的test_coding_agentflow 则用循环自动批准所有挂起的中断i.restart(resumed_metadata{tool_approved: True})让 Agent 无人值守跑完任务。七、小结这套模式如何落到自己的项目入口文件server.py——导入 Agent → 循环serve_agent(agent)挂载 →serve_flow挂辅助 flow → uvicorn 起 :8080共享注册表_ai.py 中单个Genkit(plugins[GoogleAI(), Middleware()])实例导入即注册端点生成serve_agent自动派生/{name}、/{name}/getSnapshot、/{name}/abort后两者以存在 store 为前提实现见 handler.py运行方式cd py/samples/agents uv sync genkit start -- uv run testapp/server.pyDev UI 在:4000HTTP 在:8080仅需GEMINI_API_KEY上线路径将store从InMemorySessionStore/FileSessionStore替换为genkit_google_cloud.exp的FirestoreSessionStore()前端路径与路由保持不变。适用前提与限制Agents 属于genkit.exp实验性 APItestapp 依赖 Gemini 模型GEMINI_API_KEYCORS 配置allow_origins[*]面向本地跨源前端开发场景生产部署时建议按 handler.py 中context_dependency的扩展点接入真实的鉴权依赖。【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表