AI Agent 系统中的常用 Workflow 模式(2) Evaluator-Optimizer模式

Agent 中的 Evaluator-Optimizer 模式

在大语言模型（LLM）智能体的开发中，如何确保生成的结果质量高、符合要求，是一个核心挑战。本文将结合 Anthropic 团队的研究经验，深入分析 evaluator_optimizer 模式的实现与应用，为开发者提供构建高效智能体的实用指南。

什么是 Evaluator-Optimizer 模式

Evaluator-Optimizer 模式是一种工作流设计，通过"生成-评估-优化"的循环过程，不断提升 LLM 输出的质量。根据 Anthropic 的定义，这种模式属于工作流类型的智能体系统，通过预定义的代码路径编排 LLM 和工具的交互。

核心实现分析

我们来看一下 evaluator_optimizer 目录中的代码实现：

1. 核心工作流程

def loop(
    task: str, evaluator_prompt: str, generator_prompt: str, max_attempts: int = 10
) -> tuple[str, str]:
    """
    循环评估-优化
    :param task: 任务描述
    :param evaluator_prompt: 评估提示词
    :param generator_prompt: 生成提示词
    :param max_attempts: 最大尝试次数
    :return: 最终响应
    """
    memory = []
    chain_of_thought = []

    thoughts, result = generate(generator_prompt, task)
    memory.append(result)
    chain_of_thought.append(
        {
            "thoughts": thoughts,
            "result": result,
        }
    )
    attempt = 0
    while True:
        evaluation, feedback = evaluate(evaluator_prompt, result, task)
        if evaluation == "PASS":
            return result, chain_of_thought
        attempt += 1
        if attempt >= max_attempts:
            return result, chain_of_thought
        # context 只包含结果和上一次的反馈，不保留思考过程
        context = "n".join(
            [
                "Previous attemps:",
                *[f"- {m}" for m in memory],
                f"nFeedback: {feedback}",
            ]
        )
        thoughts, result = generate(generator_prompt, task, context)
        memory.append(result)
        chain_of_thought.append(
            {
                "thoughts": thoughts,
                "result": result,
            }
        )

2. 生成与评估函数

def generate(prompt: str, task: str, context: str = "") -> tuple[str, str]:
    """
    生成响应
    :param prompt: 提示词
    :param task: 任务描述
    :param context: 上下文
    :return: 响应
    """
    full_prompt = (
        f"{prompt}n{context}nTask: {task}" if context else f"{prompt}nTask: {task}"
    )
    response = llm_call(full_prompt)
    thoughts = extract_xml(response, "thoughts")
    result = extract_xml(response, "result")
    print("n开始生成响应n")
    print(f"提示词:n {full_prompt} n")
    print(f"思考:n {thoughts}n")
    print(f"结果:n {result}n")
    print("n响应生成完成n")
    return thoughts, result


def evaluate(prompt: str, content: str, task: str) -> tuple[str, str]:
    """
    评估响应
    :param prompt: 提示词
    :param content: 响应内容
    :param task: 任务描述
    :return: 评估结果
    """
    full_prompt = f"{prompt}nOriginal task: {task}nContent to evaluate: {content}"
    response = llm_call(full_prompt)
    evaluation = extract_xml(response, "evaluation")
    feedback = extract_xml(response, "feedback")
    print("n开始评估响应n")
    print(f"提示词:n {full_prompt} n")
    print(f"评估:n {evaluation}n")
    print(f"反馈:n {feedback}n")
    print("n评估完成n")
    return evaluation, feedback

3. 辅助函数

def llm_call(
    prompt: str, system_prompt: str = "", model="qwen-plus", debug=False
) -> str:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt},
    ]
    if debug:
        print(messages)
    reponse = dashscope.Generation.call(
        api_key="************",
        model=model,
        messages=messages,
        temperature=0.8,
        top_p=0.8,
        enable_thinking=False,
        stream=False,
        incremental_output=True,
        response_format={"type": "text"},
        result_format="message",  # or text
        enable_search=False,
        max_tokens=4096,
    )
    return reponse.output.choices[0].message.content


def extract_xml(text: str, tag: str) -> str:
    """
    Extracts the content of the specified XML tag from the given text. Used for parsing structured responses

    Args:
        text (str): The text containing the XML.
        tag (str): The XML tag to extract content from.

    Returns:
        str: The content of the specified XML tag, or an empty string if the tag is not found.
    """
    match = re.search(f"<{tag}>(.*?)</{tag}>", text, re.DOTALL)
    return match.group(1) if match else ""

工作原理与优势

工作原理

1. 初始化：系统首先生成一个初始响应
2. 评估：评估器对生成的响应进行评估，给出评估结果和改进建议
3. 优化：生成器根据评估反馈，生成改进后的响应
4. 循环：重复评估-优化过程，直到获得满意的结果或达到最大尝试次数

核心优势

1. 结构化反馈：通过 XML 标签提取结构化的评估结果和反馈
2. 记忆机制：保存之前的尝试结果，为后续优化提供参考
3. 可控性：通过预定义的评估标准和优化流程，确保结果质量
4. 透明度：详细的打印信息，使整个过程可观察、可调试

应用场景

根据 Anthropic 的研究，Evaluator-Optimizer 模式特别适合以下场景：

1. 代码生成与优化：如示例中的栈实现，通过评估-优化循环，不断改进代码质量
2. 内容创作：如文章、报告等，通过评估反馈提升内容质量
3. 问题求解：复杂问题的逐步优化解决方案
4. 决策制定：通过多轮评估，做出更合理的决策

代码优化建议

在分析代码实现后，我发现几个可以优化的点：

1. 错误处理：添加对 LLM 响应格式错误的处理
2. 并行处理：对于多个任务，可以使用 ThreadPoolExecutor 并行处理
3. 评估标准可配置：将评估标准参数化，使其更灵活
4. 自动停止条件：除了人工干预和最大尝试次数外，可以添加基于评估分数的自动停止条件

实际应用示例

让我们看一下示例中的任务：实现一个具有 push、pop 和 getMin 方法的栈，所有操作都应为 O(1) 时间复杂度。

初始生成：可能生成一个基本实现，但 getMin 操作不是 O(1) 评估：评估器指出 getMin 操作的时间复杂度问题 优化：生成器根据反馈，使用辅助栈实现 O(1) 的 getMin 操作 最终结果：获得一个符合要求的栈实现

总结

Evaluator-Optimizer 模式是构建高效智能体的重要工具，它通过结构化的评估-优化循环，不断提升 LLM 输出的质量。正如 Anthropic 所强调的，最成功的智能体系统往往采用简单可组合的模式，而不是复杂的框架。

这种模式的核心价值在于：

1. 质量保证：通过多轮评估确保输出质量
2. 过程透明：整个优化过程可观察、可调试
3. 适应性强：可应用于多种任务类型
4. 易于实现：代码结构清晰，易于理解和扩展

通过合理应用 Evaluator-Optimizer 模式，开发者可以构建出更可靠、更高效的智能体系统，为各种应用场景提供高质量的解决方案。