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/genkitGenkit 是 Google 开源的 AI 应用框架其 Python SDKpy/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.也就是说它提供两件事一个统一入口生成文本/结构化数据、调用工具、定义与运行 flow工作流、构建 Agent都通过同一个Genkit实例上的方法完成而不用针对每个模型供应商写不同客户端代码可插拔的模型与基础设施GeminiGoogle AI、Vertex AI、OpenAI、Anthropic、Ollama、Amazon Bedrock 均以插件形式接入Cloud Trace、Firestore 等可选组件按需启用。包元信息可以从 pyproject.toml 中得到更精确的约束这也是复现本文示例的适用前提项目取值依据包名 / 版本genkit/0.11.0pyproject.toml#L80-L83Python 版本3.10classifiers 声明支持 3.10 ~ 3.14pyproject.toml#L82关键依赖pydantic2.10.5、opentelemetry-api/sdk、httpx、starlette、uvicorn、anyio等pyproject.toml#L41-L66构建/分类使用hatchling构建声明Framework :: Pydantic :: 2、Typing :: Typedpyproject.toml#L107-L112其中pydantic2这一点很重要后文示例中的结构化输出完全建立在 Pydantic v2 的BaseModel之上opentelemetry-*与uvicorn/starlette/sse-starlette依赖则解释了 SDK 内建的 OpenTelemetry 追踪能力与 Dev UI 反射服务的运行底座。二、安装README 给出的安装命令使用uvPython 包管理器一条命令同时装入核心包和 Google 模型插件uv add genkit genkit-google-genaigenkit核心 SDK本仓库 py/packages/genkit 对应的 PyPI 包genkit-google-genaiGemini 模型插件对应本仓库 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()], modelGoogleAI.gemini_model(gemini-flash-latest)) class Issue(BaseModel): title: str Field(descriptionShort title) severity: str Field(descriptioncritical, warning, or info) suggestion: str Field(descriptionHow to fix it) ai.flow() async def review(code: str) - Issue: result await ai.generate( promptfReview this code:\n{code}, output_schemaIssue, ) return result.output async def main() - None: print((await review(eval(user_input))).model_dump_json(indent2)) if __name__ __main__: ai.run_main(main())3.1 逐段解读1创建实例并注册插件与默认模型ai Genkit(plugins[GoogleAI()], modelGoogleAI.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 等注册为 formatmodel...把该引用注册为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(descriptionShort title) severity: str Field(descriptioncritical, warning, or info) suggestion: str Field(descriptionHow to fix it)Field(description...)的描述会进入生成 schema成为给模型的字段级说明——这是 Pydantic v2 Genkit 结构化输出的标准用法。severity若需要枚举约束也可以改用Literal[critical, warning, info]让 schema 层直接收窄取值。3ai.flow()注册工作流ai.flow() async def review(code: str) - Issue: result await ai.generate( promptfReview this code:\n{code}, output_schemaIssue, ) return result.outputflow装饰器定义见 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-L106ai.generate(..., output_schemaIssue)的返回值类型是ModelResponse[Issue]result.text是原始文本result.output是已经反序列化并通过 Pydantic 校验的Issue实例这也是示例中直接return result.output的原因。4run_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 CtrlC 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_backgroundpy/packages/genkit/src/genkit/_ai/_aio.py#L862-L929在 daemon 线程中做几件事检查环境变量GENKIT_REFLECTION_V2_SERVER若 CLI 以 v2 模式启动运行时并提供了 WebSocket URL则改走 v2 JSON-RPC 客户端ReflectionServerV2否则创建 ASGI 反射应用create_reflection_asgi_app用uvicorn绑定127.0.0.1上的一个随机空闲端口bind((127.0.0.1, 0))服务就绪后通过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(promptWrite a haiku about rain., output_schemaIssue) 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-L1386timeout参数用于控制 channel 超时。四、generate的完整参数面README 示例只用到了prompt与output_schema但Genkit.generate的完整签名py/packages/genkit/src/genkit/_ai/_aio.py#L1114-L1192远比这丰富全部为关键字参数参数说明model模型引用或名称缺省用构造时的默认模型prompt/system/messages用户提示 / 系统指令字符串或Part列表即多模态内容块/ 多轮消息历史tools工具列表元素可以是工具名字符串或Tool对象源码注释说明用协变的Sequence类型以便list[Tool]与list[str]都能传入tool_choice/return_tool_requests工具选择策略是否把未执行的工具请求返回给调用方供应用自行处理工具循环resume_respond/resume_restart/resume_metadata工具中断interrupt恢复相关参数配合define_interrupt使用config模型配置可以是ModelConfigDict、具体配置的BaseModel或普通Mapping框架会按模型注册的config_schema校验assert_correct_config_classmax_turns模型与工具往返的最大轮数context本次调用的上下文数据缺省时取当前ActionRunContextoutput_schemaPydantic 模型或 JSON schema dict决定output的类型与校验output_format/output_content_type/output_instructions/output_constrained输出格式与约束控制内置 format 见 py/packages/genkit/src/genkit/_ai/_formatsuse中间件链BaseMiddleware实例或MiddlewareRefdocs传入的Document列表从实现看每次generate调用会创建一个调用作用域的 child registryself.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()], modelGoogleAI.gemini_model(gemini-flash-latest)) ai.tool() async def current_weather(city: str) - str: return fSunny in {city} ai.flow() async def my_flow(prompt: str) - str: res await ai.generate(promptprompt, 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-L203load_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_pathload_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 即可执行多语言对照同一框架还有 JSjs/genkit与 Gogo/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_schemaModel)做结构化生成 →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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考