技术热点落地:Google AX Agent Executor — 让 AI Agent 跑通生产环境(2026-05-26)
技术热点落地:Google AX Agent Executor — 让 AI Agent 跑通生产环境
适用场景与目标
适用场景:
- 需要运行”长时运行”AI Agent(分钟级到天级不断续的任务)
- Agent 需跨多系统调用工具、调用技能(Skill),并在网络中断、人为介入后恢复
- 团队从 AI 原型验证转向生产级 Agent 部署,但不想被单一云厂商绑定
- 需要在自托管环境运行多 Agent 协同,且要求运行时具备容错和可审计性
核心目标: 通过 Google 刚开源的 AX(Agent Executor),构建一套可本地部署、支持中断恢复、具备沙箱隔离能力的 Agent 基础设施,不必依赖 Google Cloud 服务即可上手。
最小可行方案(MVP)
1. 环境准备
# 基础依赖:Python 3.11+,Git
python --version # >= 3.11
git --version
# 推荐用 venv 隔离(避免污染全局)
python -m venv ax-env
source ax-env/bin/activate
# 克隆 AX 仓库(截至 2026-05-25,官方已开源)
git clone https://github.com/google/agent-executor.git
cd agent-executor
2. 安装核心依赖
# 必装:运行时核心 + 观测能力(Prometheus metrics)
pip install ax-core ax-observability
# 推荐:本地沙箱支持(独立进程隔离 Tool 执行)
pip install ax-sandbox
# 可选:与 Gemini 模型对接(替换为其他 LLM provider 同理)
pip install google-genai
# 验证安装
axctl --version
3. 定义你的第一个 Agent
创建文件 my_agent.py:
import ax
from ax.runtime import Agent, Skill, Tool
# 定义一个 Tool(Agent 调用外部能力的最小单元)
@Tool(name="fetch_weather", description="查询城市天气")
def fetch_weather(city: str) -> str:
"""这是一个受沙箱保护的 Tool,不会泄露主进程环境变量"""
return f"{city} 今日晴,气温 22°C"
# 定义一个 Skill(可组合的技能单元)
class SummarizeSkill(Skill):
name = "summarize"
description = "对文本做摘要"
def execute(self, text: str) -> str:
# 实际生产中可接 Gemini 或本地 LLM
return f"[摘要] {text[:50]}..."
# 构建 Agent
agent = Agent(
model="gemini-2.5-flash", # 可替换为 Claude/OAI 模型
tools=[fetch_weather],
skills=[SummarizeSkill()],
enable_resumption=True, # ⭐ 关键:开启断点续传能力
)
# 启动运行(阻塞式,适合 CLI 验证)
result = ax.run(agent, input="北京天气怎么样?并对结果做摘要")
print(result.output)
4. 启动分布式运行时(Single Node MVP)
# 启动 AX Runtime 服务(默认端口 8080,Prometheus metrics 端口 9090)
axctl runtime start --port 8080
# 查看 Agent 状态和历史轨迹
axctl agent list
axctl agent trajectory <agent-id> # 查看完整执行路径,用于 Debug
5. 配置长期任务 & 中断恢复
编辑 ax.yaml(放在项目根目录):
runtime:
resumption:
enabled: true
checkpoint_interval_ms: 30000 # 每 30s 打一次检查点
storage: local # MVP 用本地目录;生产换 S3/GCS
sandbox:
enabled: true
per_tool_isolation: true # 每个 Tool 独立进程,Tool 之间互不影响
observability:
prometheus_port: 9090
log_level: info
# 重启运行时,加载配置
axctl runtime restart --config ax.yaml
# 模拟一次运行中断(Ctrl+C 模拟)
# 重新连接时,从检查点恢复
axctl agent resume <agent-id>
关键实现细节
架构概览
┌─────────────────────────────────────────────────────────┐
│ AX Runtime (axctl) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Resumption │ │ Scheduler │ │ Observability │ │
│ │ Manager │ │ │ │ (Prometheus) │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
│ ┌─────────────────────────────────────────────────────┐│
│ │ Agent Harness (隔离执行层) ││
│ │ [Tool A] [Tool B] [Skill X] [LLM Client] ││
│ └─────────────────────────────────────────────────────┘│
│ ┌─────────────────────────────────────────────────────┐│
│ │ Sandbox Layer (per-tool 进程隔离) ││
│ └─────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────┘
Resumption(断点续传)原理
AX 在每个 checkpoint 将 Agent 的完整状态序列化:
- 执行轨迹(Trajectory):已完成的 Thought/Action/Observation 序列
- 工作内存(Working Memory):当前上下文
- Tool 会话密钥:外部 API token、文件句柄等
恢复时,AX 重新实例化 Tool 和 LLM Client,复现到断点位置继续执行。前提条件:Tool 必须是幂等的(同一输入重复调用结果一致)。
多 Agent 协作(进阶)
from ax.runtime.multi_agent import AgentGroup
orchestrator = Agent(
model="gemini-2.5-flash",
skills=[]
)
worker_a = Agent(model="gemini-2.5-flash", tools=[...])
worker_b = Agent(model="gemini-2.5-flash", tools=[...])
group = AgentGroup(
orchestrator=orchestrator,
workers={"worker_a": worker_a, "worker_b": worker_b},
protocol="ax://protocol/v1" # 标准化 Agent 间通信协议
)
result = ax.run(group, input="帮我分析竞品 A 和竞品 B 的公开财报")
print(result.output)
常见坑与规避清单
| 坑 | 表现 | 规避方法 |
|---|---|---|
| Tool 非幂等 | Agent 恢复后重复调用 API,产生副作用(发重邮件、重复下单) | 设计 Tool 时必须保证幂等;使用唯一 request_id 字段去重 |
| Checkpoint 过于频繁 | 本地磁盘写入成为瓶颈(高并发场景) | 调整 checkpoint_interval_ms,文件密集型任务设大些(60s+) |
| Sandbox 资源泄露 | Tool 进程在异常时未释放,积累后导致 OOM | ax sandbox set --max-processes 20 限制并发沙箱数;设置 timeout_ms |
| LLM API Token 泄露 | Tool 代码里硬编码了 os.getenv("API_KEY"),沙箱无法隔离主进程 secrets | 将 secrets 注入沙箱环境变量,而非主进程;通过 ax secrets create 管理 |
| 长轨迹检索慢 | trajectory 命令在运行几小时后超长列表卡顿 | 使用 axctl agent trajectory --from <checkpoint-id> 而非从头拉取 |
| Config 格式不兼容 | 新版 AX yaml schema 与旧 runtime 不兼容导致启动失败 | 查看 axctl schema validate 确认配置合规后再重启 |
| 断点续传后 Tool 结果”对不上” | 外部系统(如 DB)在检查点之后被外部改动 | 业务层使用乐观锁或 version 字段;AX 不保证分布式一致性 |
| 多 Agent 死锁 | 多个 Worker Agent 相互等待对方释放资源 | 避免循环依赖;设置 timeout_ms 强制退出等待 |
成本 / 性能 / 维护权衡
成本
| 部署方式 | 适用规模 | 主要成本项 |
|---|---|---|
| 本地单节点(本地磁盘 checkpoint) | MVP / 开发验证 | 机器成本(无云费用),适合 < 10 并发 Agent |
| 本地单节点 + 对象存储 | 中等规模(50 并发) | 对象存储费用(如 MinIO 或 S3),Checkpoint 持久化 |
| Kubernetes 集群 + 高可用 | 生产级规模 | K8s 集群运维成本,etcd 用于 HA checkpoint 存储 |
性能
- Checkpoint 开销:实测每个 checkpoint 序列化约 200ms~2s(取决于轨迹长度),生产中推荐
30s~60s间隔,而非更密 - Sandbox 启动延迟:每个 Tool 进程冷启动约 500ms~2s;建议提前预热(
--sandbox pool-size=5) - Resumption 延迟:恢复约 1~5s(取决于状态大小),网络中断场景可接受
维护
- 升级策略:AX 遵循语义化版本,大版本升级前先在 staging 环境跑全套回归;建议 pinned 到 minor version
- 备份:Checkpoint 目录定期备份;S3/GCS 可配置 lifecycle 规则自动归档旧检查点
- 监控关键指标:
ax_agent_checkpoints_total(Prometheus):检查点频率ax_sandbox_active_processes:当前活跃沙箱进程数ax_agent_resumption_success_rate:恢复成功率(应 > 99%)
一周内可执行行动清单
Day 1:本地跑通 MVP
-
git clone+pip install装好 AX - 运行官方 Quickstart 示例,验证 Agent 能正常执行 Tool 调用
- 用
axctl runtime start启动服务,用axctl agent list确认状态
Day 2:接入自有 LLM + 断点续传验证
- 将模型换成你已有的 API(Claude/Gemini/OAI 兼容接口)
- 配置
ax.yaml开启resumption.enabled: true - 跑一个需要 > 30s 的任务,手动
kill进程后resume,确认输出正确
Day 3:Tool 安全沙箱 & Secrets 管理
- 对所有 Tool 启用
per_tool_isolation - 使用
ax secrets create注册你的 API Key,避免硬编码 - 验证 Tool 异常退出不会影响主 Agent 进程
Day 4:可观测性搭建
- 配置 Prometheus 抓取
axctl runtime start --observability.prometheus_port=9090 - 导入 Grafana Dashboard 模板(官方已提供)
- 确认
ax_agent_resumption_success_rate指标正常采集
Day 5:生产选型评估
- 评估你的 Agent 规模(并发量 / 运行时长),选择合适部署模式
- 制定 Checkpoint 存储方案(本地 vs 对象存储)
- 输出内部技术方案文档(适宜场景 / 成本估算 / 迁移步骤)
小结: Google AX(Agent Executor)的开源填补了”从 AI 原型到生产 Agent 运行时”这个工程化空缺。它的核心价值不是模型能力,而是让 Agent 执行具备可靠性、可观测和可恢复——这正是企业级部署最难解决的三个问题。上手门槛低,建议从单节点 MVP 开始跑通全链路,再按需扩展到 Kubernetes 高可用架构。