Genkit Python 实战用 output_schema 与 output_format 实现类型安全的结构化输出【免费下载链接】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 官方的output-formats示例py/samples/output-formats/README.md展开讲清楚generate()如何通过output_schema把模型响应直接变成 Pydantic 模型、output_format的 enum/json/array 等内置格式各有什么区别以及generate_stream()中「流式 chunk 也是带类型的」这一关键机制。读完你可以直接复现这个示例并理解响应解析背后 Formatter 的实现原理从而在自己的 Genkit 应用中安全地消费结构化输出而不再手动解析 JSON 字符串。示例概览与运行方式该示例位于 py/samples/output-formats/目录结构很简单src/main.py演示 5 种输出场景的完整代码pyproject.toml声明依赖genkit与genkit-google-genai要求 Python 3.10README 中给出的运行命令如下前提已申请 Gemini API Keyexport GEMINI_API_KEYyour-api-key uv sync uv run src/main.py示例创建的 AI 实例使用了GoogleAI插件和gemini-flash-latest模型见 main.py 第 26 行ai Genkit( plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest), )核心概念output_schema 就是你在 response.output 上拿到的类型README 的核心结论只有一句话output_schemais the Pydantic model you get back onresponse.output.output_schema就是你在response.output上取回的 Pydantic 模型。也就是说当你给ai.generate(...)传入一个 Pydantic 模型类例如Country作为output_schema时返回的response.output不是一个待你json.loads的字符串而是已经实例化、已经过校验的Country对象。这一点在 main.py 第 59-64 行有明确注释# JSON: response.output is a Country, not a string you parse. country await ai.generate( promptGive quick facts about Japan., output_schemaCountry, ) print(country.output)类型层面同样有保证从 genkit/_ai/_aio.py 中generate与generate_stream的大量类型重载overload签名看传入output_schema: type[OutputT]时返回值类型分别是ModelResponse[OutputT]与ModelStreamResponse[OutputT]——静态类型检查器因此能对response.output的属性访问给出完整补全。逐场景解读 main.py 中的 5 种输出方式1. 默认纯文本不传output_schema时走默认文本路径直接读haiku.texthaiku await ai.generate(promptWrite a haiku about coding.) print(haiku.text)2. enum让模型从固定集合里选一个值output_formatenum配合一个str, Enum类型示例中的Sentiment取值 POSITIVE/NEGATIVE/NEUTRALreview await ai.generate( promptClassify this review: This product broke after one day., output_formatenum, output_schemaSentiment, ) print(review.output)底层实现位于 genkit/_ai/_formats/_enum.py有两个值得注意的细节schema 校验EnumFormat.handle()要求 schema 的type必须是string或enum否则抛出INVALID_ARGUMENT错误第 81-85 行响应清洗message_parser会用正则剥掉模型输出里的引号并 strip返回裸的枚举字符串避免你拿到NEGATIVE这样的带引号值第 87-93 行。3. json默认的 Pydantic 模型路径不显式指定output_format时只要提供了 schema默认即按 JSON 处理content_type为application/json。示例定义了Country(name, capital, population)模型并直接获得实例化对象。实现上genkit/_ai/_formats/_json.py 的JsonFormat做了两件事有 schema 时向模型注入「输出必须是符合以下 schema 的 JSON」的指令把 schemajson.dumps(indent2)后拼进 instructions见 第 109-118 行解析响应时不直接json.loads而是调用extract_json从可能带有 markdown 代码块等噪声的文本中稳健地提取出 JSON第 79-107 行。4. 流式 JSONchunk 本身也是带类型的部分对象这是 README 第二段强调的能力generate_stream(..., output_schemaCountry)流式返回的chunk 也是Country类型——字段可能还是None或者只是已到达内容的前缀而(await sr.response).output是最终完成并经过校验的对象。sr ai.generate_stream( promptGive quick facts about Japan., output_schemaCountry, ) async for chunk in sr: if chunk.output and chunk.output.name: print(fstreaming name: {chunk.output.name}) final_country (await sr.response).output print(final_country)对应实现中JSON 格式的chunk_parser使用extract_json(chunk.accumulated_text, throw_on_bad_jsonFalse)累积文本还构不成合法 JSON 时返回None对应「字段可能还是 None 或前缀」的现象合法后逐步填出完整对象见 _json.py 第 93-107 行。这也解释了为什么示例里要先if chunk.output and chunk.output.name判空再打印。5. array / jsonllist[T] 的 items schema 写法对于「列表」类型的输出schema 需要描述为array。示例中的技巧是用 Pydantic 的TypeAdapter为list[Book]生成 JSON schema 传给output_formatarray拿到结果后再用validate_python转回Book模型列表books await ai.generate( promptList 3 famous fantasy books., output_formatarray, output_schemaTypeAdapter(list[Book]).json_schema(), ) print(TypeAdapter(list[Book]).validate_python(books.output))底层 genkit/_ai/_formats/_array.py 的ArrayFormat有两个关键点schema 必须是array类型否则抛INVALID_ARGUMENT第 89-93 行流式解析采用增量游标chunk_parser记录上一轮文本长度作为 cursor对累积文本增量提取 JSON 对象从而在流式过程中逐个吐出完整 item第 110-121 行。底层机制Formatter 体系一览上述所有格式都由统一的 Formatter 体系驱动位于 py/packages/genkit/src/genkit/_ai/_formats/格式实现文件content_type解析策略text_text.py-直接取文本enum_enum.pytext/enum去引号清洗json_json.pyapplication/jsonextract_json容错提取array_array.pyapplication/json增量游标逐 item 提取jsonl_jsonl.py-逐行 JSON 提取五个格式实例在init.py 的built_in_formats列表中注册。每个格式是一个FormatDef携带FormatterConfigformat、content_type、constrained、default_instructions等字段调用handle(schema)时产出一个Formatter其包含message_parser完整响应解析、chunk_parser流式 chunk 解析和注入给模型的instructions见 _types.py 第 30-122 行。从源码结构看这一设计的意义在于无论模型输出多么「噪声」SDK 侧的解析器负责容错提取而output_schema提供的 Pydantic 类型负责最终实例化与校验——调用方拿到的始终是类型化的 Python 对象而非裸字符串。小结把 Pydantic 模型直接作为output_schema传给ai.generate()response.output即为已校验的模型实例output_format可选enum、json默认、array、jsonl、text各格式对 schema 形态有不同要求enum 需 stringenumarray 需 array 类型generate_stream(..., output_schemaT)的 chunk.output 是「带洞的」部分对象最终结果通过(await sr.response).output获取list[T]场景用TypeAdapter(list[T])生成 items schema并在取回后用validate_python转回模型列表深入实现可阅读 py/packages/genkit/src/genkit/_ai/_formats/ 下的五个 Format 实现与 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），仅供参考