很多智能体工作坊会提供一个完整仓库，让参与者复制命令、运行代码，然后以一排绿色勾号结束。纽约 DevFest Community Workshop 尝试了另一条路线：约 80 名工程师不再把注意力放在样板代码上，而是集中理解图式工作流、状态持久化、人工审批、评估门禁和受约束的自动修复。
这组能力决定了智能体能否从演示走向生产。一个能回答问题的 Agent 并不难做；难的是让它运行数小时甚至数天，安全地暂停和恢复，并在模型输出偏离预期时阻止错误继续扩散。
Prompt 不是执行边界
工作坊点出了智能体系统中一个常被忽略的问题：自然语言指令本质上是概率性建议，而不是可强制执行的安全边界。
例如，下面的提示词表达了业务意图，却不能提供真正的预算约束：
你是一名广告竞价智能体。
请提高转化率，但绝对不要让单次出价超过 100 美元。
如果数据异常，请谨慎处理。
模型仍可能误读单位、忽略上下文，或在工具调用时传入错误参数。生产系统应把“不得超过 100 美元”落实为普通代码、数据库约束或策略引擎规则：
def enforce_bid_policy(proposed_bid: float, hard_limit: float = 100.0) -> float:
if proposed_bid < 0:
raise ValueError('bid cannot be negative')
return min(proposed_bid, hard_limit)
Prompt 可以帮助模型形成建议，但最终执行必须经过确定性校验。对于付款、权限变更、部署和数据删除等高风险操作，还要加入人工审批，不能把最终决定交给一段措辞更强硬的提示词。
长任务的关键：把状态和计算分开
工作坊使用 Google 的 Agent Development Kit、Gemini Enterprise Agent Platform 上的 Memory Bank 与 RAG Engine，以及 Veo 3.1 等组件探索智能体工作流。其中最重要的架构原则之一，是将持久状态与活跃计算解耦。
假设一个流程生成竞价方案后，需要等待运营人员批准。审批可能要几个小时。如果进程一直占用容器、线程或模型会话，不仅成本高，重启后也很难恢复。更稳妥的设计是：
- 每个节点完成后写入检查点；
- 遇到人工审批时保存状态并退出；
- 审批事件到达后重新加载状态；
- 从明确的下一节点继续，而不是重放整个流程。
状态中至少应记录：
- 当前工作流节点和版本；
- 输入、工具调用结果与关键决策；
- 审批状态和审批人；
- 重试次数、幂等键与超时时间；
- 评估结果及自动修复历史。
Memory Bank 一类组件可以承担长期记忆，RAG Engine 可为运行节点提供检索上下文，而真正的业务进度仍应使用结构明确、可审计的状态保存。不要把聊天记录误当作工作流数据库。
一个可运行的“暂停—审批—评估—修复”示例
下面是一个仅依赖 Python 标准库的教学示例。它不是 Google ADK 的官方 API，而是用于展示工作坊强调的架构模型：状态落盘、进程退出、人工批准、确定性评估，以及受约束的参数修复。
将以下内容保存为 agent_workflow.py：
import json
import sys
from pathlib import Path
STATE_FILE = Path('state.json')
def initial_state():
return {
'stage': 'plan',
'approved': False,
'budget': 500.0,
'auction_history': [90.0, 110.0, 100.0],
'max_bid': 120.0,
'proposed_bid': None,
'patches': [],
}
def load_state():
if not STATE_FILE.exists():
return initial_state()
return json.loads(STATE_FILE.read_text())
def save_state(state):
STATE_FILE.write_text(json.dumps(state, indent=2))
def run():
state = load_state()
while True:
if state['stage'] == 'plan':
history = state['auction_history']
raw_bid = sum(history) / len(history) * 1.15
state['proposed_bid'] = round(raw_bid, 2)
state['stage'] = 'waiting_approval'
save_state(state)
print('Paused: review state.json, then run approve.')
return
if state['stage'] == 'waiting_approval':
print('Still waiting for human approval.')
return
if state['stage'] == 'execute':
state['proposed_bid'] = min(
state['proposed_bid'], state['max_bid']
)
state['stage'] = 'validate'
save_state(state)
if state['stage'] == 'validate':
safe_limit = round(state['budget'] * 0.20, 2)
if state['proposed_bid'] > safe_limit:
patch = {
'field': 'max_bid',
'old': state['max_bid'],
'new': safe_limit,
'reason': 'bid exceeded 20% of total budget',
}
state['patches'].append(patch)
state['max_bid'] = safe_limit
state['stage'] = 'execute'
save_state(state)
print('Eval failed; applied constrained patch:', patch)
continue
state['stage'] = 'done'
save_state(state)
print('Completed safely:', state['proposed_bid'])
return
if state['stage'] == 'done':
print('Workflow already completed.')
return
def approve():
state = load_state()
if state['stage'] != 'waiting_approval':
raise SystemExit('Workflow is not waiting for approval.')
state['approved'] = True
state['stage'] = 'execute'
save_state(state)
print('Approved. Run the workflow again to resume.')
def reset():
STATE_FILE.unlink(missing_ok=True)
print('State cleared.')
if __name__ == '__main__':
command = sys.argv[1] if len(sys.argv) > 1 else 'run'
{'run': run, 'approve': approve, 'reset': reset}[command]()
按以下顺序运行：
python agent_workflow.py reset
python agent_workflow.py run
cat state.json
python agent_workflow.py approve
python agent_workflow.py run
cat state.json
第一次执行会在审批节点退出，因此等待期间没有常驻计算。批准后，流程从 execute 节点恢复。评估器发现建议出价超过预算的 20%，便只修改允许修复的 max_bid 参数，再重新执行和验证。
生产实现需要把本地 JSON 替换成具备事务能力的存储，并为每个外部操作加入幂等键。自动修复也不应任意改写代码；更安全的方式是限定可修改字段、设置上下界、保留旧版本，并要求修复结果重新通过评估门禁。
自进化不是“让 Agent 随便改自己”
工作坊还通过 BigQuery 中的历史数据构建自主数据工程与竞价流程，并用评估门禁捕获支出异常。这里的“自进化”更适合理解为受控闭环：
观测运行结果
→ 生成候选调整
→ 在隔离环境中评估
→ 通过策略与预算门禁
→ 小范围发布
→ 持续监控并允许回滚
模型可以提出补丁，但不能同时担任提案者、审批者和审计者。尤其当智能体能够查询 BigQuery、调用外部服务或调整竞价时，应分离以下权限：
- 读取历史数据；
- 生成执行计划；
- 修改运行参数；
- 发起真实交易；
- 批准高风险动作。
评估也不能只检查回答是否“看起来合理”。竞价场景应验证预算上限、累计支出、单位一致性和异常波动；数据管道则要检查 schema、行数变化、空值比例与重复写入。
落地前检查这几件事
团队开始构建长时间运行的多智能体系统时，可以先完成以下检查：
- 流程可恢复：任一节点崩溃后，都能从最近检查点继续。
- 副作用幂等：重试不会重复付款、发消息或写入数据。
- 规则在代码中：预算、权限和合规要求不只存在于 Prompt。
- 审批会真正暂停：等待期间不占用昂贵的活跃计算资源。
- 修复受到约束：Agent 只能修改白名单参数，并且必须重新评估。
- 状态可审计：能回答谁在何时批准了什么、系统为何做出决定。
- 组件可替换：模型、检索系统和执行器之间通过稳定接口连接。
Workbench 这类工作坊的价值，不是让参与者多记住几个 SDK 方法，而是迫使工程师面对智能体最难的部分：状态、边界、恢复、评估和责任归属。框架会变化，但这些架构问题不会消失。