CMA：Prompt 版本管理与回滚

服务端 Prompt 版本控制和回滚机制

🌐 查看英文原文 | 源码 Notebook

Prompt 版本控制和回滚

想象你是一个 PM，你的产品支持系统使用 LLM 将收到的工单路由到正确的团队。你想调整路由 prompt，使更多与 API 相关的工单发送到平台团队。以前，prompt 存在于你的代码库中。更改它需要 PR、CI 运行和部署。回滚也需要相同的更改。Managed Agents 改为在服务端保存 prompt。每次 agents.update 都会产生一个新的不可变版本，session 通过 ID 选择使用哪个版本。你仍然审查和批准 prompt 更改，但你批准的是配置中的版本号而不是代码 diff，你的运行服务无需重新构建即可获取更改。如果出现问题，只需将调用者指回旧版本即可回滚。

在本 cookbook 中，我们将创建一个支持工单分诊 Agent，更新系统提示，并在性能下降时回滚。

完成后你将： - 创建了一个 agent 并看到它返回为 version: 1 - 在标记测试集上对特定版本评分 - 发布了 v2 prompt 并看到版本号变化 - 在没有部署的情况下将回退的 Agent 回滚到 v1

前提条件

必需： - Python 3.11+ - anthropic>=0.91.0 - Anthropic API 密钥 - 本目录中的 example_data/prompt_versioning_and_rollback/support_tickets.jsonl fixture

环境配置

%%capture
%pip install -q "anthropic>=0.91.0" python-dotenv
import json
import os
import time
from collections import defaultdict
from pathlib import Path

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()
if not os.getenv("ANTHROPIC_API_KEY"):
    raise RuntimeError("Set ANTHROPIC_API_KEY in your environment or .env file")

MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-4-6")
client = Anthropic()

创建 agent（版本 1）

我们将首先创建一个 Environment（agent 运行所在的容器模板）和 agent 本身。系统提示很短：将每个工单分类到团队和优先级，并仅以 JSON 响应。

env = client.beta.environments.create(name="ticket-triage-env")
ENV_ID = env.id

V1_SYSTEM = """You are a support-ticket triage agent for a usage-billed API product.
Read the ticket and respond with ONLY a single line of raw JSON (no code fences, no prose):
{"team": "<billing|auth|api-platform|dashboard>", "priority": "<P1|P2|P3>"}
Route based on the customer's actual problem, not surface keywords."""

agent = client.beta.agents.create(name="ticket-triage", model=MODEL, system=V1_SYSTEM)
AGENT_ID = agent.id
print(f"env={ENV_ID}  agent={AGENT_ID} v{agent.version}")
env=env_017FKqTdU6dFgEe6zbbVQvE5  agent=agent_011CZo6t9Cr6Eg5TeFFPWPpi v1

agents.create 在我们没有要求的情况下给我们返回了 version: 1。我们之后做的每次更新都会产生一个具有相同 id 的新版本，那个数字就是我们将用来固定 session 和回滚的。

加载标记的测试集

下面我们有二十个工单，每个团队五个，正确的路由已经标记。这些是我们将用来评分的基准真相。

fixture = Path("example_data/prompt_versioning_and_rollback/support_tickets.jsonl")
tickets = [json.loads(line) for line in fixture.read_text().splitlines()]
teams = sorted({t["team"] for t in tickets})
print(f"{len(tickets)} tickets across {len(teams)} teams: {teams}")
20 tickets across 4 teams: ['api-platform', 'auth', 'billing', 'dashboard']

对版本 1 评分

让我们通过 v1 运行每个工单，看看它的表现。下面的 triage 辅助函数打开一个固定到特定版本的 session，发送一个工单，轮询 events.list 直到 session 变为空闲，并从 agent.message 事件中解析 JSON 判定。

本 cookbook 中重要的部分是 sessions.create 上的 agent= 参数。传递一个纯字符串（agent=AGENT_ID）会给你最新的版本。传递 {"type": "agent", "id": ..., "version": ...} 固定到一个确切的版本，这正是我们进行受控比较所需要的。

def triage(version: int, ticket: dict) -> dict:
    """Run one ticket through a pinned agent version and return its verdict."""
    session = client.beta.sessions.create(
        agent={"type": "agent", "id": AGENT_ID, "version": version},
        environment_id=ENV_ID,
    )
    try:
        prompt = "Subject: " + ticket["subject"] + "\n\n" + ticket["body"]
        client.beta.sessions.events.send(
            session.id,
            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
        )
        deadline = time.time() + 60
        while time.time() < deadline:
            events = client.beta.sessions.events.list(session.id).data
            if events and events[-1].type == "session.status_idle":
                break
            time.sleep(1)
        else:
            raise TimeoutError(f"session {session.id} did not idle within 60s")
        agent_events = [e for e in events if e.type == "agent.message"]
        reply = "".join(b.text for e in agent_events for b in e.content)
        return json.loads(reply)
    finally:
        try:
            client.beta.sessions.archive(session.id)
        except Exception:  # noqa: S110
            pass


def score(version: int) -> dict:
    """Evaluate all tickets against the given version and return per-team accuracy."""
    hits = defaultdict(lambda: [0, 0])
    for t in tickets:
        pred = triage(version, t)
        hits[t["team"]][1] += 1
        if pred.get("team") == t["team"]:
            hits[t["team"]][0] += 1
    return dict(hits)


v1_scores = score(version=1)
print("v1 results:")
for team, (correct, total) in sorted(v1_scores.items()):
    print(f"  {team:14s} {correct}/{total}")
v1 results:
  api-platform   5/5
  auth           5/5
  billing        4/5
  dashboard      5/5

模型表现很好，几乎所有的路由都是正确的。

发布版本 2

现在 PM 发布了她的更改：一条路由规则告诉 Agent，任何关于 API 使用或速率限制的内容都属于平台团队。agents.update 将 agent 更新为新的系统提示。

V2_SYSTEM = V1_SYSTEM + (
    "\n\nROUTING RULE: If the ticket text mentions API usage, rate limits, quotas, "
    "or request volume, route to api-platform. Apply this rule before any other "
    "consideration; do not second-guess it based on the rest of the ticket."
)
agent = client.beta.agents.update(AGENT_ID, version=agent.version, system=V2_SYSTEM)
print(f"agent {AGENT_ID} now at v{agent.version}")
agent agent_011CZo6t9Cr6Eg5TeFFPWPpi now at v2

代码审查去哪了？

我们刚用一个 API 调用更改了 agent。在本 notebook 中，这完全没有审查，这对于演示来说可以，但不是你运行生产环境的方式。

agents.update 没有内置的审批工作流。工作空间中的任何密钥都可以调用它。如果调用者传递的是简单的 agent ID 而不是固定的版本，他们将在下一个 session 中开始使用新 prompt。这就是不需要部署的代价，与你使用功能标志或任何其他通过 API 而不是通过代码管理的配置所做的权衡相同。

将审查步骤放回的模式：生产调用者始终固定到显式版本，而那个固定的数字是受变更控制的东西。任何人都可以创建 v2、v3、v10；这些版本放在服务器上没有流量。提升意味着更新告诉生产调用者传递哪个版本的任何配置，而该更新通过你的正常审查流程。此流程使创建版本变得廉价并保持你的 SDLC 完整。它仍然是一个胜利，因为你不需要更新生产代码，只需更新配置值，更改就会被所有运行器获取。

对版本 2 评分

相同的评估，固定到 v2。新规则很宽泛，在一个按使用计费的 API 产品上，计费工单也谈论 API 使用。

v2_scores = score(version=agent.version)

for team in sorted(v1_scores):
    c1, n1 = v1_scores[team]
    c2, n2 = v2_scores[team]
    flag = "  <-- regressed" if c2 < c1 else ""
    print(team)
    print(f"  v1: {c1}/{n1}")
    print(f"  v2: {c2}/{n2}{flag}")
api-platform
  v1: 5/5
  v2: 5/5
auth
  v1: 5/5
  v2: 5/5
billing
  v1: 4/5
  v2: 2/5  <-- regressed
dashboard
  v1: 5/5
  v2: 5/5

回滚

计费回退了。版本 1 仍然在服务器上。回滚不是部署；调用者只需回到传递 version: 1。

billing = [t for t in tickets if t["team"] == "billing"]
rerun = [triage(version=1, ticket=t).get("team") for t in billing]
print(f"billing tickets via version=1: {rerun.count('billing')}/{len(billing)} correct")
billing tickets via version=1: 4/5 correct

由于版本存在于服务端，如果 PM 想继续迭代 v2 而生产环境保持在 v1 上，v2 仍然在那里，或者将一小部分流量路由到它作为金丝雀。当修复准备好时，她可以创建 v3 并运行相同的流程来提升它。

清理

client.beta.agents.archive(AGENT_ID)
client.beta.environments.archive(ENV_ID)
print("archived")
archived

回顾

这里的机制很小（创建、更新、固定、重新固定），但 prompt 成为版本化的服务端资源使你能够独立于应用程序代码评估和提升 prompt，可以通过上面展示的 update 运行或通过更改配置中的版本。

有一些值得带入你的工作流程的事项：

  • 让生产调用者固定到显式版本，而不是简单的 agent ID。新版本在你提升之前保持不可见。
  • 将固定的版本号视为更改 prompt 的门控。创建版本是探索性的；更新生产固定版本才是需要通过审查的。
  • 对于更高风险的 agent，在完全提升之前，将一部分流量路由到新版本并进行比较（如本 notebook 所做）。你可以有效地将此版本控制用作功能标志。

后续步骤

  • client.beta.agents.versions.list(AGENT_ID) 返回 agent 的每个版本。
  • managed_agents/ 中的其他 notebook 涵盖了 session、自定义工具和端到端模式。
Tip

中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。