Genkit Python Agent State 完全指南三层会话状态、客户端托管与输出脱敏【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills本文基于仓库内skills/cloud/genkit-python技能中 agents-state.md 整理展开用于在 Genkit Python 中为多轮 Agent 会话引入类型化自定义状态custom state从会话的三层结构消息、状态、产物出发讲解无 Store 的客户端托管模式、有 Store 的持久化模式以及如何通过工具回调更新状态、如何在流式输出中做增量补丁最后落到输出脱敏等安全细节。读完本文你将能够在自己的 Genkit Python 应用中设计能路由、能驱动 UI、能被安全地裁剪后再返回给客户端的 Agent 会话状态。会话携带的三层数据Genkit Python 的 Agent 会话Session不止是一堆聊天消息。根据 agents-state.md 的定义一个会话由三层数据组成messages多轮对话的消息历史custom state由state_schema声明类型产品自定义状态例如路由标识、当前工单号、用户等级等业务上下文artifacts命名产物报告、文件、代码片段详见 agents-artifacts.md。这三层在chat对象上分别以chat.messages、chat.state、chat.artifacts暴露。一个关键行为是当你为 Agent 声明了state_schema之后chat.state以及流式输出中的chunk.custom都会以该 Pydantic 模型的类型返回而不是裸字典这让状态在运行时是类型化的可以直接访问字段。class Profile(BaseModel): name: str tier: str free agent ai.define_agent( nameprofileAgent, systemGreet the user by name when you know it., state_schemaProfile, ) chat agent.chat(stateProfile(nameAda, tierpro)) await chat.send(Hello) print(chat.state.name) # Ada类型化访问而非 dict两种状态管理模式有 Store 与无 Store理解状态的关键是历史由谁持有。Genkit 提供两种模式两者在 API 使用上有明显差异有 Store 模式—— 历史由 Genkit 持有通过snapshot_id或session_id恢复会话agent.load_chat(snapshot_id...)不能再用chat(state...)手动播种状态——状态只能由运行时/tool 写入这是服务端持有历史的推荐方式也是分支branching、后台任务detach等高级能力的前提参见 agents-sessions.md 与 agents-branching.md。无 Store 模式—— 历史由你的应用持有由你自行播种状态并在每次轮次之间手动往返三层数据可以在agent.chat()里直接传入state、messages、artifacts。无 Store 模式的核心写法原文代码直接可运行chat agent.chat(stateProfile(nameAda, tierpro)) await chat.send(Hello) resumed agent.chat( messageschat.messages, statechat.state, artifactschat.artifacts )也就是说没有 Store 时session_id/snapshot_id都是None应用自己把上一轮的结果原样传回下一轮即可这等价于 agents.md 中Without a store一节的 echo 示例做法chat agent.chat() await chat.send(My name is Ada. Remember it.) resumed agent.chat( messageschat.messages, statechat.state, artifactschat.artifacts ) await resumed.send(What is my name? One word.)自定义状态不会自动注入模型一个容易踩的误区自定义状态是给你的产品用的路由、UI 展示默认不会注入给模型。即使声明了state_schema模型也看不到chat.state除非你显式把它放进 system prompt 或消息里。更重要的约束是仅仅声明state_schema并不会自动填充chat.state。必须有一方在某轮会话中调用update_custom状态才会真正产生值。官方建议在普通 Agentai.define_agent中优先通过工具tool来调用update_custom因为这样你仍然保留中间件middleware链路如果改用define_custom_agent手工编排就需要自己承担更多运行时职责详见 agents-custom.md。客户端托管类型化状态下面是一个完整的客户端托管示例应用负责持有状态并通过chat(state...)播种一个类型化状态。注意这里没有传入store因此可以播种from pydantic import BaseModel from genkit import Genkit from genkit_google_genai import GoogleAI ai Genkit(plugins[GoogleAI()], modelgoogleai/gemini-flash-latest) class Profile(BaseModel): name: str tier: str free agent ai.define_agent( nameprofileAgent, systemGreet the user by name when you know it., state_schemaProfile, ) chat agent.chat(stateProfile(nameAda, tierpro)) await chat.send(Hello) print(chat.state.name)要点回顾state_schemaProfile声明状态的类型骨架chat(stateProfile(...))完成播种仅限无 Store 场景由于声明了 schemachat.state返回的是Profile实例可以直接print(chat.state.name)。Store 中间件从工具里更新状态当启用了 Store 与中间件Middleware()之后状态更新要走ai.current_session()。核心 API 是会话对象上的sess.update_custom(mutator)update_custom接收一个异步函数mutator该函数接收当前 custom state返回新状态mutator 必须是 async如果回调收到的是 dict需要自己强转成你的模型用model_validate。下面这段是原文的支持工单完整示例工具openCase在每次被调用时从ai.current_session()拿到当前会话然后把case_id写入CaseState。同时注意ToolApproval(allowed_tools[openCase])允许该工具自动执行不打断用户其余工具才需要人工审批中间件细节见 agents-human-in-the-loop.md。from pydantic import BaseModel, Field from genkit import Genkit from genkit.agent import InMemorySessionStore from genkit_google_genai import GoogleAI from genkit_middleware import Middleware, ToolApproval class CaseState(BaseModel): case_id: str status: str open class OpenCaseInput(BaseModel): case_id: str Field(descriptionSupport case id) ai Genkit(plugins[GoogleAI(), Middleware()], modelgoogleai/gemini-flash-latest) ai.tool(nameopenCase, descriptionOpen or update the support case id.) async def open_case(input: OpenCaseInput) - dict: sess ai.current_session() if sess is None: return {ok: False, error: no session} async def mutate(c: object) - CaseState: base c if isinstance(c, CaseState) else CaseState.model_validate(c or {}) return CaseState(case_idinput.case_id, statusbase.status) await sess.update_custom(mutate) return {ok: True, case_id: input.case_id} agent ai.define_agent( namesupportOps, systemSupport ops. Call openCase when asked. Be brief., tools[open_case], state_schemaCaseState, use[ToolApproval(allowed_tools[openCase])], storeInMemorySessionStore(), )这段代码里值得展开的细节ai.current_session()可能返回None——例如在无会话上下文的地方如独立 flow调用工具时务必做空值保护mutator 的幂等写法c if isinstance(c, CaseState) else CaseState.model_validate(c or {})同时兼容已类型化与收到 dict两种输入既安全又健壮Store 选择InMemorySessionStore()重启即丢失若需落盘可用FileSessionStore(./.snapshots)每个快照一个 JSON 文件并可配置max_persisted_chain_length剪枝与reject_ambiguous_session防分支歧义见 agents-sessions.md配套的运行时导入一览可参考 SKILL.mdfrom genkit.agent import InMemorySessionStore, ...、from genkit_middleware import Middleware, ToolApproval, ...。Live patches流式增量更新在自定义 Agentagents-custom.md里update_custom还有一个特殊行为更新会以chunk.custom的形式流式推送给客户端。这意味着你可以在回合进行中持续发布状态增量客户端可以实时感知例如展示正在生成…的进度计数。async def bump(c): return {turns: (c or {}).get(turns, 0) 1} await sess.update_custom(bump)结合前面有 schema 时chunk.custom以模型类型返回的行为这里的自定义状态更新与流式协议是同一套数据通路服务端每次调用update_custom客户端流里就会收到对应的chunk.custom补丁方便驱动 UI 进度、轮次计数等实时展示。输出脱敏state_transform 与 chunk_transform当服务端要响应客户端时并不总是希望把完整状态原样交给客户端。Genkit 提供两个出口变换state_transform改变客户端看到的会话状态快照。必须返回一个完整的SessionStatechunk_transform逐 chunk 改写流式输出。返回None表示丢弃该 chunk。关键设计这两个变换只影响客户端所见不改变 Store 中保存的内容——也就是说落盘/持久化始终保留原始完整状态脱敏只发生在离开服务端的边界上。这在包含密钥等敏感字段的业务里非常实用from genkit.agent import SessionState def redact(state: SessionState) - SessionState: custom dict(state.custom or {}) if api_key in custom: custom[api_key] REDACTED return SessionState(messagesstate.messages, customcustom, artifactsstate.artifacts) agent ai.define_agent(..., state_schemaSecretState, state_transformredact, store...)注意事项state_transform不能只返回部分状态必须构造完整的SessionState(messages..., custom..., artifacts...)因为state_transform是同步函数签名如上例适合做字符串替换、字段抹除等纯变换若需要异步或复杂过滤可结合chunk_transform在流式路径上做裁剪该能力与自定义状态不进模型配合可以形成完整的安全边界敏感数据只存在 Store只在必要时通过工具读入对话且出口再脱敏一次。与其他 Agent 能力的协同关系自定义状态不是孤立功能它嵌入在 Genkit Agent 的整体会话模型中会话与持久化状态是会话三层的中间层Store 决定谁来持有历史、能否播种详见 agents-sessions.md分支每个快照都是不可变检查点从同一 checkpoint 派生不同方向时不同分支可以各自维护状态见 agents-branching.md后台任务detach出的后台任务同样有独立会话与状态完成后再load_chat(snapshot_id...)读取见 agents-background.md自定义 Agentdefine_custom_agent中通过sess.get_custom()/sess.update_custom()直接管理状态并可用ctx.send_chunk(AgentStreamChunk(...))自定义流式输出见 agents-custom.mdHTTP 服务genkit_fastapi的serve_agent可以把多用户各自的状态安全地隔离在服务端客户端只转发凭证见 agents-http.md。小结与上手路径在 Genkit Python 中管理 Agent 状态核心记住四条规则会话 messages custom statestate_schema artifacts 三层声明 schema 后chat.state与chunk.custom都是类型化对象有 Store 时不能chat(state...)播种状态由运行时和工具写入无 Store 时由应用自行播种并往返messages/state/artifacts想让状态被写入必须在回合里调用update_custom优先在普通 Agent 的工具中通过ai.current_session()完成以保留中间件对外输出前用state_transform/chunk_transform脱敏且不影响 Store 内保留的原始数据。开始动手前先按 setup.md 用uv建好项目Python 3.10uv add genkit genkit-google-genaiAgent 中间件场景再加genkit-middleware再回到 agents.md 完成一个带state_schema的最小 Agent最后用genkit start -- uv run src/main.py跑起来并在终端用genkit trace:get traceId --format json验证状态与工具调用的真实链路运行方式详见 SKILL.md。【免费下载链接】skillsAgent Skills for Google products and technologies项目地址: https://gitcode.com/GitHub_Trending/skills29/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考