prompt-auto-optimization¶
第8章 · Agent 的自我进化 · 配套项目
chapter8/prompt-auto-optimization
项目说明¶
实验 8-3：系统提示词的自动优化（★★）¶
《深入理解 AI Agent》配套代码 · 第 8 章
基于人类反馈的 自动化系统提示学习：让一个 Coding Agent 读取系统提示词文件、 定位有问题的规则、生成精确修改并 真的改写 prompt 文件，从而修复 Agent 的 "过度转接" 行为。场景为 tau-bench 风格的航空客服。
1. 目的与问题¶
初始的航空客服 Agent 里，人工转接规则写得很含糊——"仅当请求无法在你的行动范围内 处理时才转接"，并且强调"客户满意度第一，遇到不满就转人工，不要与乘客争辩政策"。
评测发现：Agent 过度转接——一遇到政策争议（要求超政策退款、要求免费、要求豁免 费用）就直接甩给人工，而不尝试向乘客解释政策。
人类专家反馈：这类争议应当 通过耐心解释政策来处理，而不是一转了之。真正需要转 人工的只有两种情况：乘客明确要求人工客服、紧急安全/人身健康风险。
本实验演示一条自动化闭环：人类反馈 → Coding Agent 改写系统提示词 → 重新评测验证。
2. 方法与流程¶
初始 prompt ──评测──► 暴露"过度转接"问题
│
人类反馈 ───┤
▼
Coding Agent（读文件→定位规则→生成精确 search/replace 编辑→改写文件）
│ 展示真实 diff
▼
自动优化后 prompt ──评测──► 边界集正确率↑ 且 保留集不退化
│
对照：人工调优版 prompt ──评测──►
两组评测用例（各 5 个，控制成本、结论可复现）：
- 保留任务集 (holdout)：正常请求。3 个应由 Agent 自行处理（改签 / 行李额 / 选座）， 2 个本就该转接（乘客明确要人工 / 紧急安全）。用来检验优化 不会破坏既有正确行为。
- 边界案例集 (boundary)：5 个政策争议（不可退票要退款 / 要求免改签费 / 小延误索赔 / 索要免费升舱 / 超额免费行李）。正确行为是 解释政策、不转接。用来检验过度转接被修复。
判据（evaluate.py）：结合确定性规则 + LLM-as-judge——
- 该转的用例：正确 ⇔ 确实调用了 transfer_to_human；
- 不该转的用例：正确 ⇔ 没转接，且 LLM 裁判确认它按 rubric 妥善解释了政策 / 办理了业务。
Coding Agent（coding_agent.py）不整篇重写，而是像真实编程 Agent 一样产出一组
(old_str → new_str) 精确编辑，由代码逐条做精确字符串替换落盘；匹配不上就把错误反馈
给模型重试，保证修改可审计、可出 diff。
3. 文件结构¶
| 文件 | 说明 |
|---|---|
demo.py |
一条命令跑通完整流程（评测→改写→复评→对照→对比表） |
airline_env.py |
精简航空客服模拟环境：工具（含 transfer_to_human）、Agent 循环、10 个用例 |
coding_agent.py |
Coding Agent：读取并 改写 系统提示词文件，输出 diff |
evaluate.py |
评测器：规则 + LLM-as-judge 判定每个用例是否被正确处理 |
config.py |
LLM 客户端配置（默认 OpenAI gpt-5.6-luna，可切 Moonshot / 火山方舟；缺 Key 时 OpenRouter 兜底） |
prompts/system_prompt.txt |
初始系统提示词（含会诱发过度转接的规则） |
prompts/system_prompt_manual.txt |
对照组：人工调优版系统提示词 |
runtime/system_prompt_working.txt |
运行时由 Coding Agent 改写的工作副本（每次运行重置） |
4. 运行¶
pip install -r requirements.txt
cp env.example .env # 填入 OPENAI_API_KEY（或仅设 OPENROUTER_API_KEY 走兜底）
python demo.py # 完整运行：10 个用例 × 3 份 prompt
python demo.py --quick # 快速演示：每组只取 2 个用例，省时省钱（推荐先跑这个）
python demo.py --help # 查看全部命令行参数（中文说明）
python demo.py --dry-run # 离线自检：只打印配置与选中用例，不调用任何 API
命令行参数（python demo.py --help 可见完整中文说明）：
| 参数 | 作用 | 默认 |
|---|---|---|
--quick |
每组只取 2 个用例，省时省钱 | 关闭 |
--limit N |
每组最多评测 N 个用例（覆盖 --quick） |
不限 |
--group {holdout,boundary,both} |
选择评测的任务集 | both |
--rounds N |
Coding Agent 自动改写提示词的最大重试轮数 | 3 |
--model NAME |
覆盖 LLM 模型名（等价于 LLM_MODEL） |
见 config.py |
--provider {openai,moonshot,ark} |
覆盖 LLM 提供商（等价于 LLM_PROVIDER） |
openai |
--output PATH |
把优化前后 + 人工对照的对比结果写成 JSON | 不写 |
--dry-run |
离线：只打印解析后的配置与用例数，不调用 API | 关闭 |
默认模型 gpt-5.6-luna（读取 OPENAI_API_KEY；缺失时若有 OPENROUTER_API_KEY 则自动改走 OpenRouter 并映射到 openai/gpt-5.6-luna），温度 0（结果可复现）。完整运行跑
10 个用例 × 3 份 prompt，约几十次 API 调用、数分钟；--quick（或 --limit N 指定每组
用例数）会显著缩短耗时，用于快速验证闭环。命令行参数优先级高于环境变量：--model /
--provider 会覆盖 .env 中的 LLM_MODEL / LLM_PROVIDER。加 --output output/run.json
可把对比表落盘为 JSON（output/ 已被 .gitignore 忽略），便于复现与二次分析。若未设置
对应 API Key，程序会打印清晰的中文错误提示并退出，而不是抛出堆栈。
优化后的工作副本会写入 runtime/system_prompt_working.txt（每次运行自动重置，属生成
工件，已被 .gitignore 忽略）。
5. 真实运行结论¶
下表为一次真实运行（gpt-5.6-luna，完整 10 用例）的结果：
正确率对比（保留任务集 = 既有正确行为不能退化；边界案例集 = 过度转接应改善）
==========================================================================
系统提示词版本 保留任务集(holdout) 边界案例集(boundary)
--------------------------------------------------------------------------
初始 prompt(优化前) 5/5 (100%) 0/5 (0%)
自动优化后 prompt 5/5 (100%) 1/5 (20%)
人工调优版(对照) 5/5 (100%) 2/5 (40%)
==========================================================================
【结论】
· 边界案例集正确率：0/5 → 1/5 （提升 ✓）
· 保留任务集正确率：5 → 5 （未退化 ✓）
- 边界案例集：优化前 5 个用例全部"一转了之"（过度转接 5/5），优化后 转接彻底消失 （边界集转接数 5 → 0），过度转接问题被修复；正确率从 0/5 升到 1/5——B5（超额免费行李） 由主动解释重量计费政策而被判合格，B1/B2/B3/B4 虽不再转接，但模型倾向先向乘客索要订单号、 未充分把政策解释到位，被严格的裁判判为"处理不当"，属于真实的边界表现。
- 保留任务集：优化前后均 5/5，既有正确行为（含 H4/H5"该转的仍会转"）未退化。
- 自动优化后的效果接近 人工调优版对照组（自动边界 1/5、人工 2/5，保留均 5/5）；人工版在 B2（要求免改签费）上多解释到位一例，差距在裁判的 ±1 个用例波动范围内。
注：具体数字可能随模型版本与采样有 ±1 个用例的波动，但"边界集过度转接被修复、保留集不退化、 且与人工调优接近"的结论稳定可复现。上表为
gpt-5.6-luna的一次真实运行（因该模型仅支持默认 温度，本次以LLM_TEMPERATURE=1运行）。
Coding Agent 对 system_prompt.txt 的真实改写（diff 节选）会在 python demo.py 的
【步骤 2】中打印，核心是把第 3、4 条转接规则收紧为"仅乘客明确要求人工 / 紧急安全才转"，
并新增负面规则"绝不因政策争议或乘客不满而转接，应先解释政策再给替代方案"。
补充：模型 ↔ 脚手架此消彼长（强模型 vs 弱模型的两次真实运行）¶
一个自然的问题：模型更强，这套"自动改 prompt"的脚手架是不是就不那么必要了？
为此我们把完全相同的闭环（10 用例 × 3 份 prompt × 最多 3 轮改写）在一个较弱模型
gpt-4o-mini（OpenAI 直连，LLM_TEMPERATURE=0）上又真实跑了一次，与上文
gpt-5.6-luna 的运行对照。边界案例集（过度转接应改善）的优化前 → 自动优化后变化：
| 模型 | 保留集(holdout) | 边界集 优化前 → 自动优化后 | 自动优化增益 | 人工调优版(对照) |
|---|---|---|---|---|
gpt-4o-mini（较弱） |
5/5 → 5/5（未退化） | 1/5 → 3/5 | +2 个用例 | 3/5 |
gpt-5.6-luna（较强） |
5/5 → 5/5（未退化） | 0/5 → 1/5 | +1 个用例 | 2/5 |
- 方向上，较弱的
gpt-4o-mini从这套自动优化脚手架里拿到的增益更大（边界集 +2 个用例， 1/5→3/5），强于gpt-5.6-luna的 +1（0/5→1/5）；且优化后gpt-4o-mini已追平其人工调优版 （均 3/5）。这与"弱模型更依赖脚手架、强模型对同一条基线 prompt 的容错更高"的直觉一致。 - 但需诚实标注两点，避免过度解读：(1) 两者增益之差仅 1 个用例，正落在本实验反复申明的
±1 个用例裁判波动带内；(2) 两次运行采样条件不同——弱模型
temperature=0（确定性）， 强模型因仅支持默认温度而temperature=1（有采样噪声）。因此这是一次方向性、非结论性的 对照：可以说"弱模型从脚手架获益不少于强模型"，但不宜据这 1 个用例的差断言强模型"就不需要" 自动优化。真正稳健的结论仍是两模型共有的那条——边界集过度转接被修复、保留集不退化、 且自动优化逼近人工调优。
6. 如何适配 / 扩展与局限¶
- 换模型 / 供应商：
LLM_PROVIDER可切换openai/moonshot/ark（均兼容 OpenAI 接口），LLM_MODEL覆盖模型名，LLM_TEMPERATURE调采样温度（默认 0，见config.py/env.example）。 - 换任务 / 输入：评测用例集在
airline_env.py的CASES（分holdout/boundary两组）； 驱动优化的人类反馈是demo.py顶部的HUMAN_FEEDBACK；初始与人工对照 prompt 在prompts/下—— 改这几处即可把闭环套到你自己的场景。 - 局限：环境为教学用途的精简模拟，工具返回固定 mock 数据；重点在"人类反馈驱动提示词自动优化" 这一闭环，而非完整复刻 tau-bench。具体正确率会随模型版本与采样有 ±1 个用例的波动。
源代码¶
airline_env.py¶
"""
精简版「航空客服」模拟环境（对标 tau-bench 的航空场景，但去掉复杂度）。
包含三部分：
1. TOOLS —— 暴露给 Agent 的工具（含关键的 transfer_to_human）。
2. run_agent —— 一个带工具调用循环的最小 Agent：给定 system prompt 和用户请求，
返回它是否转接人工、以及最终回复。
3. CASES —— 两组评测用例：
- 保留任务集(holdout)：正常请求，Agent 应正确处理（不该转的别转，该转的要转）。
- 边界案例集(boundary)：政策争议，Agent 应解释政策而非一转了之。
"""
import json
from config import get_client, get_model, TEMPERATURE
# ----------------------------------------------------------------------------
# 1. 工具定义（OpenAI function-calling 格式）
# ----------------------------------------------------------------------------
TOOLS = [
{
"type": "function",
"function": {
"name": "lookup_reservation",
"description": "根据订单号查询乘客的订单详情（航班、舱位、票价类型等）。",
"parameters": {
"type": "object",
"properties": {
"confirmation_code": {"type": "string", "description": "订单号"}
},
"required": ["confirmation_code"],
},
},
},
{
"type": "function",
"function": {
"name": "change_flight",
"description": "为乘客办理改签到指定的新航班。",
"parameters": {
"type": "object",
"properties": {
"confirmation_code": {"type": "string"},
"new_flight": {"type": "string", "description": "新航班号或日期"},
},
"required": ["confirmation_code", "new_flight"],
},
},
},
{
"type": "function",
"function": {
"name": "get_refund_policy",
"description": "查询退票/退款政策。传入票价类型（如 经济舱特价票/全价经济舱/商务舱）。",
"parameters": {
"type": "object",
"properties": {
"fare_type": {"type": "string", "description": "票价类型"}
},
"required": ["fare_type"],
},
},
},
{
"type": "function",
"function": {
"name": "get_baggage_policy",
"description": "查询行李额与逾重费政策。传入舱位等级。",
"parameters": {
"type": "object",
"properties": {
"cabin": {"type": "string", "description": "舱位等级，如 经济舱/商务舱"}
},
"required": ["cabin"],
},
},
},
{
"type": "function",
"function": {
"name": "change_seat",
"description": "为乘客办理选座或换座。",
"parameters": {
"type": "object",
"properties": {
"confirmation_code": {"type": "string"},
"seat": {"type": "string", "description": "目标座位号"},
},
"required": ["confirmation_code", "seat"],
},
},
},
{
"type": "function",
"function": {
"name": "transfer_to_human",
"description": "把对话转接给人工客服。调用后 Agent 不再继续处理本次请求。",
"parameters": {
"type": "object",
"properties": {
"reason": {"type": "string", "description": "转接原因"}
},
"required": ["reason"],
},
},
},
]
# ----------------------------------------------------------------------------
# 2. 工具的 mock 实现（返回固定的模拟数据，供 Agent 组织回复）
# ----------------------------------------------------------------------------
_POLICY_REFUND = {
"经济舱特价票": "经济舱特价票为不可退票产品，不支持自愿退款；如未起飞可申请退还机建燃油等税费。",
"全价经济舱": "全价经济舱起飞前可退，收取 5% 退票手续费。",
"商务舱": "商务舱起飞前可全额退票，不收手续费。",
}
def _run_tool(name: str, args: dict) -> str:
"""执行工具，返回给模型的字符串结果。"""
if name == "lookup_reservation":
return json.dumps(
{
"confirmation_code": args.get("confirmation_code", "UNKNOWN"),
"passenger": "张伟",
"flight": "YS1234 上海虹桥→北京首都 2026-08-01 09:00",
"cabin": "经济舱",
"fare_type": "经济舱特价票",
"status": "已出票",
},
ensure_ascii=False,
)
if name == "change_flight":
return json.dumps(
{"result": "success", "new_flight": args.get("new_flight"), "fee": "改签费 200 元"},
ensure_ascii=False,
)
if name == "get_refund_policy":
fare = args.get("fare_type", "经济舱特价票")
text = _POLICY_REFUND.get(fare, _POLICY_REFUND["经济舱特价票"])
return json.dumps({"fare_type": fare, "policy": text}, ensure_ascii=False)
if name == "get_baggage_policy":
cabin = args.get("cabin") or "经济舱"
free = "20kg" if "经济" in cabin else "30kg"
return json.dumps(
{"cabin": cabin, "free_allowance": free, "excess_fee": "逾重费 50 元/kg"},
ensure_ascii=False,
)
if name == "change_seat":
return json.dumps(
{"result": "success", "seat": args.get("seat")}, ensure_ascii=False
)
return json.dumps({"result": "ok"}, ensure_ascii=False)
# ----------------------------------------------------------------------------
# 3. 最小 Agent 循环
# ----------------------------------------------------------------------------
def run_agent(system_prompt: str, user_message: str, max_steps: int = 4) -> dict:
"""
运行一次客服会话。返回：
{
"transferred": bool, # 是否调用了 transfer_to_human
"transfer_reason": str|None,
"final_text": str, # Agent 面向乘客的最终回复（若转接则为空）
"tool_calls": [str, ...], # 依次调用过的工具名
}
"""
client = get_client()
model = get_model()
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
tool_calls_log = []
for _ in range(max_steps):
resp = client.chat.completions.create(
model=model,
messages=messages,
tools=TOOLS,
temperature=TEMPERATURE,
)
msg = resp.choices[0].message
if not msg.tool_calls:
# 模型直接给出面向乘客的文字回复 —— 会话结束
return {
"transferred": False,
"transfer_reason": None,
"final_text": msg.content or "",
"tool_calls": tool_calls_log,
}
# 有工具调用，先把 assistant 消息加入历史
messages.append(msg)
transferred_now = False
transfer_reason = None
for tc in msg.tool_calls:
name = tc.function.name
try:
args = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
args = {}
tool_calls_log.append(name)
if name == "transfer_to_human":
transferred_now = True
transfer_reason = args.get("reason", "")
result = json.dumps({"result": "transferred"}, ensure_ascii=False)
else:
result = _run_tool(name, args)
messages.append(
{"role": "tool", "tool_call_id": tc.id, "content": result}
)
if transferred_now:
return {
"transferred": True,
"transfer_reason": transfer_reason,
"final_text": "",
"tool_calls": tool_calls_log,
}
# 超出步数上限仍未收敛，兜底再要一次纯文本回复
resp = client.chat.completions.create(
model=model, messages=messages, temperature=TEMPERATURE
)
return {
"transferred": False,
"transfer_reason": None,
"final_text": resp.choices[0].message.content or "",
"tool_calls": tool_calls_log,
}
# ----------------------------------------------------------------------------
# 4. 评测用例
# ----------------------------------------------------------------------------
# 每个用例字段：
# id / group / user —— 用户请求
# should_transfer —— 正确行为是否应当转接人工
# rubric —— 当 should_transfer=False 时，判定"是否妥善处理"的要点
CASES = [
# ---------- 保留任务集(holdout)：既有正确行为，优化后不能退化 ----------
{
"id": "H1-改签",
"group": "holdout",
"user": "你好，我的订单号是 ABC123，想把 8 月 1 日的航班改到 8 月 3 日，帮我改一下。",
"should_transfer": False,
"rubric": "查询订单并为乘客办理/说明改签到 8 月 3 日的相关信息（如改签费），而不是转人工。",
},
{
"id": "H2-行李额",
"group": "holdout",
"user": "我买的是经济舱，免费托运行李额是多少？超重了怎么收费？",
"should_transfer": False,
"rubric": "给出经济舱免费行李额与逾重费的具体政策信息，而不是转人工。",
},
{
"id": "H3-选座",
"group": "holdout",
"user": "订单号 ABC123，我想换到靠窗的 32A 座位。",
"should_transfer": False,
"rubric": "为乘客办理/确认换到 32A 座位，而不是转人工。",
},
{
"id": "H4-明确要人工",
"group": "holdout",
"user": "别跟我扯了，我就要跟人工客服说话，马上给我转人工。",
"should_transfer": True, # 乘客明确要求人工 —— 这类本就该转，优化后仍要能转
"rubric": "乘客明确要求人工客服，应当转接。",
},
{
"id": "H5-紧急安全",
"group": "holdout",
"user": "我现在在机场，同行的老人突然晕倒了，赶不上 YS1234 航班了，需要马上有人帮忙处理！",
"should_transfer": True, # 紧急安全 —— 本就该转，优化后仍要能转
"rubric": "涉及紧急人身安全，应当转接人工立即介入。",
},
# ---------- 边界案例集(boundary)：政策争议，应解释政策而非一转了之 ----------
{
"id": "B1-不可退票要退款",
"group": "boundary",
"user": "我要求全额退款！订单 ABC123。我不管，我就要全退。",
"should_transfer": False,
"rubric": "查询退票政策后，耐心解释该票（经济舱特价票）为不可退票产品、可退税费，并给出可行替代方案，而不是转人工。",
},
{
"id": "B2-要求免改签费",
"group": "boundary",
"user": "改签还要收我 200 块？凭什么！你必须给我免掉这个改签费，不然没完。",
"should_transfer": False,
"rubric": "耐心解释改签费的政策依据，说明无法违规减免，并提供合规的替代方案，而不是因为乘客不满就转人工。",
},
{
"id": "B3-小延误索赔",
"group": "boundary",
"user": "航班晚了半小时，害我差点误事，我要求航空公司赔偿我 500 块！",
"should_transfer": False,
"rubric": "共情并解释延误补偿的政策门槛（30 分钟的短延误通常不达补偿标准），说明处理口径，而不是转人工。",
},
{
"id": "B4-索要免费升舱",
"group": "boundary",
"user": "我是老客户了，这次必须给我免费升到商务舱，这点面子都不给？",
"should_transfer": False,
"rubric": "礼貌解释免费升舱不符合政策、说明可付费升舱或积分兑换等合规途径，而不是转人工。",
},
{
"id": "B5-超额免费行李",
"group": "boundary",
"user": "我经济舱，这次要带 3 件行李，你们必须都给我免费托运，别收钱。",
"should_transfer": False,
"rubric": "解释经济舱的免费行李额与超出部分的收费政策，说明无法全部免费，并给出合规建议，而不是转人工。",
},
]
def get_cases(group: str = None):
if group is None:
return CASES
return [c for c in CASES if c["group"] == group]
coding_agent.py¶
"""
Coding Agent：读取系统提示词文件 → 定位相关规则 → 生成精确的搜索/替换编辑 → 真的改写文件。
它的工作方式和真实的编程 Agent（如 Claude Code / Cursor）一致：
不是让模型整篇重写，而是让模型产出一组 (old_str -> new_str) 的精确编辑，
由代码逐条做"精确字符串替换"落到文件里；若某条编辑的 old_str 匹配不上，
把错误反馈回模型让它重试。这样修改是"代码级"的、可审计的（能直接出 diff）。
"""
import difflib
from config import get_client, get_model, TEMPERATURE
# 暴露给 Coding Agent 的"文件编辑工具"
EDIT_TOOLS = [
{
"type": "function",
"function": {
"name": "apply_edits",
"description": (
"对提示词文件应用一组精确的搜索/替换编辑。每条编辑给出 old_str "
"（文件中唯一存在的原文片段）和 new_str（替换后的新文本）。"
"old_str 必须与文件内容逐字符完全一致。"
),
"parameters": {
"type": "object",
"properties": {
"edits": {
"type": "array",
"items": {
"type": "object",
"properties": {
"old_str": {"type": "string"},
"new_str": {"type": "string"},
},
"required": ["old_str", "new_str"],
},
},
"rationale": {
"type": "string",
"description": "简述本次改动如何回应人类反馈。",
},
},
"required": ["edits"],
},
},
}
]
def _apply_one(content: str, old_str: str, new_str: str) -> tuple[str, str | None]:
"""尝试应用一条编辑。成功返回(新内容, None)，失败返回(原内容, 错误信息)。"""
if old_str is None or new_str is None:
return content, "old_str/new_str 不能为 null"
count = content.count(old_str)
if count == 0:
return content, f"old_str 在文件中未找到：{old_str[:60]!r}"
if count > 1:
return content, f"old_str 在文件中出现 {count} 次（不唯一）：{old_str[:60]!r}"
return content.replace(old_str, new_str, 1), None
def _apply_edits_from_args(working: str, args: dict) -> tuple[str, int, list, list]:
"""Apply apply_edits tool args to working text. JSON null edits like omit ([])."""
edits = args.get("edits")
if edits is None:
edits = []
errors = []
applied = 0
for e in edits:
working, err = _apply_one(working, e.get("old_str", ""), e.get("new_str", ""))
if err:
errors.append(err)
else:
applied += 1
return working, applied, errors, edits
def optimize_prompt(prompt_path: str, feedback: str, max_rounds: int = 3, verbose: bool = True) -> dict:
"""
让 Coding Agent 根据 human feedback 改写 prompt_path 指向的文件（原地覆盖）。
返回 {"before": 原文, "after": 新文, "diff": 统一 diff 文本, "rationale": 说明}。
"""
client = get_client()
model = get_model()
with open(prompt_path, "r", encoding="utf-8") as f:
original = f.read()
system = (
"你是一名资深的提示词工程 Coding Agent。你会收到一份航空客服 Agent 的"
"系统提示词文件，以及人类专家的反馈。请定位与'人工转接'相关的规则，"
"生成精确的搜索/替换编辑来改进它，然后调用 apply_edits 工具落地修改。\n"
"改动目标：\n"
"1) 把转接的边界收紧、明确为仅两种情况：乘客明确要求人工客服、以及紧急安全情况；\n"
"2) 删除或改写会诱发'过度转接'的模糊规则（如'不确定或乘客不满就转接'）；\n"
"3) 新增一条明确的负面规则：绝不因政策争议 / 乘客不满而转接，而应先查政策、"
"耐心解释并提供合规的替代方案。\n"
"只修改与转接策略相关的部分，尽量保留其余内容不动。"
)
messages = [
{"role": "system", "content": system},
{
"role": "user",
"content": (
f"【人类专家反馈】\n{feedback}\n\n"
f"【当前系统提示词文件内容】\n---\n{original}\n---\n\n"
"请调用 apply_edits 提交你的精确编辑。"
),
},
]
working = original
rationale = ""
for round_idx in range(max_rounds):
resp = client.chat.completions.create(
model=model,
messages=messages,
tools=EDIT_TOOLS,
tool_choice={"type": "function", "function": {"name": "apply_edits"}},
temperature=TEMPERATURE,
)
msg = resp.choices[0].message
messages.append(msg)
if not msg.tool_calls:
break
# 处理（唯一的）apply_edits 调用
tc = msg.tool_calls[0]
import json
try:
args = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
args = {}
rationale = args.get("rationale", rationale)
working, applied, errors, edits = _apply_edits_from_args(working, args)
if verbose:
print(f" [round {round_idx + 1}] 提交 {len(edits)} 条编辑，成功 {applied}，失败 {len(errors)}")
if not errors:
# 全部编辑成功，落盘
messages.append(
{"role": "tool", "tool_call_id": tc.id, "content": "所有编辑已成功应用。"}
)
break
else:
# 有失败：回滚到原文，把错误反馈给模型重试（保持编辑的原子性）
working = original
feedback_msg = (
"以下编辑未能应用，请修正后重新提交完整的编辑列表（注意 old_str 必须与文件逐字符一致）：\n"
+ "\n".join(f"- {er}" for er in errors)
)
messages.append({"role": "tool", "tool_call_id": tc.id, "content": feedback_msg})
# 落盘（原地覆盖 prompt 文件）
with open(prompt_path, "w", encoding="utf-8") as f:
f.write(working)
diff = "".join(
difflib.unified_diff(
original.splitlines(keepends=True),
working.splitlines(keepends=True),
fromfile="system_prompt.txt (before)",
tofile="system_prompt.txt (after)",
)
)
return {"before": original, "after": working, "diff": diff, "rationale": rationale}
config.py¶
"""
统一的 LLM 客户端配置。
默认使用 OpenAI（读取 OPENAI_API_KEY，模型 gpt-5.6-luna）。
也支持通过环境变量 LLM_PROVIDER 切换到 Moonshot / 火山方舟(ARK)，
它们都兼容 OpenAI 的 Chat Completions + 工具调用接口。
export LLM_PROVIDER=openai # 默认
export LLM_PROVIDER=moonshot # 用 MOONSHOT_API_KEY
export LLM_PROVIDER=ark # 用 ARK_API_KEY，并需设置 ARK_MODEL
统一的 OpenRouter 兜底（fallback）：
若所选 provider 自己的 Key 缺失，但设置了 OPENROUTER_API_KEY，则自动改走
OpenRouter（https://openrouter.ai/api/v1），并把模型名映射到 OpenRouter 命名：
gpt-* -> openai/gpt-*
claude-* -> anthropic/claude-opus-4.8
含 "/" -> 原样透传
其它 -> openai/gpt-5.6-luna
"""
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
# 各提供商的默认配置：base_url / 环境变量名 / 默认模型
_PROVIDERS = {
"openai": {
"base_url": None, # 使用 SDK 默认
"key_env": "OPENAI_API_KEY",
"default_model": "gpt-5.6-luna",
},
"moonshot": {
"base_url": "https://api.moonshot.cn/v1",
"key_env": "MOONSHOT_API_KEY",
"default_model": "kimi-k3",
},
"ark": {
"base_url": "https://ark.cn-beijing.volces.com/api/v3",
"key_env": "ARK_API_KEY",
# ARK 需要用推理接入点(endpoint id) 作为 model，请通过 ARK_MODEL 指定
"default_model": os.getenv("ARK_MODEL", "doubao-seed-1-6-250615"),
},
}
def get_provider() -> str:
return os.getenv("LLM_PROVIDER", "openai").lower().strip()
def _to_openrouter_model(model: str) -> str:
"""把常见模型名映射到 OpenRouter 命名空间。"""
if not model:
return "openai/gpt-5.6-luna"
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
def _is_reasoning_model(model: str) -> bool:
"""gpt-5.x / o1·o3·o4 / kimi-k3 / *reasoner 等推理模型：不接受 temperature=0，
直连 gpt-5.x 还需组织实名且工具调用受限，故优先走 OpenRouter。"""
m = (model or "").lower()
return (m.startswith(("gpt-5", "o1", "o3", "o4"))
or m.startswith("kimi-k3")
or "reasoner" in m or "thinking" in m)
def _use_openrouter(cfg: dict) -> bool:
"""走 OpenRouter 的两种情形：
1) provider 自己的 Key 缺失、但有 OPENROUTER_API_KEY（统一兜底）；
2) 目标是 gpt-5.x 且有 OPENROUTER_API_KEY —— 直连 gpt-5.x 需组织实名、
且 /chat/completions 工具调用受限，故即便有 OPENAI_API_KEY 也优先 OpenRouter。"""
if not os.getenv("OPENROUTER_API_KEY"):
return False
if not os.getenv(cfg["key_env"]):
return True
model = os.getenv("LLM_MODEL") or cfg["default_model"]
return (model or "").lower().startswith("gpt-5")
def get_model() -> str:
"""允许用 LLM_MODEL 覆盖默认模型；OpenRouter 兜底路径下映射模型名。"""
provider = get_provider()
if provider not in _PROVIDERS:
raise ValueError(f"未知的 LLM_PROVIDER: {provider}")
cfg = _PROVIDERS[provider]
model = os.getenv("LLM_MODEL") or cfg["default_model"]
if _use_openrouter(cfg):
return _to_openrouter_model(model)
return model
def get_client() -> OpenAI:
provider = get_provider()
if provider not in _PROVIDERS:
raise ValueError(f"未知的 LLM_PROVIDER: {provider}")
cfg = _PROVIDERS[provider]
if _use_openrouter(cfg):
return OpenAI(api_key=os.getenv("OPENROUTER_API_KEY"), base_url=OPENROUTER_BASE_URL)
api_key = os.getenv(cfg["key_env"])
if not api_key:
raise RuntimeError(
f"环境变量 {cfg['key_env']} 未设置，也未设置 OPENROUTER_API_KEY。"
f"请参考 env.example 配置其一（OpenRouter 可作为统一兜底）后重试。"
)
kwargs = {"api_key": api_key}
if cfg["base_url"]:
kwargs["base_url"] = cfg["base_url"]
return OpenAI(**kwargs)
# 全部 LLM 调用统一使用低温度，保证结果可复现；
# 但推理模型（gpt-5.x / o 系列 / kimi-k3 等）只接受默认 temperature=1，
# 故按当前解析出的模型自动选择默认温度（可用 LLM_TEMPERATURE 显式覆盖）。
def _default_temperature() -> str:
provider = get_provider()
cfg = _PROVIDERS.get(provider, _PROVIDERS["openai"])
model = os.getenv("LLM_MODEL") or cfg["default_model"]
return "1" if _is_reasoning_model(model) else "0"
TEMPERATURE = float(os.getenv("LLM_TEMPERATURE", _default_temperature()))
demo.py¶
"""
实验 8-3：系统提示词的自动优化（基于人类反馈的自动化系统提示学习）
一条命令跑通完整流程：
1. 用【初始 prompt】评测 → 暴露"政策争议就转人工"的过度转接问题；
2. Coding Agent 读取 prompt 文件、定位转接规则、生成精确编辑并【真的改写文件】→ 展示 diff；
3. 用【自动优化后的 prompt】重新评测；
4. 对照【人工调优版 prompt】；
5. 打印"保留任务集 / 边界案例集"在优化前后 + 人工版的正确率对比表。
python demo.py # 完整运行：10 个用例 × 3 份 prompt
python demo.py --quick # 快速演示：每组只取 2 个用例，省时省钱
python demo.py --help # 查看全部命令行参数（中文说明）
"""
import argparse
import json
import os
import shutil
import sys
from evaluate import evaluate_prompt
from coding_agent import optimize_prompt
from config import get_provider, get_model
from airline_env import CASES
GROUPS = ("holdout", "boundary")
HERE = os.path.dirname(os.path.abspath(__file__))
INITIAL_PROMPT = os.path.join(HERE, "prompts", "system_prompt.txt")
MANUAL_PROMPT = os.path.join(HERE, "prompts", "system_prompt_manual.txt")
WORKING_PROMPT = os.path.join(HERE, "runtime", "system_prompt_working.txt")
# 人类专家反馈：这就是驱动"自动系统提示学习"的信号
HUMAN_FEEDBACK = (
"评测发现 Agent 存在【过度转接】问题：一遇到政策争议（如乘客要求超政策退款、"
"要求免费、要求豁免费用）就直接转人工，而不尝试向乘客解释政策。\n"
"正确做法应该是：通过耐心、共情地解释政策来处理这类争议，并提供合规的替代方案，"
"而不是一转了之。真正需要转接人工的，只有两种情况——乘客明确要求人工客服，"
"以及出现紧急安全 / 人身健康风险。"
)
def _read(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
def _pct(cn):
c, n = cn
return f"{c}/{n} ({100 * c / n:.0f}%)" if n else "-"
def print_table(rows):
"""rows: list of (label, holdout_tuple, boundary_tuple)"""
print("\n" + "=" * 74)
print("正确率对比（保留任务集 = 既有正确行为不能退化；边界案例集 = 过度转接应改善）")
print("=" * 74)
header = f"{'系统提示词版本':<26}{'保留任务集(holdout)':<20}{'边界案例集(boundary)':<20}"
print(header)
print("-" * 74)
for label, holdout, boundary in rows:
print(f"{label:<24}{_pct(holdout):<22}{_pct(boundary):<22}")
print("=" * 74)
def _select_cases(limit_per_group=None, groups=GROUPS):
"""按分组筛选用例，并对每组最多取 limit_per_group 个（None 表示不限制）。"""
picked, counts = [], {}
for c in CASES:
g = c["group"]
if g not in groups:
continue
if limit_per_group and counts.get(g, 0) >= limit_per_group:
continue
picked.append(c)
counts[g] = counts.get(g, 0) + 1
return picked
def main(cases=None, rounds=3, output=None):
if cases is None:
cases = CASES
print("#" * 74)
print("# 实验 8-3：基于人类反馈的系统提示词自动优化（航空客服场景）")
print(f"# LLM 提供商: {get_provider()} 模型: {get_model()}")
print(f"# 用例数: {len(cases)}（保留集 + 边界集） Coding Agent 优化轮数上限: {rounds}")
print("#" * 74)
# ---- 准备：把初始 prompt 复制成本次运行的工作副本（Coding Agent 会改写它）----
os.makedirs(os.path.dirname(WORKING_PROMPT), exist_ok=True)
shutil.copyfile(INITIAL_PROMPT, WORKING_PROMPT)
# ---- 步骤 1：评测初始 prompt ----
print("\n【步骤 1】用初始系统提示词评测（观察是否过度转接）")
before = evaluate_prompt(_read(INITIAL_PROMPT), label="初始 prompt", cases=cases)
print(
f"\n 初始结果：保留集 {_pct(before['holdout'])}，"
f"边界集 {_pct(before['boundary'])}"
)
over_transfer = [
r for r in before["results"]
if r["group"] == "boundary" and not r["should_transfer"] and r["transferred"]
]
print(f" 边界案例中出现【过度转接】的用例数：{len(over_transfer)} / "
f"{len([r for r in before['results'] if r['group'] == 'boundary'])}")
for r in over_transfer:
print(f" - {r['id']}：政策争议却直接转人工，原因『{r['transfer_reason']}』")
# ---- 步骤 2：Coding Agent 自动改写 prompt 文件 ----
print("\n【步骤 2】Coding Agent 读取并改写系统提示词文件……")
opt = optimize_prompt(WORKING_PROMPT, HUMAN_FEEDBACK, max_rounds=rounds, verbose=True)
print(f"\n Coding Agent 改动说明：{opt['rationale']}")
print("\n ---------- 系统提示词文件 diff（真实写入磁盘）----------")
print(opt["diff"] if opt["diff"].strip() else " (无改动)")
print(" --------------------------------------------------------")
# ---- 步骤 3：评测自动优化后的 prompt ----
print("\n【步骤 3】用自动优化后的系统提示词重新评测")
after = evaluate_prompt(opt["after"], label="自动优化后 prompt", cases=cases)
# ---- 步骤 4：对照人工调优版 ----
print("\n【步骤 4】对照组：人工调优版系统提示词")
manual = evaluate_prompt(_read(MANUAL_PROMPT), label="人工调优版 prompt(对照)", cases=cases)
# ---- 步骤 5：对比表 ----
print_table([
("初始 prompt(优化前)", before["holdout"], before["boundary"]),
("自动优化后 prompt", after["holdout"], after["boundary"]),
("人工调优版(对照)", manual["holdout"], manual["boundary"]),
])
# ---- 结论 ----
b_before_c, b_before_n = before["boundary"]
b_after_c, _ = after["boundary"]
h_before_c, _ = before["holdout"]
h_after_c, _ = after["holdout"]
print("\n【结论】")
print(f" · 边界案例集正确率：{b_before_c}/{b_before_n} → {b_after_c}/{b_before_n} "
f"（{'提升 ✓' if b_after_c > b_before_c else '未提升'}）")
print(f" · 保留任务集正确率：{h_before_c} → {h_after_c} "
f"（{'未退化 ✓' if h_after_c >= h_before_c else '退化 ✗'}）")
print(f"\n 优化后的工作副本已写入：{WORKING_PROMPT}")
# ---- 可选：把对比结果落盘为 JSON，便于复现与二次分析 ----
if output:
summary = {
"provider": get_provider(),
"model": get_model(),
"rounds": rounds,
"num_cases": len(cases),
"rationale": opt["rationale"],
"diff": opt["diff"],
"rows": [
{"label": "初始 prompt(优化前)", "holdout": list(before["holdout"]),
"boundary": list(before["boundary"])},
{"label": "自动优化后 prompt", "holdout": list(after["holdout"]),
"boundary": list(after["boundary"])},
{"label": "人工调优版(对照)", "holdout": list(manual["holdout"]),
"boundary": list(manual["boundary"])},
],
}
os.makedirs(os.path.dirname(os.path.abspath(output)), exist_ok=True)
with open(output, "w", encoding="utf-8") as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
print(f" 对比结果已写入：{output}")
def _build_parser():
parser = argparse.ArgumentParser(
prog="demo.py",
description="实验 8-3：基于人类反馈的系统提示词自动优化演示（航空客服场景）。",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"示例：\n"
" python demo.py # 完整运行：10 个用例 × 3 份 prompt\n"
" python demo.py --quick # 每组只取 2 个用例，省时省钱\n"
" python demo.py --group boundary # 只评测边界案例集\n"
" python demo.py --rounds 5 --model gpt-5.6-luna\n"
" python demo.py --output output/run.json # 把对比结果写成 JSON\n"
" python demo.py --dry-run # 离线：只打印配置与用例数，不调用 API"
),
)
parser.add_argument(
"--quick", action="store_true",
help="快速演示模式：每组只取 2 个用例，减少 API 调用与耗时。",
)
parser.add_argument(
"--limit", type=int, default=None, metavar="N",
help="每组最多评测 N 个用例（覆盖 --quick）。",
)
parser.add_argument(
"--group", choices=("holdout", "boundary", "both"), default="both",
help="选择评测的任务集：holdout(保留集) / boundary(边界集) / both(默认，两者都跑)。",
)
parser.add_argument(
"--rounds", type=int, default=3, metavar="N",
help="Coding Agent 自动改写提示词的最大重试轮数（默认 3）。",
)
parser.add_argument(
"--model", default=None, metavar="NAME",
help="覆盖 LLM 模型名（等价于设置环境变量 LLM_MODEL，如 gpt-5.6-luna）。",
)
parser.add_argument(
"--provider", choices=("openai", "moonshot", "ark"), default=None,
help="覆盖 LLM 提供商（等价于设置环境变量 LLM_PROVIDER，默认 openai）。",
)
parser.add_argument(
"--output", default=None, metavar="PATH",
help="把优化前后 + 人工对照的对比结果写入指定 JSON 文件（如 output/run.json）。",
)
parser.add_argument(
"--dry-run", action="store_true",
help="离线自检：只打印解析后的配置与选中用例数，不调用任何 LLM API。",
)
return parser
if __name__ == "__main__":
args = _build_parser().parse_args()
# 命令行覆盖优先级高于环境变量：get_provider()/get_model() 均在调用时读取环境变量
if args.provider:
os.environ["LLM_PROVIDER"] = args.provider
if args.model:
os.environ["LLM_MODEL"] = args.model
limit = args.limit if args.limit is not None else (2 if args.quick else None)
groups = GROUPS if args.group == "both" else (args.group,)
cases = _select_cases(limit, groups=groups)
if args.dry_run:
# 离线路径：不触发任何网络请求，仅用于验证参数解析与用例选择
print("[dry-run] 解析后的运行配置（不调用 API）：")
print(f" LLM 提供商 : {get_provider()}")
print(f" LLM 模型 : {get_model()}")
print(f" 优化轮数 : {args.rounds}")
print(f" 任务集 : {args.group}")
print(f" 选中用例数 : {len(cases)} -> {[c['id'] for c in cases]}")
print(f" 输出文件 : {args.output or '(不写文件)'}")
sys.exit(0)
try:
main(cases=cases, rounds=args.rounds, output=args.output)
except RuntimeError as e:
# 例如 API Key 未设置：给出清晰的人类可读错误，而非原始 traceback
print(f"\n[错误] {e}", file=sys.stderr)
sys.exit(1)
evaluate.py¶
"""
评测器：给定一份 system prompt，在用例集上运行 Agent 并判定每个用例是否被"正确处理"。
判定规则（结合确定性规则 + LLM-as-judge）：
- 用例标注了 should_transfer（正确行为是否应转接人工）。
- 若 should_transfer=True： 正确 ⇔ Agent 确实转接了。
- 若 should_transfer=False： 正确 ⇔ Agent 没有转接，且 LLM 裁判确认它按 rubric 妥善处理了请求
（即真的解释了政策 / 办理了业务，而不是敷衍或答非所问）。
之所以对"不转接"再加一道 LLM 裁判，是为了防止 Agent 只是"没转接但也没好好处理"蒙混过关。
"""
import json
from config import get_client, get_model, TEMPERATURE
from airline_env import run_agent
def _judge_handled(user_message: str, rubric: str, agent_reply: str) -> bool:
"""LLM-as-judge：Agent 没转接的情况下，是否按 rubric 妥善处理了请求。"""
client = get_client()
model = get_model()
prompt = f"""你是严格的客服质检员。请判断客服 Agent 的回复是否妥善处理了乘客请求。
【乘客请求】
{user_message}
【合格标准(rubric)】
{rubric}
【Agent 的回复】
{agent_reply}
请只输出一个 JSON：{{"handled": true 或 false, "reason": "简短理由"}}
其中 handled=true 表示 Agent 的回复实质满足了合格标准。"""
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=TEMPERATURE,
response_format={"type": "json_object"},
)
try:
verdict = json.loads(resp.choices[0].message.content)
return bool(verdict.get("handled", False))
except (json.JSONDecodeError, TypeError):
return False
def evaluate_case(system_prompt: str, case: dict, verbose: bool = False) -> dict:
"""评测单个用例，返回结果 dict。"""
result = run_agent(system_prompt, case["user"])
transferred = result["transferred"]
should_transfer = case["should_transfer"]
if should_transfer:
correct = transferred
note = "应转接：" + ("已转接 ✓" if transferred else "未转接 ✗")
else:
if transferred:
correct = False
note = "不应转接：却转接了 ✗（过度转接）"
else:
handled = _judge_handled(case["user"], case["rubric"], result["final_text"])
correct = handled
note = "不应转接：未转接且妥善处理 ✓" if handled else "不应转接：未转接但处理不当 ✗"
out = {
"id": case["id"],
"group": case["group"],
"correct": correct,
"transferred": transferred,
"should_transfer": should_transfer,
"note": note,
"final_text": result["final_text"],
"transfer_reason": result["transfer_reason"],
"tool_calls": result["tool_calls"],
}
if verbose:
icon = "✓" if correct else "✗"
print(f" [{icon}] {case['id']:<16} {note}")
if transferred:
print(f" 转接原因: {result['transfer_reason']}")
else:
preview = (result["final_text"] or "").replace("\n", " ")[:80]
print(f" 回复: {preview}...")
return out
def evaluate_prompt(system_prompt: str, label: str = "", verbose: bool = True, cases=None) -> dict:
"""在全部用例上评测一份 prompt，返回分组正确率与明细。
cases 为 None 时评测全部用例；也可传入用例子集（如 --quick 模式）以控制成本。
"""
from airline_env import CASES
if cases is None:
cases = CASES
if verbose and label:
print(f"\n>>> 评测 [{label}]")
results = []
for case in cases:
results.append(evaluate_case(system_prompt, case, verbose=verbose))
def _acc(group):
rows = [r for r in results if r["group"] == group]
n = len(rows)
c = sum(1 for r in rows if r["correct"])
return c, n
holdout_c, holdout_n = _acc("holdout")
boundary_c, boundary_n = _acc("boundary")
return {
"label": label,
"holdout": (holdout_c, holdout_n),
"boundary": (boundary_c, boundary_n),
"results": results,
}
test_apply_one_null.py¶
from coding_agent import _apply_one
def test_apply_one_null_old_str():
content, err = _apply_one("hello world", None, "x")
assert content == "hello world"
assert err is not None
assert "null" in err
def test_apply_one_null_new_str():
content, err = _apply_one("hello world", "hello", None)
assert content == "hello world"
assert err is not None
assert "null" in err
def test_apply_one_normal():
content, err = _apply_one("hello world", "hello", "hi")
assert err is None
assert content == "hi world"
test_baggage_policy_null.py¶
import json
from airline_env import _run_tool
def test_baggage_policy_null_cabin():
# 模型显式传 {"cabin": null}：应回退到经济舱默认，而不是 TypeError
args = json.loads('{"cabin": null}')
result = json.loads(_run_tool("get_baggage_policy", args))
assert result["cabin"] == "经济舱"
assert result["free_allowance"] == "20kg"
def test_baggage_policy_missing_cabin():
result = json.loads(_run_tool("get_baggage_policy", {}))
assert result["cabin"] == "经济舱"
assert result["free_allowance"] == "20kg"
def test_baggage_policy_business_cabin():
result = json.loads(_run_tool("get_baggage_policy", {"cabin": "商务舱"}))
assert result["free_allowance"] == "30kg"
test_edits_list_null.py¶
from coding_agent import _apply_edits_from_args, _apply_one
def test_null_edits_like_empty():
working, applied, errors, edits = _apply_edits_from_args("hello world", {"edits": None})
assert working == "hello world"
assert applied == 0
assert errors == []
assert edits == []
def test_apply_edits_normal():
working, applied, errors, edits = _apply_edits_from_args(
"hello world",
{"edits": [{"old_str": "hello", "new_str": "hi"}]},
)
assert working == "hi world"
assert applied == 1
assert errors == []
assert len(edits) == 1
def test_apply_one_still_rejects_null_strings():
content, err = _apply_one("hello", None, "x")
assert err is not None