Genkit Python SDK 实战指南用类型安全 Flow、结构化输出与可观测性构建生产级 AI 应用【免费下载链接】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本指南以 py/README.md 为主线系统讲解 Google 开源框架 Genkit 的 Python SDK从三步快速上手安装、配置 API Key、编写首个应用到 Flow、Tool、结构化输出等核心原语再到 OpenTelemetry 可观测性、多模型提供商接入与 ASGI/WSGI 部署最后完整覆盖仓库开发工作流前置条件、目录结构、just py命令、示例运行。读完本文你将掌握用 Python 构建类型安全、可观测、可部署的生成式 AI 应用与 Agent 的完整方法并能在当前仓库的 py/ 目录下直接参与开发与贡献。一、Genkit Python SDK 是什么Genkit 是 Google 开源的生产级 AI 应用框架支持 JavaScript、Go、Dart、Python 多语言其 Python SDK本仓库 py/packages/genkit 下名为genkit的包定位为用 Python 构建生产就绪的 AI 应用提供类型安全的 Flow、结构化输出与集成式可观测性。从 SDK 的公开 API 表面见 py/packages/genkit/src/genkit/init.py可以看出开发者日常打交道的核心入口集中在Genkit主类与FlowAction的别名应用编排的顶层入口Action/StreamResponse可注册、可调用的执行单元及其流式响应工具体系Tool、tool装饰器、ToolRunContext、Interrupt、response/respond_to_interrupt/restart_tool等内容与模型类型Message、Part、Role、ToolRequest、Media、Document、ModelInfo、ModelRequest、ModelResponse等错误与插件体系GenkitError、PublicError、Plugin。一句话概括一套统一的ai.generateAPI 承载生成、工具调用、结构化输出与 Agent 能力本地 Dev UI 提供实时可观测性Vertex AI、Cloud Trace、Firestore、OpenAI、Anthropic、Ollama、Bedrock 等按需接入。二、快速开始三个步骤写出第一个 AI 应用原文档给出的入门路径极其简洁共三步。第 1 步安装 SDK 与模型提供商插件uv add genkit genkit-google-genai这里一次性安装了核心 SDKgenkit与 Google AIGemini提供商插件genkit-google-genai。Genkit 采用核心 插件架构核心包只负责 Flow、Tool、生成、追踪等基础能力具体模型由独立的提供商包提供。如果你偏好使用pip等价地执行pip install genkit genkit-google-genai亦可而在 py/packages/genkit/README.md 中官方同样推荐以这条uv add命令作为标准安装方式。第 2 步设置 API Keyexport GEMINI_API_KEYyour-api-keygenkit-google-genai插件会读取GEMINI_API_KEY环境变量完成认证。生产环境中请通过密钥管理服务注入切勿把 Key 硬编码进代码或提交到仓库。第 3 步创建 AI 应用from genkit import Genkit from genkit_google_genai import GoogleAI # 1. 初始化 Genkit 并加载 Google AIGemini插件 ai Genkit(plugins[GoogleAI()]) # 2. 定义一个类型安全的工具 ai.tool(descriptionGet current weather for a city) def get_weather(city: str) - str: return fSunny, 72°F in {city} # 3. 定义一个可观测的 Flow ai.flow() async def plan_trip(destination: str) - str: response await ai.generate( modelgoogleai/gemini-flash-latest, promptfSuggest activities in {destination} given the weather., tools[get_weather], ) return response.text # Based on the sunny weather in Seattle...这段示例浓缩了 Genkit 的四个关键设计Genkit(plugins[GoogleAI()])初始化应用级单例负责插件注册、注册表Registry维护与运行时管理。ai.tool(description...)注册工具函数签名即工具的输入/输出 schemacity: str - str无需手写 JSON Schema模型会自动获得类型化工具定义。除了像这里直接传函数对象也可以先注册再用注册名引用——SDK 文档字符串中展示了tools[current_weather]的等价写法见 py/packages/genkit/src/genkit/init.py。ai.flow()注册 Flow异步函数被注册为可追踪、可流式、可通过 Dev UI 或反射协议调度的执行单元。ai.generate(...)统一生成 APImodel指定模型引用googleai/gemini-flash-latest表示 Google AI 提供商的 Gemini Flash 最新版prompt传提示词tools挂载工具模型在需要时会自动调用工具再汇总回答。补充默认模型与结构化输出的另一种写法若在初始化时就指定默认模型generate调用可以省略model参数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), )配合 Pydantic 模型可实现强类型结构化输出示例源自 py/packages/genkit/README.mdclass 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())output_schemaIssue让 SDK 约束模型输出为合法的IssueJSON 并校验为 Pydantic 实例Flow 的返回值类型也直接标注为Issue——类型安全贯穿输入、生成与输出全链路。三、核心原语源码深挖Flow、Tool 与 Generate 的底层机制3.1ai.flow()装饰器与注册链路Genkit.flow()装饰器py/packages/genkit/src/genkit/_ai/_aio.py#L210-L260支持三个可选参数参数类型默认值作用namestr \| None函数名自定义 Flow 注册名Dev UI 与调用方以此引用descriptionstr \| None函数 docstring 推导Flow 的描述信息chunk_typetype \| NoneNone流式分块的类型标注传入后返回类型变为Action[InputT, OutputT, ChunkT]底层由_FlowDecoratorpy/packages/genkit/src/genkit/_ai/_decorators.py#L34-L57调用define_flow完成注册。define_flowpy/packages/genkit/src/genkit/_core/_flow.py#L61-L79做了三件事强制校验函数必须是async协程否则抛TypeError以函数名或显式name为注册名注册ActionKind.FLOW类型的 Action注入span_metadata{flow:name: flow_name}—— 这正是 Flow 可观测性的来源每次执行都会在追踪链路中留下带flow:name标签的 Span。由此可以推断Flow 的本质是一个注册在全局 Registry 中的可追踪 Action因此天然支持 Dev UI 远程调用、反射协议Reflection Protocol暴露、流式分块与中途 span 查看。3.2ai.tool()函数签名即 SchemaGenkit.tool()装饰器py/packages/genkit/src/genkit/_ai/_aio.py#L299-L327签名如下def tool( self, name: str | None None, description: str | None None, *, input_schema: type[BaseModel] | dict[str, object] | None None, ) - Callable[[Callable[..., Any]], Tool]:其中函数返回值注解被 SDK 用作工具输出的outputSchemainput_schema可显式覆盖传入 Pydantic 类或 JSON Schema 字典缺省时由函数参数注解自动推导支持同步与异步两种函数体define_tool见 py/packages/genkit/src/genkit/_ai/_tools.py#L735-L760。工具体系不止于此__init__.py还导出了Interrupt、ToolRunContext、respond_to_interrupt、restart_tool等对应工具中断-人工审批/继续执行的高级交互模式相关能力在仓库示例 py/samples/tool-interrupts 中有完整演示。3.3ai.generate()统一生成 API 的能力矩阵Genkit主类围绕生成提供了多条方法py/packages/genkit/src/genkit/_ai/_aio.py#L1008-L1400方法用途generate()单轮生成返回ModelResponse.text/.outputgenerate_stream()流式生成边生成边产出分块embed()/embed_many()文本向量化RAG 场景generate_operation()后台长任务式生成配合 Operation 轮询在生成内部实现py/packages/genkit/src/genkit/_ai/_generate.py中有一个值得注意的常量DEFAULT_MAX_TURNS 5这意味着默认情况下一次generate中模型与工具之间的多轮交互Agent 循环上限为 5 轮防止工具调用无限循环。这是从源码确认的默认行为可按需在生成选项中调整。四、为什么选择 Genkit四大核心能力展开原文档概括了四个核心卖点下面逐一结合仓库证据展开。4.1 类型安全Pydantic 贯穿全链路工具输入输出由函数签名 /input_schema自动推导见 3.2结构化输出由output_schemaPydanticModel约束并校验见第 2 节补充示例Flow 泛型Action[InputT, OutputT]在装饰器层通过多个overload重载实现精确的类型推断py/packages/genkit/src/genkit/_ai/_decorators.py#L47-L57配合 pyright/ty 等静态检查器可在编译期捕获输入输出类型错误。4.2 多模型提供商一套 API 切换所有主流模型Genkit 以核心 提供商插件解耦模型接入。当前仓库 py/packages 目录下已包含官方集成包genkit-google-genaiGemini/Vertex AI、genkit-anthropicClaude、genkit-openai、genkit-ollama、genkit-amazon-bedrock、genkit-vertexai以及genkit-a2ui、genkit-django、genkit-fastapi、genkit-flask、genkit-google-cloud、genkit-evaluators、genkit-middleware等。切换提供商只需替换插件与模型引用业务代码中的generate/flow调用完全不变。各提供商均有对应可运行示例例如 py/samples/anthropic-sample、py/samples/ollama-sample、py/samples/amazon-bedrock-sample。4.3 集成可观测性OpenTelemetry 追踪 Dev UISDK 内置基于 OpenTelemetry 的追踪与评估指标核心实现在 py/packages/genkit/src/genkit/_core/_trace包含默认导出器_default_exporter.py、实时 Span 处理器_realtime_processor.py、日志导出器_log_exporter.py等Flow 注册时自动注入的flow:namespan 元数据正是追踪链路的来源见 3.1。本地调试时使用 Genkit Developer UI由 genkit-tools/cli 提供genkit命令genkit start -- uv run entrypoint.py启动后在浏览器中打开 Dev UI即可实时查看每个 Flow 执行的 Span 树、工具调用明细、耗时与评估指标也可以在 UI 中直接触发已注册的 Flow 和 Agent 进行交互测试。4.4 随处部署ASGI/WSGI 与 ServerlessGenkit Flow 本质是注册的 Action可统一暴露为标准 Web 服务。仓库提供了genkit-fastapi/genkit-flask/genkit-django把 Genkit 应用桥接为 FastAPI、Flask、Django 应用run_main()py/packages/genkit/src/genkit/_ai/_aio.py#L955-L960运行用户主协程在开发环境下会同时阻塞以维持反射服务Reflection Server供 Dev UI 连接_core/_reflection.py/_reflection_v2.py提供 Reflection 协议使 Flow 以标准 JSON-RPC 风格被外部调用。因此同一个 Flow 既可本地脚本运行也可包装为 ASGI/WSGI 应用部署到 Cloud Run 或任意 Serverless 平台。对应示例见 py/samples/fastapi-bugbot、py/samples/flask-hello、py/samples/django-hello。五、仓库开发工作流从源码运行与贡献原文档的后半部分面向想在当前仓库中运行示例或参与 SDK 开发的读者。5.1 前置条件Python 3.10CI 覆盖 3.10–3.14见 py/justfile 中test-nox的说明uv极快的 Python 包与项目管理器仓库依赖解析使用uv.lockjust现代命令运行器just/justfile将常用开发命令集中封装。5.2 工作区结构py/ ├── bin/ # CI/CD 与发布自动化脚本 ├── docs/ # Playbook 与 API 参考模板[py/docs](https://link.gitcode.com/i/0a6afc00b2fa61b38b99c17d0a3492fb) ├── packages/ # 核心框架与官方集成[py/packages](https://link.gitcode.com/i/ef81dc610585b5cd97e1c4d5881a797d) ├── samples/ # 可运行示例应用[py/samples](https://link.gitcode.com/i/891dabfecab8710e8886d1c8f46a7b3e) ├── scripts/ # 维护与校验脚本 ├── tests/ # 跨包集成测试套件 ├── justfile # 命令运行器快捷方式just py command ├── noxfile.py # 多版本测试自动化3.10–3.14 ├── pyproject.toml # 工作区元数据与工具依赖 └── uv.lock # 依赖锁定文件packages/下每个子包如genkit、genkit-google-genai均遵循src/包名/tests/布局并带有各自的pyproject.toml便于独立发布与依赖隔离。samples/目录当前收录了 20 余个示例覆盖 Agentpy/samples/agents、评估py/samples/evaluators、中间件py/samples/middleware、提示词py/samples/prompts、多模态py/samples/google-genai-media、追踪py/samples/tracing等主题。5.3 开发命令just py在仓库根目录执行just py command或在py/目录下直接just command。核心命令如下完整定义见 py/justfile命令底层执行用途just py syncuv syncuv pip check安装并校验工作区依赖just py lintruff check、ruff format、ty check、pyrefly check、pyright packages/格式化 Lint 类型检查对应 CI 的lint-and-format/type-checkjust py fmtruff formatruff check --fix自动格式化并修复 Lint 错误just py testpytest .运行单元测试just py test-nox用 nox 覆盖 Python 3.10–3.14 多版本just py checkcheck_consistency.py校验工作区版本一致性值得注意的还有just py sample交互式示例选择器、just py generate-schema从 JSON Schema 重新生成typing.py、just py build/publish构建与发布到 PyPI等命令见 py/justfile。5.4 运行示例cd samples/sample-name genkit start -- uv run entrypoint.py以 py/samples/agents 为例进入目录、执行上述命令后Genkit 会启动 Dev UI 并运行示例入口脚本浏览器打开 Dev UI 即可直接与已注册的 Flow、Agent 交互观察实时 Span。5.5 文档与维护API Reference完整的类与方法签名请查阅 py/docs/index.md贡献与规范编码约定、提交规范与类型检查规则见仓库根的 CONTRIBUTING.md发布流程维护者发布流程见 py/docs/release_playbook.md。六、开源协议Genkit Python SDK 采用 Apache 2.0 许可证见 py/README.md 与 py/packages/genkit/LICENSE可自由用于商业与个人项目。【免费下载链接】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），仅供参考