企业官网建设流程全解析

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

README 中给出的运行命令如下（前提：已申请 Gemini API Key）：

export GEMINI_API_KEY=your-api-key uv sync uv run src/main.py

示例创建的 AI 实例使用了GoogleAI插件和gemini-flash-latest模型（见 main.py 第 26 行）：

ai = Genkit( plugins=[GoogleAI()], model=GoogleAI.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( prompt='Give quick facts about Japan.', output_schema=Country, ) 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.text：

haiku = await ai.generate(prompt='Write a haiku about coding.') print(haiku.text)

2. enum：让模型从固定集合里选一个值

output_format='enum'配合一个str, Enum类型（示例中的Sentiment，取值 POSITIVE/NEGATIVE/NEUTRAL）：

review = await ai.generate( prompt='Classify this review: This product broke after one day.', output_format='enum', output_schema=Sentiment, ) 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(indent=2)后拼进 instructions，见 第 109-118 行）；
• 解析响应时不直接json.loads，而是调用extract_json从可能带有 markdown 代码块等噪声的文本中稳健地提取出 JSON（第 79-107 行）。

4. 流式 JSON：chunk 本身也是带类型的部分对象

这是 README 第二段强调的能力：generate_stream(..., output_schema=Country)流式返回的chunk 也是Country类型——字段可能还是None或者只是已到达内容的前缀；而(await sr.response).output是最终完成并经过校验的对象。

sr = ai.generate_stream( prompt='Give quick facts about Japan.', output_schema=Country, ) async for chunk in sr: if chunk.output and chunk.output.name: print(f'streaming name: {chunk.output.name}') final_country = (await sr.response).output print(final_country)

对应实现中，JSON 格式的chunk_parser使用extract_json(chunk.accumulated_text, throw_on_bad_json=False)：累积文本还构不成合法 JSON 时返回None（对应「字段可能还是 None 或前缀」的现象），合法后逐步填出完整对象（见 _json.py 第 93-107 行）。这也解释了为什么示例里要先if chunk.output and chunk.output.name判空再打印。

5. array / jsonl：list[T] 的 items schema 写法

对于「列表」类型的输出，schema 需要描述为array。示例中的技巧是：用 Pydantic 的TypeAdapter为list[Book]生成 JSON schema 传给output_format='array'，拿到结果后再用validate_python转回Book模型列表：

books = await ai.generate( prompt='List 3 famous fantasy books.', output_format='array', output_schema=TypeAdapter(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/：

五个格式实例在init.py 的built_in_formats列表中注册。每个格式是一个FormatDef（携带FormatterConfig：format、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 需 string+enum，array 需 array 类型）；
• generate_stream(..., output_schema=T)的 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