当 AI Agent 进入生产环境:从模型调用走向可运营系统

2026-10-02 35 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

QCon San Francisco 2026 将邀请 Airbnb、OpenAI、Netflix、Honeycomb 等工程组织的实践者,讨论 AI Agent 承担更多任务之后,生产系统应该如何构建、运行与演进。真正困难的部分已经不只是“让模型返回正确答案”,而是把一个具有不确定性、会调用工具、可能持续运行多步的组件,纳入成熟的软件工程体系。

Agent 一旦能读写数据、执行命令或触发业务流程,就不再是普通的聊天界面。它更像一个权限边界模糊、执行路径动态生成的后台服务。工程团队需要重新审视可靠性、可观测性、成本控制和变更管理。

Agent 服务改变了哪些生产假设

传统服务通常拥有相对明确的调用图:请求进入 API,经过业务逻辑和数据库,再返回结果。Agent 的执行图则可能由模型在运行时决定:它会选择工具、根据工具结果修改计划,甚至重复调用同一个接口。

这带来几类明显变化:

  • 延迟不再由一次调用决定:一次用户请求可能展开成多轮模型推理和多个外部 API 调用。
  • 失败不再只有成功与异常:Agent 可能顺利执行,却选择了错误工具,或生成了语法正确但业务含义错误的参数。
  • 成本成为请求级指标:Token、搜索、沙箱运行和第三方 API 都会叠加成本。
  • 权限需要动态收紧:能够“调用工具”并不意味着可以无限制调用所有工具。
  • 重试可能制造副作用:重复发送邮件、创建工单或执行退款,比普通 HTTP 重试更危险。

因此,Agent 平台不能只记录最终回答。至少还要记录模型版本、提示词版本、工具选择、参数摘要、执行时长、重试次数、Token 用量和最终状态。

把每次运行看成一条受约束的工作流

一种更稳妥的设计,是把 Agent 运行建模为有预算、有截止时间、可中断的工作流。模型可以提出下一步动作,但运行时负责决定动作是否被允许。

可以为每次运行设置四类硬边界:

  1. 步骤预算:防止 Agent 在规划与工具调用之间无限循环。
  2. 时间预算:超过截止时间后停止新动作,并尽力取消下游调用。
  3. 成本预算:限制 Token 或外部服务费用。
  4. 权限预算:按用户、任务和环境生成工具白名单。

高风险动作还应经过策略检查或人工确认。例如,读取订单状态可以自动执行,而退款、删除资源和发送批量通知则需要审批。这个边界应由确定性的代码控制,而不是仅靠提示词要求模型“谨慎操作”。

一个可运行的最小执行器

下面的 Python 示例不依赖第三方包,演示步骤上限、工具白名单、参数校验、幂等键和审计日志。示例中的 planner 是确定性的模拟器;接入真实模型时,可以替换 plan_next_action,但应保留运行时的控制逻辑。

将代码保存为 agent_runtime.py,然后运行 python agent_runtime.py:

import json
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class RunContext:
    run_id: str
    deadline: float
    max_steps: int
    allowed_tools: set[str]

ORDERS = {"A-100": {"status": "paid", "total": 42.0}}
processed_actions: set[str] = set()

def get_order(order_id: str) -> dict[str, Any]:
    return ORDERS.get(order_id, {"status": "not_found"})

def create_support_ticket(order_id: str, reason: str, idempotency_key: str) -> dict[str, Any]:
    if idempotency_key in processed_actions:
        return {"status": "duplicate_ignored"}
    processed_actions.add(idempotency_key)
    return {"status": "created", "ticket_id": "T-900", "order_id": order_id, "reason": reason}

TOOLS: dict[str, Callable[..., dict[str, Any]]] = {
    "get_order": get_order,
    "create_support_ticket": create_support_ticket,
}

def audit(event: str, **fields: Any) -> None:
    print(json.dumps({"event": event, **fields}, ensure_ascii=False))

def plan_next_action(step: int, state: list[dict[str, Any]]) -> dict[str, Any]:
    # 用真实模型接入替换这里,并要求模型返回结构化 JSON。
    if step == 0:
        return {"tool": "get_order", "args": {"order_id": "A-100"}}
    if step == 1:
        return {
            "tool": "create_support_ticket",
            "args": {"order_id": "A-100", "reason": "customer requested review"},
        }
    return {"final": "订单已查询,并已创建人工复核工单。"}

def run_agent() -> str:
    ctx = RunContext(
        run_id=str(uuid.uuid4()),
        deadline=time.monotonic() + 5,
        max_steps=4,
        allowed_tools={"get_order", "create_support_ticket"},
    )
    state: list[dict[str, Any]] = []

    for step in range(ctx.max_steps):
        if time.monotonic() >= ctx.deadline:
            raise TimeoutError("agent deadline exceeded")

        action = plan_next_action(step, state)
        audit("action_planned", run_id=ctx.run_id, step=step, action=action)

        if "final" in action:
            audit("run_completed", run_id=ctx.run_id, steps=step + 1)
            return str(action["final"])

        tool_name = action.get("tool")
        if tool_name not in ctx.allowed_tools or tool_name not in TOOLS:
            raise PermissionError(f"tool not allowed: {tool_name}")

        args = dict(action.get("args", {}))
        if tool_name == "create_support_ticket":
            args["idempotency_key"] = f"{ctx.run_id}:{step}:{tool_name}"

        started = time.monotonic()
        result = TOOLS[tool_name](**args)
        duration_ms = round((time.monotonic() - started) * 1000, 2)
        audit(
            "tool_completed",
            run_id=ctx.run_id,
            step=step,
            tool=tool_name,
            duration_ms=duration_ms,
            result=result,
        )
        state.append({"action": action, "result": result})

    raise RuntimeError("agent step budget exhausted")

if __name__ == "__main__":
    print(run_agent())

这个示例离完整平台仍有距离,但它体现了一个关键分工:模型负责建议动作,运行时负责授权、执行和留痕。接入生产环境时,还可以增加 JSON Schema 校验、分布式追踪、持久化检查点、熔断器和人工审批队列。

可观测性不能停留在一次请求的耗时

Agent 的观测单位应该同时覆盖“运行”和“步骤”。只看 API 的总延迟,很难回答究竟是模型推理慢、工具服务慢,还是 Agent 发生了循环。

建议至少建立以下指标:

  • agent_run_duration_seconds:一次完整运行的耗时;
  • agent_steps_total:每次运行的步骤数分布;
  • tool_call_duration_seconds:按工具划分的延迟;
  • tool_call_errors_total:工具失败和策略拒绝次数;
  • agent_tokens_total:按模型、租户和任务统计 Token;
  • agent_budget_exhausted_total:时间、步骤或成本预算耗尽次数;
  • agent_human_escalation_total:转人工的比例与原因。

日志中不要无条件保存完整提示词和工具结果。它们可能包含个人信息、访问令牌或客户业务数据。更稳妥的做法是默认记录结构化元数据,对敏感字段进行脱敏,并为原始内容设置独立权限和保留期限。

上线时用工程门槛代替演示效果

Agent 的演示通常展示最佳路径,而生产环境必须覆盖失败路径。团队可以从低风险、只读任务开始,再逐步开放写操作。上线前建议检查:

  • 是否能重放一次运行,并定位每一步使用的模型、提示词和工具版本;
  • 所有产生副作用的工具是否支持幂等,或明确禁止自动重试;
  • 是否存在步骤、时间、成本和并发上限;
  • 高风险动作是否由独立策略层或人工审批控制;
  • 模型或提示词升级是否经过离线评测、影子流量和小比例灰度;
  • 出现异常时,是否能立即禁用某个模型、工具或 Agent 能力;
  • 审计数据是否兼顾排障需要与隐私要求。

AI Agent 不会让传统生产工程失效,反而会放大它的重要性。可靠的 Agent 系统依然需要超时、隔离、幂等、可观测性和渐进式发布,只是这些机制必须从固定调用链扩展到动态决策链。真正成熟的系统,不是让 Agent 拥有最多工具,而是让每一次自主行动都可限制、可解释、可停止。


相关推荐