谁来为 AI 值班？智能体归属权的未解危机

The New Stack | DevOps, Open Source, and Cloud Native News 2026-10-22T21:42:30.263761

凌晨 2 点 07 分，一个 AI 智能体开始出故障。应用还在运行，但由于下游 API 返回错误，请求开始超时。这个智能体已经重试了两次，消耗的 token 比平时更多，响应时间也大约翻了一倍。此时第一个该问的问题不是「我们该用哪个模型？」，而是「现在这个系统归谁管？」

在构建 AI 智能体时，这个问题很容易被忽略。团队往往把精力投入到提示词、检索、工具和模型选择上，但真正决定一个智能体是成为可靠的服务、还是又一个运维风险来源的，是它在生产环境中的归属权。「团队往往把精力投入到提示词、检索、工具和模型选择上，但真正决定一个智能体是成为可靠的服务、还是又一个运维风险来源的，是它在生产环境中的归属权。」

一个生产环境中的智能体需要有负责人、边界、遥测（telemetry，即运行数据的采集与监控）、故障策略、升级路径，并且要明确区分哪些环节系统可以自动处理、哪些必须由人来介入。

生产环境中的智能体不应该从推理直接跳到执行。每一次工具调用都应该经过明确的控制、保持可观测，并且有清晰的人工介入路径。下面是一套构建这些控制的实用方法。

第一步：把智能体当作生产服务来对待

智能体应该和其他任何生产服务一样，拥有相同的运维身份。首先要定义它的归属元数据。

from pydantic import BaseModel

class AgentService(BaseModel):
    name: str
    owner_team: str
    service_tier: str
    escalation_channel: str

agent_service = AgentService(
    name="incident-command-agent",
    owner_team="platform-engineering",
    service_tier="critical",
    escalation_channel="#platform-oncall",
)

) 这看起来简单，但改变了讨论的框架。Agent 不再是跑在某个 Python 进程里的实验，而是一个有明确负责团队和升级路径的服务。“Agent 不再是跑在某个 Python 进程里的实验，而是一个有明确负责团队和升级路径的服务。”之后，你可以把同一份元数据挂到日志、追踪、仪表盘、告警和部署记录上。

第 2 步：定义 Agent 能做什么

归属权也意味着要明确 Agent 被允许做什么。一个只能读取监控数据的 Agent，风险远小于一个能重启生产服务、变更基础设施或部署代码的 Agent。所以我倾向于显式工具白名单（allowlist），只允许 Agent 执行你刻意授予的操作。

ALLOWED_TOOLS = {
    "get_service_health",
    "get_recent_deployments",
    "search_runbooks",
    "summarize_incident",
}

class ToolAuthorizationError(Exception):
    """Raised when an agent attempts to use an unapproved tool."""

def authorize_tool(tool_name: str) -> None:
    if tool_name not in ALLOWED_TOOLS:
        raise ToolAuthorizationError(
            f"Tool '{tool_name}' is not authorized."
        )

指导原则是默认拒绝。新增一项能力应当是经过考量的工程决策，而不是因为某个工具恰好对 Agent 可用就顺带发生。对于破坏性操作，授权应更严格。重启生产服务或部署代码，必须经过人工认证审批，不能由 Agent 自行判断。

第 3 步：给 Agent 执行加上预算

一个可以无限推理下去的 Agent 很难运维。每个生产工作流都应当对执行时间、模型调用次数、工具调用次数设限，最好还能限制 token 消耗。

import os

class ExecutionBudget:
    def __init__(
        self,
        max_steps: int | None = None,
        max_tools: int | None = None,
    ):
        self.steps = 0
        self.tool_calls = 0
        self.max_steps = (
            max_steps
            if max_steps is not None
            else int(os.getenv("MAX_AGENT_STEPS", "5"))
        )
        self.max_tools = (
            max_tools
            if max_tools is not None
            else int(os.getenv("MAX_TOOL_CALLS", "8"))
        )

if self.max_steps <= 0 or self.max_tools <= 0:
raise ValueError(
"Execution budget limits must be positive."
)
def check_step(self) -> None:
if self.steps >= self.max_steps:
raise RuntimeError(
"Agent step budget exceeded."
)
self.steps += 1
def check_tool_call(self) -> None:
if self.tool_calls >= self.max_tools:
raise RuntimeError(
"Agent tool-call budget exceeded."
)
self.tool_calls += 1

执行限制应该做成可配置的，而不是埋在应用代码深处。工程师可以通过构造函数传入这些限制，给不同的 agent 分配不同预算，而无需改动全局设置。诊断类 agent 可能只需要三步，而更复杂的研究工作流往往需要多得多。如果没有传入限制，类会退回到读取环境变量。真正重要的是：每个工作流都有明确的上限，并且在执行全程都强制同一个预算。

第四步：让模型调用可观测

负责人看不到的东西，就没法管。至少要记录这些信息：

• 请求延迟
• 使用的模型
• 成功还是失败
• 工具调用情况
• 可获取时的 token 用量
• 关联 ID 或请求 ID
• 升级事件

用一个简单的包装函数，就能划出一个有用的观测边界。

import logging
import time
import uuid

logger = logging.getLogger(__name__)

def invoke_model(model, messages):
    request_id = str(uuid.uuid4())
    started = time.perf_counter()

    # Extract the model name across different provider libraries.
    model_name = (
        getattr(model, "model_name", None)
        or getattr(model, "model", None)
        or "unknown_model"
    )

    try:
        response = model.invoke(messages)
        elapsed = time.perf_counter() - started
        logger.info(
            "agent_model_call_success",
            extra={
                "request_id": request_id,
                "model": model_name,
                "latency_seconds": round(elapsed, 3),
            },
        )
        return response
    except Exception:
        elapsed = time.perf_counter() - started
        logger.exception(
            "agent_model_call_failed",
            extra={
                "request_id": request_id,
                "model": model_name,
                "latency_seconds": round(elapsed, 3),
            },
        )

请求 ID 在涉及多个服务的故障排查中尤其有用。工程师应当能从一条告警，追溯到对应的 Agent 请求，再进一步查到背后的模型调用和工具调用。在生产环境中，还要在这个日志层之上补充分布式追踪和指标，才能更完整地看清系统行为和性能。

第 5 步：把“失败”和“升级”分开

一次模型调用失败，不代表整个服务就该跟着挂掉。反过来，模型调用成功，也不代表 Agent 给出的建议可以自动执行。这是两个不同的判断。

def handle_agent_failure(error: Exception) -> dict:
    logger.error(
        "Agent execution failed; escalating to human.",
        exc_info=(
            type(error),
            error,
            error.__traceback__,
        ),
    )
    return {
        "status": "escalated",
        "message": (
            "The AI agent could not complete the operation. "
            "Human intervention is required."
        ),
        "error_type": type(error).__name__,
    }

这个区分在真实故障中很关键。Agent 可能挂了，但工程团队仍然应该能靠底层的工作流和运行手册（runbook）继续响应。这就是“优雅降级”在实践中的样子。

第 6 步：高风险操作必须经人工审批

企业级 Agent 最有用的模式之一，就是把“给建议”和“执行”分开。比如：

遥测数据
    ↓
AI Agent
    ↓
诊断
    ↓
建议操作
    ↓
风险检查
    ↓
人工审批
    ↓
生产环境变更

Agent 可以建议：“把数据库连接池调大。”但它不应该仅仅因为模型生成了这个建议，就直接把变更推到生产环境。

一个简单的审批边界长这样：

def execute_action(
    action: str,
    approved_by: str | None,
) -> dict:
    if not approved_by:
        return {
            "status": "awaiting_approval",
            "action": action,
        }
    return {
        "status": "approved",
        "action": action,
        "approved_by": approved_by,
    }

真正的系统需要更复杂的实现，但关键是架构上的边界划分。「Agent 可以拥有推理能力，但不拥有授权。」Agent 可以拥有推理能力，但不拥有授权。

第 7 步：给 Agent 加健康检查

生产环境的 Agent 和其他服务一样，需要一个健康检查端点。用 FastAPI 可以这样写：

import os
import httpx
from fastapi import FastAPI, Response, status

app = FastAPI()
SERVICE_NAME = "incident-command-agent"

@app.get("/health/live")
async def liveness():
    """检查应用进程是否还在响应。"""
    return {
        "status": "alive",
        "service": SERVICE_NAME,
    }

async def check_provider() -> bool:
    """检查已配置的模型提供方健康端点。"""
    health_url = os.getenv("MODEL_PROVIDER_HEALTH_URL")
    if not health_url:
        return False
    try:
        async with httpx.AsyncClient(timeout=2.0) as client:
            result = await client.get(health_url)
            return result.status_code == 200
    except httpx.RequestError:
        return False

@app.get("/health/ready")
async def readiness(response: Response):
    """检查所需依赖是否可用。"""
    if not await check_provider():
        response.status_code = (
            status.HTTP_503_SERVICE_UNAVAILABLE
        )
        return {
            "status": "unhealthy",
            "reason": "Model provider unavailable",
        }
    return {
        "status": "ready",
        "service": SERVICE_NAME,

}

存活性（liveness）和就绪性（readiness）的用途不同。存活性端点确认应用进程能正常响应，就绪性端点则根据依赖服务是否可用，判断服务能否处理请求。这个例子里，就绪性端点会检查配置好的模型服务商健康检查 URL，检查失败就返回 HTTP 503。这个检查刻意做得轻量、设置了超时，避免带来不必要的延迟或成本。生产环境里，团队应当按服务商支持的健康检查机制来调整，同时把向量数据库、配置服务这类关键依赖也纳入检查。把两种检查分开，Kubernetes 这类编排平台才能区分两种情况：应用需要重启，还是暂时无法提供服务。

第 8 步：安全控制失效时，默认拒绝

最糟糕的故障之一，是安全机制已经失效，agent 却还在继续运行。例如：

def run_controlled_agent(
    tool_name: str,
    model,
    messages,
    budget: ExecutionBudget,
):
    try:
        authorize_tool(tool_name)

        # Reuse the budget across agent steps and tool calls.
        budget.check_step()
        budget.check_tool_call()

        return invoke_model(
            model,
            messages,
        )

    except ToolAuthorizationError:
        logger.exception(
            "Agent authorisation check failed."
        )
        return {
            "status": "blocked",
            "message": (
                "The requested operation is not authorised."
            ),
        }

    except RuntimeError:
        logger.exception(
            "Agent execution budget exceeded."
        )
        return {
            "status": "escalated",
            "message": (
                "The agent exceeded its execution budget. "
                "Human investigation is required."
            ),
        }

    except Exception as exc:

return handle_agent_failure(exc)

注意，不同的失败会导致不同的处理结果。授权失败应当阻止请求的操作，而执行预算耗尽则应触发升级（escalation），而不是再次重试。临时性的模型故障，则应按照系统既定的失败处理策略执行。

另一个重要细节：执行预算必须在整个 agent 工作流中持久存在。如果每次调用都新建一个预算，计数器就会被重置，反复调用就能绕过原本设定的限制。正确做法是只创建一次预算，然后把它传入每一次后续调用，如下所示。

# 为整个 agent 工作流创建一个预算。
budget = ExecutionBudget(
    max_steps=5,
    max_tools=8,
)

# 在后续每次调用中复用同一个预算。
for tool_name in planned_tools:
    result = run_controlled_agent(
        tool_name=tool_name,
        model=model,
        messages=messages,
        budget=budget,
    )
    if isinstance(result, dict) and result.get(
        "status"
    ) in {"blocked", "escalated"}:
        break

多次调用共享同一个预算后，每一次计入的步骤和工具调用都会累加到工作流的总限制上。一旦达到上限，agent 就会停止，而不是拿着一个新预算继续跑。这些控制措施只有在整个执行过程中始终有效才有意义，尤其是当 agent 反复遇到失败的时候。

第 9 步：定义凌晨两点的运行手册

到了这一步，运维责任才真正落地。假设 agent 在凌晨两点开始出故障，值班工程师应该能迅速回答五个问题：

• agent 本身是否健康？
• 模型供应商是否健康？
• 检索或工具依赖是否在失败？
• agent 是否超出了正常的成本或延迟范围？
• 如果我们把它关掉，会发生什么？

最后一个问题最关键。每个生产环境的 AI 系统都应该有一条清晰的降级路径。对于一个事件响应 agent 来说，这条路径可能是：

AI Agent 不可用
    ↓
事件仍被创建
    ↓
遥测数据仍然可用
    ↓
标准运行手册被激活
    ↓

人类工程师继续排查。AI 系统应该让运维更快，而不是变成组织离不开、又管不住的东西。那么，这个 agent 归谁？答案绝不能简单一句“AI 团队”。生产环境的归属必须写清楚。具体要有人对下面这些负责：部署；基础设施；模型配置；工具权限；监控；故障响应；成本；安全；回滚；以及停用 agent 的决策权。

这不等于让一个人把所有事都揽下来。它意味着每个人都清楚自己的角色，真出故障时，团队不用浪费时间争论该谁上。AI agent 的真正考验：跑一个 AI agent 最轻松的时刻，是上班时间一切正常的时候。真正的考验在凌晨两点。一个达到生产标准的 agent，应该有能被呼叫的值班负责人、能防止失控行为的执行预算、能解释清楚“发生了什么”的可观测性、能约束其行为的授权边界，以及一套让工程师可以脱离它继续工作的兜底方案。

目标不是把人踢出流程，而是把界限划清楚：agent 的职责到哪里为止，人在哪里必须接手。因为凌晨两点出事时，问题不是模型够不够聪明，而是系统设计得够不够好，好到值班的人知道发生了什么、agent 做了什么、下一步该做什么。

查看原文