托管代理教程:提示版本管理与回滚
技术背景
在传统开发中,提示词通常硬编码在代码库中,修改需要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选择要使用的版本。你仍然需要审查和批准提示更改,但你批准的是配置中的版本号,而不是代码差异,你的运行服务无需重新构建即可获取更改。如果出现问题,只需将调用者指回旧版本即可回滚。
在本教程中,我们将创建一个支持工单分类代理,更新系统提示,并在性能下降时回滚。
最终你将:
- 创建一个代理,并看到它返回为
version: 1 - 针对标记的测试集对特定版本进行评分
- 发布v2提示,并观察版本号变化
- 无需部署即可将性能下降的代理回滚到v1
先决条件
必需:
- Python 3.11+
anthropic>=0.91.0- Anthropic API密钥
- 此目录中的
example_data/prompt_versioning_and_rollback/support_tickets.jsonl测试数据文件
设置
%%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 运行,也可以通过更改配置中的版本来实现。
有几件事值得带入你的工作流程:
- 让生产环境的调用者固定到明确的版本,而不是裸的代理 ID。新版本在推广之前保持不可见。
- 将固定的版本号视为更改提示词的关卡。创建版本是探索性的;更新生产环境的固定版本才是需要经过审查的。
- 对于高风险代理,在完全推广之前,将一部分流量路由到新版本并进行比较(如本笔记本所做)。你可以有效地将此版本控制用作功能开关。
后续步骤
client.beta.agents.versions.list(AGENT_ID)返回代理的所有版本。managed_agents/中的其他笔记本涵盖了会话、自定义工具和端到端模式。
常见问题
常见错误: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工程实践。
- --
内容来源与验证
- 内容等级:PREMIUM_A
- 质量等级:A
- 难度:中级
- 验证等级:V1
- 原文作者:markn-ant
- 原文日期:2026-04-07
- 来源许可:MIT
- 来源版本:
01c17898654d05f084715c54ed4caa6a9a41672e - 查看原文
站内还有更多完整实战教程
加入会员,解锁全部教程、完整代码与持续更新内容。
加入会员