用 define_custom_agent 实现 Genkit Python 自定义编排从固定 Prompt 循环到完全可控的 Agent【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skillsGenkit Pythonskills/cloud/genkit-python的define_agent会替你运行一套固定的「系统提示词 工具循环」适合大多数对话场景但当你需要**多步骤编排、自定义流式输出、手动管理消息/状态/工件Artifacts**时就需要直接接管回合循环。本文以 agents-custom.md 为骨架系统讲解 Beta 版ai.define_custom_agentAPI 的解剖结构、完整示例、流式/非流式两种handle_turn写法并结合仓库内 Agents 系列文档给出可直接复制运行的代码与底层原理说明。读完你就能写出一个「自己说了算」的 Agent手动决定模型调用、工具执行、消息持久化与流式转发而调用方依然只需chat.send/send_stream。为什么需要define_custom_agentai.define_agent的运行模型是固定的系统提示词 模型 工具列表 自动的工具调用循环每轮用户消息由框架调度。这对「标准聊天助手」够用但对以下场景不够灵活多步骤编排需要先查询 A再根据结果决定调用 B流程由你的代码决定而非模型自由发挥自定义流式想把模型流式分片model_chunk以你自己的节奏转发给调用方手动管理消息/状态/工件何时把模型回复写进历史、何时更新自定义状态、何时产出工件全部由你掌控混合调用在一个回合里既用generate又用generate_stream甚至自己实现工具循环。此时使用ai.define_custom_agent你注册一个 async 函数作为 Agent 的编排内核调用方仍通过chat.send/send_stream与之交互接口契约不变。Anatomy自定义 Agent 的解剖结构自定义 Agent 的核心是一个 async 函数其骨架如下源自 agents-custom.mdasync def fn(sess: SessionRunner, ctx: ActionRunContext) - AgentResult: async def handle_turn(inp: AgentInput, turn_ctx: TurnContext) - TurnResult | None: # 用户消息此时已在 sess.get_messages() 中运行时已自动追加 # 在这里做模型调用 / 工具调用 / 自定义逻辑… # 需要流式时用 ctx.send_chunk(AgentStreamChunk(...)) # 需要持久化回复时用 sess.add_messages(...) return TurnResult(finish_reasonAgentFinishReason.STOP) await sess.run(handle_turn) return await sess.result() agent ai.define_custom_agent(namecustomCoder, fnfn, storestore)关键点逐条拆解handle_turn每轮用户消息执行一次。在它运行之前运行时已把inp.message追加进历史所以函数体内直接sess.get_messages()就能拿到含最新用户输入的完整历史。返回值返回一个TurnResult或None。常见finish_reason为STOP、INTERRUPTED、FAILED如果函数体内抛出异常本次调用会以FAILED结束。turn_ctx仅在有 store 时有效提供snapshot_id可在其下命名外部工作树/目录、parent_snapshot_id、turn_index。没有 store 时snapshot_id为None。这与 agents-sessions.md 中「每个回合写入一个不可变快照」的模型一致——snapshot_id正是分支Branching与后台任务的基础。会话读写从sess上可调用get_messages/add_messages/set_messages、get_custom/update_custom、get_artifacts/add_artifacts。注意add_messages与add_artifacts都接收列表。流式输出通过ctx.send_chunk(AgentStreamChunk(...))把分片发给调用方的turn.stream。完整示例一个手动编排的customCoder下面是一个带流式转发的完整可运行示例源自 agents-custom.md 并补齐上下文from genkit import ActionRunContext, FinishReason, Genkit, Message from genkit.agent import ( AgentFinishReason, AgentInput, AgentResult, AgentStreamChunk, InMemorySessionStore, SessionRunner, TurnContext, TurnResult, ) from genkit_google_genai import GoogleAI ai Genkit(plugins[GoogleAI()]) store InMemorySessionStore() async def custom_coder_fn(sess: SessionRunner, ctx: ActionRunContext) - AgentResult: async def handle_turn(inp: AgentInput, turn_ctx: TurnContext) - TurnResult | None: history await sess.get_messages() messages [Message(m) for m in history] if history else None stream_resp ai.generate_stream( modelgoogleai/gemini-flash-latest, systemConcise coding assistant., messagesmessages, ) async for chunk in stream_resp.stream: ctx.send_chunk(AgentStreamChunk(model_chunkchunk)) res await stream_resp.response if res.message: await sess.add_messages([res.message]) fr ( AgentFinishReason.STOP if res.finish_reason FinishReason.STOP else AgentFinishReason.UNKNOWN ) return TurnResult(finish_reasonfr) await sess.run(handle_turn) return await sess.result() agent ai.define_custom_agent(namecustomCoder, fncustom_coder_fn, storestore) chat agent.chat() await chat.send(What is a Python list comprehension?)这个例子里值得注意的细节模型 ID 使用带前缀形式googleai/gemini-flash-latest与仓库 SKILL.md 中的约定一致需要设置GEMINI_API_KEY。历史恢复sess.get_messages()拿到原始消息后包成Message再传给generate_stream保证模型看到完整上下文。流式转发AgentStreamChunk(model_chunkchunk)逐片转发调用方用chat.send_stream(...)时能实时收到.stream。持久化回复sess.add_messages([res.message])把模型回复写入历史。文档明确强调必须用sess.add_messages持久化模型回复否则下一轮handle_turn看不到上一轮的回答。结束原因映射把模型的FinishReason映射为 Agent 层的AgentFinishReasonSTOP/UNKNOWN。补充自定义状态与工件在自定义 Agent 里手动管理状态和工件同样直观详见 agents-state.md 与 agents-artifacts.md状态sess.update_custom(bump)传入一个 async 变更函数且会把变更流式推送到chunk.custom。例如await sess.update_custom(lambda c: {turns: (c or {}).get(turns, 0) 1})。工件await sess.add_artifacts([Artifact(namereport.md, parts[Part(TextPart(textbody))])])注意同样要传列表读取时TextPart的文本在.root.text。文档还提醒自定义状态用于你的产品逻辑路由、UI不会自动注入给模型除非你把它拼进系统提示词或消息里。非流式handle_turn让ai.generate帮你跑工具循环如果不需要逐片转发handle_turn可以写得非常简单——直接用ai.generate由框架替你执行工具循环源自 agents-custom.mdasync def handle_turn(inp: AgentInput, ctx: TurnContext) - TurnResult | None: history [Message(m) for m in await sess.get_messages()] # ai.generate 会自动执行工具调用循环 res await ai.generate( messageshistory list(inp.messages or []), tools[get_time], systemYou are helpful., ) return TurnResult(messagesres.messages, finish_reasonAgentFinishReason.STOP)这里的要点inp.messages与历史拼接inp.messages携带本轮输入与已持久化的历史拼在一起喂给模型。tools[get_time]工具参数建议用小型 Pydantic 模型声明哪怕只有一个字段空输入需要空子类而非裸BaseModelagents.md。TurnResult(messagesres.messages, ...)直接把模型的全部输出消息放进TurnResult框架会处理持久化工具调用的输入输出规范见 agents.md 的 Tool inputs 小节。从固定循环到自定义两种 Agent 的选择边界结合 agents.md 可梳理出选择依据需求define_agentdefine_custom_agent固定 prompt 工具循环✅ 内置✅ 自己写多步骤编排代码控制流程❌ 受限✅ 完全掌控自定义流式转发节奏❌ 框架决定✅ctx.send_chunk手动管理消息/状态/工件❌ 自动✅sess系列方法中间件use[...]✅需自行处理自定义内核绕过标准调度调用方接口chat.send/send_stream相同chat.send/send_stream需要注意define_agent的便捷来自它替你处理了消息持久化、工具循环、中间件调度切到define_custom_agent后这些都要自己负责——尤其是忘记add_messages会导致上下文断裂以及中断INTERRUPTED后调用方要靠resume恢复详见 agents-human-in-the-loop.md 的finish_reason INTERRUPTED→ 构建restart/respondparts →chat.resume流程。运行与调试从 CLI 验证自定义 Agent自定义 Agent 与普通 Agent 一样不能直接用genkit flow:run运行flow:run只跑ai.flow()。要快速验证一轮可以把它包进一次性 flowagents.mdai.flow() async def try_custom_agent(message: str) - str: return (await agent.chat().send(message)).text # genkit flow:run try_custom_agent What is a Python list comprehension? -- uv run src/main.py开发调试建议使用genkit start -- uv run src/main.py捕获每个 Genkit action 的 trace再用genkit trace:list/genkit trace:get traceId检查模型输入输出与工具调用SKILL.md。这样即使handle_turn里的编排逻辑完全自定义模型 I/O 依然全程可见、可审计。小结define_custom_agent是 Genkit Python Agents 体系中「完全控制」的入口注册一个 async 编排函数在handle_turn里手动完成模型调用、工具执行、流式转发与消息持久化最后返回TurnResult声明回合结束原因。它适合多步骤编排、自定义流式、手动管理状态与工件的场景而调用方接口保持不变。与 agents.md、agents-state.md、agents-artifacts.md、agents-sessions.md、agents-human-in-the-loop.md 组合阅读即可搭建从状态管理、分支恢复到人工审批的完整自定义 Agent 应用。【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考