Google Agent Executor & Substrate：开源生产级AI Agent运行时的革命性突破

# 传统模型：无状态请求-响应
def handle_request(user_input: str) -> str:
    # 每次请求都是全新的上下文
    context = load_context()  # 需要手动加载
    response = llm.generate(context, user_input)
    save_context(context)  # 需要手动保存
    return response

# Agent Executor模型：持久化状态
from google.agent_executor import AgentExecutor, WorkflowState

class MyAgentWorkflow:
    def __init__(self):
        self.executor = AgentExecutor(
            checkpoint_enabled=True,
            event_logging=True,
            durable_execution=True
        )
    
    async def run(self, task_id: str, initial_input: dict):
        """
        启动一个持久化的工作流
        工作流可以在任何点中断和恢复
        """
        workflow = await self.executor.create_workflow(
            workflow_id=task_id,
            initial_state={
                "input": initial_input,
                "step": 0,
                "memory": [],
                "results": {}
            }
        )
        
        # 工作流会持续执行，支持中断和恢复
        result = await workflow.execute()
        return result

# 事件日志示例
event_log = [
    WorkflowCreated(timestamp="2026-05-28T10:00:00Z", workflow_id="wf_001"),
    StepStarted(step_id=1, step_name="analyze_requirement", input={"query": "分析销售数据"}),
    ToolCallRequested(tool="sql_executor", params={"query": "SELECT * FROM sales"}),
    ToolCallCompleted(tool="sql_executor", result={"rows": 1500, "columns": 12}),
    StateSnapshot(state={"analysis": {...}, "confidence": 0.85}),
    # 工作流可能在此时中断
]

# 恢复执行
await workflow.resume(from_event=4)

from google.agent_executor.snapshot import StateSnapshotManager

class AdvancedSnapshotManager(StateSnapshotManager):
    """高级快照管理器"""
    
    def __init__(self, storage_backend, retention_count=10):
        self.storage = storage_backend
        self.retention_count = retention_count
        self.snapshots = {}
    
    async def create_snapshot(self, workflow_id: str, state: dict) -> str:
        """创建状态快照"""
        snapshot_id = generate_snapshot_id()
        snapshot = {
            "workflow_id": workflow_id,
            "snapshot_id": snapshot_id,
            "timestamp": current_timestamp(),
            "state": state,
            "checksum": compute_checksum(state)
        }
        
        # 存储快照
        await self.storage.put(f"snapshots/{workflow_id}/{snapshot_id}", snapshot)
        
        # 清理旧快照
        await self._cleanup_old_snapshots(workflow_id, snapshot_id)
        
        return snapshot_id
    
    async def restore_snapshot(self, workflow_id: str, snapshot_id: str) -> dict:
        """恢复指定快照"""
        snapshot = await self.storage.get(f"snapshots/{workflow_id}/{snapshot_id}")
        if not snapshot:
            raise ValueError(f"Snapshot {snapshot_id} not found")
        
        # 验证快照完整性
        if compute_checksum(snapshot["state"]) != snapshot["checksum"]:
            raise CorruptionError("Snapshot checksum mismatch")
        
        return snapshot["state"]
    
    async def get_latest_snapshot(self, workflow_id: str) -> Optional[dict]:
        """获取最新快照"""
        snapshot_ids = await self.storage.list(f"snapshots/{workflow_id}")
        if not snapshot_ids:
            return None
        
        latest_id = max(snapshot_ids)
        return await self.restore_snapshot(workflow_id, latest_id)

from google.agent_executor.branching import TrajectoryBrancher

class TestDifferentStrategies:
    """测试不同策略的分支测试"""
    
    def __init__(self, executor: AgentExecutor):
        self.executor = executor
        self.brancher = TrajectoryBrancher(executor)
    
    async def run_ab_test(self, workflow_id: str, strategy_a: str, strategy_b: str):
        """
        基于历史状态创建A/B测试分支
        """
        # 获取当前工作流的最新状态
        current_state = await self.executor.get_state(workflow_id)
        
        # 创建分支A：使用策略A
        branch_a_id = await self.brancher.create_branch(
            parent_workflow_id=workflow_id,
            parent_snapshot_id=current_state["snapshot_id"],
            branch_id="strategy_a_test",
            override_config={"strategy": strategy_a}
        )
        
        # 创建分支B：使用策略B
        branch_b_id = await self.brancher.create_branch(
            parent_workflow_id=workflow_id,
            parent_snapshot_id=current_state["snapshot_id"],
            branch_id="strategy_b_test",
            override_config={"strategy": strategy_b}
        )
        
        # 并行执行两个分支
        result_a = await self.executor.execute_workflow(branch_a_id)
        result_b = await self.executor.execute_workflow(branch_b_id)
        
        # 比较结果
        return {
            "strategy_a_result": result_a,
            "strategy_b_result": result_b,
            "winner": self._determine_winner(result_a, result_b)
        }

from google.agent_executor.approval import ApprovalManager, ApprovalRequest

class EnterpriseApprovalWorkflow:
    """企业级审批工作流"""
    
    def __init__(self):
        self.approval_mgr = ApprovalManager()
    
    async def run_with_approvals(self, task: dict):
        """带审批的工作流执行"""
        workflow = await self.executor.create_workflow(
            initial_state={"task": task}
        )
        
        # 定义需要审批的关键步骤
        approval_points = {
            "financial_transfer": {
                "threshold": 10000,
                "approvers": ["finance_manager", "ceo"],
                "timeout": 3600  # 1小时超时
            },
            "data_deletion": {
                "threshold": "any",
                "approvers": ["dpo", "data_owner"],
                "timeout": 7200
            }
        }
        
        async for step in workflow.step_generator():
            # 检查是否需要审批
            if step.name in approval_points:
                approval_req = ApprovalRequest(
                    workflow_id=workflow.id,
                    step_id=step.id,
                    action=step.name,
                    details=step.details,
                    required_approvers=approval_points[step.name]["approvers"],
                    timeout=approval_points[step.name]["timeout"]
                )
                
                # 暂停工作流，等待审批
                approved = await self.approval_mgr.request_approval(approval_req)
                
                if not approved:
                    workflow.cancel(f"Rejected by {approval_req.approvers}")
                    return {"status": "rejected", "reason": approval_req.rejection_reason}
            
            # 继续执行
            await workflow.execute_step(step)
        
        return workflow.final_result

# Agent Substrate CRD (Custom Resource Definition) 示例
apiVersion: agent.googleapis.com/v1
kind: AgentGroup
metadata:
  name: customer-service-agents
  namespace: production
spec:
  agentType: "customer-support"
  replicaCount: 100
  maxConcurrentTasks: 10
  resources:
    requests:
      cpu: "500m"
      memory: "512Mi"
    limits:
      cpu: "2000m"
      memory: "2Gi"
  persistence:
    enabled: true
    storageClass: "fast-ssd"
    accessMode: "ReadWriteMany"
  scaling:
    minReplicas: 10
    maxReplicas: 1000
    targetCPUUtilization: 70
  security:
    serviceAccount: "agent-sa"
    networkPolicy: "agent-network-policy"
    podSecurityPolicy: "restricted"

from kubernetes import client
from agent_substrate.snapshot import PodSnapshotManager

class AgentPodSnapshotManager(PodSnapshotManager):
    """Agent Pod快照管理器"""
    
    def __init__(self, k8s_client: client.ApiClient):
        self.k8s = k8s_client
        self.snapshot_api = client.CustomObjectsApi(k8s_client)
    
    async def create_pod_snapshot(self, pod_name: str, namespace: str) -> str:
        """创建Pod快照"""
        # 获取Pod的完整规格
        pod = self.k8s.read_namespaced_pod(pod_name, namespace)
        
        # 获取Pod的所有Volumes
        volumes = self._capture_volumes(pod.spec.volumes)
        
        # 获取Pod的网络状态
        network_state = await self._capture_network_state(pod)
        
        # 获取Agent的内存状态
        agent_state = await self._capture_agent_state(pod)
        
        snapshot = {
            "apiVersion": "agent.googleapis.com/v1",
            "kind": "AgentPodSnapshot",
            "metadata": {
                "generateName": f"{pod_name}-snapshot-"
            },
            "spec": {
                "originalPod": pod.metadata.name,
                "originalNode": pod.spec.node_name,
                "volumes": volumes,
                "networkState": network_state,
                "agentState": agent_state
            }
        }
        
        # 创建快照资源
        snapshot_obj = self.snapshot_api.create_namespaced_custom_object(
            group="agent.googleapis.com",
            version="v1",
            namespace=namespace,
            plural="agentpodsnapshots",
            body=snapshot
        )
        
        return snapshot_obj["metadata"]["name"]
    
    async def restore_pod_from_snapshot(
        self, 
        snapshot_name: str, 
        namespace: str,
        target_node: str = None
    ) -> str:
        """从快照恢复Pod"""
        snapshot = self.snapshot_api.get_namespaced_custom_object(
            group="agent.googleapis.com",
            version="v1",
            namespace=namespace,
            plural="agentpodsnapshots",
            name=snapshot_name
        )
        
        # 重建Pod规格
        new_pod_spec = self._reconstruct_pod_spec(snapshot["spec"])
        
        # 如果指定了目标节点，设置节点亲和性
        if target_node:
            new_pod_spec.affinity = client.V1Affinity(
                node_affinity=client.V1NodeAffinity(
                    required_during_scheduling_ignored_during_execution=client.V1NodeSelector(
                        node_selector_terms=[
                            client.V1NodeSelectorTerm(
                                match_expressions=[
                                    client.V1NodeSelectorRequirement(
                                        key="kubernetes.io/hostname",
                                        operator="In",
                                        values=[target_node]
                                    )
                                ]
                            )
                        ]
                    )
                )
            )
        
        # 创建新Pod
        new_pod = client.V1Pod(
            api_version="v1",
            kind="Pod",
            metadata=client.V1ObjectMeta(
                generate_name=f"{snapshot['spec']['originalPod']}-restored-"
            ),
            spec=new_pod_spec
        )
        
        created_pod = self.k8s.create_namespaced_pod(namespace, new_pod)
        return created_pod.metadata.name

from agent_substrate.scheduler import AgentScheduler
from agent_substrate.load_balancer import AgentLoadBalancer

class HyperscaleAgentManager:
    """超大规模Agent管理器"""
    
    def __init__(
        self,
        scheduler: AgentScheduler,
        load_balancer: AgentLoadBalancer,
        state_store,
        cache_layers
    ):
        self.scheduler = scheduler
        self.load_balancer = load_balancer
        self.state_store = state_store
        self.cache = cache_layers
    
    async def register_agent(self, agent: AgentMetadata) -> str:
        """注册新Agent"""
        # 生成唯一ID
        agent_id = self.scheduler.generate_agent_id()
        
        # 检查缓存
        cached = await self.cache.get(f"agent:{agent.type}")
        if cached and cached.count < cached.capacity:
            # 直接分配到现有Pod
            return await self._assign_to_pod(agent_id, cached.pod_id)
        
        # 注册到状态存储
        agent_record = {
            "id": agent_id,
            "type": agent.type,
            "version": agent.version,
            "registered_at": current_timestamp(),
            "status": "active"
        }
        await self.state_store.put(f"agents/{agent_id}", agent_record)
        
        # 更新索引缓存
        await self.cache.put(f"agent_index:{agent.type}", agent_id)
        
        return agent_id
    
    async def route_tool_call(self, tool_call: ToolCallRequest) -> ToolCallResponse:
        """路由工具调用到合适的Agent"""
        # L1缓存：检查热点Tool的最近路由
        cache_key = f"route:{tool_call.tool_name}:{tool_call.tenant_id}"
        cached_route = await self.cache.get(cache_key)
        
        if cached_route:
            return await self._execute_tool_call(cached_route.agent_id, tool_call)
        
        # 查询状态存储获取可用Agent
        agent_ids = await self.state_store.get(
            f"agent_index:{tool_call.tool_name}"
        )
        
        if not agent_ids:
            raise NoAvailableAgentError(f"No agent available for tool: {tool_call.tool_name}")
        
        # 负载均衡选择
        selected_agent = await self.load_balancer.select(
            candidates=agent_ids,
            criteria=tool_call.routing_criteria
        )
        
        # 缓存路由结果
        await self.cache.put(cache_key, selected_agent, ttl=60)
        
        return await self._execute_tool_call(selected_agent, tool_call)

from agent_substrate.security import (
    AgentSecurityManager,
    SandboxConfig,
    NetworkPolicy,
    IAMPolicy
)

class SecureAgentDeployment:
    """安全Agent部署配置"""
    
    def __init__(self):
        self.security_mgr = AgentSecurityManager()
    
    def create_sandbox_config(self) -> SandboxConfig:
        """创建沙箱配置"""
        return SandboxConfig(
            runtime="gke-sandbox",  # 或 "kata-containers"
            filesystem_readonly=True,
            filesystem_whitelist=[
                "/tmp/agent_workspace",
                "/var/agent_data"
            ],
            network_mode="none",  # 完全禁用网络，或使用 "whitelist"
            allowed_networks=[
                "internal-mesh",
                "agent-api-backend"
            ],
            resources=ResourceLimits(
                cpu_period=100000,
                cpu_quota=200000,
                memory_limit=2 * 1024 * 1024 * 1024,  # 2GB
                pids_limit=100
            ),
            syscalls_filter=[
                # 允许的系统调用
                "read", "write", "open", "close",
                "brk", "mmap", "mprotect",
                "sched_yield", "nanosleep"
                # 禁止的危险系统调用
            ]
        )
    
    def create_network_policy(self) -> NetworkPolicy:
        """创建网络策略"""
        return NetworkPolicy(
            policy_type="default_deny",  # 默认拒绝所有
            ingress_rules=[
                # 只允许来自API Gateway的流量
                AllowedSource(
                    pod_selector={"component": "api-gateway"},
                    ports=[8080]
                )
            ],
            egress_rules=[
                # 只允许访问特定的外部服务
                AllowedDestination(
                    pod_selector={"component": "agent-backend"},
                    ports=[8080, 9090]
                ),
                AllowedDestination(
                    dns=["internal-mesh.svc.cluster.local"],
                    ports=[443]
                )
            ]
        )
    
    def create_iam_policy(self) -> IAMPolicy:
        """创建IAM策略"""
        return IAMPolicy(
            service_account="agent-service-account",
            role_bindings=[
                RoleBinding(
                    role="roles/agent.tool-access",
                    # 只允许访问特定的工具
                    allowed_tools=["sql-query", "file-read", "http-request"]
                ),
                RoleBinding(
                    role="roles/data.reader",
                    # 只允许读取特定的数据源
                    allowed_resources=[
                        "projects/*/datasets/reporting_data",
                        "projects/*/buckets/reports"
                    ]
                ),
                # 禁止的操作
                DeniedBinding(
                    role="roles/owner",
                    reason="Agent must never have owner privileges"
                )
            ]
        )

from langchain.agents import AgentExecutor as LCAgentExecutor
from google.agent_executor import AgentExecutor, Adapter

# 创建Agent Executor适配器
adapter = Adapter(
    executor=AgentExecutor(
        checkpoint_enabled=True,
        event_logging=True
    )
)

# 使用LangChain的ReAct Agent
agent = create_langchain_react_agent(
    model=ChatGoogleGenerativeAI(model="gemini-2.0"),
    tools=my_tools,
    prompt=my_prompt
)

# 通过适配器运行，支持持久化
chain = LCAgentExecutor(
    agent=agent,
    tools=my_tools,
    adapter=adapter  # 注入Agent Executor能力
)

# 执行，支持中断和恢复
result = await chain.arun("分析本月的销售数据")

from langgraph.graph import StateGraph
from google.agent_executor import AgentExecutor, CheckpointSaver

# 定义状态图
graph = StateGraph(SalesAnalysisState)

graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report", report_node)

graph.set_entry_point("research")
graph.add_edge("research", "analyze")
graph.add_edge("analyze", "report")
graph.set_finish_point("report")

# 编译，启用Agent Executor的checkpoint
compiled_graph = graph.compile(
    checkpoint=CheckpointSaver(
        executor=AgentExecutor(checkpoint_enabled=True)
    )
)

# 运行，状态自动持久化
async for state in compiled_graph.astream(initial_state):
    print(state)

from agent_substrate.deployment import (
    DeploymentConfig,
    CloudProvider,
    HybridConfig
)

# GKE部署配置
gke_config = DeploymentConfig(
    provider=CloudProvider.GKE,
    project="my-gcp-project",
    region="us-central1",
    features=[
        "autopilot",
        "security_posture",
        "binary_authorization"
    ]
)

# 混合云配置
hybrid_config = HybridConfig(
    control_plane=DeploymentConfig(
        provider=CloudProvider.GKE,
        project="my-gcp-project",
        region="us-central1"
    ),
    data_planes=[
        DataPlaneConfig(
            provider=CloudProvider.AWS,
            region="us-east-1",
            agent_count=10000
        ),
        DataPlaneConfig(
            provider=CloudProvider.AZURE,
            region="eastus",
            agent_count=10000
        ),
        DataPlaneConfig(
            provider=CloudProvider.SELF_HOSTED,
            location="on-prem-datacenter",
            agent_count=50000
        )
    ],
    data_routing=DataRoutingPolicy(
        prefer_local=True,
        fallback_cloud=CloudProvider.GKE
    )
)

import asyncio
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum

from google.agent_executor import AgentExecutor, WorkflowState
from agent_substrate import AgentGroup, ScalingConfig
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.tools import tool

# ========== 工具定义 ==========

@dataclass
class CustomerContext:
    """客户上下文"""
    customer_id: str
    tier: str  # 'basic', 'premium', 'enterprise'
    account_age_days: int
    previous_tickets: list = field(default_factory=list)

@tool
def lookup_customer(customer_id: str) -> dict:
    """查询客户信息"""
    # 实际实现中会调用CRM API
    return {
        "customer_id": customer_id,
        "tier": "premium",
        "account_age_days": 365,
        "previous_tickets": ["TICKET-001", "TICKET-002"]
    }

@tool
def get_order_history(customer_id: str) -> list:
    """获取订单历史"""
    return [
        {"order_id": "ORD-001", "date": "2026-05-01", "amount": 299.99},
        {"order_id": "ORD-002", "date": "2026-05-15", "amount": 149.99}
    ]

@tool
def create_support_ticket(
    customer_id: str,
    subject: str,
    description: str,
    priority: str
) -> str:
    """创建支持工单"""
    ticket_id = f"TICKET-{hash(subject) % 100000}"
    return ticket_id

@tool
def send_email(to: str, subject: str, body: str) -> bool:
    """发送邮件"""
    return True

# ========== Agent实现 ==========

class CustomerServiceAgent:
    """智能客服Agent"""
    
    def __init__(self):
        # 初始化LLM
        self.llm = ChatGoogleGenerativeAI(
            model="gemini-2.0-pro",
            temperature=0.7
        )
        
        # 初始化Agent Executor
        self.executor = AgentExecutor(
            checkpoint_enabled=True,
            event_logging=True,
            durable_execution=True
        )
        
        # 工具列表
        self.tools = [
            lookup_customer,
            get_order_history,
            create_support_ticket,
            send_email
        ]
        
        # 业务规则
        self.escalation_threshold = {
            "basic": 500,  # 金额超过500需要升级
            "premium": 1000,
            "enterprise": float("inf")  # 不升级
        }
    
    def _build_system_prompt(self, customer_tier: str) -> str:
        """构建系统提示词"""
        tier_prompts = {
            "basic": "你是一位专业客服，代表公司为客户提供优质服务。",
            "premium": "你是高级客服，可以提供更详细的技术支持和解决方案。",
            "enterprise": "你是企业级客服专家，可以提供全方位的企业支持和定制化服务。"
        }
        
        return f"""
{tier_prompts.get(customer_tier, tier_prompts['basic'])}

工作流程：
1. 首先查询客户信息，了解客户背景
2. 根据客户问题类型收集必要信息
3. 提供解决方案或创建工单
4. 对于高价值客户(订单超过${self.escalation_threshold.get(customer_tier, 500)})，
   自动升级到高级客服

沟通原则：
- 使用友好、专业的语气
- 复杂问题用简单语言解释
- 永远不要承诺无法兑现的事情
"""
    
    async def handle_customer_request(
        self,
        workflow_id: str,
        customer_id: str,
        request: str
    ) -> dict:
        """处理客户请求"""
        
        # 获取客户上下文
        customer = lookup_customer.invoke(customer_id)
        customer_context = CustomerContext(
            customer_id=customer_id,
            tier=customer["tier"],
            account_age_days=customer["account_age_days"]
        )
        
        # 构建初始状态
        initial_state = {
            "customer": customer_context,
            "request": request,
            "conversation_history": [],
            "escalated": False,
            "resolution": None
        }
        
        # 创建工作流
        workflow = await self.executor.create_workflow(
            workflow_id=workflow_id,
            initial_state=initial_state
        )
        
        try:
            # 执行处理流程
            result = await self._process_with_retry(workflow)
            return {
                "status": "resolved",
                "workflow_id": workflow_id,
                "result": result
            }
        except HumanApprovalRequired as e:
            # 需要人工审批（如退款超过阈值）
            return {
                "status": "pending_approval",
                "workflow_id": workflow_id,
                "approval_details": e.details
            }
        except Exception as e:
            return {
                "status": "failed",
                "workflow_id": workflow_id,
                "error": str(e)
            }
    
    async def _process_with_retry(
        self,
        workflow: WorkflowState,
        max_retries: int = 3
    ) -> dict:
        """带重试的处理逻辑"""
        
        for attempt in range(max_retries):
            try:
                # 构建提示词
                prompt = self._build_workflow_prompt(workflow.current_state)
                
                # 调用LLM
                response = await self.llm.agenerate([prompt])
                
                # 解析响应并执行工具调用
                actions = self._parse_llm_response(response)
                
                for action in actions:
                    if action.type == "tool_call":
                        result = await workflow.execute_tool(
                            action.tool_name,
                            action.parameters
                        )
                        workflow.add_to_history(action, result)
                    elif action.type == "escalate":
                        workflow.update_state({"escalated": True})
                    elif action.type == "resolve":
                        workflow.update_state({
                            "resolution": action.resolution
                        })
                
                # 检查是否完成
                if workflow.current_state.get("resolution"):
                    return workflow.current_state["resolution"]
                
            except ToolCallError as e:
                if attempt == max_retries - 1:
                    raise
                # 重试逻辑
                await asyncio.sleep(2 ** attempt)
        
        raise MaxRetriesExceededError("Failed after maximum retries")

# ========== 部署配置 ==========

def create_production_deployment() -> AgentGroup:
    """创建生产部署配置"""
    
    return AgentGroup(
        name="customer-service-prod",
        agent_type="customer-support",
        scaling=ScalingConfig(
            min_replicas=10,
            max_replicas=1000,
            target_concurrent_requests=100,
            scale_up_cooldown=60,
            scale_down_cooldown=300
        ),
        resources=ResourceRequirements(
            requests={"cpu": "1", "memory": "2Gi"},
            limits={"cpu": "4", "memory": "8Gi"}
        ),
        security=SecurityConfig(
            sandbox_enabled=True,
            network_policy="customer-service-netpol",
            service_account="customer-service-sa"
        ),
        persistence=PersistenceConfig(
            enabled=True,
            checkpoint_interval_seconds=30,
            max_checkpoints_per_workflow=100
        )
    )

# ========== 主程序 ==========

async def main():
    agent = CustomerServiceAgent()
    deployment = create_production_deployment()
    
    # 启动Agent Group
    await deployment.deploy()
    
    # 处理客户请求
    result = await agent.handle_customer_request(
        workflow_id="wf_cs_20260528_001",
        customer_id="CUST-12345",
        request="我想申请退货，订单号ORD-002"
    )
    
    print(f"处理结果: {result}")

if __name__ == "__main__":
    asyncio.run(main())