AI 智能体上线后由谁值班：生产归属的九个工程步骤

凌晨2点07分，一个AI智能体开始出现故障。

应用仍在运行，但请求正在超时，因为某个下游API正在返回错误。该智能体已经重试了两次，消耗的token比平时更多，响应时间也大约翻了一倍。

此时第一个问题不应该是：

“我们应该用哪个模型？”

而应该是：

“现在谁负责这个系统？”

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

“团队往往把精力投入到提示词、检索、工具和模型选择上，但生产归属决定了智能体是成为一个可靠的服务，还是又一个运营风险来源。”

一个生产级智能体需要负责人、边界、遥测、故障策略、升级路径，以及明确区分系统可以自动化什么、什么需要人工介入。

生产级智能体不应拥有一条从推理直达行动的路径。每一次工具调用都应经过明确的控制、保持可观测，并有一条通往人工介入的既定路径。

以下是一种构建这些控制的实用方法。

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

智能体应当拥有与任何其他生产服务相同的运营身份。

首先定义它的归属元数据。

Python 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", )

这看起来可能很简单，但它改变了对话的性质。

智能体不再是一个在某个Python进程里运行的实验。它是一个有明确负责团队和升级路径的服务。

“智能体不再是一个在某个Python进程里运行的实验。它是一个有明确负责团队和升级路径的服务。”

你之后可以把同样的元数据附加到日志、追踪、仪表盘、告警和部署记录上。

第2步：定义智能体被允许做什么

归属也意味着要明确智能体被允许做什么。

一个只能读取遥测数据的智能体，其风险远低于一个可以重启生产服务、更改基础设施或部署代码的智能体。

这就是为什么我倾向于使用显式的工具允许列表，它把智能体限制在你刻意授予的操作范围内。

Python 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." )

指导原则是默认拒绝。

添加新能力应当是一项有意的工程决策，而不是因为某个工具对智能体变得可用就顺带发生。

对于破坏性操作，授权应当更加严格。生产重启或部署应当要求经过身份验证的人工批准，而不是由智能体自行判断。

第3步：为智能体执行设置预算

一个可以无限推理的智能体很难运营。

每个生产工作流都应当对执行时间、模型调用、工具调用，最好还有token消耗设置限制。

Python 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 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

执行限制应当可配置，而不是埋在应用代码里。通过构造函数传入这些限制，工程师可以为不同的智能体分配不同的预算，而无需更改全局设置。一个诊断型智能体可能只需要三步，而一个更复杂的研究工作流可能需要多得多。当未提供限制时，该类会回退到环境变量。重要的是每个工作流都有一个明确的上限，并且在其整个执行过程中都强制执行同一预算。

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

负责人无法运营他们看不见的东西。

在可用的情况下记录token使用量；

关联ID或请求ID；

升级事件。

一个简单的包装器提供了一个有用的边界。

Python 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), }, )

raise

当一次事故涉及多个服务时，请求ID就显得尤为重要。

工程师应当能够从告警追踪到智能体请求，再追踪到底层的模型调用和工具调用。

在生产环境中，应当用分布式追踪和指标来补充这一日志层，以便更全面地了解系统行为和性能。

第5步：将失败与升级区分开

一次模型调用失败并不一定意味着整个服务都应该失败。

同样，一次成功的模型调用也不意味着智能体的建议应当自动执行。

这些是不同的决策。

Python 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__, }

在真实事故中，这种区分很重要。

智能体可能会宕机，但工程团队仍应能够借助底层工作流和运行手册进行响应。这就是优雅降级在实践中的样子。

第6步：让高风险操作经过人工审批

对企业智能体而言，最有用的模式之一就是将建议与执行分开。

例如：

智能体可能会建议：

“增大数据库连接池。”

但它不应该仅仅因为模型生成了这条建议，就把该变更推送到生产环境。

下面是一个简单的审批边界示例：

Python 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, }

真实系统需要更复杂的实现，但真正重要的是这一架构边界。

“智能体可以负责推理，但不能负责授权。”

智能体可以负责推理，但不能负责授权。

第7步：为智能体提供健康检查

生产环境中的智能体和其他任何服务一样，都需要一个健康检查端点。

使用FastAPI：

`Python import os

import httpx from fastapi import FastAPI, Response, status

app = FastAPI()

SERVICE_NAME = "incident-command-agent"

@app.get("/health/live") async def liveness(): """Check whether the application process is responsive.""" return { "status": "alive", "service": SERVICE_NAME, }

async def check_provider() -> bool: """Check a configured provider health endpoint.""" 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): """Check whether required dependencies are available.""" 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, }

存活检查和就绪检查的用途不同。存活端点确认应用进程能够响应，而就绪检查则根据所需依赖的可用性来判断服务能否处理请求。

在这个示例中，就绪端点会检查一个已配置的模型提供商健康URL，并在检查失败时返回HTTP 503。该检查被有意设计得轻量且有时间限制，以避免引入不必要的延迟或成本。在生产环境中，团队应将其调整为适配提供商所支持的健康检查机制，并覆盖其他关键依赖，例如向量数据库或配置服务。

将这两类检查分开，可以让Kubernetes等编排平台区分一个需要重启的应用和一个暂时无法处理流量的应用。

第8步：安全控制失效时默认阻断

最糟糕的故障模式之一，就是在安全机制失效后仍让智能体继续运行。

例如：

Python 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)

注意，不同的失败会产生不同的结果。授权失败应阻止所请求的操作，而执行预算耗尽应触发升级，而不是再次重试。与此同时，临时性的模型故障应遵循系统定义的故障处理策略。

另一个重要细节：执行预算必须在智能体的整个工作流中持续存在。为每次调用创建新的预算会重置其计数器，使重复调用能够绕过预期的限制。相反，应只创建一次预算，并将其传入每一次后续调用，如下所示。

Python # 为整个智能体工作流创建一个预算。 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

由于在多次调用之间共享同一个预算，每个计入的步骤和工具调用都会计入工作流的总体限制。一旦达到该限制，智能体就会停止，而不是带着新的预算继续运行。这些控制措施只有在整个执行过程中始终保持有效才能发挥作用，尤其是在智能体反复遇到失败时。

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

这正是运维归属真正落地的地方。

假设智能体在凌晨 2 点开始出现故障。

值班工程师应当能够迅速回答五个问题：

智能体本身是否健康？

模型提供商是否健康？

检索或工具依赖是否出现故障？

智能体是否超出了正常的成本或延迟范围？

如果我们将其禁用会发生什么？

最后一个问题最为重要。

每个生产环境的 AI 系统都应当有一条清晰的降级路径。

对于事件响应智能体而言，这可能意味着：

AI 智能体不可用 ↓ 事件仍被创建 ↓ 遥测数据仍然可用 ↓ 标准运行手册被激活 ↓ 人类工程师继续调查

AI 系统应当加快运营速度，而不是成为组织离开它就无法运转的东西。

那么，谁来负责这个智能体？

答案绝不应仅仅是：

“AI 团队。”

生产环境的归属需要明确。

禁用智能体的决定。

这并不意味着必须由一个人处理所有事情。它意味着每个人都足够清楚自己的角色，以至于当事件发生时，团队不会浪费时间弄清楚该由谁来负责。

对 AI 智能体的真正考验

运行 AI 智能体最容易的时候是工作时间，一切正常运转之时。

真正的考验出现在凌晨 2 点。

一个达到生产就绪标准的智能体应当有一位可以被呼叫的负责人、一个防止失控行为的执行预算、能够解释所发生情况的可观测性、约束其可执行操作的授权边界，以及一个让工程师能够在没有它的情况下继续工作的回退方案。

目标并不是把人类排除在循环之外。

“一个达到生产就绪标准的智能体应当有一位可以被呼叫的负责人、一个执行预算、能够解释所发生情况的可观测性、授权边界，以及一个让工程师能够在没有它的情况下继续工作的回退方案。”

而是要明确智能体的责任在哪里结束，以及人类需要在何处介入。

因为当凌晨 2 点出问题时，问题不在于模型是否足够聪明。

而在于系统是否设计得足够好，让值班人员知道发生了什么、智能体做了什么，以及接下来该怎么做。

文章《谁来为 AI 值班？这是智能体归属尚未解决的危机》首发于 The New Stack。