企业官网建设流程全解析

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把实验性（experimental）Agent 变成可被前端消费的 HTTP 后端，以及 Agent 的快照（getSnapshot）与中止（abort）端点是如何在源码层面自动生成的。

一、testapp 是什么：一个「即插即用」的 Python Agent 后端

testapp的定位写得很直白（见 testapp/README.md）：

A 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.

翻译成三个关键设计点：

1. 单进程多 Agent：一个 FastAPI 应用实例承载本目录下的全部 Agent，路由路径与 Agent 名一一对应；
2. 会话存储默认本地化：会话快照存放在内存或磁盘文件里，本地开发只需要GEMINI_API_KEY一个环境变量；
3. 生产存储可替换：正式部署时，把会话存储换成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 确认）：

• Python>=3.10；
• 依赖：genkit、genkit-google-genai、genkit-middleware、genkit-fastapi、fastapi>=0.100.0、uvicorn[standard]>=0.24.0、httpx>=0.27.0、pydantic>=2.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.md）：

• Dev UI：http://localhost:4000（Genkit 开发者界面，可点选运行各个 flow、观察流式输出）；
• HTTP API：http://localhost:8080（server.py 末尾以uvicorn.run(app, host='127.0.0.1', port=8080)启动）。

三、server.py 逐段解析：挂载列表如何生成

3.1 共享 Genkit 实例：_ai.py

所有 Agent 文件都从一个共享实例注册自己，见 _ai.py：

from 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()], model=DEFAULT_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(title='Genkit 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，分别对应独立的示例文件：

两个值得注意的实现细节：

• 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

除了 Agent，server 还挂载了两个"普通 flow"，服务于编码 Agent 的文件浏览器 UI：

app.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_dependency=context_dependency) if agent.store is not None: # 注册 {agent.name}_snapshot 与 {agent.name}_abort 两个 Action # 挂载到 {base}/getSnapshot 与 {base}/abort

可以确认的行为：

1. turn 路由：Agent 本身作为 Action 挂载在/{agent.name}（testapp 中经prefix='/api'后即为/api/<name>）；
2. 快照与中止是条件性的：只有当agent.store is not None（即 Agent 有 server 端会话存储）时，才会额外生成{agent.name}_snapshot和{agent.name}_abort两个 Action，分别挂载到/api/<name>/getSnapshot与/api/<name>/abort；
3. context_dependency：允许传入一个 FastAPI 依赖，其解析值会作为 Action 上下文，应用到 turn、getSnapshot 与 abort 三条路由上——这是复用现有Depends鉴权 / 资源注入机制的扩展点；
4. 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( name='weatherAgent', system='You are an assistant helping with weather information. Use the getWeather tool.', tools=[get_weather], store=FileSessionStore('./.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 的说明，当前仓库给出的存储演进路径是：

切换方式是"同一插槽替换"：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(name='delegate_to_researcher', description='Hand 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( name='codingAgent', ... use=[ ToolApproval(allowed_tools=['list_files', 'read_file']), # 只读工具自动放行 Filesystem(root_dir=str(WORKSPACE_DIR), allow_write_access=True), # 沙箱文件工具 ], store=FileSessionStore('./.snapshots-coding'), max_turns=30, )

顺序是关键：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.py，Dev UI 在:4000，HTTP 在:8080，仅需GEMINI_API_KEY；
• 上线路径：将store=从InMemorySessionStore/FileSessionStore替换为genkit_google_cloud.exp的FirestoreSessionStore()，前端路径与路由保持不变。

适用前提与限制：Agents 属于genkit.exp实验性 API；testapp 依赖 Gemini 模型（GEMINI_API_KEY）；CORS 配置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