QCon AI Boston 2026 释放出的信号很明确:AI Agent 的工程重点正在从“怎样写出更好的提示词”转向“怎样让 Agent 在生产环境中稳定、可控、可评估地运行”。上下文管理、包围 Agent 的安全 harness,以及贯穿开发和上线流程的评测体系,开始成为真正决定系统质量的基础设施。
提示词只是入口,平台才是运行边界
在原型阶段,团队往往把主要精力放在 system prompt、工具描述和少量示例上。但进入生产环境后,问题会迅速扩展:
- Agent 获得了哪些上下文,这些数据是否仍然有效?
- 模型准备调用什么工具,参数是否越权?
- 一次任务消耗了多少 token、时间和外部 API 配额?
- 模型失败后应该重试、降级,还是转交人工?
- 新模型或新提示词上线后,原有任务是否发生回归?
这些问题不能靠继续堆叠提示词解决。更可行的形态是把模型放进一个受控平台:模型负责生成候选决策,平台负责身份认证、上下文装配、工具授权、执行限制、日志记录和结果评测。
可以把生产 Agent 拆成五个层次:
- 上下文层:检索、筛选、压缩并标注信息来源和时效。
- 模型层:生成计划、结构化参数或最终回答。
- Harness 层:验证输入输出,控制工具、预算、超时和重试。
- 执行层:真正访问数据库、内部服务或第三方 API。
- 评测与观测层:记录轨迹,运行离线回归和在线质量检查。
这种拆分的关键不是增加抽象,而是明确责任:模型输出不等于授权,工具调用建议也不等于工具已经执行。
Harness:把 Agent 当作不完全可信的执行者
这里的 harness 可以理解为包围 Agent 的运行时控制层。它既不是一条更长的 system prompt,也不只是敏感词过滤器,而是一组由代码强制执行的策略。
一个实用的 harness 通常需要处理:
- 工具白名单和逐工具参数校验;
- 用户身份、租户和数据权限传递;
- 最大步骤数、token 预算、费用和超时;
- 提示注入与敏感数据泄漏检测;
- 写操作确认、幂等键和审计日志;
- 模型异常、工具异常与策略拒绝的分类处理。
例如,客服 Agent 可以查询订单,但退款操作不应仅凭模型生成一句“调用 refund”。Harness 应校验订单归属、退款金额、业务状态和人工审批条件。对于删除数据、转账、发布内容等不可逆操作,还应把“生成计划”和“执行变更”拆成两个阶段。
安全边界也不应只部署在模型前面。检索到的文档、工具返回值乃至网页内容都可能携带恶意指令,因此上下文进入模型前要过滤,模型输出进入工具前还要再次验证。
上下文管理是数据工程问题
Agent 的上下文窗口再大,也不意味着应该把所有历史记录塞进去。未经治理的长上下文会引入旧数据、相互冲突的指令、无关噪声和更高成本。
生产系统可以为每一段上下文附加元数据:
source:数据来自用户、内部数据库还是外部网页;timestamp:内容产生或更新的时间;trust_level:可信等级;tenant_id:所属租户;expires_at:失效时间;permissions:当前调用者是否有权读取。
装配上下文时,平台应先做权限过滤,再按任务相关性、可信度和新鲜度排序,最后执行 token 预算裁剪。摘要可以降低成本,但不能成为唯一事实来源;涉及金额、身份和状态的数据,最好保留结构化原值及来源引用。
可以这样实践:实现一个最小 Harness
下面是一个只依赖 Python 标准库的最小示例。它不调用真实大模型,而是模拟模型提出工具调用,再由 harness 执行白名单、参数、权限和预算检查。这里明确做一个假设:模型层已经能输出形如 {"tool": "lookup_order", "arguments": {...}} 的结构化结果。
将以下内容保存为 agent_harness.py,然后运行 python agent_harness.py:
from dataclasses import dataclass
from typing import Any, Callable
@dataclass(frozen=True)
class RequestContext:
user_id: str
tenant_id: str
roles: frozenset[str]
class PolicyError(Exception):
pass
class AgentHarness:
def __init__(self, max_steps: int = 3) -> None:
self.max_steps = max_steps
self.steps = 0
self.tools: dict[str, Callable[..., dict[str, Any]]] = {
"lookup_order": self.lookup_order,
"refund_order": self.refund_order,
}
def execute(self, proposal: dict[str, Any], ctx: RequestContext) -> dict[str, Any]:
self.steps += 1
if self.steps > self.max_steps:
raise PolicyError("agent step budget exceeded")
tool_name = proposal.get("tool")
arguments = proposal.get("arguments")
if tool_name not in self.tools:
raise PolicyError(f"tool is not allowed: {tool_name!r}")
if not isinstance(arguments, dict):
raise PolicyError("arguments must be an object")
self.validate_arguments(tool_name, arguments)
self.authorize(tool_name, ctx)
print({
"event": "tool_call_approved",
"tool": tool_name,
"user_id": ctx.user_id,
"tenant_id": ctx.tenant_id,
})
return self.tools[tool_name](ctx=ctx, **arguments)
@staticmethod
def validate_arguments(tool_name: str, arguments: dict[str, Any]) -> None:
required = {
"lookup_order": {"order_id"},
"refund_order": {"order_id", "amount"},
}[tool_name]
if set(arguments) != required:
raise PolicyError(
f"invalid fields for {tool_name}: expected {sorted(required)}"
)
if not isinstance(arguments["order_id"], str):
raise PolicyError("order_id must be a string")
if tool_name == "refund_order":
amount = arguments["amount"]
if not isinstance(amount, (int, float)) or not 0 < amount <= 100:
raise PolicyError("automatic refund must be between 0 and 100")
@staticmethod
def authorize(tool_name: str, ctx: RequestContext) -> None:
if tool_name == "refund_order" and "refund_operator" not in ctx.roles:
raise PolicyError("refund requires refund_operator role")
@staticmethod
def lookup_order(order_id: str, ctx: RequestContext) -> dict[str, Any]:
return {
"order_id": order_id,
"tenant_id": ctx.tenant_id,
"status": "paid",
}
@staticmethod
def refund_order(
order_id: str, amount: float, ctx: RequestContext
) -> dict[str, Any]:
return {
"order_id": order_id,
"refunded": amount,
"tenant_id": ctx.tenant_id,
}
if __name__ == "__main__":
context = RequestContext(
user_id="user-42",
tenant_id="tenant-a",
roles=frozenset({"support_agent"}),
)
model_proposal = {
"tool": "refund_order",
"arguments": {"order_id": "ORD-1001", "amount": 25.0},
}
try:
result = AgentHarness().execute(model_proposal, context)
print({"result": result})
except PolicyError as exc:
print({"event": "tool_call_rejected", "reason": str(exc)})
当前用户没有 refund_operator 角色,因此示例会拒绝退款。把角色集合改为下面的内容即可观察授权成功路径:
roles=frozenset({"support_agent", "refund_operator"})
真实项目还应把工具参数定义为 JSON Schema 或强类型模型,将审计事件写入集中日志,并为写操作加入幂等键。模型生成的参数即使通过 JSON 解析,也不能跳过业务校验。
Evals 要覆盖轨迹,而不只是最终答案
传统接口测试通常判断状态码和返回字段,Agent 评测则需要观察整个执行轨迹。最终答案看起来正确,并不代表过程可接受:模型可能读取了越权数据、调用了多余工具,或在偶然条件下碰巧得到答案。
建议至少建立四类评测:
- 任务成功率:用户目标是否完成,关键事实是否正确;
- 策略合规率:是否使用允许的工具和数据,是否触发必要确认;
- 轨迹质量:步骤数、重复调用、失败重试和无效检索;
- 运行指标:端到端延迟、token 消耗、工具错误率和单任务成本。
评测集不应只有理想输入。还要加入过期文档、冲突指令、缺少权限、工具超时、格式错误和提示注入等案例。每次修改模型、提示词、检索策略或工具描述后,都应运行同一套回归集,并保存失败轨迹以便定位差异。
在线评测则应谨慎使用。自动评分模型适合做趋势监控和样本筛选,但不应单独承担高风险业务的放行决策。退款、医疗、金融或权限变更等场景仍需要确定性规则和人工复核。
上线前的工程检查
团队不必一开始就建设庞大的 Agent 平台,但应优先补齐最容易形成事故的控制点:
- 工具调用使用结构化协议,并在服务端校验参数;
- 模型身份与最终用户身份分离,权限按用户和租户计算;
- 为步骤数、时间、token 和费用设置硬上限;
- 对写操作提供确认、幂等、审计和人工接管;
- 上下文携带来源、时效、权限和可信度信息;
- 保存可脱敏的执行轨迹,建立固定回归集;
- 模型、提示词、工具和检索配置全部版本化;
- 准备超时、模型不可用和外部工具失败时的降级路径。
从提示词走向平台,并不意味着提示词不再重要。它意味着团队承认模型只是系统中的一个概率组件。真正可靠的生产 Agent,需要由平台限定能力,由 harness 执行边界,再由 evals 持续证明变更没有破坏既有行为。