基于Kubernetes的Agentic工作负载调度与编排：ax原语解析

1. 从“ax”这个标题说起一个被低估的调度原语第一次看到“ax”这个标题很多人会以为是某个命令行工具的缩写或者某个内部项目的代号。但把热搜词摊开来看——ax、agentic、orchestration、kubernetes、workspace——这几个词凑在一起指向的其实是一个非常具体的技术命题在 Kubernetes 之上为 agentic 工作负载做一层调度与编排的抽象层。我最早接触这类需求是在做一个多智能体协作平台的时候。当时团队把一堆 agent 塞进 K8s 的 Pod 里跑每个 agent 有自己的 workspace、自己的工具调用链、自己的上下文状态。跑着跑着问题就来了agent 之间要通信要共享中间结果要按依赖顺序执行还要在某个 agent 挂掉之后能恢复现场。K8s 原生那套 Deployment、Service、Job 的模型处理无状态服务很顺手但面对“有状态、有依赖、有生命周期”的 agent 编排就显得力不从心。“ax”要解决的就是这层 gap。它不是要替代 K8s而是在 K8s 之上做一层agentic orchestration的调度层把 agent 当成一等公民来对待给每个 agent 分配独立的 workspace管理它们之间的依赖关系处理失败重试和状态恢复。热搜里出现的 “ax调度” 这个词说的就是这个层面的东西。这篇文章适合谁看如果你正在做多智能体系统、正在用 K8s 跑 AI 工作负载、或者正在设计一个需要动态编排任务的后端架构那这篇内容应该能给你一些可以直接抄作业的思路。如果你只是听说过 agentic 这个词但还没动手那也可以把它当成一个从零到一的架构拆解来看。2. 为什么 K8s 原生调度不够用agentic 负载的特殊性2.1 无状态假设与有状态现实的冲突K8s 的设计哲学里有一条很核心的假设Pod 是无状态的随时可以被替换。这个假设对 Web 服务、API 网关这类负载非常成立一个 Pod 挂了ReplicaSet 立刻拉起一个新的流量切过去用户无感知。但 agent 不是这样的。一个 agent 在执行任务的过程中会积累大量的中间状态它读过的文件、它调用过的工具返回结果、它当前的推理上下文、它和其他 agent 交换过的消息。这些东西如果丢了任务就得从头再来。你当然可以把这些状态外置到 Redis 或者数据库里但那样每次读写都是一次网络往返延迟上去了而且状态的一致性管理会变得非常复杂。“ax”这层抽象要做的第一件事就是给每个 agent 一个workspace。这个 workspace 不是简单的 emptyDir而是一个有生命周期管理的持久化空间。agent 启动时挂载执行过程中读写任务结束后可以选择保留或清理。热搜里那个 “claudes workspace requires the virtual machine platform on windows” 虽然说的是另一个场景但反映的问题是一样的workspace 不是随便给个目录就行它需要有明确的隔离性和资源边界。2.2 依赖编排从 DAG 到动态调度K8s 原生的 Job 和 CronJob 能处理简单的批处理任务但 agent 之间的依赖关系往往不是静态的 DAG。举个例子agent A 负责检索资料agent B 负责分析agent C 负责生成报告。表面上看是 A→B→C 的线性依赖但实际上 B 可能需要根据 A 返回的内容决定要不要再让 A 去检索一轮C 可能需要在 B 分析到一半的时候就拿到部分结果开始起草。这种动态依赖用 K8s 原生的方式很难表达。你当然可以用 Argo Workflows 或者 Tekton 这类工具但它们的设计初衷是 CI/CD 流水线对 agent 这种“长时间运行、状态持续变化、依赖动态生成”的负载来说抽象层级不太匹配。“ax”的思路是引入一个orchestrator层它维护一张运行时的依赖图而不是静态的 DAG。每个 agent 完成任务后orchestrator 根据返回结果决定下一步调度哪些 agent、给它们分配什么参数、挂载哪个 workspace。这个 orchestrator 本身可以跑在 K8s 里也可以跑在外面通过 API 和 K8s 交互。2.3 资源画像agent 不是均匀的还有一个容易被忽略的点不同的 agent 对资源的需求差异极大。一个做文本检索的 agent 可能只需要 0.5 核 CPU 和 512MB 内存但一个做代码生成的 agent 可能需要 4 核 CPU 和 8GB 内存甚至需要 GPU。K8s 的 ResourceQuota 和 LimitRange 能设上限但做不到按 agent 类型动态调整。“ax”在调度的时候会维护一个agent 资源画像。每个 agent 在注册的时候声明自己的资源需求orchestrator 在调度时根据当前集群的可用资源和 agent 的优先级做匹配。这个画像不是静态的而是可以根据历史执行数据动态调整的。比如某个 agent 过去十次执行平均用了 2.3 核 CPU那下次调度时就按 2.5 核来预留留一点 buffer。3. 核心架构拆解ax 的分层设计3.1 控制面与数据面的分离“ax”的架构可以粗略分成两层控制面负责调度决策数据面负责实际执行。控制面里跑的是 orchestrator、状态管理器、资源画像服务数据面里跑的是一个个 agent 的 Pod每个 Pod 挂载自己的 workspace。这种分离的好处是控制面可以独立于 K8s 集群存在。你可以把 orchestrator 跑在一台小机器上管理一个几百节点的 K8s 集群。控制面挂了已经跑起来的 agent 不受影响只是新的调度请求会排队。数据面的 agent 挂了控制面负责重新调度并从 workspace 里恢复状态。控制面和数据面之间通过一个轻量的 agent runtime 通信。这个 runtime 跑在每个 agent Pod 里负责向控制面汇报心跳、拉取任务、上报执行结果。它不处理业务逻辑只做通信和状态同步所以资源占用很小。3.2 Workspace 的生命周期管理Workspace 是“ax”里最核心的抽象之一。它的生命周期可以分成四个阶段创建当 orchestrator 决定调度一个 agent 时先创建一个 workspace。这个 workspace 可以是一个 PVC也可以是一个 hostPath取决于你的存储方案。如果是临时任务可以用 emptyDir如果需要持久化就用 PVC。挂载agent Pod 启动时把 workspace 挂载到容器内的指定路径。这个路径通常是/workspaceagent 的所有读写操作都在这个目录下进行。快照在 agent 执行的关键节点runtime 可以对 workspace 做快照。快照可以是文件系统级别的也可以是应用级别的。文件系统级别的快照用 CSI 的 VolumeSnapshot 就能做应用级别的快照需要 agent 自己实现。清理任务完成后根据策略决定是保留 workspace 还是删除。保留的话下次同一个 agent 启动时可以复用省去重新初始化的时间删除的话释放存储资源。这里有个坑要注意workspace 的并发访问。如果两个 agent 同时挂载同一个 workspace可能会出现写冲突。解决办法是给 workspace 加锁或者干脆让每个 agent 用独立的 workspace通过消息传递来共享数据。我倾向于后者因为锁的管理成本太高而且容易出死锁。3.3 调度器的决策逻辑调度器的核心逻辑可以用一句话概括在满足依赖和资源约束的前提下选择优先级最高的 agent 执行。但实际实现起来要考虑的因素很多。首先是依赖检查。每个 agent 在注册时会声明自己的前置依赖比如“我需要 agent A 的输出作为输入”。调度器在执行前会检查这些依赖是否满足。如果 A 还没跑完这个 agent 就进入等待队列。其次是资源匹配。调度器会查询当前集群的可用资源和 agent 的资源画像做匹配。如果资源不够agent 就排队等待。这里可以用优先级队列高优先级的 agent 先调度。然后是亲和性。有些 agent 需要和特定的 agent 跑在同一个节点上以减少网络延迟有些 agent 需要和特定的 agent 分开以避免资源争抢。这些亲和性规则可以在 agent 注册时声明调度器在决策时考虑。最后是失败处理。如果 agent 执行失败调度器要根据重试策略决定是重新调度、跳过、还是终止整个流程。重试策略可以按 agent 配置比如“最多重试 3 次每次间隔 30 秒”。4. 实操从零搭一个最小可用的 ax 调度层4.1 环境准备与依赖安装先说一下我的实验环境一个三节点的 K8s 集群版本 v1.26.0每个节点 4 核 16GB。这个配置跑十几个 agent 没问题再多就要加节点了。需要提前装好的东西Kubernetes 集群v1.24 以上因为要用到一些新的 APIkubectl 配置好能访问集群Helm用来装一些依赖组件一个存储方案我用的 local-path-provisioner简单够用如果你是从零开始可以用 kind 或者 minikube 快速起一个本地集群。kind 的好处是启动快资源占用小适合做实验。minikube 的好处是功能全支持的特性多。我两个都用过做 ax 的实验用 kind 就够了。# 用 kind 创建一个三节点集群 kind create cluster --name ax-lab --config - EOF kind: Cluster apiVersion: kind.x-k8s.io/v1alpha4 nodes: - role: control-plane - role: worker - role: worker EOF集群起来之后装一个 local-path-provisioner 作为默认的 StorageClasskubectl apply -f https://raw.githubusercontent.com/rancher/local-path-provisioner/master/deploy/local-path-storage.yaml kubectl patch storageclass local-path -p {metadata: {annotations:{storageclass.kubernetes.io/is-default-class:true}}}4.2 定义 Agent 的 CRD“ax”的核心抽象是 Agent 和 Workspace我用 CRD 来定义它们。先定义 WorkspaceapiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: name: workspaces.ax.io spec: group: ax.io versions: - name: v1alpha1 served: true storage: true schema: openAPIV3Schema: type: object properties: spec: type: object properties: size: type: string storageClass: type: string retainPolicy: type: string enum: [Retain, Delete] scope: Namespaced names: plural: workspaces singular: workspace kind: Workspace shortNames: - ws再定义 AgentapiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: name: agents.ax.io spec: group: ax.io versions: - name: v1alpha1 served: true storage: true schema: openAPIV3Schema: type: object properties: spec: type: object properties: image: type: string command: type: array items: type: string dependencies: type: array items: type: string resources: type: object properties: cpu: type: string memory: type: string workspace: type: string retryPolicy: type: object properties: maxRetries: type: integer backoffSeconds: type: integer scope: Namespaced names: plural: agents singular: agent kind: Agent shortNames: - ag这两个 CRD 定义好之后就可以用 kubectl 来创建 Agent 和 Workspace 了。但光有 CRD 还不够还需要一个 controller 来监听这些资源的变化并做出相应的调度决策。4.3 写一个最小化的 OrchestratorOrchestrator 我用 Python 写因为生态好写起来快。核心逻辑就是一个循环监听 Agent 资源的变化检查依赖匹配资源创建 Pod。import time import logging from kubernetes import client, config, watch logging.basicConfig(levellogging.INFO) logger logging.getLogger(ax-orchestrator) class Orchestrator: def __init__(self): config.load_kube_config() self.custom_api client.CustomObjectsApi() self.core_api client.CoreV1Api() self.namespace default def list_agents(self): return self.custom_api.list_namespaced_custom_object( groupax.io, versionv1alpha1, namespaceself.namespace, pluralagents ) def check_dependencies(self, agent): deps agent[spec].get(dependencies, []) for dep in deps: dep_agent self.custom_api.get_namespaced_custom_object( groupax.io, versionv1alpha1, namespaceself.namespace, pluralagents, namedep ) status dep_agent.get(status, {}).get(phase, Pending) if status ! Succeeded: return False return True def create_pod(self, agent): name agent[metadata][name] spec agent[spec] pod client.V1Pod( metadataclient.V1ObjectMeta( namefax-agent-{name}, labels{ax.io/agent: name} ), specclient.V1PodSpec( containers[ client.V1Container( nameagent, imagespec[image], commandspec.get(command, []), resourcesclient.V1ResourceRequirements( requests{ cpu: spec[resources][cpu], memory: spec[resources][memory] } ), volume_mounts[ client.V1VolumeMount( nameworkspace, mount_path/workspace ) ] ) ], volumes[ client.V1Volume( nameworkspace, persistent_volume_claimclient.V1PersistentVolumeClaimVolumeSource( claim_namespec[workspace] ) ) ], restart_policyNever ) ) self.core_api.create_namespaced_pod( namespaceself.namespace, bodypod ) logger.info(fCreated pod for agent {name}) def run(self): while True: agents self.list_agents() for agent in agents[items]: name agent[metadata][name] status agent.get(status, {}).get(phase, Pending) if status ! Pending: continue if not self.check_dependencies(agent): logger.info(fAgent {name} dependencies not met, skipping) continue self.create_pod(agent) self.update_status(name, Running) time.sleep(5) def update_status(self, name, phase): self.custom_api.patch_namespaced_custom_object_status( groupax.io, versionv1alpha1, namespaceself.namespace, pluralagents, namename, body{status: {phase: phase}} ) if __name__ __main__: orch Orchestrator() orch.run()这个 orchestrator 很粗糙但能跑通基本流程。实际生产环境要考虑的东西多得多并发控制、错误处理、状态持久化、多命名空间支持等等。4.4 部署一个示例 Agent假设我们要部署一个简单的 agent它读取 workspace 里的一个文件处理后写回另一个文件。先创建 WorkspaceapiVersion: ax.io/v1alpha1 kind: Workspace metadata: name: demo-workspace spec: size: 1Gi storageClass: local-path retainPolicy: Retain再创建 AgentapiVersion: ax.io/v1alpha1 kind: Agent metadata: name: demo-agent spec: image: busybox:latest command: [sh, -c, echo hello from agent /workspace/output.txt sleep 10] dependencies: [] resources: cpu: 500m memory: 256Mi workspace: demo-workspace retryPolicy: maxRetries: 3 backoffSeconds: 30创建好之后orchestrator 会在 5 秒内检测到这个 Agent检查依赖没有依赖然后创建一个 Pod。Pod 启动后执行命令把结果写到 workspace 里。你可以用kubectl exec进去看kubectl exec -it ax-agent-demo-agent -- cat /workspace/output.txt应该能看到hello from agent。5. 踩坑记录那些文档里不会写的问题5.1 Workspace 挂载失败权限与路径的坑第一次跑的时候Pod 一直起不来报错permission denied。查了半天发现是 local-path-provisioner 创建的目录权限不对。默认情况下它创建的目录是 root 所有但容器里的进程可能以非 root 用户运行写不进去。解决办法有两个一是让容器以 root 运行不推荐有安全风险二是在 PVC 的 mountOptions 里指定 uid 和 gid。我用的是第二种apiVersion: v1 kind: PersistentVolumeClaim metadata: name: demo-workspace spec: accessModes: - ReadWriteOnce storageClassName: local-path resources: requests: storage: 1Gi mountOptions: - uid1000 - gid1000但 local-path-provisioner 对 mountOptions 的支持有限有些版本不生效。如果遇到这种情况可以在 Pod 的 securityContext 里指定 fsGroupsecurityContext: fsGroup: 1000这样 K8s 在挂载卷的时候会自动把目录的组改成 1000容器里的进程就能写了。5.2 Agent 卡在 Pending资源不足的排查思路有一次我创建了 20 个 agent结果一半卡在 Pending。用kubectl describe pod一看报错Insufficient cpu。集群总共 12 核每个 agent 要 1 核理论上能跑 12 个但实际只跑了 8 个。原因是节点上还有系统组件在跑占了一部分资源。排查这类问题的步骤kubectl describe node看每个节点的 Allocated resources了解实际可用资源kubectl get pods --all-namespaces -o wide看哪些 Pod 占了资源调整 agent 的资源请求或者加节点我后来把 agent 的 CPU 请求从 1 核降到 500m就能跑 20 个了。但要注意请求降低之后实际使用可能会超导致节点负载过高。所以最好还是加节点或者用 cluster-autoscaler 自动扩缩容。5.3 状态不一致Orchestrator 重启后的恢复Orchestrator 本身是个无状态服务重启之后内存里的调度状态就丢了。如果重启时正好有 agent 在跑重启后 orchestrator 会认为这些 agent 还是 Pending可能会重复调度。解决办法是把调度状态持久化到 etcd 或者数据库里。我一开始图省事直接把状态写在 Agent 的 status 字段里靠 K8s 的 etcd 来持久化。这样 orchestrator 重启后从 Agent 的 status 里读状态就能恢复。但这里有个 race conditionorchestrator 更新 status 和创建 Pod 不是原子的。如果更新了 status 但创建 Pod 失败agent 就会卡在 Running 状态但实际上没有 Pod 在跑。解决办法是加一个 reconcile 循环定期检查 Running 状态的 agent 是否有对应的 Pod 存在没有就重新创建。5.4 网络通信Agent 之间怎么交换数据Agent 之间交换数据有几种方式共享 workspace简单但有并发问题通过 orchestrator 中转orchestrator 维护一个消息队列agent 把结果发给 orchestratororchestrator 再转发给依赖它的 agent直接网络通信agent 之间通过 Service 互相访问我推荐第二种因为 orchestrator 本来就负责调度顺便做消息中转很自然。而且 orchestrator 可以做消息的持久化agent 挂了消息不丢。具体实现上可以用 Redis 或者 NATS 作为消息中间件。agent 完成任务后把结果写到 Redis 的某个 key 里orchestrator 监听到 key 的变化触发下游 agent 的调度。6. 性能调优与扩展思路6.1 调度延迟优化默认的 5 秒轮询间隔对于实时性要求高的场景太慢了。可以改成 watch 机制监听 Agent 资源的变化有变化时立即触发调度。K8s 的 watch API 很成熟用起来也简单w watch.Watch() for event in w.stream( self.custom_api.list_namespaced_custom_object, groupax.io, versionv1alpha1, namespaceself.namespace, pluralagents ): agent event[object] # 处理事件这样调度延迟可以降到毫秒级。但要注意watch 连接可能会断需要加自动重连逻辑。6.2 批量调度与并发控制当 agent 数量很多时逐个调度效率太低。可以批量处理一次拉取所有 Pending 的 agent按优先级排序然后并发创建 Pod。并发数要控制不然会把 K8s API Server 打挂。我一般设成 10也就是同时最多创建 10 个 Pod。from concurrent.futures import ThreadPoolExecutor def batch_schedule(self, agents): with ThreadPoolExecutor(max_workers10) as executor: futures [] for agent in agents: if self.check_dependencies(agent): futures.append(executor.submit(self.create_pod, agent)) for f in futures: f.result()6.3 资源画像的动态调整前面提到过agent 的资源画像可以根据历史数据动态调整。具体做法是每次 agent 执行完成后记录它的实际资源使用量可以从 cAdvisor 或者 metrics-server 拿然后计算移动平均值更新画像。def update_resource_profile(self, agent_name, actual_cpu, actual_memory): profile self.get_profile(agent_name) alpha 0.3 # 平滑因子 profile[cpu] alpha * actual_cpu (1 - alpha) * profile[cpu] profile[memory] alpha * actual_memory (1 - alpha) * profile[memory] self.save_profile(agent_name, profile)这样跑一段时间后画像会越来越准调度决策也会越来越合理。6.4 多集群调度单集群的资源总是有限的。当 agent 数量超过单集群承载能力时就需要多集群调度。思路是在每个集群里跑一个 agent runtimeorchestrator 根据各集群的可用资源决定把 agent 调度到哪个集群。这需要 orchestrator 维护一个全局的资源视图知道每个集群有多少可用资源。可以用 Karmada 或者 Clusternet 这类多集群管理工具来做底层支撑orchestrator 只需要调用它们的 API 就行。热搜里提到的 “karmada正式毕业” 这个事对做多集群调度的人来说是个好消息。Karmada 提供了统一的 API 来管理多个 K8s 集群包括跨集群的调度、故障转移、流量管理。用它来做 ax 的多集群底座能省不少事。7. 常见问题速查表问题现象可能原因排查方法解决方案Pod 卡在 Pending资源不足kubectl describe pod看 Events降低资源请求或加节点Pod 启动后立即退出命令执行失败kubectl logs看输出检查命令和镜像Workspace 写不进去权限问题kubectl exec进去ls -la设置 fsGroup 或 mountOptionsAgent 重复调度Orchestrator 状态丢失检查 Agent status持久化调度状态调度延迟高轮询间隔太长看 orchestrator 日志改用 watch 机制依赖检查失败上游 agent 未完成检查上游 agent status等待或手动触发上游网络不通Service 配置错误kubectl get svc检查修正 Service selector存储卷挂载失败PVC 未绑定kubectl get pvc检查检查 StorageClass 配置8. 一些个人体会做 ax 这层调度抽象最大的感受是不要试图用 K8s 原生的一切。K8s 提供了很好的基础能力但 agentic 负载有它的特殊性硬套原生模型会很别扭。该自定义 CRD 就自定义该写 controller 就写 controller该用外部 orchestrator 就用外部 orchestrator。另一个体会是状态管理是核心难点。agent 的状态比普通服务复杂得多workspace 的状态、依赖的状态、调度的状态每一层都要考虑持久化和恢复。我在这上面踩的坑最多也花了最多时间。最后从最小可用开始。不要一上来就设计一个支持多集群、动态资源画像、复杂依赖编排的完整系统。先跑通一个 agent 的调度再加第二个再加依赖再加资源匹配。每一步都验证过再往下走比一次性设计一个大而全的系统要靠谱得多。这个方向后续还可以往几个方向扩展一是和 agentic RAG 结合把检索增强生成的工作流也纳入调度范围二是做更细粒度的资源隔离比如用 cgroup v2 做 CPU 和内存的硬限制三是做跨集群的故障转移当一个集群挂了自动把 agent 迁移到其他集群。这些我都在陆续尝试有新的进展再分享。