LangGraph 系列 11：使用评估器优化器循环改进模型输出

本篇继续使用预先定义的图结构，解决一类需要反复改进输出的问题：模型先生成内容，评估器按照明确标准检查；如果没有通过，就把具体反馈交回生成器继续修改。

这种模式叫作 Evaluator-optimizer Workflow，也就是评估器优化器工作流。它的重点不是简单地多调用几次模型，而是让每次重试都有上一轮的评估依据：

生成内容 -> 评估内容 -> 通过则结束 -> 不通过则携带反馈重新生成

本文使用中文冷笑话作为案例。我们先让本地 Qwen3 负责生成、用固定评估规则验证循环和停止条件，再验证 Qwen3 的结构化输出，最后把生成器与评估器都交给模型。

1. 认识评估器优化器 Workflow

普通模型调用只生成一次，结果是否满足要求通常由调用方在工作流之外判断。评估器优化器把“生成、检查和修改”都放进工作流：

1. 生成器根据输入生成内容。
2. 评估器按照预先定义的标准检查内容。
3. 通过时结束工作流。
4. 不通过时返回具体反馈。
5. 生成器读取反馈并重新生成。
6. 达到最大次数仍未通过时停止。

这种模式适合有明确验收标准、允许反复修改的任务，例如标题改写、格式检查、内容完整性检查和代码审查建议。如果验收标准无法说清楚，评估器也很难稳定判断；如果一次生成已经足够，额外评估只会增加延迟和成本。

它与普通模型调用、Agent 的主要区别，是下一步由谁决定：

方式	执行路径	下一步由谁决定	适用场景
普通模型调用	输入 → 模型 → 输出	没有后续步骤	一次生成即可完成
评估器优化器	生成 → 评估 → 接受或重试	程序预先定义的路由规则	标准明确、允许修改
Agent	模型可选择工具和后续动作	模型结合上下文动态决定	路径难以预先确定

本文实现的是 Workflow：虽然图中存在循环，但节点、边和停止条件都由代码提前定义。这与 LangGraph Evaluator-optimizer Workflow 的模式一致。

这里的“评估器”是工作流中的在线评估节点。每次运行时，它只检查本轮刚生成的内容，是否符合要求。

2. 设计循环、State 与停止条件

2.1 节点与执行路径

完整流程由三个业务节点和一个路由函数组成：

• generate_joke：生成笑话；重试时读取上一轮反馈。
• evaluate_joke：按照固定标准返回等级和反馈。
• route_after_evaluation：决定接受、重试还是因达到上限而停止。
• finish：把真实停止原因写入 State。

执行路径如下：

2.2 State 保存什么

冷笑话工作流需要保存输入、本轮结果、评估反馈和停止信息：

字段	作用
topic	用户指定的主题
joke	生成器本轮返回的笑话字符串
feedback	评估器的判断理由或修改建议
grade	固定为 funny 或 not funny
attempts	已经生成的次数
stop_reason	最终因为通过还是达到上限而停止

这些字段使用默认覆盖行为即可。新一轮笑话应该替换旧笑话，新一轮反馈也应该替换旧反馈，因此不需要列表 Reducer。节点只返回自己要修改的字段，LangGraph 会把局部更新合并进 State。

2.3 两种次数限制不要混淆

案例同时使用两类限制：

• 业务次数上限：MAX_ATTEMPTS 或 State 中的 max_attempts，表示最多允许生成多少次。达到上限属于正常业务结果，停止原因为 max_attempts。
• 图递归限制：调用时传入的 recursion_limit，用于防止错误路由造成无限循环。触发时会抛出异常，它不是正常的业务分支。

因此，业务代码必须自己判断生成次数，不能只依赖 recursion_limit。路由判断也要先检查是否已经通过，再检查次数上限：最后一次生成如果通过，应该得到 accepted，而不是 max_attempts。

3. 测试评估器

第一个案例已经使用本地 Qwen3 生成笑话，但暂时不让模型负责评估。固定评估规则只根据生成次数返回结果：

• 第一次生成后，评估器固定返回不通过，并写入修改反馈。
• 如果允许重试，Qwen3 会读取反馈重新生成，第二次评估固定返回通过。
• 如果最多只生成一次，工作流直接按次数上限结束。

这种设计能够确认真实模型是否收到了反馈，同时让反馈重试、成功结束和次数上限三个核心行为保持稳定。此脚本需要本地 Qwen3 服务，请先按照第 4.2 节启动并检查服务。

下面是 01_deterministic_evaluator_loop.py 的完整代码：

"""使用本地 Qwen3 生成笑话，并用固定评估规则演示循环和停止条件。

生成内容来自真实模型，评估结果仍由固定规则产生，方便稳定观察反馈重试。
"""
import httpx
from typing import Literal, NotRequired, TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph

# 生成节点调用本地 MLX-LM 提供的 OpenAI 兼容接口。
generator_model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0.7,
    max_tokens=256,
    # 禁止读取系统代理，确保请求直接发送到本机 Qwen3。
    http_client=httpx.Client(trust_env=False),
)

class JokeState(TypedDict):
    """定义冷笑话工作流在各节点之间传递的状态。"""

    # 调用工作流时需要提供的输入。
    topic: str
    max_attempts: int

    # 以下字段由各个 Node 在运行过程中逐步写入。
    joke: NotRequired[str]
    feedback: NotRequired[str]
    grade: NotRequired[Literal["funny", "not funny"]]
    attempts: NotRequired[int]
    stop_reason: NotRequired[str]

def generate_joke(state: JokeState) -> dict:
    """调用本地 Qwen3 生成笑话，重试时参考上一轮反馈。"""

    attempts = state.get("attempts", 0) + 1
    feedback = state.get("feedback")

    # 第一次没有 feedback，只根据主题生成；重试时要求模型针对反馈改进。
    if feedback:
        instruction = f"""请围绕“{state['topic']}”重新写一个简短的中文冷笑话。
上一版没有通过，评估反馈是：{feedback}
请针对反馈改进，使用问答结构、双关或结尾反转，只输出笑话正文。"""
    else:
        instruction = f"""请围绕“{state['topic']}”写一个简短的中文冷笑话。
笑话需要简洁、容易理解，只输出笑话正文。"""

    response = generator_model.invoke(
        "你是一名擅长写简短中文冷笑话的创作者。\n\n" + instruction
    )

    # 模型返回 AIMessage，State 中只保存它的文本内容。
    return {"joke": str(response.content).strip(), "attempts": attempts}


def evaluate_joke(state: JokeState) -> dict:
    """用固定规则模拟评估器，方便稳定观察循环行为。"""

    if state["attempts"] >= 2:
        return {"grade": "funny", "feedback": "笑话包含问答结构和反转，可以通过。"}

    return {
        "grade": "not funny",
        "feedback": "这只是一句陈述，请增加问答结构和结尾反转。",
    }


def route_after_evaluation(
    state: JokeState,
) -> Literal["accepted", "retry", "max_attempts"]:
    """根据评估结果和生成次数决定结束还是重试。"""

    # 判断顺序很重要：先检查是否通过，再检查是否达到次数上限。
    if state["grade"] == "funny":
        return "accepted"
    if state["attempts"] >= state["max_attempts"]:
        return "max_attempts"
    return "retry"


def finish(state: JokeState) -> dict:
    """写入稳定的停止原因，便于调用方判断工作流如何结束。"""

    if state["grade"] == "funny":
        stop_reason = "accepted"
    else:
        stop_reason = "max_attempts"
    return {"stop_reason": stop_reason}


# 第 1 步：创建图并注册三个 Node。
builder = StateGraph(JokeState)
builder.add_node("generate_joke", generate_joke)
builder.add_node("evaluate_joke", evaluate_joke)
builder.add_node("finish", finish)

# 第 2 步：添加固定 Edge 和条件 Edge。
builder.add_edge(START, "generate_joke")
builder.add_edge("generate_joke", "evaluate_joke")
builder.add_conditional_edges(
    "evaluate_joke",
    route_after_evaluation,
    # 路由函数返回业务值，path_map 再把业务值映射到真实 Node。
    {
        "accepted": "finish",
        "retry": "generate_joke",
        "max_attempts": "finish",
    },
)
builder.add_edge("finish", END)

# 第 3 步：编译后才得到可以 invoke() 的工作流。
workflow = builder.compile()


def run_case(max_attempts: int) -> None:
    """运行一个案例，并打印最终 State。"""

    print(f"\n最大生成次数：{max_attempts}")
    result = workflow.invoke(
        {
            "topic": "程序员",
            "max_attempts": max_attempts,
        }
    )

    print("笑话：", result["joke"])
    print("生成次数：", result["attempts"])
    print("评估等级：", result["grade"])
    print("停止原因：", result["stop_reason"])


if __name__ == "__main__":
    # 第一个案例允许重试，因此第二次生成后通过评估。
    run_case(max_attempts=3)
    # 第二个案例只允许生成一次，因此未通过时直接按次数上限结束。
    run_case(max_attempts=1)

运行：

cd /Users/bianhn/git/llm-learning
source .venv_langgraph/bin/activate
python langgraph/p11_evaluator_optimizer/01_deterministic_evaluator_loop.py

笑话正文由模型生成，因此每次运行可能不同。一次实际输出如下：

最大生成次数：3
笑话： 为什么程序员喜欢用黑暗模式？
因为他们在“亮”着的时候，总觉得“黑”得不够彻底。
生成次数： 2
评估等级： funny
停止原因： accepted

最大生成次数：1
笑话： 我问程序员：“你有没有女朋友？”
他说：“有啊，她是我的 Bug，不过我还没找到她。”
生成次数： 1
评估等级： not funny
停止原因： max_attempts

两个案例使用同一张编译后的图，只改变输入中的 max_attempts。虽然笑话内容不固定，但第一个案例会在第二次生成后得到 accepted，第二个案例会在第一次生成后得到 max_attempts。这说明停止条件属于业务状态的一部分，可以由调用方按任务设置。

4. Qwen3 验证评估器

确定图的循环没有问题后，再单独验证模型是否能稳定返回评估结果。这里的顺序很重要：先定义返回结构，再启动并检查模型服务，最后运行固定笑话测试。

4.1 定义 Feedback

评估节点不应该返回一段难以解析的自然语言，而应该返回固定结构：

class Feedback(BaseModel):
    grade: Literal["funny", "not funny"]
    feedback: str

Literal 把等级限制为两个合法值，Pydantic 负责校验结果类型。路由函数因此可以直接判断 grade，不需要猜测模型写的是“通过”“还不错”还是其他表达。

4.2 启动并检查 Qwen3

打开一个终端，在项目根目录启动 MLX-LM 的 OpenAI 兼容服务：

cd /Users/bianhn/git/llm-learning
source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model Qwen3-14B-AWQ-4bit-MLX \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

服务启动后先检查模型列表：

curl http://127.0.0.1:18080/v1/models

只有这个请求能够返回模型信息，才继续运行 Python 脚本。

4.3 运行固定笑话测试

with_structured_output(Feedback, method="function_calling") 会要求模型按照 Feedback 的字段返回结果，并把响应解析成 Pydantic 对象。下面是 02_test_structured_evaluator.py 的完整代码：

"""使用本地 Qwen3 评估一条固定冷笑话，验证 Pydantic 结构化输出。"""

from typing import Literal

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


MODEL_NAME = "Qwen3-14B-AWQ-4bit-MLX"
BASE_URL = "http://127.0.0.1:18080/v1"

class Feedback(BaseModel):
    """约束评估器只能返回固定等级和具体反馈。"""

    grade: Literal["funny", "not funny"] = Field(
        description="笑话是否达到基本的冷笑话标准"
    )
    feedback: str = Field(description="判断理由；不通过时给出具体修改建议")

# ChatOpenAI 可以调用 MLX-LM 提供的 OpenAI 兼容接口。
model = ChatOpenAI(
    model=MODEL_NAME,
    base_url=BASE_URL,
    api_key="not-needed",
    temperature=0,
    max_tokens=256,
)

# function_calling 让模型按照 Feedback 的字段返回结果。
# 返回值会经过 Pydantic 校验，不再是普通字符串。
evaluator = model.with_structured_output(Feedback, method="function_calling")

def main() -> None:
    """把固定笑话交给结构化评估器并打印结果。"""

    result = evaluator.invoke(
        """请评估下面的中文冷笑话。

评估标准：
1. 内容与主题相关；
2. 有清晰的双关、误解或结尾反转；
3. 表达完整，读者能够理解笑点。

笑话：程序员最讨厌哪一种声音？需求变更的提示音。

如果不通过，请给出一条可以直接执行的修改建议。"""
    )

    # result 已经是 Feedback 对象，可以直接通过属性读取字段。
    print("返回类型：", type(result).__name__)
    print("评估等级：", result.grade)
    print("评估反馈：", result.feedback)


if __name__ == "__main__":
    main()

运行：

source .venv_langgraph/bin/activate
python langgraph/p11_evaluator_optimizer/02_test_structured_evaluator.py

一次实际输出如下：

返回类型： Feedback
评估等级： funny
评估反馈： 该笑话符合所有评估标准：内容与程序员相关，有清晰的结尾反转，表达完整且能够理解笑点。

反馈文字可能变化，但 result 应该是 Feedback 对象，grade 只能是 funny 或 not funny。如果这一步失败，应先检查模型服务和结构化输出兼容性，不要立即调试 LangGraph 循环。

5. 构建完整评估器优化器

现在把真实生成器、结构化评估器和条件路由组合起来。

5.1 生成与反馈重试

generate_joke 每执行一次，就把 attempts 加一。第一次没有反馈，提示词只包含主题；重试时把上一轮 feedback 加入提示词，并明确要求针对建议修改。

模型返回的是 AIMessage，但 State 中的 joke 字段设计为字符串，因此节点只保存 response.content。保持字段类型稳定，可以让后续评估节点和输出代码更简单。

5.2 结构化评估

evaluate_joke 把主题、当前笑话和三项标准交给结构化评估器。返回的 Feedback 对象不会直接写入 State，而是拆成 grade 和 feedback 两个字段：

return {"grade": result.grade, "feedback": result.feedback}

路由函数只读取合并后的 State，不修改 State。

5.3 路由与停止节点

条件路由的判断顺序是：

1. grade == "funny" 时返回 accepted。
2. 未通过且 attempts >= MAX_ATTEMPTS 时返回 max_attempts。
3. 其余情况返回 retry。

accepted 和 max_attempts 都进入 finish，由该节点写入稳定的 stop_reason。调用方不需要根据最后一条反馈猜测工作流为何结束。

5.4 完整代码

下面是 03_qwen3_evaluator_optimizer_workflow.py 的完整代码：

"""使用本地 Qwen3 完成冷笑话的生成、评估、反馈重试和停止判断。"""

from typing import Literal, NotRequired, TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field

MODEL_NAME = "Qwen3-14B-AWQ-4bit-MLX"
BASE_URL = "http://127.0.0.1:18080/v1"
MAX_ATTEMPTS = 3

class JokeState(TypedDict):
    """定义评估器优化器工作流中的共享状态。"""

    # 用户输入。
    topic: str

    # 工作流运行过程中产生的字段。
    joke: NotRequired[str]
    feedback: NotRequired[str]
    grade: NotRequired[Literal["funny", "not funny"]]
    attempts: NotRequired[int]
    stop_reason: NotRequired[str]


class Feedback(BaseModel):
    """约束评估节点返回的等级和反馈。"""

    grade: Literal["funny", "not funny"] = Field(
        description="笑话是否达到基本的冷笑话标准"
    )
    feedback: str = Field(description="判断理由；不通过时给出具体修改建议")


# 生成器温度较高以增加变化，评估器温度为 0 以保持判断稳定。
generator_model = ChatOpenAI(
    model=MODEL_NAME,
    base_url=BASE_URL,
    api_key="not-needed",
    temperature=0.7,
    max_tokens=256,
)

evaluator_model = ChatOpenAI(
    model=MODEL_NAME,
    base_url=BASE_URL,
    api_key="not-needed",
    temperature=0,
    max_tokens=256,
)

# 评估器必须返回符合 Feedback Schema 的结构化结果。
structured_evaluator = evaluator_model.with_structured_output(
    Feedback,
    method="function_calling",
)

def generate_joke(state: JokeState) -> dict:
    """根据主题生成笑话；重试时必须参考上一次评估反馈。"""

    attempts = state.get("attempts", 0) + 1
    feedback = state.get("feedback")

    # 第一次生成没有反馈；重试时把上一轮反馈加入提示词。
    if feedback:
        instruction = f"""请围绕“{state['topic']}”重新写一个简短的中文冷笑话。
上一版没有通过，评估反馈是：{feedback}
请针对反馈改进，使用问答结构、双关或结尾反转，只输出笑话正文。"""
    else:
        instruction = f"""请围绕“{state['topic']}”写一个简短的中文冷笑话。
笑话需要有问答结构、双关或结尾反转，只输出笑话正文。"""

    # 传入普通字符串即可获得 AIMessage，最终只把 content 写入 State。
    response = generator_model.invoke(
        "你是一名擅长写简短中文冷笑话的创作者。\n\n" + instruction
    )

    # State 中的 joke 字段保存字符串，不直接保存 AIMessage 对象。
    return {"joke": str(response.content).strip(), "attempts": attempts}


def evaluate_joke(state: JokeState) -> dict:
    """按照固定标准评估当前笑话，并返回 Pydantic 校验后的结果。"""

    result = structured_evaluator.invoke(
        f"""请评估下面的中文冷笑话。

主题：{state['topic']}
笑话：{state['joke']}

评估标准：
1. 内容与主题相关；
2. 有清晰的双关、误解或结尾反转；
3. 表达完整，读者能够理解笑点。

达到三项标准时 grade 返回 funny；否则返回 not funny，并给出一条可以直接执行的修改建议。"""
    )

    return {"grade": result.grade, "feedback": result.feedback}


def route_after_evaluation(
    state: JokeState,
) -> Literal["accepted", "retry", "max_attempts"]:
    """评估通过则结束，否则在次数范围内返回生成节点。"""

    if state["grade"] == "funny":
        return "accepted"
    if state["attempts"] >= MAX_ATTEMPTS:
        return "max_attempts"
    return "retry"


def finish(state: JokeState) -> dict:
    """记录工作流的真实停止原因。"""

    if state["grade"] == "funny":
        stop_reason = "accepted"
    else:
        stop_reason = "max_attempts"
    return {"stop_reason": stop_reason}


# 第 1 步：创建图并注册生成、评估和结束 Node。
builder = StateGraph(JokeState)
builder.add_node("generate_joke", generate_joke)
builder.add_node("evaluate_joke", evaluate_joke)
builder.add_node("finish", finish)

# 第 2 步：使用固定 Edge 和条件 Edge 组成循环。
builder.add_edge(START, "generate_joke")
builder.add_edge("generate_joke", "evaluate_joke")
builder.add_conditional_edges(
    "evaluate_joke",
    route_after_evaluation,
    {
        "accepted": "finish",
        "retry": "generate_joke",
        "max_attempts": "finish",
    },
)
builder.add_edge("finish", END)

# 第 3 步：编译得到可执行工作流。
workflow = builder.compile()


def main() -> None:
    """运行工作流，并打印 invoke() 返回的最终 State。"""

    final_state = workflow.invoke(
        {"topic": "程序员"},
        # 业务次数上限负责正常停止；recursion_limit 是防止路由错误的最后保护。
        config={"recursion_limit": 20},
    )

    print("主题：", final_state["topic"])
    print("笑话：", final_state["joke"])
    print("生成次数：", final_state["attempts"])
    print("评估等级：", final_state["grade"])
    print("评估反馈：", final_state["feedback"])
    print("停止原因：", final_state["stop_reason"])

if __name__ == "__main__":
    main()

运行：

source .venv_langgraph/bin/activate
python langgraph/p11_evaluator_optimizer/03_qwen3_evaluator_optimizer_workflow.py

主流程使用 invoke() 直接取得最终 State，不需要手工合并每个节点的局部更新。生成模型使用较高温度增加变化，评估模型使用温度 0 尽量保持判断稳定；温度 0 也不代表所有推理环境下都绝对确定。

结果:

一次实际运行在第一轮就通过了：

主题： 程序员
笑话： 为什么程序员不喜欢在厨房里工作？  
因为那里总是有很多“bug”要处理。
生成次数： 1
评估等级： funny
评估反馈： 该笑话符合所有三项标准：内容与程序员主题相关，使用了“bug”这一双关语，表达完整且笑点清晰。
停止原因： accepted

笑话文本、反馈内容和生成次数都可能变化，不应该对它们做固定断言。稳定规则只有：

• attempts 不大于 MAX_ATTEMPTS。
• grade 只能是 funny 或 not funny。
• stop_reason 只能是 accepted 或 max_attempts。

如果想观察每个节点提交了哪些字段，可以把主流程中的 invoke() 临时换成下面的流式调用：

for update in workflow.stream(
    {"topic": "程序员"},
    config={"recursion_limit": 20},
    stream_mode="updates",
):
    print(update)

stream_mode="updates" 返回节点的局部更新，形状类似 {"generate_joke": {"joke": "...", "attempts": 1}}。它适合观察循环过程，但每一项不是完整 State，也不应该在示例中手工维护第二份 State。需要最终结果时直接使用 invoke()；需要观察过程时使用 stream()。

accepted 和 max_attempts 都是可预期的业务结束方式。前者表示内容通过，后者表示工作流已在成本边界内停止，并不代表 LangGraph 执行失败。

6. 成本、局限与常见问题

6.1 调用次数与延迟

每一轮通常包含一次生成调用和一次评估调用。生成了 N 次时，模型调用次数大约是：

总调用次数 ≈ 2 × N

因此，增加 MAX_ATTEMPTS 会同时增加延迟和计算成本。生产任务应根据输出价值设置较小的业务上限，并记录每轮次数、等级、反馈和停止原因。

6.2 评估器并不等于绝对正确

评估器优化器仍有这些边界：

• 标准不明确时，同一内容可能得到不同判断。
• 生成器与评估器使用同一个模型时，可能存在自我评估偏差。
• “符合写作标准”不代表事实一定正确；事实性任务仍需检索、工具或规则校验。
• 生成器可能表面改写内容，却没有真正采纳反馈。
• 本地 OpenAI 兼容服务对结构化输出方式的支持可能不同。

更重要的任务可以组合确定性规则、独立评估模型和人工抽检，而不是把所有判断都交给一个模型。

6.3 常见问题

问题	原因与处理方式
路由无法稳定判断结果	不要解析自然语言，使用 Pydantic 和 Literal 限定结构
State 字段类型前后不一致	joke 保存 response.content 字符串，不保存整个 AIMessage
每轮生成几乎没有变化	重试提示词必须显式包含上一轮 feedback
工作流一直循环	增加业务次数上限，并保留合理的 recursion_limit
结构化输出解析失败	先单独运行第 4 章脚本，确认服务支持 function_calling
请求连接失败	先启动 Qwen3，再用 curl /v1/models 检查服务
最后一次通过却显示达到上限	路由时先判断 grade，再判断 attempts

结构化输出也可以使用其他方法，但本文选择 function_calling，因为它在当前 MLX-LM OpenAI 兼容服务上已经验证通过。更换模型服务后，应先重新运行独立测试脚本。

7. 总结

评估器优化器 Workflow 的核心可以概括为四点：

1. 生成器负责产出候选结果，评估器负责按照明确标准判断。
2. 不通过时，具体反馈通过 State 交回生成器，形成有依据的重试。
3. 条件边负责接受、重试和达到上限三种路径，finish 记录稳定的停止原因。
4. 业务次数上限负责控制成本，recursion_limit 只作为图执行的最后保护。

开发这类工作流时，可以先用真实生成器和固定评估规则验证反馈循环，再单独验证结构化评估器，最后接入完整模型循环。这样遇到问题时，可以快速判断错误来自路由、模型服务还是结构化输出。