ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Genkit Python 开发示例详解:结构化输出、流式生成、Flows、Tools 与 Embeddings 实战指南

Genkit Python 开发示例详解:结构化输出、流式生成、Flows、Tools 与 Embeddings 实战指南 Genkit Python 开发示例详解结构化输出、流式生成、Flows、Tools 与 Embeddings 实战指南【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills导读本文基于当前仓库skills/cloud/genkit-python技能包中的 examples.md 文档系统讲解 Genkit Python SDK 中最常用、最小化的 API 调用模式。你可以从中掌握结构化输出JSON / Array / Enum、文本流式生成、媒体图片输入输出、Flows 与流式 Flows、Tools 工具调用以及 Embeddings 向量化等核心能力的标准写法并了解如何结合genkit start与 Dev UI 进行调试验证。所有示例均以 Google AIGoogleAI、googleai/...为默认提供商其他提供商Vertex AI、Anthropic 等只需替换对应插件与模型前缀即可套用同一套模式。环境准备公开导入与最小初始化在编写任何 Genkit 代码前先确保依赖与运行环境就绪。按 setup.md 的说明始终使用虚拟环境推荐uv并安装核心依赖uv init uv venv --python 3.12 .venv source .venv/bin/activate uv add genkit genkit-google-genai export GEMINI_API_KEYyour_key_here对应最小pyproject.toml的[project]块依赖不锁版本由解析器选取兼容版本[project] name my-app version 0.1.0 requires-python 3.10 dependencies [ genkit, genkit-google-genai, ]公共导入规范examples.md 特别强调只使用公开包与公共模块——如genkit、genkit_google_genai、genkit_fastapi、genkit_middleware、genkit_evaluators以及genkit.agent、genkit.embedder、genkit.evaluator、genkit.model等严禁导入内部模块如genkit._core。PyPI 上的包名使用连字符genkit-google-genai导入模块名使用下划线genkit_google_genai。from genkit import Genkit, ActionRunContext from genkit_google_genai import GoogleAI ai Genkit(plugins[GoogleAI()], modelgoogleai/gemini-flash-latest)这里的ai实例是后续所有示例的公共入口生成、流式、工具、Flows、Embedding 都经由它调用。注意模型 ID 必须带提供商前缀googleai/gemini-flash-latest而非裸gemini-flash-latest这也是 common-errors.md 中列出的高频错误之一。对于多轮 Agent 场景从genkit.agent导入相关 API详见 agents.md。结构化输出让模型返回可校验的 JSON结构化输出是 Genkit 最常用的能力之一。只需定义 PydanticBaseModel作为输出 Schema并在ai.generate中声明output_formatjson与output_schema模型返回的结果便会被自动解析为对应模型实例通过response.output直接访问。from pydantic import BaseModel, TypeAdapter class CityInfo(BaseModel): name: str population: int country: str response await ai.generate( promptGive facts about Tokyo., output_formatjson, output_schemaCityInfo, ) city response.output数组输出用 TypeAdapter 包装若希望模型返回对象数组先用TypeAdapter生成 JSON Schema再以output_formatarray调用schema TypeAdapter(list[CityInfo]).json_schema() response await ai.generate( promptList 3 cities., output_formatarray, output_schemaschema, )支持的输出格式Genkit 共支持五种output_format格式适用场景text纯文本默认行为读取response.textjson单个 JSON 对象配合output_schema自动反序列化arrayJSON 数组Schema 需用TypeAdapter(list[T]).json_schema()生成enum枚举取值jsonl每行一个 JSON 对象JSON Linesresponse.output返回的是 Schema 对应的模型实例普通文本请读取response.text。不要混淆.text与.output这也是 common-errors.md 中 Missing.json/.messageon the response 一节的要点。文本流式生成逐 token 打印与提前退出流式生成返回一个GenerateResponseStream对象sr它同时暴露异步可迭代的.stream与最终响应await sr.responsesr ai.generate_stream(promptTell me a story.) async for chunk in sr.stream: if chunk.text: print(chunk.text, end, flushTrue) final await sr.response # final.text一个关键设计你不必排干.stream才能结束本轮——即使提前break跳出循环await sr.response依然会正常完成。这与 Agent 的send_stream语义一致见 agents.md 中turn.stream/turn.response的用法。因此流式 提前中断 拿最终结果是安全且受支持的组合。需要注意generate_stream返回的对象不能直接 await要 await 的是.response详见 common-errors.md 的 Streaming TypeError 一节。文本与媒体部件读取模型产出的图片/音频模型不仅产出文本还可能产出媒体内容图片、音频等。非流式场景下直接遍历response.media# Non-streaming response await ai.generate(prompt...) for media in response.media: print(media.content_type, (media.url or )[:80])流式场景下媒体通常在最终响应上才完整出现因此在遍历流的同时最终结果上读取final.mediafrom genkit import MediaPart sr ai.generate_stream(prompt...) async for chunk in sr.stream: if chunk.text: print(chunk.text, end, flushTrue) final await sr.response for media in final.media: print(media.content_type, (media.url or )[:80])若需要访问原始消息部件结构可通过final.message.content逐部件判断类型if final.message: for part in final.message.content: if isinstance(part.root, MediaPart) and part.root.media: print(part.root.media.content_type)这里的MediaPart是 Genkit 消息模型中的媒体部件详见下文发送图片输入part.root保存实际部件内容。发送图片/媒体输入构造多部件用户消息上一节讨论的是模型产出的媒体要向 Gemini发送一张图片需要手工构造一条用户Message其中包含文本部件与媒体部件MediaPart媒体来源可以是 URL 或 data URIfrom genkit import Media, MediaPart, Message, Part, Role, TextPart image_url https://example.com/cat.jpg # or data:image/png;base64,... msg Message( roleRole.USER, content[ Part(rootTextPart(textWhat is in this image?)), Part(rootMediaPart(mediaMedia(urlimage_url, content_typeimage/jpeg))), ], ) response await ai.generate(messages[msg]) print(response.text)使用视觉模型与 URL 注意事项建议选用具备视觉能力的模型例如googleai/gemini-flash-latest。URL 图片由服务端抓取fetch 发生在 Google 侧因此被防盗链hotlink拦截或缩略图地址往往会导致失败优先使用直链图片 URL 或 data URI。本地图片推荐转成 base64 data URI 传入import base64 b64 base64.b64encode(Path(cat.png).read_bytes()).decode() url fdata:image/png;base64,{b64}流式 结构化输出边输出边解析 JSON将前两种能力组合一边流式生成一边获得最终的结构化对象。模型先写一段故事再输出对故事的分析 JSONclass StoryAnalysis(BaseModel): title: str genre: str summary: str sr ai.generate_stream( promptWrite a short story then analyze it., output_formatjson, output_schemaStoryAnalysis, ) async for chunk in sr.stream: if chunk.text: print(chunk.text, end, flushTrue) final await sr.response analysis final.outputfinal.output会被解析为StoryAnalysis实例。这种流式展示 结构化收尾的模式非常适合既要实时反馈又要机器可读结果的场景例如把故事正文流式呈现给用户同时把元数据存入数据库。Flows可观测、可调试、可复用的生成单元Flow 是 Genkit 的核心抽象之一用ai.flow()装饰的异步函数即成为一个可观测的 action可以在 Dev UI 与 trace 中查看其输入输出与内部调用链。它要求入参使用 PydanticBaseModel单一标量入参同样建议包成模型便于 trace 与 CLI 调用class SummarizeInput(BaseModel): text: str ai.flow() async def summarize(input: SummarizeInput) - str: response await ai.generate(promptfSummarize: {input.text}) return response.text流式 Flows用 ctx.send_chunk 逐块推送在 Flow 内部消费流式生成并通过ActionRunContext.send_chunk把每个文本块实时推给调用方Dev UI、CLI 或 HTTP SSE 流ai.flow() async def stream_story(subject: str, ctx: ActionRunContext) - str: sr ai.generate_stream(promptfStory about {subject}.) full async for chunk in sr.stream: if chunk.text: ctx.send_chunk(chunk.text) full chunk.text return full这里ctx: ActionRunContext是 Flow 的第二个参数。关键是必须调用ctx.send_chunk()——在 fastapi.md 中强调若不调用ctx.send_chunkFlow 虽然照常运行但 HTTP 客户端收不到任何流式事件只能等到最终结果。用 CLI 验证 Flow运行genkit start启动运行时后可以使用自终止的flow:run命令快速验证单个 Flowgenkit start -- uv run src/main.py # 另一终端 genkit flow:run summarize {text: ...} -- uv run src/main.pyflow:run运行一次、打印Trace ID后自动退出适合 CI 与自动化场景注意它只能运行ai.flow()不能直接运行 Agent详见 agents.md。更多 CLI 调试手段见后文调试与验证。Tools函数调用让模型动手Tool 让模型在推理过程中调用你的函数。examples.md 给出两条硬性规则参数必须是 PydanticBaseModel——即使只有一个字段也要包成模型。裸的str/float标量参数会在 Gemini 上得到 400 错误详见 common-errors.md 的 Tool schema must be an object。使用ai.tool()装饰器而不是ai.define_tool()后者已废弃/缺失见 define_toolmissing 一节。class WeatherInput(BaseModel): city: str ai.tool() async def get_weather(input: WeatherInput) - str: return fSunny in {input.city} response await ai.generate(promptWeather in Paris?, tools[get_weather])把 tool 列表通过tools[...]传给ai.generate模型便会自行决定何时调用。Tool 的返回值会被注入模型上下文继续推理。工具在 Agent 中的进阶用法在 Agent 场景agents.md中Tool 还常与中间件配合ToolApproval(allowed_tools[...])控制哪些工具可自由执行、哪些需要人工审批Filesystem(root_dir..., allow_write_accessTrue)提供文件读写能力Retry()对不稳定的模型 HTTP 调用进行重试。可以声明ai.tool(namegetWeather, description...)显式指定工具名与描述并在define_agent(tools[get_weather], ...)中挂载。Embeddings把文本转成向量嵌入Embedding用于语义检索、向量数据库、RAG 等场景。示例使用GeminiEmbeddingModels枚举构造带前缀的 embedder ID然后调用ai.embed单个文本或ai.embed_many批量文本from genkit_google_genai import GeminiEmbeddingModels embedder fgoogleai/{GeminiEmbeddingModels.GEMINI_EMBEDDING_001} embeddings await ai.embed(embedderembedder, contentThe sky is blue.) vector embeddings[0].embedding embeddings await ai.embed_many( embedderembedder, content[The sky is blue., Grass is green.], )常用的 embedder ID 包括googleai/gemini-embedding-001googleai/gemini-embedding-exp-03-07ai.embed_many返回的列表中每个元素的.embedding即对应输入文本的向量。调试与验证用 traces 而非盲跑examples.md 与 dev-workflow.md 一致强调直接uv run运行程序不会捕获开发期 trace等于盲跑。推荐的运行方式是让genkit start包裹你的程序genkit start -- uv run src/main.pygenkit start会在后台为每一个 Genkit action 采集 trace并在 Dev UI默认 http://localhost:4000中可视化它同样转发 stdio交互式 CLI 工具不受影响。genkit start会一直运行直到你按 CtrlC 退出这是预期行为。无 UI 的终端调试可用 trace 命令genkit trace:list # 列出最近 trace ID genkit trace:get traceId # 查看完整 trace输入输出、工具调用、延迟、错误 genkit trace:get traceId --format json # 机器可读 JSON可管道给 jq默认输出面向人类阅读若需程序化处理请务必加--format json。验证工具是否真的被调用、模型输入输出是什么等是genkit start trace 相比直接运行脚本的核心优势。常见坑位速查结合 common-errors.md 与上文示例整理高频错误及对策问题正确做法from genkit_google_genai import GoogleAI失败uv add genkit genkit-google-genai后重新导入流式返回对象直接 await 报 TypeErrorawaitsr.response/turn.response不要 await 流句柄本身工具参数是裸标量导致 400包成 PydanticBaseModel如class WeatherInput(BaseModel)define_tool不存在使用ai.tool()模型 ID 无前缀使用googleai/gemini-flash-latest想拿结构化结果却读.text结构化输出读response.output纯文本读response.textgenkit start下事件循环问题使用ai.run_main(main())作为入口小结本文覆盖了 Genkit Python 中最核心的八类 API 用法结构化输出含数组与枚举、文本流式、媒体部件读取、图片输入、流式 结构化组合、Flows普通与流式、Tools 以及 Embeddings。这些模式全部基于公开 API 编写可直接复制运行其他模型提供商只需替换插件如genkit_vertexai与模型前缀。进一步学习可阅读同目录下的 agents.md多轮 Agent、dotprompt.md.prompt模板文件、fastapi.mdHTTP 服务与 SSE 流式与 evals.md评测完整技能入口见 SKILL.md。【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表