自进化智能体：自主智能体再训练实战手册

本教程面向有编程基础的开发者，深入讲解如何构建自进化智能体系统。以医疗文档处理为例，通过人工反馈和LLM评判，实现智能体的持续优化。内容涵盖基线智能体设计、评估方法、提示优化策略（包括OpenAI Evals平台、静态元提示和GEPA），并提供完整代码实现。读者将学会如何诊断智能体性能瓶颈，设计可测量的反馈信号，并构建自动化再训练循环，以提升智能体在生产环境中的准确性和适应性。

技术背景

智能体系统在概念验证后常因依赖人工诊断边缘案例而陷入瓶颈。本教程旨在解决此问题，通过可重复的再训练循环，捕获问题、从反馈中学习，并将改进推广到类似工作流。适用于要求准确性、可审计性和快速迭代的领域，如医疗文档处理。

核心原理

核心原理是构建一个闭环反馈系统：基线智能体生成输出，由人工或LLM评判者评估，根据评估结果优化提示，再生成新版本智能体，循环直至达到性能阈值。关键组件包括：基线智能体、评估器（如化学名称检查、长度偏差、余弦相似度、LLM评判）、元提示优化器、版本管理。数据流为：输入数据→智能体生成→评估器评分→反馈→元提示优化→更新智能体。

架构分析

系统架构采用模块化设计，包含：1) 基线智能体（如摘要器、合规检查器）；2) 评估模块（多种评分器组合）；3) 优化模块（元提示智能体或GEPA）；4) 版本管理（VersionedPrompt类）；5) 编排循环（self_evolving_loop）。设计思想强调可观测性、可回滚性和自动化，通过追踪和日志监控优化过程。

完整实战教程

概述

智能体系统在概念验证后常常陷入瓶颈，因为它们依赖人类来诊断边缘案例并纠正失败。本手册介绍了一种可重复的再训练循环，用于捕获这些问题、从反馈中学习，并将改进推广到类似生产的工作流中。我们将该方法应用于一个受监管的医疗文档处理任务，但这些模式可推广到任何要求准确性、可审计性和快速迭代的领域。

你将学到什么

本笔记本适合谁

如何使用本笔记本

  1. 从第1节开始，了解医疗用例、基线智能体和系统架构。
  2. 使用第2节在OpenAI Evals界面中练习提示优化，并收集结构化反馈。
  3. 运行第3节，使用评分器、评估和再训练逻辑自动化优化循环。
  4. 参考附录中的可复用提示、配置和评估模板，根据你的环境定制工作流。

本笔记本是模块化的——你可以独立或顺序运行各节，以适应你自己的智能体再训练循环。

1. 用例概述：医疗保健中的自进化智能体

问题定义

在本教程中，我们聚焦于一个真实世界的用例：为制药公司起草监管文件。这些组织必须准备并向监管机构（例如美国食品药品监督管理局）提交大量文件，以获得新药批准。这些提交的准确性和速度至关重要，因为它们直接影响救命治疗能够多快惠及患者。

监管文件起草是一个高度复杂、迭代且精确驱动的过程，需要深厚的科学、医学和合规专业知识。尽管有先进的写作工具，它仍然劳动密集且容易出错。智能体系统通过协助研究综合、内容生成和文档结构化提供了巨大的杠杆作用，但仍需要人类专家来确保事实准确性和法规合规性。

关键挑战是设计一个反馈循环，使这些智能体系统能够迭代学习并随时间优化模型行为。这样的系统可以逐渐将人类努力从详细修正转向高层监督，在保持监管提交所需严格标准的同时提高效率。

自进化智能体

下图说明了通过反馈、元提示和评估持续改进AI智能体的迭代过程。该循环结合了人类判断或使用LLM作为评判者的自动化反馈，以迭代地提升性能。

<br><em>图 1 - 展示自动化智能体改进的自进化循环示意图。</em>

该过程包括以下步骤：

  1. 基线智能体

    2. 该过程从基线智能体开始。在本笔记本中，我们使用一个刻意简化的示例（一个总结文档各部分的智能体）来说明迭代改进循环。在真实世界或企业环境中，基线智能体可能复杂得多。它生成的摘要作为后续评估和优化的初始基准。

  1. 人工反馈（或 LLM 作为评判者）

    2. 然后，基线智能体的输出由人工评审员（例如，在生产环境中）和/或自动化 LLM 作为评判者 系统进行评估。此步骤收集定量和定性反馈，以指示智能体在多大程度上实现其目标——例如，如果我们测试摘要的长度，反馈可能是“摘要太长”或由评估生成的数字分数（通常在 0 到 1 之间），用于评估摘要是否少于 500 字。

  1. 评估与聚合分数

    2. 基于收集的反馈，生成新的提示并通过评估（Evals）进行测试。这些测试根据预定义标准衡量性能，并将结果合并为反映整体性能的聚合分数。循环持续进行，直到分数超过目标阈值（例如，0.8）或达到最大重试次数（例如，max_retry = 10）。如果达到重试限制，则会提醒工程师需要手动改进。

  1. 更新后的基线智能体

    2. 一旦改进版本达到目标性能，它将取代原始基线智能体。这个更新后的智能体成为下一次迭代的基础，支持学习、反馈和优化的持续循环。

数据集概述

用于评估的数据集包含从 _Sample CMC Section for Hyperpolarized Pyruvate (13C) Injection_ 中提取的约 70 个部分，该文档可公开获取 此处。该数据集提供了真实、领域特定的内容，适合测试科学摘要和法规遵从行为。

基线智能体概述

为了使本指南自包含且易于复现，我们简化了法规起草用例，同时保留了其基本复杂性。在生产环境中，典型的法规撰写智能体由多个专门的子智能体组成，负责起草、数据分析、合规检查、引用生成和事实核查等任务。

在本指南中，我们缩小了法规撰写智能体的范围，专注于系统的自愈方面。我们的法规撰写智能体由两个子智能体组成：

对于本手册的其余部分，我们实现了 Summarizer 代理的简化版本（参见下文 代理设置 部分）。或者，您可以复用使用 AgentBuilder 创建的代理代码。如果您希望直接从 AgentBuilder 界面复现该代理，以下是使用的关键提示和参数：

两个代理均配置了默认参数 - 使用 GPT-5，低推理力度，输出格式为文本。

评估方法

要评估基线代理，主要有两种方法：

  1. 收集人工反馈。 此方法涉及通过 OpenAI Evals 平台（或为特定应用构建的自定义 UI）收集人类用户的反馈。它最适合生产环境或当领域专家（SME）在真实场景中与工具交互时进行试点。此方法有助于发现开发过程中可能未识别的边缘情况。在 Evals 平台上，用户可以给出点赞或点踩评分，并分享关于摘要的定性反馈。
  1. 使用 LLM 作为裁判。 此选项通常在开发阶段使用，无需 SME 的时间即可实现快速反馈循环。LLM 作为裁判 使用 LLM 根据预定义标准自动评估和评分代理的输出。它还可用于监控模型漂移（例如，在生产环境中）或验证模型及模型版本之间的更改（例如，在 gpt-5 和 gpt-5-mini 之间切换）。

本手册演示了这两种方法：

_注意：Evals 平台尚不提供以编程方式检索用户反馈的 API。_

2. 使用 OpenAI Evals 平台

OpenAI Evals 平台为提示优化和评估提供了一个直观的界面。本节演示了从数据集上传到迭代提示改进的完整工作流程，展示了如何在实施自动化解决方案之前，利用平台的视觉界面来优化你的提示。

步骤 1：上传数据集

要开始使用 OpenAI 评估平台，你首先需要上传数据集：

  1. 点击 + 创建 按钮
  2. 定义数据集名称
  3. 上传 CSV 文件并选择要保留的列
  4. 上传

你的数据集应包含需要总结的文档或文档部分。每一行代表一个将由你的系统处理的输入。

步骤 2：探索你的数据

上传后，你可以探索你的数据集。点击数据集名称以探索上传的数据。这允许你在继续配置提示之前，验证数据格式是否正确，并包含预期的内容。

步骤 3：配置初始提示

这是你定义初始系统提示并配置数据如何流经模型的地方。

<br><em>图 3 - 平台的“新建提示”界面，显示模型配置、变量和系统消息设置。</em>

配置步骤

  1. 系统提示：添加定义模型任务和行为的系统消息（此提示将被优化）
  2. 用户提示模板：添加用户消息的提示消息模板，使用诸如 {{<列名>}} 的变量，这些变量将被数据集中的实际数据替换
  3. 模型选择：选择用于生成的模型（例如，gpt-4.1、gpt-5）
  4. 温度：配置创造性与确定性之间的平衡

您可以从一个非常简单的提示开始，以展示优化过程的力量。例如，仅以“总结”开头，可以展示系统如何从最简起点演变。

第4步：生成输出

一旦您的提示配置完成，就可以在数据集上生成输出。提示将对每一行运行一次，输出将生成在新的输出列中。

  1. 点击“生成输出”
  2. 平台将对所有样本运行您的提示
  3. 结果将出现在新的输出列中

平台将处理数据集中的每一行，将模板变量替换为实际值，并使用您的系统提示调用模型。这将创建您可以评估的基线输出。

第5步：审查和评估

评估是您提供结构化反馈以指导提示改进的地方。

审查输出

  1. 添加评估列（如果未自动添加）- 点击“列”→“注释”→“添加”：
  1. 提供评分和反馈 - 为每个输出添加您的评估。

根据输出的质量，您可以选择好或坏的评分，并根据您希望答案如何改进来解释您的分数。例如：

（评分）| 反馈
- （好）很好，但应仅提供答案。输出不应包含标题或除答案以外的任何文本。
- （坏）信息很好，但应以项目符号呈现。
- （好）总结很好；很清晰。
- （坏）回答时使用项目符号以提高可读性。分别总结每个小节。
  1. 保存注释 - 您的反馈将与评估运行一起保存。

    2. <br><em>图 4 - 评估界面，显示生成的输出以及用于标注的评分和反馈列。</em>

这种结构化反馈成为自动提示优化的基础。

第 6 步：优化提示

收集反馈后，平台可以自动生成改进后的提示。

  1. 点击 “优化”
  2. 新提示版本将在新标签页中生成
  3. 点击 “查看提示” 查看改进后的版本

    4. <br><em>图 5 - 平台生成的改进提示，显示详细说明和要求。</em>

第 7 步：迭代与比较

准备好改进后的提示后，开始新一轮迭代以衡量改进效果。

  1. 点击 "生成输出"
  2. 查看新结果，并对任何剩余问题提供反馈
  3. 如有需要，再次点击 "优化"
  4. 重复操作，直到满意为止

平台的标签结构允许您比较各迭代的性能。您可以轻松看到输出如何从初始提示演变为优化版本。

<br><em>图 6 - 优化后提示词的反馈与评估结果，显示输出质量的提升。</em>

何时停止迭代

持续优化循环，直到：

这种基于平台的方法为理解提示词优化提供了极佳的基础，之后再转向自动化实现。可视化界面使更改的影响和优化过程一目了然。

3. 使用LLM作为评判者的自进化循环

本节介绍一种完全自动化的评估工作流，通过OpenAI API使用LLM作为评判者，无需任何用户界面。这种方法能够对智能体性能进行可扩展的、程序化的评估，支持快速迭代和生产环境中的持续模型监控。

# gepa and litellm are only required for the Section 4.b (prompt optimization with GEPA)
%pip install --upgrade openai openai-agents pydantic pandas gepa litellm python-dotenv -qqq
%load_ext dotenv
%dotenv

# Place your API key in a file called .env
# OPENAI_API_KEY=sk-...

评估创建

为了评估基线摘要代理，我们使用四个互补的评分器，它们在确定性检查与语义判断之间取得平衡。

| 评分器 | 类型 | 通过阈值 | 检查内容 | 原因 |

|---|---|---:|---|---|

| 化学字符串名称 | python | 0.8 | 章节中出现的任何精确化学名称是否出现在摘要中。 | 强制保留关键领域实体，使摘要不会遗漏具有化学意义的术语。 |

| 摘要长度 | python | 0.85 | 与预期100字长度的反向偏差。 | 保持摘要简洁且具有可比性，减少可能掩盖内容缺陷的冗长。 |

| 余弦相似度 | text_similarity | 0.85 | 章节与摘要文本之间的余弦相似度。 | 确保摘要始终锚定在源内容上，而不是在语义上漂移。 |

| LLM作为评判者 | score_model | 0.85 | 由模型作为评估者根据规则驱动的分数。 | 捕捉基于规则的指标遗漏的细微质量信号，提高整体鲁棒性。 |

注意

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

data_source_config = {
    "type": "custom",
    "item_schema": {
        "type": "object",
        "properties": {"section": {"type": "string"}, "summary": {"type": "string"}},
        "required": ["section", "summary"],
    },
    "include_sample_schema": False,
}

testing_criteria = [
    {
        "type": "python",
        "name": "chemical_name_grader",
        "image_tag": "2025-05-08",
        "pass_threshold": 0.8,
        "source": r"""def grade(sample: dict, item: dict) -> float:
    section = item["section"]
    summary = item["summary"]
    CHEMICALS_MASTER = ["[1-¹³C]Pyruvic acid","[1-¹³C]Pyruvate","¹²C Pyruvic acid","Sodium [1-¹³C]pyruvate","Sodium pyruvate (¹²C)","AH111501 (Trityl radical)","Tris{8-carboxyl-2,2,6,6-tetra[2-(1-methoxyethyl)]-benzo(1,2-d:4,5-d’)bis(1,3)dithiole-4-yl}methyl acid","AH111501 sodium salt","Methyl, tris[8-carboxy-2,2,6,6-tetrakis(2-methoxyethyl)benzo[1,2-d:4,5-d’]bis[1,3]dithiol-4-yl]-, trisodium salt","AH111501 trisodium salt","AH111576","2,2′,2″,2‴-(4,8-Dibromobenzo[1,2-d:4,5-d′]bis([1,3]dithiole)-2,2,6,6-tetrayl)tetraethanol","AH111586","4,8-Dibromo-2,2,6,6-tetrakis(2-methoxyethyl)benzo[1,2-d:4,5-d′]bis([1,3]dithiole)","AH111709","AH111743","AH112615","4,4-Bis-hydroxymethyl-2-methyl-oxazolidine-2-carboxylic acid","AH112623","Parapyruvate","2-Hydroxy-2-methyl-4-oxo-pentanedioic acid","AH113127","(4-Hydroxymethyl-oxazolidin-4-yl)-methanol","AH113462/E","Enol lactone","AH113462/K","Keto lactone","Acetyl bromide","Methanol","Dimethyl sulfoxide","DMSO","Tetrahydrofuran","THF","Acetonitrile","ACN","Diethyl ether","Et₂O","N,N-Dimethylacetamide","DMA","1,3-Dimethyl-2-imidazolidinone","DMI","Hydrochloric acid","HCl","Sodium hydroxide","NaOH","Disodium ethylenediaminetetraacetate","Na₂EDTA","Ethylenediaminetetraacetic acid","EDTA","Tris(hydroxymethyl)aminomethane","TRIS","Trometamol","Trifluoroacetic acid","TFA","Toluene","Heptane","Ethyl acetate","Ethanol","Water","H₂O","Sodium chloride","NaCl","Cuprous [1-¹³C]cyanide","Cu¹³CN","Gadolinium","Gd","Tin","Sn","Phosphorus","P","Carbon dioxide","CO₂","Sodium [1-13C]pyruvate","[1-13C]Pyruvic acid","1-13C pyruvate"]

    # Identify the chemicals present in the section
    present = [chem for chem in CHEMICALS_MASTER if chem in section]

    # If no chemicals present, consider it satisfied
    if not present:
        return 1.0

    correct = 0
    for chem in present:
        # Only count as correct if the exact chemical string appears in the summary
        if chem in summary:
            correct += 1

    return correct / len(present)""",
    },
    {
    "type": "python",
    "name": "word_length_deviation_grader",
    "image_tag": "2025-05-08",
    "pass_threshold": 0.85,
    "source": r"""
def grade(sample: dict, item: dict) -> float:
    summary = item["summary"]
    word_count = len(summary.split())
    
    expected_summary_length = 100
    tolerance = 0.2  # 20% band around target
    
    # relative deviation
    deviation = abs(word_count - expected_summary_length) / expected_summary_length
    
    # If within tolerance band → full score
    if deviation <= tolerance:
        return 1.0
    
    # Outside band → score decays linearly, capped at 0
    # e.g., deviation 0.3 → score 0.8, deviation 1.0+ → 0.0
    score = 1.0 - (deviation - tolerance)
    return max(0.0, score)
""",
},
    {
        "name": "cosine_similarity",
        "type": "text_similarity",
        "input": "{{ item.summary }}",
        "reference": "{{ item.section }}",
        "evaluation_metric": "cosine",
        "pass_threshold": 0.85,
    },
    {
        "name": "llm_as_judge",
        "type": "score_model",
        "model": "gpt-4.1",
        "input": [
            {
                "role": "system",
                "content": (
                    "You are an expert technical summarization evaluator. "
                    "Evaluate whether the summary captures and preserves the important technical facts and specific details from the section, allowing for occasional minor rewording or omissions of less important points, but not major technical inaccuracies or information loss.\n\n"
                    "Scoring Guidelines:\n"
                    "- Return a numerical score between 0 and 1 (with up to two decimal places).\n"
                    "- A score of 1 means the summary is almost flawless: it is comprehensive, highly faithful, and technically accurate, with virtually no important or meaningful details missing, and no significant misstatements or distortions.\n"
                    "- 0.75-0.99 indicates excellent work: all main facts are represented, but there may be trivial omissions or very minor rewording that do not materially affect understanding.\n"
                    "- 0.5-0.75 indicates good but imperfect: most technical information is retained and correctly presented, some less critical details might be missing or slightly rephrased, but overall fidelity is preserved.\n"
                    "- 0.3-0.5 means significant information is missing, or some technical inaccuracies are present, but the summary retains a reasonable portion of key facts.\n"
                    "- 0.0-0.3 means there are major omissions, misunderstandings, or a failure to capture the most important technical content.\n\n"
                    "Respond only with a single number between 0 and 1 indicating summary quality by these criteria."
                ),
            },
            {
                "role": "user",
                "content": (
                    "Section:\n{{item.section}}\n"
                    "Summary:\n{{sample.output_text}}"
                ),
            },
        ],
        "range": [0, 1],
        "pass_threshold": 0.85,
    },
]

eval = client.evals.create(
    name="self_evolving_eval",
    data_source_config=data_source_config,
    testing_criteria=testing_criteria,
)
print(f"Created Eval: {eval.id}")

你应该会在输出中看到一个评估ID，例如 eval_...。这是我们刚刚创建的评估的ID（如下所示）

<br><em>图 7 - 平台的评估界面，显示数据源配置和测试标准设置。</em>

评分器评分与解析

接下来，我们需要对摘要代理的输出运行评估，并解析评估的评分器得分结果。为此，我们将使用几个辅助函数：

import time
import json

def run_eval(eval_id: str, section: str, summary: str):
  """Creates a run of the eval with the input section and output summary."""
  return client.evals.runs.create(
    eval_id=eval_id,
    name="self-evolving-eval",
    data_source={
      "type": "jsonl",
      "source": {
        "type": "file_content",
        "content": [
          {
            "item": {
              "section": section,
              "summary": summary,
            }
          }
        ],
      },
    },
  )


def poll_eval_run(eval_id: str, run_id: str, max_polls = 10):
    """
    Polls the evaluation run until completion or timeout.

    This function exists to handle asynchronous behavior in the eval service by
    periodically checking run status. It balances responsiveness and resource use by
    polling at fixed intervals rather than blocking indefinitely. The retry limit
    prevents runaway loops in cases where the service never returns a completed status.
    """
    run = None
    for attempt in range(1, max_polls + 1):
        run = client.evals.runs.retrieve(eval_id=eval_id, run_id=run_id)
        if run.status == "completed":
            break
        if attempt == max_polls:
            print("Exceeded retries, aborting")
            break

        time.sleep(5)

    run_output_items = client.evals.runs.output_items.list(
        eval_id=eval_id, run_id=run_id
    )
    return run_output_items


def parse_eval_run_output(items):
    """Extract all grader scores and any available conclusion outputs."""
    all_results = []

    for item in items.data:
        for result in item.results:
            grader_name_full = result.name
            score = result.score
            passed = result.passed
            reasoning = None
            try:
                sample = result.sample
                if sample:
                    content = result.sample["output"][0]["content"]
                    content_json = json.loads(content)
                    steps = content_json["steps"]
                    reasoning = " ".join([step["conclusion"] for step in steps])
            except Exception:
                pass

            all_results.append(
                {
                    "grader_name": grader_name_full,
                    "score": score,
                    "passed": passed,
                    "reasoning": reasoning,
                }
            )

    return all_results

现在我们可以使用之前创建的 eval ID，对任意输入章节和摘要输出运行评分器。这构成了反馈循环的主干，该循环将启动提示优化例程。

评估执行运行

让我们通过直接提供一个章节和生成的摘要来测试我们的评估。

EVAL_ID = eval.id #Created eval ID from above cell
SECTION = "3.2.S.1 General Information ([1-13C]pyruvic acid) The active ingredient in Hyperpolarized Pyruvate (13C) Injection is hyperpolarized [1-13C]pyruvate. The drug substance is defined as [13C]pyruvic acid, which is neutralized to [1-13C]pyruvate during the compounding process. In several pre-clinical and clinical studies and during evaluation of stability, pyruvic acid has been used instead of [1-13C]pyruvic acid (see Sections 3.2.P.2.2.1 Formulation Development for Hyperpolarized Pyruvate (13C) Injection and Section 8.1 Introduction for Item 8 Pharmacology and Toxicology Info). In the Section 3.2.S Drug Substance, data are presented for both pyruvic acid and for [1-13C]pyruvic acid. For simplicity, the terminology used in headings and captions is [1-13C]pyruvic acid. Batches containing pyruvic acid are specified by footnotes. 3.2.S.1.1 Nomenclature ([1-13C]pyruvic acid) The drug substance used for compounding of Hyperpolarized Pyruvate (13C) Injection is [1-13C]pyruvic acid. Company code: W6578 Chemical name: [1-13C]pyruvic acid CAS registry number: 127-17-3 3.2.S.1.2 Structure ([1-13C]pyruvic acid) Figure 1 Structure of [1-13C]pyruvic acid Molecular formula: C H O 3 4 3 Molecular weight: 89.06 3.2.S.1.3 General Properties ([1-13C]pyruvic acid) Appearance: Colorless to yellow, clear, viscous liquid pKa:Ka:aranWater solubility: Complete The structure of [1-13C]pyruvic acid has been confirmed by spectroscopic analysis (see Section 3.2.S.3.1 Elucidation of Structure and other Characteristics)."
SUMMARY = "The active ingredient in Hyperpolarized Pyruvate (13C) Injection is hyperpolarized [1-13C]pyruvate, derived from [1-13C]pyruvic acid (neutralized during compounding). Both pyruvic acid and [1-13C]pyruvic acid were used in studies and stability evaluations, but the documentation refers to [1-13C]pyruvic acid unless otherwise noted. The drug substance ([1-13C]pyruvic acid, CAS 127-17-3) is a colorless to yellow, clear, viscous liquid with a molecular formula C3H4O3 and molecular weight 89.06. Its structure has been confirmed by spectroscopic analysis, and it is completely soluble in water."

eval_run = run_eval(EVAL_ID, section=SECTION, summary=SUMMARY)
run_output = poll_eval_run(eval_id=EVAL_ID, run_id=eval_run.id)

grader_scores = parse_eval_run_output(run_output)
print(grader_scores)

你应该会在输出中看到一列评分器的分数，例如：


运行此脚本，我们可以看到大多数评分器都通过了，除了 `chemical_name_grader`。接下来，我们将以编程方式识别这一改进摘要代理的机会。

_注意：在本地运行时，除 `chemical_name_grader` 之外的评分器可能最初会失败。这很正常，因为评分器最初可能会失败，但结果应通过反馈循环得到改善。早期失败仅反映模型在收敛到更准确结果之前调整其响应。_

### 仪表盘可观测性
评估运行和结果也可以在 OpenAI 仪表盘中查看：
![](assets/images/276a3f7b4e923f0.png)<br><em>图 8 - 评估仪表板，显示评估运行和结果。</em>

我们还可以深入查看特定的评估运行：
![](assets/images/6be59aad71bc6494.png)<br><em>图 9 - 详细的评估运行结果，显示评分器得分和性能指标。</em>

## 智能体设置

现在我们已经设置好了评估和评分器，可以回到我们的摘要智能体。
为简单起见，我们将在下面提供一个简单智能体的代码。您也可以使用 `AgentBuilder`，如图 2 所示，并从 UI 导出代码。

我们还需要一个元提示优化智能体来优化我们的提示，以及一些简单的工具来处理提示版本：
- `PromptVersionEntry`：一个 pydantic 模型，用于在生产环境中跟踪提示及其元数据的变化
- `VersionedPrompt`：一个工具类，用于跟踪提示版本，这在生产环境中分析提示的演变以及确保在出现回归时有回退历史时非常重要

from datetime import datetime

from typing import Any, Optional

from pydantic import BaseModel, Field, ConfigDict, field_validator

class PromptVersionEntry(BaseModel):

"""Data model for a prompt and associated data for observability"""

version: int = Field(

  1. ., ge=0, description="Version number of the prompt (increments)"

    2. )

    model: str = Field(

    "gpt-5",

    min_length=1,

    description="The model version to use for this version of the prompt, defaults to gpt-5",

    )

    prompt: str = Field(

  2. ., min_length=1, description="The prompt text for this version"

    3. )

    timestamp: datetime = Field(

    default_factory=datetime.utcnow,

    description="UTC timestamp when this version was created",

    )

    eval_id: Optional[str] = Field(

    None, description="ID of the evaluation associated with this prompt version"

    )

    run_id: Optional[str] = Field(

    None, description="ID of the run associated with this prompt version"

    )

    metadata: Optional[dict[str, Any]] = Field(

    None, description="Free-form metadata dict (e.g., section, summary)"

    )

model_config = ConfigDict(

str_strip_whitespace=True, validate_assignment=True, extra="forbid"

)

@field_validator("prompt")

@classmethod

def prompt_not_blank(cls, v: str) -> str:

if not v.strip():

raise ValueError("prompt must not be blank or only whitespace")

return v

class VersionedPrompt:

"""Manages a collection of prompt versions and provides controlled updates and rollbacks."""

def __init__(

self,

initial_prompt: str,

model: Optional[str] = "gpt-5",

eval_id: Optional[str] = None,

run_id: Optional[str] = None,

metadata: Optional[dict[str, Any]] = None,

):

if not initial_prompt or not initial_prompt.strip():

raise ValueError("initial_prompt must be non-empty")

self._versions: list[PromptVersionEntry] = []

first_entry = PromptVersionEntry(

version=0,

prompt=initial_prompt,

model=model,

eval_id=eval_id,

run_id=run_id,

metadata=metadata,

)

self._versions.append(first_entry)

def update(

self,

new_prompt: str,

model: Optional[str] = "gpt-5",

eval_id: Optional[str] = None,

run_id: Optional[str] = None,

metadata: Optional[dict[str, Any]] = None,

) -> PromptVersionEntry:

if not new_prompt or not new_prompt.strip():

raise ValueError("new_prompt must be non-empty")

version = self.current().version + 1

entry = PromptVersionEntry(

version=version,

prompt=new_prompt,

model=model,

eval_id=eval_id,

run_id=run_id,

metadata=metadata,

)

self._versions.append(entry)

return entry

def current(self) -> PromptVersionEntry:

return self._versions[-1]

def revert_to_version(self, version: int) -> PromptVersionEntry:

idx = None

for i, entry in enumerate(self._versions):

if entry.version == version:

idx = i

break

if idx is None:

raise ValueError(f"No version found with version={version}")

self._versions = self._versions[: idx + 1]

return self._versions[-1]


接下来，我们将创建初始的摘要和提示优化代理。

_注意：我们创建了一个包装器来跟踪摘要代理中的提示更改，因为它在生产中预计会不断演变，而元提示代理的提示在本示例中将保持静态。_

from agents import Agent

METAPROMPT_TEMPLATE = """

Context:

Original prompt:

{original_prompt}

Section:

{section}

Summary:

{summary}

Reason to improve the prompt:

{reasoning}

Task:

Write a new summarization prompt that is significantly improved and more specific than the original.

The new prompt should instruct the model to produce concise yet comprehensive technical summaries that precisely preserve all explicit information from the source text. It should emphasize the inclusion of all named entities, quantities, compounds, and technical terminology without paraphrasing or omission. The resulting prompt should read like a clear, directive system message for a technical summarization assistant—structured, unambiguous, and generalizable across scientific or regulatory document sections.

"""

metaprompt_agent = Agent(

name="MetapromptAgent", instructions="You are a prompt optimizer."

)

summarization_prompt = VersionedPrompt(

initial_prompt="""You are a summarization assistant.

Given a section of text, produce a summary."""

)

def make_summarization_agent(prompt_entry: PromptVersionEntry) -> Agent:

return Agent(

name="SummarizationAgent",

instructions=prompt_entry.prompt,

model=prompt_entry.model,

)

summarization_agent = make_summarization_agent(summarization_prompt.current())

Cache eval results by section + summary so repeated attempts do not trigger redundant grader runs.

eval_cache: dict[tuple[str, str], list[dict[str, Any]]] = {}

Track the highest-scoring candidate that also passes the lenient score threshold.

best_candidate: dict[str, Any] = {

"score": float("-inf"),

"prompt": summarization_prompt.current().prompt,

"model": summarization_prompt.current().model,

"summary": None,

"metadata": None,

"version": summarization_prompt.current().version,

"passed_lenient": False,

"total_score": float("-inf"),

}

Aggregate per-version performance so we can pick the strongest total scorer at the end.

aggregate_prompt_stats: dict[int, dict[str, Any]] = {}


### 编排与监控

到目前为止，我们已经创建了：
- 包含4个评分器的评估，用于评估输出并为每个评分器生成分数
- 一个带有版本化提示类的摘要代理，用于跟踪提示和模型的变更
- 一个元提示优化代理，将根据一组推理尝试更新提示

现在，这些不同的功能可以组合起来，通过OpenAI仪表板中的Agent追踪来编排自我进化的循环。

请记住，这是一个简化的示例。在真实场景中，您需要确保对优化尝试设置防护措施，并在触发防护措施时通知人类。

_注意：由于cookbook的实际限制，我们通过输入静态数据集并使用`print`语句代替真正的可观测性来模拟数据流。_

### 编排工具

与前面的部分一样，我们将创建一些工具来管理反馈循环的编排逻辑。

import asyncio

from typing import Any, Optional

from agents import Runner

LENIENT_PASS_RATIO = 0.75 # 75% of graders must pass (binary)

LENIENT_AVERAGE_THRESHOLD = 0.85 # 85% average score across graders

def reset_best_candidate() -> None:

"""Reset the best candidate tracker for a new optimization run."""

global best_candidate

current = summarization_prompt.current()

best_candidate = {

"score": float("-inf"),

"prompt": current.prompt,

"model": current.model,

"summary": None,

"metadata": None,

"version": current.version,

}

def reset_best_trackers() -> None:

"""Reset both the best-candidate tracker and aggregate stats."""

reset_best_candidate()

aggregate_prompt_stats.clear()

def update_best_candidate(

*,

average_score: Optional[float] = None,

prompt_text: str,

model_name: str,

summary_text: str = None,

metadata: dict[str, Any] = None,

lenient_passed: bool = False,

prompt_version: int = None,

total_score: Optional[float] = None,

score: Optional[float] = None,

) -> None:

"""Persist the best lenient-passing candidate."""

global best_candidate

if prompt_version is None:

prompt_version = summarization_prompt.current().version

if average_score is None:

average_score = score

if average_score is None:

return

if lenient_passed:

best_candidate.update(

{

"score": average_score,

"prompt": prompt_text,

"model": model_name,

"summary": summary_text,

"metadata": metadata,

"version": prompt_version,

"total_score": total_score if total_score is not None else average_score,

}

)

def apply_best_candidate_if_needed() -> Agent:

"""Ensure summarization_prompt reflects the best prompt candidate."""

if best_candidate["score"] > float("-inf"):

current = summarization_prompt.current()

target = best_candidate

Only update if different

if (

current.prompt != target["prompt"]

or current.model != target["model"]

or current.version != target.get("version")

):

summarization_prompt.update(

new_prompt=target["prompt"],

model=target["model"],

metadata=target.get("metadata"),

)

target["version"] = summarization_prompt.current().version

return make_summarization_agent(summarization_prompt.current())

return make_summarization_agent(summarization_prompt.current())

def record_aggregate_prompt_score(

*,

prompt_version: int,

prompt_text: str,

model_name: str,

average_score: float,

total_score: Optional[float] = None,

) -> None:

"""Accumulate per-version grader scores for aggregate selection."""

stats = aggregate_prompt_stats.setdefault(

prompt_version,

{

"version": prompt_version,

"prompt": prompt_text,

"model": model_name,

"total_score": 0.0,

"total_average": 0.0,

"count": 0,

},

)

stats["total_score"] += total_score if total_score is not None else average_score

stats["total_average"] += average_score

stats["count"] += 1

stats["prompt"] = prompt_text

stats["model"] = model_name

def select_best_aggregate_prompt() -> Optional[dict[str, Any]]:

"""Return the prompt version with the highest cumulative score."""

if not aggregate_prompt_stats:

return None

return max(

aggregate_prompt_stats.values(),

key=lambda entry: (

entry.get("total_score", float("-inf")),

entry.get("version", -1),

),

)

async def get_eval_grader_score(eval_id: str, section: str, summary: str):

"""Retrieve grader scores for a section-summary pair with caching."""

cache_key = (section, summary)

if cache_key in eval_cache:

return eval_cache[cache_key]

eval_run = run_eval(eval_id=eval_id, section=section, summary=summary)

run_output = poll_eval_run(eval_id=eval_id, run_id=eval_run.id)

results = parse_eval_run_output(run_output)

eval_cache[cache_key] = results

return results

def calculate_grader_score(grader_scores):

"""Simple average score of all graders from the eval."""

if not grader_scores:

return 0.0

score_sum = 0.0

for entry in grader_scores:

score_sum += entry.get("score", 0.0)

return score_sum / len(grader_scores)

def calculate_total_grader_score(grader_scores):

"""Sum of all grader scores for aggregate tracking."""

if not grader_scores:

return 0.0

return sum(entry.get("score", 0.0) for entry in grader_scores)

DEFAULT_PASSING_FEEDBACK = (

"All graders passed; tighten factual coverage, chemical completeness, and conciseness."

)

def is_lenient_pass(grader_scores, average_score: float) -> bool:

if not grader_scores:

return False

passed_count = sum(1 for entry in grader_scores if entry.get("passed"))

total_graders = len(grader_scores)

if total_graders and (passed_count / total_graders) >= LENIENT_PASS_RATIO:

return True

return average_score >= LENIENT_AVERAGE_THRESHOLD

def collect_grader_feedback(grader_scores):

"""Consolidate grader reasoning into actionable feedback for the metaprompt agent."""

feedback_lines = []

for entry in grader_scores:

grader = entry.get("grader_name", "")

passed = entry.get("passed", False)

reasoning = entry.get("reasoning")

if not passed:

if grader.startswith("chemical_name_grader"):

feedback_lines.append(

"Not all chemical names in the input section were included in the summary."

)

elif grader.startswith("word_length_deviation_grader"):

feedback_lines.append(

"The summary length deviates too much from the expected length."

)

elif grader.startswith("cosine_similarity"):

feedback_lines.append(

"The summary is not sufficiently similar to the source section (cosine similarity too low)."

)

elif grader.startswith("llm_as_judge") and reasoning:

feedback_lines.append(reasoning)

if not feedback_lines:

feedback_lines.append(DEFAULT_PASSING_FEEDBACK)

return "".join(feedback_lines)


### 自进化循环

现在，为了模拟一连串的摘要请求，我们将输入一个准备好的数据集，并观察优化如何从一个朴素的提示词开始进化。

> 所引用的 dataset.csv 可以在 Github 仓库中找到。

import pandas as pd

from agents import Agent, trace

EVAL_ID = eval.id #Created eval ID from above cell

MAX_OPTIMIZATION_RETRIES = 3

async def self_evolving_loop(summarization_agent: Agent) -> Agent:

print(f"Starting self-evolving loop | Initial prompt v{summarization_prompt.current().version}")

print(f"Prompt:{summarization_prompt.current().prompt}")

print("-" * 80)

reset_best_trackers()

df = pd.read_csv("data/dataset.csv")

with trace("Self-evolving Optimization Workflow"):

for _, row in df.head().iterrows():

content = row.get("content")

if pd.isna(content) or (isinstance(content, str) and not content.strip()):

continue

section_number = str(row["section_number"])

section = str(content)

current_version = summarization_prompt.current().version

print(f"[Section {section_number}] Using prompt v{current_version}")

optimization_success = False

for attempt in range(1, MAX_OPTIMIZATION_RETRIES + 1):

print(f" Attempt {attempt}: evaluating summary...")

summary_result = await Runner.run(summarization_agent, section)

summary = summary_result.final_output

grader_scores = await get_eval_grader_score(eval_id=EVAL_ID, summary=summary, section=section)

average_score = calculate_grader_score(grader_scores)

total_score = calculate_total_grader_score(grader_scores)

lenient_passed = is_lenient_pass(grader_scores, average_score)

print(

f" Scores — avg={average_score:.3f}, total={total_score:.3f}, lenient_passed={lenient_passed}"

)

record_aggregate_prompt_score(

prompt_version=summarization_prompt.current().version,

prompt_text=summarization_prompt.current().prompt,

model_name=summarization_prompt.current().model,

average_score=average_score,

total_score=total_score,

)

update_best_candidate(

average_score=average_score,

prompt_text=summarization_prompt.current().prompt,

model_name=summarization_prompt.current().model,

summary_text=summary,

metadata={

"section": section_number,

"average_score": average_score,

"grader_results": grader_scores,

"prompt_version": summarization_prompt.current().version,

},

lenient_passed=lenient_passed,

prompt_version=summarization_prompt.current().version,

)

if lenient_passed:

optimization_success = True

print(f" Passed with prompt v{summarization_prompt.current().version}")

break

print(" Failed eval. Improving prompt...")

eval_feedback = collect_grader_feedback(grader_scores)

metaprompt_result = await Runner.run(

metaprompt_agent,

input=METAPROMPT_TEMPLATE.format(

original_prompt=summarization_prompt.current().prompt,

section=section,

summary=summary,

reasoning=eval_feedback,

),

)

improved_prompt = metaprompt_result.final_output

summarization_prompt.update(

new_prompt=improved_prompt,

metadata={"section": section, "summary": summary},

)

summarization_agent = make_summarization_agent(summarization_prompt.current())

print(f" Prompt improved → v{summarization_prompt.current().version}")

if not optimization_success:

print(

" All attempts failed; keeping latest prompt version "

f"v{summarization_prompt.current().version} for the next section."

)

summarization_agent = apply_best_candidate_if_needed()

print("" + "-" * 80)

print("Completed optimization loop.")

print(f"Final prompt version: v{summarization_prompt.current().version}")

if best_candidate["score"] > float("-inf"):

print(

f"Best lenient prompt: v{best_candidate.get('version')} (avg={best_candidate['score']:.3f})"

)

aggregate_best = select_best_aggregate_prompt()

if aggregate_best:

per_section = (

aggregate_best.get("total_average", 0.0) / aggregate_best.get("count", 1)

if aggregate_best.get("count")

else 0.0

)

print(

f"Aggregate best prompt: v{aggregate_best.get('version')} "

f"(total={aggregate_best.get('total_score', 0.0):.3f}, avg/section={per_section:.3f}, model={aggregate_best.get('model', 'unknown')})"

)

print(f"Final prompt:{summarization_prompt.current().prompt}")

return summarization_agent

summarization_agent = await self_evolving_loop(summarization_agent)


**最终提示词的选择方式**

- 每次评估都会记录评分者的平均分、评分者总分以及尝试是否通过了宽松标准。
- `best_candidate` 跟踪最近一次宽松通过（为了透明），但最终选择使用汇总总分，以确保我们保留整体表现最佳的提示词。
- 当循环结束时，`apply_best_candidate_if_needed` 会恢复累计评分者得分最高的提示词（并列时优先选择最新版本），从而保证最终呈现的提示词是观察到的表现最强的那个。

以下是上述代码的输出示例（节选）。

检查输出可以发现，自我进化提示词确实起作用了。有几点需要注意：
1. 优化并不总是成功的，因此能够回滚提示词版本非常重要。
2. 评分者提供的信息的保真度对于确保优化质量至关重要。

### Agent 日志与追踪

我们可以在仪表盘的日志下查看优化工作流的运行情况：
![](assets/images/ba3c1a7ecae0fda6.png)<br><em>图 10 - 仪表板中显示优化工作流运行的代理日志跟踪。</em>

并深入查看不同的代理调用：
![](assets/images/16655b00a8782507.png)<br><em>图 11 - 详细的智能体追踪，显示各个智能体调用及执行流程。</em>

### 持续监控

评估循环完成后，系统应继续监控新传入的数据，并定期在盲数据集上重新评估模型性能。这确保了随着数据分布的变化，模型保持准确和合规。

为了实现持续监控，您可以集成一个 cron 作业或轻量级调度器循环，定期检查数据源（例如，新的 PDF 上传或数据库条目）中的更新。当检测到新数据时，系统会自动触发前面描述的评估和优化循环。

例如（伪代码）：

this cell is pseudo-code and not meant to be run as-is

import time

def continuous_monitoring(interval_hours=24):

"""Periodically check for new data and trigger the evaluation loop."""

while True:

print("Checking for new data...")

if new_data_detected():

print("New data found — running evaluation and optimization loop.")

self_evolving_loop()

else:

print("No new data. Sleeping until next cycle.")

time.sleep(interval_hours * 3600)

continuous_monitoring(interval_hours=24)


这种方法使模型能够持续学习和适应，随着处理新数据而不断改进——这是保持高质量、真实世界性能的关键要求。

## 4. 更进一步

### a. 模型评估

我们现在有了一个完全自动化的循环，通过**评估**来改进我们的提示词，并在评分超过设定阈值时接受新的提示词。

在生产环境中，你可以使用类似的框架来监控你的智能体在新用户请求到来时的表现。

如上所述，这是一个简化的示例，在真实场景中，你可能需要额外的防护措施和人工参与的方法来批准新的提示词。

进一步扩展这个概念，我们还可以使用评估来测试不同的模型参数候选，如模型版本、详细程度和推理能力。要查看可考虑的全部参数集，请参阅 [Agents SDK 中的 ModelSettings 类](https://openai.github.io/openai-agents-python/ref/model_settings/#agents.model_settings.ModelSettings)。

`compare_model_candidates` 函数是一个示例，展示了如何：
1. 优化提示词
2. 使用两个或多个不同模型从优化后的提示词生成候选输出
3. 使用评估对候选输出进行评分，并选择最佳候选

它可以以最小的重构集成到 `self_evolving_loop` 函数中。

> **注意：** 模型版本的生产测试应仅限于同一系列版本（例如 gpt-5、gpt-5-mini、gpt-5-nano）。建议在生产部署前进行跨系列版本选择。

以及最终的 `self_evolving_loop` 与模型比较代码：

from agents import Agent, Runner

async def eval_agent_candidate(agent: Agent, section: str, prompt_text: str, model_name: str):

summary_result = await Runner.run(agent, section)

summary = summary_result.final_output

scores = await get_eval_grader_score(

eval_id=EVAL_ID, summary=summary, section=section

)

average = calculate_grader_score(scores)

lenient_passed = is_lenient_pass(scores, average)

passed = all(entry.get("passed") is True for entry in scores)

update_best_candidate(

average_score=average,

prompt_text=prompt_text,

model_name=model_name,

summary_text=summary,

metadata={

"section": section,

"average_score": average,

"grader_results": scores,

},

lenient_passed=lenient_passed,

)

return {"summary": summary, "scores": scores, "average": average, "passed": passed}

async def compare_model_candidates(

summarization_prompt,

eval_feedback: str,

section: str,

summary: str,

model_candidates=None,

):

"""Improve the prompt, evaluate it across candidate models, and adopt the top performer."""

if model_candidates is None:

model_candidates = ["gpt-5", "gpt-5-mini"]

metaprompt_result = await Runner.run(

metaprompt_agent,

input=METAPROMPT_TEMPLATE.format(

original_prompt=summarization_prompt.current().prompt,

section=section,

summary=summary,

reasoning=eval_feedback,

),

)

improved_prompt = metaprompt_result.final_output

async def evaluate_model(model_name: str):

candidate_agent = Agent(

name=f"SummarizationAgent:{model_name}",

instructions=improved_prompt,

model=model_name,

)

result = await eval_agent_candidate(candidate_agent, section, improved_prompt, model_name)

return model_name, candidate_agent, result

best = {

"average": float("-inf"),

"passed": False,

"agent": None,

"model": None,

"summary": None,

}

tasks = [asyncio.create_task(evaluate_model(model_name)) for model_name in model_candidates]

for task in asyncio.as_completed(tasks):

model_name, candidate_agent, result = await task

print(

f"Candidate average — {model_name}: {result['average']:.4f} "

f"(passed={result.get('passed', False)})"

)

if result["average"] > best["average"]:

best.update(

{

"average": result["average"],

"model": model_name,

"summary": result.get("summary"),

"agent": candidate_agent,

"passed": result.get("passed", False),

}

)

for task in tasks:

if not task.done():

task.cancel()

if best["passed"] and best["model"]:

summarization_prompt.update(

new_prompt=improved_prompt,

model=best["model"],

metadata={"section": section, "summary": best["summary"]},

)

print(f"Updated summarization_prompt with passing model: {best['model']}")

return make_summarization_agent(summarization_prompt.current())

print(

f"No passing models. Best candidate (model={best['model']}, "

f"avg={best['average']:.4f}) did not pass. Prompt not updated."

)

return None

async def self_evolving_loop_with_model_comparison(summarization_agent: Agent) -> Agent:

print(

f"Starting self-evolving loop | Initial prompt v{summarization_prompt.current().version}"

)

print(f"Prompt: {summarization_prompt.current().prompt}")

print(f"Model: {summarization_prompt.current().model}")

print("-" * 80)

reset_best_trackers()

df = pd.read_csv("data/dataset.csv")

with trace("Self-evolving Optimization Workflow: model comparison"):

for _, row in df.head(5).iterrows():

content = row.get("content")

if pd.isna(content) or (isinstance(content, str) and not content.strip()):

continue

section_number = str(row["section_number"])

section = str(content)

current_version = summarization_prompt.current().version

print(f"[Section {section_number}] Using prompt v{current_version}")

summary_passed = False

for attempt in range(1, MAX_OPTIMIZATION_RETRIES + 1):

print(f"\tAttempt {attempt}: evaluating summary...")

summary_result = await Runner.run(summarization_agent, section)

summary = summary_result.final_output

grader_scores = await get_eval_grader_score(

eval_id=EVAL_ID, summary=summary, section=section

)

average_score = calculate_grader_score(grader_scores)

total_score = calculate_total_grader_score(grader_scores)

lenient_passed = is_lenient_pass(grader_scores, average_score)

print(

f"\tScores — avg={average_score:.3f}, total={total_score:.3f}, lenient_passed={lenient_passed}"

)

record_aggregate_prompt_score(

prompt_version=summarization_prompt.current().version,

prompt_text=summarization_prompt.current().prompt,

model_name=summarization_prompt.current().model,

average_score=average_score,

total_score=total_score,

)

update_best_candidate(

average_score=average_score,

total_score=total_score,

prompt_text=summarization_prompt.current().prompt,

model_name=summarization_prompt.current().model,

summary_text=summary,

metadata={

"section": section_number,

"average_score": average_score,

"grader_results": grader_scores,

"prompt_version": summarization_prompt.current().version,

},

lenient_passed=lenient_passed,

prompt_version=summarization_prompt.current().version,

)

if lenient_passed:

summary_passed = True

print(

f"\tPassed with prompt v{summarization_prompt.current().version} (model={summarization_prompt.current().model})"

)

break

print("\tFailed eval. Improving prompt...")

eval_feedback = collect_grader_feedback(grader_scores)

new_agent = await compare_model_candidates(

summarization_prompt=summarization_prompt,

eval_feedback=eval_feedback,

section=section,

summary=summary,

model_candidates could be given as an argument if you want to expand options.

)

if new_agent is None:

print(

"\tNo passing model found. Optimization failed for this section."

)

summary_passed = False

else:

summarization_agent = new_agent

summary_passed = True

print(

f"\tPrompt improved → v{summarization_prompt.current().version} "

f"(model={summarization_prompt.current().model})"

)

break

if not summary_passed:

print(

"\tAll attempts failed; keeping latest prompt version "

f"v{summarization_prompt.current().version} (model={summarization_prompt.current().model}) for the next section."

)

summarization_agent = apply_best_candidate_if_needed()

print("" + "-" * 80)

print("Completed optimization loop.")

print(f"Final prompt version: v{summarization_prompt.current().version}")

print(f"Final model: {summarization_prompt.current().model}")

aggregate_best = select_best_aggregate_prompt()

if best_candidate["score"] > float("-inf"):

print(

f"Best lenient prompt: v{best_candidate.get('version')} (avg={best_candidate['score']:.3f}, model={best_candidate.get('model', 'unknown')})"

)

if aggregate_best:

per_section = (

aggregate_best.get("total_average", 0.0) / aggregate_best.get("count", 1)

if aggregate_best.get("count")

else 0.0

)

print(

f"Aggregate best prompt: v{aggregate_best.get('version')} "

f"(total={aggregate_best.get('total_score', 0.0):.3f}, avg/section={per_section:.3f}, model={aggregate_best.get('model', 'unknown')})"

)

print(f"Final prompt: {summarization_prompt.current().prompt}")

print(f"Final model: {summarization_prompt.current().model}")

return summarization_agent

summarization_agent = await self_evolving_loop_with_model_comparison(summarization_agent)


在这里，我们可以看到非常相似的输出，但额外包含了模型版本得分的信息。

### b. 使用遗传-帕累托（GEPA）进行提示优化

我们已经证明了自我进化循环是可行的，并且可以使用评估（Evals）自主改进提示。然而，我们依赖的是一个相对简单、静态的元提示来改进我们的系统提示。在本节中，我们通过使用遗传-帕累托（GEPA）[[1]](##Citations)探索一种更动态、更反身的方法——该框架对智能体轨迹进行采样，以自然语言反思它们，提出提示修订，并通过迭代反馈循环进化系统。

GEPA方法在[此处](https://doi.org/10.48550/arXiv.2507.19457)提供的论文中有所描述，为持续、自我改进的提示优化提供了一个引人注目的蓝图。下面的代码大量借鉴了[此处](https://github.com/gepa-ai/gepa)提供的GEPA GitHub仓库。

import pandas as pd

import gepa

from gepa import EvaluationBatch

Extract sections from dataset

def read_csv_content(file_path: str) -> list[dict]:

"""Read csv and return section to summarize."""

df = pd.read_csv(file_path)

return [{'content': content} for content in df['content'].tolist()]

Split dataset into training and validation sets

trainset = read_csv_content("data/dataset.csv")

val_cut = max(1, int(0.1 * len(trainset)))

valset = trainset[:val_cut] if len(trainset) > 1 else trainset


我们将复用评分器和辅助函数，只需添加一个小适配器，使我们的设置能与 GEPA 配合使用。GEPA 的 `GEPAAdapter` 让我们可以轻松地将其接入评估框架。我们定义了三个钩子：
- `evaluate`：运行摘要生成，并使用上一节定义的评分器（即 chemical_name_grader、word_length_deviation_grader、cosine_similarity、llm_as_judge）进行评分。
- `get_components_to_update`：获取 GEPA 应进化的文本字段（此处为 system_prompt）。
- `make_reflective_dataset`：打包输入、输出和反馈，用于反思。

class EvalsBackedSummarizationAdapter:

"""

Minimal adapter for GEPA:

def __init__(self, client, eval_id: str, gen_model: str = "gpt-5", user_prefix: str | None = None):

self.client = client

self.eval_id = eval_id

self.gen_model = gen_model

self.user_prefix = user_prefix or "Summarize:\n\n"

Same summarization agent as in the previous section

def _summarize(self, system_prompt: str, section: str) -> str:

resp = self.client.chat.completions.create(

model=self.gen_model,

messages=[

{"role": "system", "content": system_prompt},

{"role": "user", "content": f"{self.user_prefix}{section}"},

],

)

return resp.choices[0].message.content.strip()

Required by GEPA: run eval minibatch

def evaluate(self, inputs: list[dict], candidate: dict, capture_traces: bool = True) -> EvaluationBatch:

system_prompt = candidate["system_prompt"]

scores: list[float] = []

outputs: list[str] = []

trajectories: list[dict] = []

for item in inputs:

section = item["content"]

1) Generate with the candidate prompt

summary = self._summarize(system_prompt, section)

outputs.append(summary)

2) Grade using previous evals pipeline

run = run_eval(eval_id=self.eval_id, section=section, summary=summary)

out_items = poll_eval_run(eval_id=self.eval_id, run_id=run.id)

grader_scores = parse_eval_run_output(out_items)

3) Score + actionable feedback

scalar = calculate_grader_score(grader_scores)

feedback = collect_grader_feedback(grader_scores) or "All graders passed; keep precision and coverage."

scores.append(float(scalar))

trajectories.append(

{

"inputs": {"section": section},

"generated_output": summary,

"metrics": {

"combined": float(scalar),

"by_grader": grader_scores, # keeping for analysis if needed

},

"feedback": feedback,

}

)

return EvaluationBatch(scores=scores, outputs=outputs, trajectories=trajectories)

Required by GEPA: text field to evolve

def get_components_to_update(self, candidate: dict) -> list[str]:

return ["system_prompt"]

Required by GEPA: build the reflective dataset the reflection LM will read

def make_reflective_dataset(self, candidate: dict, eval_batch: EvaluationBatch, components_to_update: list[str]) -> dict:

examples = []

for traj in (eval_batch.trajectories or []):

examples.append(

{

"Inputs": {"section": traj["inputs"]["section"]},

"Generated Outputs": traj["generated_output"],

"Feedback": traj["feedback"],

}

)

return {"system_prompt": examples}


适配器就绪后，我们可以使用与之前自进化循环相同的起始提示（`"You are a summarization assistant. Given a section of text, produce a summary."`）和模型（此处为 `gpt-5`）运行 GEPA，以便进行比较。我们将适配器实例、种子候选以及训练/验证集提供给 `gepa.optimize(...)`。在优化过程中，GEPA 会反复调用适配器对候选进行评分，反思反馈，并最终生成进化后的最佳提示。

_注意：GEPA 可能需要大约 10-15 分钟才能完成。_

seed_candidate = {"system_prompt": "You are a summarization assistant. Given a section of text, produce a summary."}

adapter = EvalsBackedSummarizationAdapter(

client=client,

eval_id=EVAL_ID,

gen_model=summarization_prompt.current().model,

)

Keeping max_metric_calls small for the cookbook.

In practice, use a larger value to allow more optimization iterations.

result = gepa.optimize(

seed_candidate=seed_candidate,

trainset=trainset,

valset=valset,

adapter=adapter,

reflection_lm="gpt-5",

max_metric_calls=10,

track_best_outputs=True,

display_progress_bar=True

)

best_prompt = result.best_candidate["system_prompt"]

print("\n=== Best evolved instruction ===\n")

print(best_prompt)


以下是上述代码的输出示例（节选）：

在本教程中，我们探讨了提示优化的三种不同方法：

- **OpenAI 平台优化器：** 使用包含手动输入的人类反馈（点赞/点踩和文本评论）的数据集，点击 _Optimize_ 按钮，我们以最少的配置快速生成了一个强大的提示。这种方法擅长快速迭代，但无法提供生产环境所需的自动化。

- **使用静态元提示的优化：** 我们的循环结合了四个不同的评分器，实现了自动探索和迭代自我改进，无需人工干预。然而，其探索空间受限于单个静态元提示，并且评估是逐节进行的。因此，这种方法可能过度拟合即时评分器的反馈，而无法实现更广泛的泛化。

- **GEPA 优化：** 提供更结构化的搜索过程，反思性更新基于定量评分和文本反馈，同时候选提示在一个数据集上训练，在另一个数据集上验证。这种方法产生了更稳健、更通用的提示，并提供了更清晰的性能实证证据。

_注意：每种方法生成的提示示例见附录。_

根据您的使用场景，您可能优先考虑速度（OpenAI 优化器）、轻量级自动化（静态元提示）或系统性泛化（GEPA）。在实践中，将这些方法结合起来，从快速迭代开始，逐步走向反思性优化，可以同时实现敏捷性和性能。

祝编码愉快！

## 贡献者

本手册由 [Bain](https://www.bain.com) 与 [OpenAI](https://openai.com) 联合编写。

[Calvin Maguranis](https://www.linkedin.com/in/calvin-maguranis-b9956045/)
[Fanny Perraudeau](https://www.linkedin.com/in/fanny-sabran-perraudeau-494b7573/)
[Giorgio Saladino](https://www.linkedin.com/in/giorgio-saladino-202/)
[Shikhar Kwatra](https://www.linkedin.com/in/shikharkwatra/)
[Valentina Frenkel](https://www.linkedin.com/in/valentina-frenkel/)

## 引用

[1] _GEPA: 反思式提示进化可超越强化学习_ 作者：Lakshya A Agrawal, Shangyin Tan, Dilara Soylu, Noah Ziems, Rishi Khare, Krista Opsahl-Ong, Arnav Singhvi, Herumb Shandilya, Michael J Ryan, Meng Jiang, Christopher Potts, Koushik Sen, Alexandros G. Dimakis, Ion Stoica, Dan Klein, Matei Zaharia, Omar Khattab - https://arxiv.org/abs/2507.19457

## 附录

### 输出提示示例：

- **初始提示：**

You are a summarization assistant. Given a section of text, produce a summary.


- **OpenAI 平台优化器：**

You are a summarization assistant.

Task: Summarize the provided text concisely and accurately.

Output requirements:

1. Complete and Exact Information Inclusion

2. Precise Terminology and Named Entity Retention

3. All Procedural Details and Justifications

4. Regulatory and Compliance References

5. Explicit Statements of Absence, Limitations, and Applicability

6. Structural and Organizational Fidelity

7. No Paraphrasing, Summarizing, or Reinterpretation

Summary Output Objective:

Produce a summary that delivers the full technical, factual, and regulatory content and structure of the original text, reformatted by eliminating only redundant language. The summary must enable audit, regulatory review, or peer reference without loss of any explicit information or terminology from the source.

*Apply these instructions rigorously to every provided document section to ensure scientific and regulatory accuracy and completeness.*


- **GEPA 优化器**：

You are a domain-aware summarization assistant for technical pharmaceutical texts. Given a “section” of text, produce a concise, single-paragraph summary that preserves key technical facts and exact nomenclature.

Length and format

Exact names and notation

Content prioritization (if space is tight)

1) What the section is about (topic/purpose).

2) All named chemical entities and compositions (list all chemical names at least once; include concentrations/amounts if given).

3) Critical process/handling facts (e.g., aseptic processing vs terminal sterilization; ISO classifications; filtration specs; compounding/filling steps; temperatures/times/volumes; storage/administration limits).

4) Container/packaging specifics (e.g., cryovials, “sterile fluid path”).

5) Microbiological/testing/regulatory details (e.g., sterility/pyrogenicity testing timing; USP <797>; state board compliance; site/manufacturer if stated).

6) Overages/single-dose formulas and key quantities.

Numerical fidelity

Style and compression

Common domain cues to include when present

Self-check before finalizing

继续系统学习

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

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

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