托管代理教程:提示版本管理与回滚

本教程面向有编程基础的开发者,深入讲解托管代理中提示版本管理与回滚的工程实践。通过一个工单分类代理的案例,演示如何创建代理、固定版本、评分、发布新版本并在性能下降时回滚。文章强调版本控制作为服务端资源,使提示词变更无需重新部署即可生效,同时保持代码审查流程。核心机制包括`agents.update`产生不可变版本、通过`version`参数固定版本、以及将版本号作为配置项进行受控推广。教程还提供了完整的代码示例、评分方法和生产环境优化建议,帮助开发者掌握可落地的AI工程实践。

技术背景

在传统开发中,提示词通常硬编码在代码库中,修改需要PR、CI和部署,回滚同样繁琐。托管代理将提示词存储在服务端,每次更新产生新版本,会话通过ID选择版本,从而将提示词变更与代码部署解耦。这种模式解决了提示词迭代频繁、部署成本高的问题,适用于需要快速调整LLM行为(如路由、分类、生成)的场景。主要限制是agents.update无内置审批流程,需通过固定版本号实现变更控制。

核心原理

核心原理是版本化服务端资源:agents.create返回version: 1,每次agents.update生成新版本,旧版本不可变且保留。会话创建时可通过agent参数指定版本:传入字符串AGENT_ID使用最新版本,传入{"type":"agent","id":...,"version":...}固定到精确版本。评分通过triage函数实现:创建会话、发送消息、轮询事件直到空闲、解析agent.message事件中的JSON响应。数据流为:测试集→triage→评分→对比版本性能。关键设计决策是生产调用者固定版本,将版本号作为配置项,通过配置变更实现推广和回滚,保持SDLC完整性。

架构分析

系统由代理(Agent)、环境(Environment)、会话(Session)和版本(Version)组成。代理是提示词和模型的封装,环境是运行容器模板,会话是单次交互实例。版本是代理的不可变快照,通过agents.update创建。架构思想是分离提示词生命周期与代码部署:提示词作为服务端资源,通过API管理,版本号作为配置项控制流量。生产环境调用者固定版本,新版本默认无流量,通过更新配置实现金丝雀发布和回滚。这种设计将提示词变更从代码审查中解耦,但保留了对版本号的审查。

完整实战教程

提示版本管理与回滚

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

在本教程中,我们将创建一个支持工单分类代理,更新系统提示,并在性能下降时回滚。

最终你将:

先决条件

必需:

设置

%%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()

创建智能体(版本 1)

我们首先创建一个环境(智能体运行的容器模板)和智能体本身。系统提示词很短:将每个工单分类到团队和优先级,并且仅以 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}")

agents.create 在我们没有要求的情况下返回了 version: 1。之后我们进行的每次更新都会产生一个具有相同 id 的新版本,而这个数字将用于固定会话和回滚。

加载带标签的测试集

下面我们有二十张工单,每个团队五张,并已标注了正确的路由。这些是我们评分所依据的基准真相。

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}")

版本 1 评分

让我们用 v1 跑一遍每个工单,看看它的表现如何。下面的 triage 辅助函数会打开一个会话,固定到特定版本,发送一个工单,轮询 events.list 直到会话空闲,并从 agent.message 事件中解析出 JSON 判定。

对于本教程而言,关键部分是 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}")

模型表现良好,几乎所有路由都正确。

发布版本 2

现在,产品经理发布了她的变更:一条路由规则,告诉智能体任何关于 API 使用或速率限制的内容都属于平台团队。agents.update 将智能体更新到新的系统提示。

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}")

代码审查去哪了?

我们刚刚通过一个 API 调用就改变了 agent。在这个 notebook 中,这一操作没有经过任何审查,这对于演示来说没问题,但并不是你在生产环境中运行的方式。

agents.update 没有内置的审批工作流。工作区中的任何密钥都可以调用它。如果调用者传递的是裸的 agent ID 而不是固定版本,他们将在下一次会话中就开始使用新的提示词。这是不需要部署的代价,与你通过 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}")

回滚

计费出现了回退。版本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")

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

清理

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

回顾

这里的机制很小(创建、更新、固定、重新固定),但提示词成为版本化的服务端资源后,你可以独立于应用程序代码来评估和推广提示词,既可以通过如上所示的 update 运行,也可以通过更改配置中的版本来实现。

有几件事值得带入你的工作流程:

后续步骤

常见问题

常见错误:1. 未固定版本,导致生产环境意外使用新版本——解决:始终使用{"type":"agent","id":...,"version":...}。2. 会话未空闲导致超时——解决:增加轮询间隔或超时时间。3. JSON解析失败——检查响应格式,确保提示词要求纯JSON。4. 版本号错误——使用agents.versions.list确认版本。排查方法:检查会话事件日志,确认agent.message事件内容;使用score函数对比版本性能。生产优化:使用配置中心管理版本号,实现动态切换;对高风险代理进行金丝雀流量分配;监控版本性能指标,自动回滚。

企业应用扩展

可扩展方向:1. 使用agents.versions.list实现版本历史管理。2. 结合自定义工具,实现更复杂的代理行为。3. 将版本控制与CI/CD集成,自动化评估和推广流程。4. 实现基于流量的A/B测试,比较不同版本效果。5. 添加监控和告警,当版本性能下降时自动回滚。6. 探索会话管理最佳实践,如会话复用和清理。

总结

本教程面向有编程基础的开发者,深入讲解托管代理中提示版本管理与回滚的工程实践。通过一个工单分类代理的案例,演示如何创建代理、固定版本、评分、发布新版本并在性能下降时回滚。文章强调版本控制作为服务端资源,使提示词变更无需重新部署即可生效,同时保持代码审查流程。核心机制包括agents.update产生不可变版本、通过version参数固定版本、以及将版本号作为配置项进行受控推广。教程还提供了完整的代码示例、评分方法和生产环境优化建议,帮助开发者掌握可落地的AI工程实践。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程