企业官网建设流程全解析

Genkit Python SDK 实战：一套 API 打通模型生成、工具调用、结构化输出与 Agents，并内置本地 Developer UI

【免费下载链接】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 是 Google 开源的 AI 应用框架，其 Python SDK（py/packages/genkit）把模型生成（generate）、工具（tools）、结构化输出（structured output）和 Agents 收敛到同一套 API 之下，并自带一个本地 Developer UI 用于调试与观测。本文以 genkit Python 包的 README 为主体，逐行拆解其中的安装方式与完整示例，并结合仓库源码说明Genkit类、@ai.flow()、ai.generate()与run_main()背后的实际实现机制，以及 Dev UI 反射服务是如何被自动拉起的。

一、Genkit Python 是什么

README 对包本身的定位非常凝练：

Genkit is a Python SDK from Google. One API for generate, tools, structured output, and agents, plus a local Developer UI.

Vertex AI, Cloud Trace, and Firestore are there if you want them. So are OpenAI, Anthropic, Ollama, and Bedrock.

也就是说，它提供两件事：

1. 一个统一入口：生成文本/结构化数据、调用工具、定义与运行 flow（工作流）、构建 Agent，都通过同一个Genkit实例上的方法完成，而不用针对每个模型供应商写不同客户端代码；
2. 可插拔的模型与基础设施：Gemini（Google AI）、Vertex AI、OpenAI、Anthropic、Ollama、Amazon Bedrock 均以插件形式接入，Cloud Trace、Firestore 等可选组件按需启用。

包元信息可以从 pyproject.toml 中得到更精确的约束，这也是复现本文示例的适用前提：

其中pydantic>=2这一点很重要：后文示例中的结构化输出完全建立在 Pydantic v2 的BaseModel之上；opentelemetry-*与uvicorn/starlette/sse-starlette依赖则解释了 SDK 内建的 OpenTelemetry 追踪能力与 Dev UI 反射服务的运行底座。

二、安装

README 给出的安装命令使用uv（Python 包管理器），一条命令同时装入核心包和 Google 模型插件：

uv add genkit genkit-google-genai

• genkit：核心 SDK，本仓库 py/packages/genkit 对应的 PyPI 包；
• genkit-google-genai：Gemini 模型插件，对应本仓库 py/packages/genkit-google-genai。

如果你不用 Gemini 而是用其他供应商，只需把第二个包换成对应插件。这一点在 pyproject.toml 的可选依赖（[project.optional-dependencies]）中有完整清单，每一个 extra 都映射到py/packages/下的一个插件包：

[project.optional-dependencies] a2ui = ["genkit-a2ui"] amazon-bedrock = ["genkit-amazon-bedrock"] anthropic = ["genkit-anthropic"] django = ["genkit-django"] evaluators = ["genkit-evaluators"] fastapi = ["genkit-fastapi"] flask = ["genkit-flask"] google-cloud = ["genkit-google-cloud"] google-genai = ["genkit-google-genai"] middleware = ["genkit-middleware"] ollama = ["genkit-ollama"] openai = ["genkit-openai"] vertex-ai = ["genkit-vertexai"]

例如安装 OpenAI 插件可写uv add "genkit[openai]"或直接uv add genkit genkit-openai（后者即 README 推荐写法）。

三、完整示例：结构化输出的代码审查 Flow

以下是 README 中的完整示例（原文未做删改），它演示了最核心的链路：创建 Genkit 实例 → 定义 Pydantic 输出模型 → 用@ai.flow()注册 flow →ai.generate()以output_schema约束结构化输出 →run_main()启动。

from pydantic import BaseModel, Field from genkit import Genkit from genkit_google_genai import GoogleAI ai = Genkit(plugins=[GoogleAI()], model=GoogleAI.gemini_model('gemini-flash-latest')) class Issue(BaseModel): title: str = Field(description='Short title') severity: str = Field(description='critical, warning, or info') suggestion: str = Field(description='How to fix it') @ai.flow() async def review(code: str) -> Issue: result = await ai.generate( prompt=f'Review this code:\n{code}', output_schema=Issue, ) return result.output async def main() -> None: print((await review('eval(user_input)')).model_dump_json(indent=2)) if __name__ == '__main__': ai.run_main(main())

3.1 逐段解读

（1）创建实例并注册插件与默认模型

ai = Genkit(plugins=[GoogleAI()], model=GoogleAI.gemini_model('gemini-flash-latest'))

• plugins=[GoogleAI()]：把 Google 插件挂到实例上。从源码看，Genkit.init会调用_initialize_registry，逐个把插件注册进内部Registry（见 py/packages/genkit/src/genkit/_ai/_aio.py#L931-L945），同时把内置输出格式（text、json、jsonl 等）注册为 format；
• model=...：把该引用注册为defaultModel（见 py/packages/genkit/src/genkit/_ai/_aio.py#L933-L934），因此ai.generate(...)不传model参数时也会用它；
• GoogleAI.gemini_model('gemini-flash-latest')：返回一个类型化的ModelRef[GeminiConfigSchema]。其实现见 google.py——注释中特别说明：未知模型 id 也会被允许（保证新发布的 Gemini 模型在插件尚未收录时可用），但会拒绝其他模型家族的 id，防止把gemma-…之类的名字错配到 Gemini 配置 schema 上。

（2）用 Pydantic 定义输出结构

class Issue(BaseModel): title: str = Field(description='Short title') severity: str = Field(description='critical, warning, or info') suggestion: str = Field(description='How to fix it')

Field(description=...)的描述会进入生成 schema，成为给模型的字段级说明——这是 Pydantic v2 + Genkit 结构化输出的标准用法。severity若需要枚举约束，也可以改用Literal['critical', 'warning', 'info']让 schema 层直接收窄取值。

（3）@ai.flow()注册工作流

@ai.flow() async def review(code: str) -> Issue: result = await ai.generate( prompt=f'Review this code:\n{code}', output_schema=Issue, ) return result.output

• flow装饰器定义见 py/packages/genkit/src/genkit/_ai/_aio.py#L227-L260，支持name（默认取函数名）、description和chunk_type（提供后返回的 Action 会被类型化为Action[InputT, OutputT, ChunkT]，用于流式 chunk）三个参数；
• flow 本质是一个可被调用、可被 Dev UI 观测、可通过 HTTP 暴露的Action——公开 API 中Flow = Action（见 py/packages/genkit/src/genkit/init.py#L105-L106）；
• ai.generate(..., output_schema=Issue)的返回值类型是ModelResponse[Issue]：result.text是原始文本，result.output是已经反序列化并通过 Pydantic 校验的Issue实例，这也是示例中直接return result.output的原因。

（4）run_main()：开发模式下的入口

if __name__ == '__main__': ai.run_main(main())

run_main的实现在 py/packages/genkit/src/genkit/_ai/_aio.py#L955-L991，行为分两种：

• 非开发环境：等价于直接run_loop(coro)，跑完协程即退出；
• 开发环境（is_dev_environment()为真）：先 await 用户协程，然后打印Dev UI ready. Press Ctrl+C to stop.并阻塞等待 SIGINT/SIGTERM，保持后台的反射服务线程存活，从而让本地 Developer UI 能持续连上这个进程。

3.2 Dev UI 与反射服务是如何自动启动的

README 中 "plus a local Developer UI" 的说法对应的是_aio.py里的一段初始化逻辑：

if is_dev_environment(): setup_signal_handlers() self._start_reflection_background()

见 py/packages/genkit/src/genkit/_ai/_aio.py#L184-L193。_start_reflection_background（py/packages/genkit/src/genkit/_ai/_aio.py#L862-L929）在 daemon 线程中做几件事：

1. 检查环境变量GENKIT_REFLECTION_V2_SERVER：若 CLI 以 v2 模式启动运行时并提供了 WebSocket URL，则改走 v2 JSON-RPC 客户端（ReflectionServerV2）；
2. 否则创建 ASGI 反射应用create_reflection_asgi_app，用uvicorn绑定127.0.0.1上的一个随机空闲端口（bind(('127.0.0.1', 0))）；
3. 服务就绪后通过RuntimeManager.write_runtime_file()写一个运行时发现文件，Dev UI 据此找到本地进程并渲染 flow、生成请求与工具调用链路。

从源码结构看，这套设计意味着：只要你以脚本方式直接python xxx.py运行（而非嵌入 Web 框架），开发者 UI 就会自动可用，无需手写任何 HTTP 路由；而在生产环境中嵌入 FastAPI/Flask/Django 时，则可以借助 genkit-fastapi、genkit-flask、genkit-django 插件以受控方式挂载服务。

3.3 流式版本：generate_stream

示例用的是非流式的ai.generate()。同一实例上还有一对一的流式入口ai.generate_stream()（py/packages/genkit/src/genkit/_ai/_aio.py#L1306-L1386），用法与返回形态值得知道：

stream = ai.generate_stream(prompt='Write a haiku about rain.', output_schema=Issue) async for chunk in stream.stream: print(chunk.text) # 文本片段 # chunk.output 是 Issue 的"部分填充"实例：字段可能仍为 None 或前缀值 final = await stream.response # 完整的 ModelResponse[Issue] print(final.output)

其 docstring 明确提醒：带output_schema时，流中的chunk.output是目标类型的部分解析结果（partial），字段可能仍为None或前缀字符串，正式结果只应以(await sr.response).output为准。底层通过Channel把 chunk 推给消费端（见 py/packages/genkit/src/genkit/_ai/_aio.py#L1344-L1386），timeout参数用于控制 channel 超时。

四、generate的完整参数面

README 示例只用到了prompt与output_schema，但Genkit.generate的完整签名（py/packages/genkit/src/genkit/_ai/_aio.py#L1114-L1192）远比这丰富，全部为关键字参数：

从实现看，每次generate调用会创建一个调用作用域的 child registry（self.registry.new_child()），把本次内联传入的tools与use中间件注册进去，调用结束即销毁，不会污染全局 registry（见 py/packages/genkit/src/genkit/_ai/_aio.py#L1159-L1192）。这一设计保证了并发调用之间互相隔离。

围绕generate的行为在测试中有系统性覆盖，例如 generate_test.py、generate_request_construction_test.py、generate_interrupt_resume_test.py，阅读它们是了解参数语义与边界情况（如中断恢复、动态工具）的可靠入口。

五、Flow 与 Tool：让模型调用你的代码

README 只展示了 flow，但同一 API 面上@ai.tool()与其配对使用，官方 docstring 的标准组合（见 py/packages/genkit/src/genkit/init.py#L17-L39）是：

from genkit import Genkit from genkit_google_genai import GoogleAI ai = Genkit(plugins=[GoogleAI()], model=GoogleAI.gemini_model('gemini-flash-latest')) @ai.tool() async def current_weather(city: str) -> str: return f'Sunny in {city}' @ai.flow() async def my_flow(prompt: str) -> str: res = await ai.generate(prompt=prompt, tools=['current_weather']) return res.text if __name__ == '__main__': ai.run_main(my_flow('Weather in Paris?'))

要点：

• @ai.tool()（py/packages/genkit/src/genkit/_ai/_aio.py#L299-L327）把函数注册为工具，name、description可显式指定，input_schema可传 Pydantic 模型覆盖参数推导；返回注解会被模型当作outputSchema绑定；
• ai.generate(prompt=..., tools=['current_weather'])中工具以字符串名引用，框架从 registry 解析；模型发起工具请求时由 SDK 自动执行并把结果回传，直到模型产出最终文本；
• 对需要"暂停等待人工确认"的场景，可用ai.define_interrupt(...)注册中断工具（py/packages/genkit/src/genkit/_ai/_aio.py#L358-L387），随后通过generate的resume_respond/resume_restart参数恢复执行，公开 API 中还导出了Interrupt、respond_to_interrupt、restart_tool等配套类型（见 py/packages/genkit/src/genkit/init.py#L47-L56）。

除 flow 与 tool 外，Genkit实例还暴露了同一风格的定义方法：define_prompt/prompt（可执行提示模板）、define_model/define_background_model（自定义模型与长时运行模型）、define_embedder、define_evaluator、define_middleware、define_resource等，公开导出列表完整可见于 py/packages/genkit/src/genkit/init.py#L108-L177。

六、提示模板目录：prompts/的隐式加载

Genkit.__init__中还有一个容易被忽略的行为（py/packages/genkit/src/genkit/_ai/_aio.py#L195-L203）：

load_path = prompt_dir if load_path is None: default_prompts_path = Path('./prompts') if default_prompts_path.is_dir(): load_path = default_prompts_path if load_path: load_prompt_folder(self.registry, dir_path=load_path)

即：若当前目录存在./prompts/文件夹，其中的.prompt模板会自动加载注册；也可以显式传prompt_dir指定目录。之后即可用ai.prompt('name')拿到可执行提示，配合input_schema/output_schema获得类型化调用。仓库内 py/samples/prompts 提供了一个可直接运行的示例工程，展示了模板目录的组织方式。

七、验证与深入路径

• 单测：核心包测试位于 py/packages/genkit/tests，其中 tests/genkit/ai 覆盖了 generate、工具、Agent、流式与恢复等行为；pyproject.toml 配置了 pytest 的pythonpath（包含src与tests），在py/packages/genkit目录下运行 pytest 即可执行；
• 多语言对照：同一框架还有 JS（js/genkit）与 Go（go/genkit）实现，跨语言行为（如 reflection 协议）有共享的 conformance 测试规格（tests/specs），阅读 Python 实现时可对照理解协议层设计；
• 示例工程：py/samples 下按主题组织了 prompts、middleware、tool-interrupts、output-formats、agents 等示例，每个都是独立可运行的小工程。

小结

• 安装：uv add genkit genkit-google-genai；要求 Python ≥ 3.10，核心依赖 Pydantic v2（当前包版本 0.11.0）；
• 主链路：Genkit(plugins=..., model=...)→@ai.flow()注册工作流 →ai.generate(prompt=..., output_schema=Model)做结构化生成 →ai.run_main(main())启动，开发模式下 Dev UI 反射服务自动拉起；
• 供应商解耦：模型、embedder、evaluator 都经插件注入 registry，切换 OpenAI/Anthropic/Ollama/Bedrock/Vertex AI 只需替换插件包；
• 深入点：generate的完整参数面（工具、中间件、中断恢复、文档）、generate_stream的 partial 输出语义、./prompts/目录自动加载，均可在 py/packages/genkit/src/genkit/_ai/_aio.py 与对应测试中找到完整实现证据。

【免费下载链接】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