Genkit Python 后台 AgentBackground Agents / Detach实战指南把长任务交给服务端用快照句柄继续你的工作【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skillsGenkit Python 的 Agent API 提供了detach后台分离机制把一轮耗时的模型调用“交给服务端”调用立即返回一个携带snapshot_id的任务句柄等真正需要结果时再按快照加载并恢复对话。本文以 agents-background.md 为核心结合会话持久化、分支与 HTTP 服务化文档完整讲解 detach 的前置条件、核心 API、取消语义、并行研究模式与落地注意事项读完即可在真实项目中实现“提交后台任务 → 轮询结果 → 随时中止 → 恢复对话”的完整闭环。说明Background Agents / Detaching 属于 Beta / 预览 API位于genkit.agent模块下。它必须搭配 session store 使用——服务端需要把后台任务的结果写入某个地方客户端才有地方轮询和恢复。建议先阅读 agents.md 了解 Agent 基础概念。一、背景与设计动机detach 解决什么问题普通的chat.send(...)是一次同步往返模型未返回完整回复之前调用方只能阻塞等待。对于“生成一份长篇研究报告”“批量调研多个主题”“跑一遍复杂的多工具调用链”这类耗时任务同步等待会拖垮 Web 请求线程也不利于做异步 UI比如先返回任务 ID之后由前端轮询。Genkit Python 的 detach 模式把交互模型从“请求-响应”改为“提交-轮询”提交chat.detach(message)把整轮任务交给服务端异步执行立即返回一个带有snapshot_id的任务句柄DetachedTask此时任务状态为pending轮询task.wait(interval...)按固定间隔轮询 session store直到快照进入终态completed/failed/aborted等恢复任务完成后用agent.load_chat(snapshot_id...)加载最终快照拿到完整对话消息可以继续追问中止任何时候都可以通过task.abort()取消后台任务。这套机制以不可变快照snapshot_id为骨架每轮对话都会在 store 中写入一个不可变的快照这正是 分支branching 和后台任务共同的底层能力。二、前置条件必须配置 Session Store文档开篇明确声明detach需要 session store。原因很直接——后台任务是在服务端独立执行的客户端与服务端之间唯一的“信物”就是快照 ID任务结束后的结果必须持久化到 store 中客户端才能凭snapshot_id轮询到终态并读取最终消息。从 agents-sessions.md 可知Genkit Python 提供两种开箱即用的 storefrom genkit.agent import InMemorySessionStore, FileSessionStore mem_store InMemorySessionStore() # 内存存储进程重启即丢失 file_store FileSessionStore(./.snapshots) # 磁盘存储每个快照一个 JSON 文件要点InMemorySessionStore最简单适合单进程原型验证缺点是重启后快照全部丢失后台任务结果也无法跨进程读取FileSessionStore每个快照落盘为一个 JSON 文件支持进程重启后恢复也支持更多高级选项pruning FileSessionStore( ./.snapshots, max_persisted_chain_length3, # 可选修剪过长的对话链 reject_ambiguous_sessionTrue, # 可选分支后拒绝含混的 session_id )多人共享每个agent.chat()在共享的 store 上拥有独立 session按用户的session_id或snapshot_id分别恢复会话。三、核心 API 实战detach → wait → abort → load_chat原文给出了一个完整的后台报告生成示例这里完整继承并逐步拆解from genkit.agent import InMemorySessionStore agent ai.define_agent( namebackgroundAgent, modelgoogleai/gemini-flash-latest, systemSenior research analyst. Produce a comprehensive markdown report., storeInMemorySessionStore(), ) chat agent.chat() task await chat.detach(Write a report on renewable energy trends) print(task.snapshot_id) snapshot await task.wait(interval2.0) print(snapshot.status) await task.abort() done await agent.load_chat(snapshot_idtask.snapshot_id) print(done.messages)1.chat.detach(message)提交即返回detach与普通send的差别在于立即返回。它返回一个任务句柄句柄上最重要的字段是snapshot_id——这就是你在后台任务执行期间与它建立联系的凭证。服务端此刻已经把任务标记为pending并写入 store随后在后台继续处理模型调用与工具调用。注意模型 ID 使用带前缀的完整形式googleai/gemini-flash-latest这是 SKILL.md 中强调的规范需要先在环境中配置GEMINI_API_KEY。2.task.wait(interval2.0)轮询直至终态wait会以interval秒为周期轮询 store 中该快照的状态直到进入终态才返回最终快照。快照的status字段可取以下值状态含义pending仍在后台处理中继续轮询completed成功完成从快照的 messages 中读取结果failed处理过程中发生错误aborted被客户端通过abort()取消expired后台 worker 停止响应如服务端重启心跳失效视为终态任务永远无法完成轮询间隔需要权衡间隔太短会增加 store 读压力太长则用户体验上等待变久。原文示例取2.0秒适合报告生成这类秒级到分钟级的任务。3.task.abort()随时中止拿到句柄后任何时刻都可以调用await task.abort()取消后台任务。被中止的快照会标记为aborted。这一点在“提交后用户反悔”或“任务已无必要”的场景下非常重要避免白白消耗模型 token。4.agent.load_chat(snapshot_id...)按快照恢复对话任务完成后用task.snapshot_idwait返回的快照同 ID重新加载出一个可继续对话的chat对象读取done.messages即可拿到包含最终模型回复的完整消息列表也可以继续send追问细节。这与普通会话的恢复方式完全一致——见 agents.md 中load_chat注意不是chat(snapshot_id...)后者只是挂载一个恢复句柄。四、停止工作客户端停止与服务端取消是两回事“停止”在 detach 语义下要区分两个层面原文对此有非常明确的告诫客户端停止Client stopawait turn.abort() # 或对流式发送使用 asyncio.timeout 包裹 stream作用停止监听响应流。例如用asyncio.timeout包裹一个流式请求超时就放弃等待局限服务端可能仍在继续执行。因为此时你可能还没有拿到snapshot_id比如正在处理第一轮对话的中途适用时机流式读取的中途、尚未持有快照 ID、不关心后台是否继续完成的场景。服务端取消Server cancelawait chat.abort() # 或 await task.abort()前置条件必须配置了 store且已存在一个有效快照一轮对话完成后或detach之后语义真正取消服务端的后台工作把快照标记为aborted适用时机持有快照 ID 之后想要彻底回收任务资源。两个关键警告被中止的快照不是恢复点Aborted snapshots are not resume points。不要尝试从aborted快照继续对话——如果需要恢复应重新加载“最后一个正常叶子”快照上一个处于正常状态的快照第一轮对话的客户端中止陷阱在首轮对话上执行客户端 abort可能让这个 chat 永久丢失 ID——即使服务端之后保存了快照客户端侧也可能拿不到。原文给出的可恢复取消的推荐姿势是优先detachtask.abort()而不是中途硬断流。另外原文还强调abort 一定要await——“Alwaysawaitaborts — otherwise nothing happens”。异步取消如果不等待完成协程尚未执行到取消逻辑就被丢弃等于什么都没做。五、并行研究模式一个 chat 只能挂一个 detach用分支实现并发原文明确指出一个限制一个chat只能承载一个 detach。如果要对多个主题并行发起后台研究正确的做法是从一个共享检查点checkpoint为每个分支 fork 出独立的叶子对话root agent.chat() await root.send(Context both researchers should know.) checkpoint root.snapshot_id async def research(topic: str): leaf await agent.load_chat(snapshot_idcheckpoint) task await leaf.detach(fResearch {topic} in depth.) await task.wait(interval2.0) done await agent.load_chat(snapshot_idtask.snapshot_id) return topic, done.messages[-1] import asyncio results await asyncio.gather(research(Postgres), research(SQLite))拆解这个模式的要点先播种公共上下文root.send(...)让所有分支共享同一段背景知识比如“两份研究报告都要遵循同样的格式与受众”拿到检查点root.snapshot_id即不可变检查点每个分支 fork 叶子agent.load_chat(snapshot_idcheckpoint)从同一检查点分别加载独立的叶子对话互不干扰各自 detach wait每个叶子独立提交后台任务、独立轮询用asyncio.gather并发调度research是 async 函数gather让多个研究分支真正并发推进。这种“共享检查点 分支叶子 并发 gather”的写法本质上是把 agents-branching.md 中“从同一快照 fork 出多条分叉”的能力与 detach 结合起来。文档最后还特别提醒wait 之后要检查加载出的消息内容——仅凭status不足以判断结果质量completed也可能返回空或不符合预期的文本务必读取messages[-1]最后一条消息确认实际产出。六、store 解锁的能力边界何时该用 detach结合 agents-sessions.md 的能力对照表可以更清晰地定位 detach 的适用场景能力无 store有 store单 chat 多轮对话✅✅中断HITL✅✅持久化 ID /load_chat❌✅分支branching❌✅detach / 后台任务❌✅服务端取消server abort❌✅也就是说如果你的应用自己管理历史省略store参数IDs 保持None通过把messages/state/artifacts传入下一个chat(...)来接力那么 detach、分支和服务端取消都是不可用的。后台任务天然要求服务端“拥有”历史这也是 detach 以 store 为前提的根本原因。七、生产化落地把后台 Agent 通过 HTTP 暴露出去纯本地脚本可以直接await上面的代码但真实项目通常是一个服务端进程执行后台任务一个客户端进程提交与轮询。此时可借助 agents-http.md 中的 FastAPI 服务化方案。服务端serve_agentimport uvicorn from fastapi import FastAPI from genkit.agent import InMemorySessionStore from genkit_fastapi import serve_agent ai Genkit(plugins[GoogleAI()], modelgoogleai/gemini-flash-latest) agent ai.define_agent( namebackgroundAgent, systemSenior research analyst. Produce a comprehensive markdown report., storeInMemorySessionStore(), # detach 必需 ) app FastAPI() app.include_router(serve_agent(agent), prefix/api) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8080)路由默认落在/{agent.name}即/api/backgroundAgent。运行方式遵循 SKILL.md 的推荐模式以捕获完整 trace 便于排查genkit start -- uv run server.py客户端remote_agentfrom genkit.agent import remote_agent client remote_agent( urlhttp://127.0.0.1:8080/api/backgroundAgent, state_managementserver, # 服务端配置了 store则必须为 server ) chat client.chat() task await chat.detach(Write a report on renewable energy trends) print(task.snapshot_id) snapshot await task.wait(interval2.0) print(snapshot.status) done await client.load_chat(snapshot_idtask.snapshot_id) print(done.messages[-1].text)注意两点state_management必须与服务端是否配置 store 保持一致有 store 用serverURL 不要带尾部斜杠。这样detach 的提交、轮询、取消、恢复四个动作就全部跑在了 HTTP 之上客户端拿到snapshot_id即可立即返回给前端轮询。八、边界与注意事项小结Beta APIgenkit.agent模块整体属于预览性质接口可能随版本演进升级依赖时留意 changelog一个 chat 一个 detach并发任务请使用第六节的“共享检查点 fork 叶子”模式而不是在同一个 chat 上重复 detachabort 必须 await不等待的取消等于无效首轮对话优先用detachtask.abort()实现可恢复取消aborted 快照不可作为恢复点恢复请回到最后一个正常的叶子快照wait 后要读消息status completed不等于内容合格务必检查最终消息store 选择原型用InMemorySessionStore需要跨进程/重启恢复用FileSessionStore多用户按session_id/snapshot_id隔离。九、关联资料导航Agents 总览Betadefine_agent、middleware、chat、无 store 模式与 CLI 验证Sessions Persistence两种 store 的选型与高级参数Agent Branching不可变快照与分支 fork 的底层机制Agent State会话三层的类型化状态管理Agent Human-in-the-Loop中断与恢复语义与 abort 的区别Agent HTTPserve_agent与remote_agent服务化Genkit Python 总览环境准备、genkit start/genkit flow:run的运行与调试姿势【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Genkit Python 后台 Agent（Background Agents / Detach）实战指南：把长任务交给服务端，用快照句柄继续你的工作
Genkit Python 后台 AgentBackground Agents / Detach实战指南把长任务交给服务端用快照句柄继续你的工作【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skillsGenkit Python 的 Agent API 提供了detach后台分离机制把一轮耗时的模型调用“交给服务端”调用立即返回一个携带snapshot_id的任务句柄等真正需要结果时再按快照加载并恢复对话。本文以 agents-background.md 为核心结合会话持久化、分支与 HTTP 服务化文档完整讲解 detach 的前置条件、核心 API、取消语义、并行研究模式与落地注意事项读完即可在真实项目中实现“提交后台任务 → 轮询结果 → 随时中止 → 恢复对话”的完整闭环。说明Background Agents / Detaching 属于 Beta / 预览 API位于genkit.agent模块下。它必须搭配 session store 使用——服务端需要把后台任务的结果写入某个地方客户端才有地方轮询和恢复。建议先阅读 agents.md 了解 Agent 基础概念。一、背景与设计动机detach 解决什么问题普通的chat.send(...)是一次同步往返模型未返回完整回复之前调用方只能阻塞等待。对于“生成一份长篇研究报告”“批量调研多个主题”“跑一遍复杂的多工具调用链”这类耗时任务同步等待会拖垮 Web 请求线程也不利于做异步 UI比如先返回任务 ID之后由前端轮询。Genkit Python 的 detach 模式把交互模型从“请求-响应”改为“提交-轮询”提交chat.detach(message)把整轮任务交给服务端异步执行立即返回一个带有snapshot_id的任务句柄DetachedTask此时任务状态为pending轮询task.wait(interval...)按固定间隔轮询 session store直到快照进入终态completed/failed/aborted等恢复任务完成后用agent.load_chat(snapshot_id...)加载最终快照拿到完整对话消息可以继续追问中止任何时候都可以通过task.abort()取消后台任务。这套机制以不可变快照snapshot_id为骨架每轮对话都会在 store 中写入一个不可变的快照这正是 分支branching 和后台任务共同的底层能力。二、前置条件必须配置 Session Store文档开篇明确声明detach需要 session store。原因很直接——后台任务是在服务端独立执行的客户端与服务端之间唯一的“信物”就是快照 ID任务结束后的结果必须持久化到 store 中客户端才能凭snapshot_id轮询到终态并读取最终消息。从 agents-sessions.md 可知Genkit Python 提供两种开箱即用的 storefrom genkit.agent import InMemorySessionStore, FileSessionStore mem_store InMemorySessionStore() # 内存存储进程重启即丢失 file_store FileSessionStore(./.snapshots) # 磁盘存储每个快照一个 JSON 文件要点InMemorySessionStore最简单适合单进程原型验证缺点是重启后快照全部丢失后台任务结果也无法跨进程读取FileSessionStore每个快照落盘为一个 JSON 文件支持进程重启后恢复也支持更多高级选项pruning FileSessionStore( ./.snapshots, max_persisted_chain_length3, # 可选修剪过长的对话链 reject_ambiguous_sessionTrue, # 可选分支后拒绝含混的 session_id )多人共享每个agent.chat()在共享的 store 上拥有独立 session按用户的session_id或snapshot_id分别恢复会话。三、核心 API 实战detach → wait → abort → load_chat原文给出了一个完整的后台报告生成示例这里完整继承并逐步拆解from genkit.agent import InMemorySessionStore agent ai.define_agent( namebackgroundAgent, modelgoogleai/gemini-flash-latest, systemSenior research analyst. Produce a comprehensive markdown report., storeInMemorySessionStore(), ) chat agent.chat() task await chat.detach(Write a report on renewable energy trends) print(task.snapshot_id) snapshot await task.wait(interval2.0) print(snapshot.status) await task.abort() done await agent.load_chat(snapshot_idtask.snapshot_id) print(done.messages)1.chat.detach(message)提交即返回detach与普通send的差别在于立即返回。它返回一个任务句柄句柄上最重要的字段是snapshot_id——这就是你在后台任务执行期间与它建立联系的凭证。服务端此刻已经把任务标记为pending并写入 store随后在后台继续处理模型调用与工具调用。注意模型 ID 使用带前缀的完整形式googleai/gemini-flash-latest这是 SKILL.md 中强调的规范需要先在环境中配置GEMINI_API_KEY。2.task.wait(interval2.0)轮询直至终态wait会以interval秒为周期轮询 store 中该快照的状态直到进入终态才返回最终快照。快照的status字段可取以下值状态含义pending仍在后台处理中继续轮询completed成功完成从快照的 messages 中读取结果failed处理过程中发生错误aborted被客户端通过abort()取消expired后台 worker 停止响应如服务端重启心跳失效视为终态任务永远无法完成轮询间隔需要权衡间隔太短会增加 store 读压力太长则用户体验上等待变久。原文示例取2.0秒适合报告生成这类秒级到分钟级的任务。3.task.abort()随时中止拿到句柄后任何时刻都可以调用await task.abort()取消后台任务。被中止的快照会标记为aborted。这一点在“提交后用户反悔”或“任务已无必要”的场景下非常重要避免白白消耗模型 token。4.agent.load_chat(snapshot_id...)按快照恢复对话任务完成后用task.snapshot_idwait返回的快照同 ID重新加载出一个可继续对话的chat对象读取done.messages即可拿到包含最终模型回复的完整消息列表也可以继续send追问细节。这与普通会话的恢复方式完全一致——见 agents.md 中load_chat注意不是chat(snapshot_id...)后者只是挂载一个恢复句柄。四、停止工作客户端停止与服务端取消是两回事“停止”在 detach 语义下要区分两个层面原文对此有非常明确的告诫客户端停止Client stopawait turn.abort() # 或对流式发送使用 asyncio.timeout 包裹 stream作用停止监听响应流。例如用asyncio.timeout包裹一个流式请求超时就放弃等待局限服务端可能仍在继续执行。因为此时你可能还没有拿到snapshot_id比如正在处理第一轮对话的中途适用时机流式读取的中途、尚未持有快照 ID、不关心后台是否继续完成的场景。服务端取消Server cancelawait chat.abort() # 或 await task.abort()前置条件必须配置了 store且已存在一个有效快照一轮对话完成后或detach之后语义真正取消服务端的后台工作把快照标记为aborted适用时机持有快照 ID 之后想要彻底回收任务资源。两个关键警告被中止的快照不是恢复点Aborted snapshots are not resume points。不要尝试从aborted快照继续对话——如果需要恢复应重新加载“最后一个正常叶子”快照上一个处于正常状态的快照第一轮对话的客户端中止陷阱在首轮对话上执行客户端 abort可能让这个 chat 永久丢失 ID——即使服务端之后保存了快照客户端侧也可能拿不到。原文给出的可恢复取消的推荐姿势是优先detachtask.abort()而不是中途硬断流。另外原文还强调abort 一定要await——“Alwaysawaitaborts — otherwise nothing happens”。异步取消如果不等待完成协程尚未执行到取消逻辑就被丢弃等于什么都没做。五、并行研究模式一个 chat 只能挂一个 detach用分支实现并发原文明确指出一个限制一个chat只能承载一个 detach。如果要对多个主题并行发起后台研究正确的做法是从一个共享检查点checkpoint为每个分支 fork 出独立的叶子对话root agent.chat() await root.send(Context both researchers should know.) checkpoint root.snapshot_id async def research(topic: str): leaf await agent.load_chat(snapshot_idcheckpoint) task await leaf.detach(fResearch {topic} in depth.) await task.wait(interval2.0) done await agent.load_chat(snapshot_idtask.snapshot_id) return topic, done.messages[-1] import asyncio results await asyncio.gather(research(Postgres), research(SQLite))拆解这个模式的要点先播种公共上下文root.send(...)让所有分支共享同一段背景知识比如“两份研究报告都要遵循同样的格式与受众”拿到检查点root.snapshot_id即不可变检查点每个分支 fork 叶子agent.load_chat(snapshot_idcheckpoint)从同一检查点分别加载独立的叶子对话互不干扰各自 detach wait每个叶子独立提交后台任务、独立轮询用asyncio.gather并发调度research是 async 函数gather让多个研究分支真正并发推进。这种“共享检查点 分支叶子 并发 gather”的写法本质上是把 agents-branching.md 中“从同一快照 fork 出多条分叉”的能力与 detach 结合起来。文档最后还特别提醒wait 之后要检查加载出的消息内容——仅凭status不足以判断结果质量completed也可能返回空或不符合预期的文本务必读取messages[-1]最后一条消息确认实际产出。六、store 解锁的能力边界何时该用 detach结合 agents-sessions.md 的能力对照表可以更清晰地定位 detach 的适用场景能力无 store有 store单 chat 多轮对话✅✅中断HITL✅✅持久化 ID /load_chat❌✅分支branching❌✅detach / 后台任务❌✅服务端取消server abort❌✅也就是说如果你的应用自己管理历史省略store参数IDs 保持None通过把messages/state/artifacts传入下一个chat(...)来接力那么 detach、分支和服务端取消都是不可用的。后台任务天然要求服务端“拥有”历史这也是 detach 以 store 为前提的根本原因。七、生产化落地把后台 Agent 通过 HTTP 暴露出去纯本地脚本可以直接await上面的代码但真实项目通常是一个服务端进程执行后台任务一个客户端进程提交与轮询。此时可借助 agents-http.md 中的 FastAPI 服务化方案。服务端serve_agentimport uvicorn from fastapi import FastAPI from genkit.agent import InMemorySessionStore from genkit_fastapi import serve_agent ai Genkit(plugins[GoogleAI()], modelgoogleai/gemini-flash-latest) agent ai.define_agent( namebackgroundAgent, systemSenior research analyst. Produce a comprehensive markdown report., storeInMemorySessionStore(), # detach 必需 ) app FastAPI() app.include_router(serve_agent(agent), prefix/api) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8080)路由默认落在/{agent.name}即/api/backgroundAgent。运行方式遵循 SKILL.md 的推荐模式以捕获完整 trace 便于排查genkit start -- uv run server.py客户端remote_agentfrom genkit.agent import remote_agent client remote_agent( urlhttp://127.0.0.1:8080/api/backgroundAgent, state_managementserver, # 服务端配置了 store则必须为 server ) chat client.chat() task await chat.detach(Write a report on renewable energy trends) print(task.snapshot_id) snapshot await task.wait(interval2.0) print(snapshot.status) done await client.load_chat(snapshot_idtask.snapshot_id) print(done.messages[-1].text)注意两点state_management必须与服务端是否配置 store 保持一致有 store 用serverURL 不要带尾部斜杠。这样detach 的提交、轮询、取消、恢复四个动作就全部跑在了 HTTP 之上客户端拿到snapshot_id即可立即返回给前端轮询。八、边界与注意事项小结Beta APIgenkit.agent模块整体属于预览性质接口可能随版本演进升级依赖时留意 changelog一个 chat 一个 detach并发任务请使用第六节的“共享检查点 fork 叶子”模式而不是在同一个 chat 上重复 detachabort 必须 await不等待的取消等于无效首轮对话优先用detachtask.abort()实现可恢复取消aborted 快照不可作为恢复点恢复请回到最后一个正常的叶子快照wait 后要读消息status completed不等于内容合格务必检查最终消息store 选择原型用InMemorySessionStore需要跨进程/重启恢复用FileSessionStore多用户按session_id/snapshot_id隔离。九、关联资料导航Agents 总览Betadefine_agent、middleware、chat、无 store 模式与 CLI 验证Sessions Persistence两种 store 的选型与高级参数Agent Branching不可变快照与分支 fork 的底层机制Agent State会话三层的类型化状态管理Agent Human-in-the-Loop中断与恢复语义与 abort 的区别Agent HTTPserve_agent与remote_agent服务化Genkit Python 总览环境准备、genkit start/genkit flow:run的运行与调试姿势【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
 完整解析)
