如何使用 KEDA 与 Kubernetes 构建一个具备自动扩展功能、且能够适应时间变化的高性能 AI 代理系统

大多数人工智能代理在处理快速任务时表现得非常出色。当你发送一条消息后，这些代理会调用一些工具，几秒钟内就能给你回复结果。当你需要让它们总结一份文档或进行网络搜索时，这种机制确实非常有效。

但是，当任务需要花费较长时间来完成时，情况就会不同了。比如“查看我过去三天的电子邮件，为紧急事项起草回复内容，而对于与工程相关的问题，则创建相应的工单”——这类任务显然无法在短时间内完成，可能会花费几分钟、几小时甚至更长的时间。而这仅仅是一个例子而已。

这种处理流程属于典型的全流程工作模式，一旦服务器出现故障或进程重新启动，之前所做的所有工作都会丢失。没有重试机制，也没有恢复功能，你只能从头开始重新操作。

本文所要探讨的正是这个问题。

在这篇文章中，你将构建一个能够在实际环境中稳定运行的后台代理系统。你可以发送任务指令后离开，剩下的工作就会由该系统自动完成。

在底层技术层面，这个系统会运行在Kubernetes平台上，并通过KEDA实现自动扩展功能：当没有任务需要处理时，相关组件会自动关闭；而一旦有新任务到来，系统会立即重新启动这些组件。为了确保系统的稳定性和故障恢复能力，我们会使用Temporal技术；而对于代理的功能实现及工具管理，则会使用Composio框架。

本文涵盖的内容

在本教程中，你将学习如何构建一个能够在Kubernetes平台上运行、并根据实际工作负载自动扩展的后台代理系统。具体来说，你将会学到以下内容：

• 什么是代理循环机制，以及如何使用Claude和Composio来构建这样的代理循环

• 如何利用Temporal技术让那些需要长时间运行的代理任务在发生故障时仍能继续执行

• 如何设计一个将任务调度与实际执行过程分离的中间层系统

• 如何使用Docker将代理组件和中间层系统容器化

• 如何将整个系统部署到本地的Kubernetes集群中

• 如何根据队列中的任务数量，利用KEDA实现组件的自动扩展

虽然这些内容涉及一些高级技术概念，但只要你认真跟随学习，一定能够掌握很多实用的知识。

目录

• 整体规划与架构设计

  • 任务调度机制

  • 任务执行过程

  • 自动扩展功能

• 项目搭建步骤

• 应用程序的核心组件

  • 代理循环机制

  • 利用Temporal技术确保系统稳定性

  • 代理中间层系统的设计

  • 应用程序的容器化部署

  • 将系统部署到Kubernetes集群中

  • 利用KEDA实现自动扩展

• 代理组件在实际应用中的表现

• 总结与展望

整体架构规划

在开始研究代码之前，了解各个组成部分是如何协同工作的会很有帮助。

该系统被划分为两个独立的层面：一个是负责处理用户交互的控制层（Next.js前端），另一个则是执行实际任务的执行层。这两者永远不会直接相互调用，这种设计是经过深思熟虑的。

以下是整个流程的详细步骤：

任务调度过程

当用户提交一个任务目标时，系统会首先进行预检查，以确保该用户所需的所有Composio工具连接都已正常启用。如果连接正常，系统就会将任务交给Temporal处理，然后立即返回结果——用户无需等待任何响应。

注意：你不需要等待代理程序的回复，所有操作都在后台自动完成。这与普通的聊天应用不同：你只需启动任务，之后就可以忽略它了。

任务执行过程

Temporal会将任务放入队列中，然后由相应的工作进程来处理这些任务。这些工作进程会运行代理程序，利用大语言模型来分析任务目标，Composio工具则会负责执行具体的操作，最终结果会被发送回Temporal。前端系统会自动向Gateway请求状态更新，这样用户就可以随时了解任务的进展情况。

扩展性设计

KEDA会监控Temporal队列中的任务数量，并根据待处理任务的多少来动态调整工作进程的数量。当队列为空时，工作进程会减少到零；当有新的任务进入队列时，系统会自动增加工作进程的数量。这种设计非常灵活且高效！

之所以控制层从不直接接触代理程序的代码，原因很简单：某些任务的执行可能需要几分钟甚至几小时的时间，而我们并不希望这类操作发生在API层。将它们分开处理，可以确保控制层始终保持高效的运行速度，无论后台发生了什么。

此外，该应用程序还支持基于Linux CronJob机制的任务调度功能，完全不需要人工干预。因此，预检查环节就显得尤为重要——因为如果在任务调度阶段就发现错误，及时停止操作总比让任务因工具连接问题而默默失败要好得多。

以上基本上就是我们应用程序的整体架构框架。简单来说：

• Kubernetes (k8s)：编排层

• KEDA：自动扩展层

• Temporal：持久性保障层

• Composio：工具执行层

• 你选择的大语言模型（在我们的案例中是Anthropic）：推理层

项目设置指南

在开始开发之前，请确保你已经安装了以下软件：

• Docker

• k3d（用于运行本地的Kubernetes集群）

• kubectl

• Helm

• Node.js以及Python 3.11及以上版本

您还需要Anthropic和Composio的API密钥。

首先，克隆仓库：

git clone https://github.com/shricodev/kron-k8s-agent.git
cd kron-k8s-agent

接下来，创建集群、构建镜像并将它们加载到集群中：

# 创建本地集群
k3d cluster create agent --wait

# 构建两个镜像并将其导入集群
bash scripts/build-images.sh
bash scripts/load-images.sh

# 部署Temporal组件（创建temporary命名空间、Postgres数据库及相关服务器）
kubectl apply -f infra/k8s/temporal/temporal-dev.yaml

然后，创建命名空间以及所需的秘密文件。在应用程序部署之前，这些秘密文件必须是存在的，因为Pod会从中读取相应的配置信息：

# 创建agent命名空间
kubectl apply -f infra/k8s/00-namespace.yaml

# 创建包含所需密钥的secret文件
kubectl create secret generic agent-secrets -n agent \
 --from-literal=ANTHROPIC_API_KEY=sk-ant-... \
 --from-literal=COMPOSIO_API_KEY=ak_... \
 --from-literal=JWT_SECRET=$(openssl rand -hex 32)

完成上述步骤后，部署应用程序并配置自动扩展功能：

# 安装KEDA，并应用相应的扩展配置
helm repo add kedacore https://kedacore.github.io/charts
helm repo update
helm install keda kedacore/keda -n keda --create-namespace --wait

kubectl apply -f infra/k8s/40-keda-worker-scaledobject.yaml -f infra/k8s/41-gateway-hpa.yaml

最后，将gateway端口转发到本地主机，以便从您的机器上访问它：

# 将gateway端口转发到localhost:8000
kubectl -n agent port-forward svc/gateway 8000:8000

将前端应用程序的访问地址设置为http://localhost:8000，这样就可以开始执行任务了。

注意：对于这个项目来说，您不需要修改apps/worker/或apps/gateway/目录下的.env文件。这些文件仅用于在您的机器上直接运行应用程序。

在集群中，Pod会从ConfigMap中获取配置信息，而您刚刚创建的秘密文件也会在运行时作为环境变量被加载到Pod中。

应用程序中的核心组件

这个项目规模非常大。如果从头开始逐一分析每一行代码，那么阅读这些代码将会花费数小时的时间。因此，我将重点介绍那些真正让系统正常运行的核心组件。

代理循环机制

代理循环机制是整个系统的核心。每当有任务被分配执行时，就会启动这个循环流程。

async def run_agent(
    user_id: str,
    goal: str,
    toolkit_hint: str | None = None,
) -> dict:

需要三样东西：用户的ID（这样Composio就能知道应该使用哪些已连接的账户）、任务目标本身，以及一个可选的工具包提示。这个提示可以帮助确定应该加载哪些工具。如果任务明确是与Gmail相关的操作，那么传递“gmail”这个参数就可以避免加载用户所连接的所有工具。

在循环开始之前，系统会创建一个Composio会话，并为该用户获取所需的工具：

session = await create_session(user_id, toolkit_hint);
tools = await get_tools(session);

然后实际的循环才会开始运行：

for turn in range(1, settings.max_iterations + 1):
    response = await client.messages.create(
        model 설정.model,
        max_tokens 设置.max_tokens,
        system=SYSTEM_PROMPT,
        tools=tools,
        messages=messages,
    )

if response.stop_reason == "end_turn":
    return finish("completed", _extract_text(response.content))

if response.stop_reason == "tool_use":
    # 执行相应的工具，将结果添加到对话历史中，然后继续循环
    # ...

在每一轮循环中，Claude会查看任务目标以及之前的对话记录，然后决定接下来该执行什么操作。stop_reason字段可以说明当前发生了什么：

• "end_turn"：表示Claude已经完成了任务，正在返回最终答案。

• "tool_use"：表示Claude需要调用一个或多个工具。此时循环会通过Composio执行这些工具，将结果添加到对话历史中，然后继续下一轮循环。

如果某个工具的调用失败了，错误信息会被反馈回对话流程中，而不会导致整个程序崩溃：

except ComposioError as exc:
    tool_result_blocks = [
        {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": f"工具执行失败：{exc}",
            "is_error": True,
        }
    ] for block in tool_useBlocks

循环最多会运行max_iterations轮，默认值为20，这个数值在apps/worker/agent/config.py文件中定义。如果循环达到了最大次数却仍未完成任务，系统会返回maxiterations_reached状态码，而不会无限期地等待下去。

每次调用run_agent函数时，都会返回一个结构相同的字典，其中包含状态信息、任务总结以及所执行的各个步骤。这种统一的格式使得Temporal能够方便地存储和查看处理结果，我们接下来会详细讨论这一点。

利用Temporal让系统具有持久性

单独来看，这个代理循环存在一个严重的问题：如果工作进程在执行一个包含15个步骤的任务时发生崩溃，那么之前所有的进度都会丢失，用户将无法知道任务进行到了哪一步，因此必须从头开始重新执行。

工作流与活动

Temporal将代码分成了两个不同的部分：工作流和活动。

工作流定义了应该发生什么以及这些操作的执行顺序，但它本身并不会实际执行任何操作，也不会进行任何网络请求。正是这种设计使得Temporal能够在系统崩溃后安全地重新执行工作流，从而恢复之前的状态。

活动才是真正执行任务的地方。网络请求、大语言模型的调用、工具的执行——所有这些操作都发生在活动中。活动可能会失败，但可以独立地重新尝试，而不会影响工作流程的状态。

在这个项目中，`apps/worker/workflows.py`文件中的`AgentWorkflow`就是工作流程，而`apps/worker/activities.py`文件中的`run_agent_activity`则是包裹代理循环的活动。

工作流程

@workflow.defn(name="AgentWorkflow")
class AgentWorkflow:
    def init(self) -> None:
        self._status: str = "running"
        self._result: dict | None = None

当有任务被分配时，Temporal会启动这个工作流程。它会设置重试策略，并将所有的实际操作交给相应的活动来执行：

retry = RetryPolicy(
  (initial_interval = timedelta(seconds=2)),
  (backoff_coefficient = 2.0),
  (maximum_interval = timedeltaminutes=2)),
  (maximum_attempts = 5),
  (non_retryable_error_types = ["ValueError", "AuthenticationError"])
);

result = await workflow.execute_activity(
  run_agent_activity,
  (args = [user_id, goal, toolkit_hint]),
  (start_to_close_timeout = timedelta(minutes=30)),
  (retry_policy = retry),
);

将`start_to_close_timeout`设置为30分钟，并允许最多进行5次尝试，是因为代理任务确实可能需要这么长的时间来完成。根据你的工作需求，你可以调整这个时间限制。

查询工作流程的状态

Temporal的一个优点就是它提供了便捷的查询功能。工作流程能够直接暴露其当前状态和结果，而无需额外的数据库来记录这些信息：

@workflow.query
def status(self) -> str:
    return self._status

@workflow.query
def result(self) -> dict | None:
    return self._result

前端可以通过询问Temporal“工作流程X的状态是什么？”来实时获取答案。这就是前端轮询机制的工作原理。

活动

这个活动的实现非常简单，它只是包裹了`run_agent`函数，并记录下执行过程中发生的一切：

@activity.defn(name="run_agent_activity")
async def run_agent_activity(user_id: str, goal: str, toolkit_hint: str | None) -> dict:
    result = await run_agent(user_id=user_id, goal=goal, toolkit_hint=toolkit_hint)
    return result

任何涉及网络操作的步骤都发生在活动中，而不是工作流程中。这种分离机制正是Temporal能够高效运行的关键。

工作进程

worker = Worker(
            client,
            task_queue=temporal_settings.temporal_task_queue,
            workflows=[AgentWorkflow],
            activities=[run_agent_activity, notify_activity],
            max_concurrentactivities=5,
)

  @router.post("/dispatch", response_model=DispatchResponse)
  async def dispatch(
      body: DispatchRequest,
      user_id: str = Depends(current_user_id),
  ) -> DispatchResponse:
      if body.toolkit:
          access = await check_toolkit_access(user_id, body/toolkit)
          if not access["allowed"]:
              raise HTTPException(
                  status_code=status.HTTP_409_CONFLICT,
                  detail={
                      "error": "toolkit_not_connected",
                      "connect_url": access["connect_url"],
                  },
              )

      workflow_id = f"agent-{user_id}-{uuid.uuid4().hex[:8]}"
      await client.start_workflow(
          WORKFLOW_NAME,
          args=[user_id, bodygoal, body.toolkit],
          id=workflow_id,
          task_queuesettings.temporal_task_queue,
          cron_schedule=body.schedule or "",
      )
      return DispatchResponse(workflow_id=workflow_id, status="dispatched")

  async def check_toolkit_access(user_id: str, toolkit_hint: str | None) -> dict:
      if not toolkit_hint:
          return {"allowed": True}

      connected = await asyncio.to_thread(
          _has_active_account, composio, user_id, toolkit_hint
      )
      if connected:
          return {"allowed": True}

      connect_url = await asyncio.to_thread(
          _connect_link, composio, user_id, toolkit_hint
      )
      return {"allowed": False, "toolkit": toolkit_hint, "connect_url": connect_url}

  @router.get({"workflow_id"}, response_model=TaskStatusResponse)
  async def get_task(workflow_id: str, ...) -> TaskStatusResponse:
      if not _owns(workflow_id, user_id):
          raise HTTPException(status_code=404, detail="任务未找到")

      handle = client.get_workflow_handle(workflow_id, run_id=run_id)
      agent_status = await handle.query("status")

      if desc.status == WorkflowExecutionStatus.COMPLETED:
          result = await handle.query("result")

      return TaskStatusResponse(...)

FROM python:3.14-slim-bookworm AS builder

RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

COPY requirements.txt .
RUN pip install -r requirements.txt

FROM python:3.14-slim-bookworm AS runtime

ENV PYTHONUNBUFFERED=1 PATH="/opt/venv/bin:$PATH"

RUN useradd --create-home --uid 10001 app
WORKDIR /app

COPY --from=builder /opt/venv /opt/venv
COPY . .

USER app
EXPOSE 8000
CMD ["python", "main.py"]

# 使用procps命令可以让Kubernetes的存活检查机制通过pgrep来检测进程是否仍在运行
RUN apt-get update \
 && apt-get install -y --no-install-recommends procps \
 && rm -rf /var/lib/apt/lists/*

docker build \
 -f "$ROOT/docker/Dockerfilegateway" \
    -t "agent-gateway:$TAG" \
 "$ROOT/apps/gateway"

docker build \
 -f "$ROOT/docker/Dockerfile.worker" \
    -t "agent-worker:$TAG" \
 "$ROOT/apps/worker"

k3d image import "agent-gateway:dev" "agent-worker:dev" -c agent

data:
  MODEL: "claude-opus-4-8"
  MAX_TOKENS: "4096"
  MAX_ITERATIONS: "20"
  TEMPORAL_HOST: "temporal-frontend.temporal.svc.cluster.local:7233"
  TEMPORAL_TASK_QUEUE: "agent-tasks"
  GATEWAY_HOST: "0.0.0.0"
  GATEWAY_PORT: "8000"

apiVersion: v1
kind: Service
metadata:
  name: gateway
spec:
  type: ClusterIP
  ports:
    - port: 8000
      targetPort: 8000

# 该工作节点部署仅用于与Temporal进行通信
spec:
  replicas: 1
  containers:
    - name: worker
      image: agent-worker:dev
      livenessProbe:
        exec:
          command: ["pgrep", "-f", "worker.py"]
        initialDelaySeconds: 15
        periodSeconds: 20
      resources:
        requests:
          cpu: 250m
          memory: 512Mi
        limits:
          cpu: "1"
          memory: 1Gi

kubectl apply -f "$K8S/00-namespace.yaml"
kubectl apply -f "$K8S/01-configmap.yaml"
kubectl apply -f "$K8S/10-gateway-deployment.yaml"
kubectl apply -f "$K8S/20-worker-deployment.yaml"

spec:
  scaleTargetRef:
    name: worker
  minReplicaCount: 0
  maxReplicaCount: 10
  cooldownPeriod: 120
  triggers:
    - type: temporal
      metadata:
        endpoint: temporal-frontend.temporal.svc.cluster.local:7233
        namespace: default
        taskQueue: agent-tasks
        queueTypes: "workflow,activity"
        targetQueueSize: "5"
        activationTargetQueueSize: "0"

spec:
  minReplicas: 2
  maxReplicas: 6
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 60

helm install keda kedacore/keda -n keda --create-namespace --wait
kubectl apply -f infra/k8s/40-keda-worker-scaledobject.yaml -f infra/k8s/41-gateway-hpa.yaml