Genkit Python Agent State 完全指南：三层会话状态、客户端托管与输出脱敏

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 的定义，一个会话由三层数据组成：

1. messages：多轮对话的消息历史；
2. custom state（由state_schema声明类型）：产品自定义状态，例如路由标识、当前工单号、用户等级等业务上下文；
3. 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( name='profileAgent', system='Greet the user by name when you know it.', state_schema=Profile, ) chat = agent.chat(state=Profile(name='Ada', tier='pro')) 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(state=Profile(name='Ada', tier='pro')) await chat.send('Hello') resumed = agent.chat( messages=chat.messages, state=chat.state, artifacts=chat.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( messages=chat.messages, state=chat.state, artifacts=chat.artifacts ) await resumed.send('What is my name? One word.')

自定义状态不会自动注入模型

一个容易踩的误区：自定义状态是给你的产品用的（路由、UI 展示），默认不会注入给模型。即使声明了state_schema，模型也看不到chat.state，除非你显式把它放进 system prompt 或消息里。

更重要的约束是：仅仅声明state_schema并不会自动填充chat.state。必须有一方在某轮会话中调用update_custom，状态才会真正产生值。官方建议：在普通 Agent（ai.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()], model='googleai/gemini-flash-latest') class Profile(BaseModel): name: str tier: str = 'free' agent = ai.define_agent( name='profileAgent', system='Greet the user by name when you know it.', state_schema=Profile, ) chat = agent.chat(state=Profile(name='Ada', tier='pro')) await chat.send('Hello') print(chat.state.name)

要点回顾：

• state_schema=Profile声明状态的类型骨架；
• chat(state=Profile(...))完成播种（仅限无 Store 场景）；
• 由于声明了 schema，chat.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(description='Support case id') ai = Genkit(plugins=[GoogleAI(), Middleware()], model='googleai/gemini-flash-latest') @ai.tool(name='openCase', description='Open 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_id=input.case_id, status=base.status) await sess.update_custom(mutate) return {'ok': True, 'case_id': input.case_id} agent = ai.define_agent( name='supportOps', system='Support ops. Call openCase when asked. Be brief.', tools=[open_case], state_schema=CaseState, use=[ToolApproval(allowed_tools=['openCase'])], store=InMemorySessionStore(), )

这段代码里值得展开的细节：

1. ai.current_session()可能返回None——例如在无会话上下文的地方（如独立 flow）调用工具时，务必做空值保护；
2. mutator 的幂等写法：c if isinstance(c, CaseState) else CaseState.model_validate(c or {})同时兼容"已类型化"与"收到 dict"两种输入，既安全又健壮；
3. Store 选择：InMemorySessionStore()重启即丢失；若需落盘，可用FileSessionStore('./.snapshots')（每个快照一个 JSON 文件），并可配置max_persisted_chain_length剪枝与reject_ambiguous_session防分支歧义，见 agents-sessions.md；
4. 配套的运行时导入一览可参考 SKILL.md：from genkit.agent import InMemorySessionStore, ...、from genkit_middleware import Middleware, ToolApproval, ...。

Live patches：流式增量更新

在自定义 Agent（agents-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：改变客户端看到的会话状态快照。必须返回一个完整的SessionState；
• chunk_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(messages=state.messages, custom=custom, artifacts=state.artifacts) agent = ai.define_agent(..., state_schema=SecretState, state_transform=redact, 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；
• 自定义 Agent：define_custom_agent中通过sess.get_custom()/sess.update_custom()直接管理状态，并可用ctx.send_chunk(AgentStreamChunk(...))自定义流式输出，见 agents-custom.md；
• HTTP 服务：genkit_fastapi的serve_agent可以把多用户各自的状态安全地隔离在服务端，客户端只转发凭证，见 agents-http.md。

小结与上手路径

在 Genkit Python 中管理 Agent 状态，核心记住四条规则：

1. 会话 = messages + custom state（state_schema）+ artifacts 三层；声明 schema 后chat.state与chunk.custom都是类型化对象；
2. 有 Store 时不能chat(state=...)播种，状态由运行时和工具写入；无 Store 时由应用自行播种并往返messages/state/artifacts；
3. 想让状态被写入，必须在回合里调用update_custom（优先在普通 Agent 的工具中通过ai.current_session()完成，以保留中间件）；
4. 对外输出前用state_transform/chunk_transform脱敏，且不影响 Store 内保留的原始数据。

开始动手前，先按 setup.md 用uv建好项目（Python 3.10+，uv add genkit genkit-google-genai，Agent 中间件场景再加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