post cover

技术热点落地: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 进程在异常时未释放,积累后导致 OOMax 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 高可用架构。