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

langgraph structured-output workflow evaluator
langgraph
发布日期:   2026-05-28
文章字数:   6k
阅读时长:   23 分
阅读次数:  

前面已经使用 LangGraph 编写了固定路径、条件分支和并行分支。但是有一类任务不是执行一次就能结束：模型先生成内容，评估器再按照明确标准检查；如果没有通过，就把具体反馈交回生成器继续修改。

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

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

本文使用一个简单的中文冷笑话案例，把 State、结构化输出、条件路由、循环和停止条件组合起来。为了避免把图结构问题与模型问题混在一起，先使用确定性函数验证工作流，再接入本地 Qwen3。

1. 评估器优化器解决什么问题

普通模型调用只生成一次。结果是否满足要求，只能由调用方在工作流之外判断。

评估器优化器会把判断过程放进工作流：

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

评估器优化器的生成、评估、重试与停止循环

图中的 accepted、retry 和 max_attempts 是路由函数返回的业务值，不是三个节点。path_map 将两个停止结果 accepted 和 max_attempts 都映射到同一个 finish 节点，只有 retry 返回 generate_joke 继续生成。

它适合有明确验收标准、并且允许反复修改的任务，例如标题改写、格式检查、内容完整性检查和代码审查建议。

但它并不适合所有问题。如果标准无法说清楚，评估器就难以稳定判断；如果一次生成已经足够，多一次评估只会增加时间和调用成本。

这一执行模式与 LangGraph Evaluator-optimizer Workflow 的定义一致：生成器产生结果，评估器给出判断和反馈，未通过时再根据反馈重新生成。

2. 它与普通调用和 Agent 有什么区别

三种执行方式的差异主要在“下一步由谁决定”。

2.1 普通模型调用

程序提交输入，模型返回一次结果，流程随即结束：

输入 -> 模型 -> 输出

2.2 评估器优化器

程序员提前定义生成、评估、接受、重试和停止路径。模型负责生成内容和执行评估，但不会自由选择工作流之外的动作。

2.3 Agent

Agent 通常由模型根据当前消息决定下一步，例如调用哪个工具、继续思考还是返回最终答案。它比固定 Workflow 灵活，但执行路径也更难预测。

本文的流程属于 Workflow。它存在循环，但每条边和每个停止条件都由代码提前定义。

3. 本文的“评估器”指什么

这里的评估器是工作流中的一个运行时节点。用户每次执行工作流时，它都会检查本轮刚刚生成的内容。

它不是下面这些概念：

  • 不是 LangSmith 中针对数据集批量运行的离线评估。
  • 不是模型排行榜或基准测试。
  • 不是人工标注平台。

这些功能都可能使用“评估”一词，但运行位置和目标不同。本文只讨论 Workflow 内部的在线评估节点。

4. 设计工作流 State

冷笑话工作流需要保存主题、当前笑话、评估反馈和循环次数：

class JokeState(TypedDict):
    topic: str
    joke: NotRequired[str]
    feedback: NotRequired[str]
    grade: NotRequired[Literal["funny", "not funny"]]
    attempts: NotRequired[int]
    stop_reason: NotRequired[str]

各字段的作用如下：

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

Topic、Joke、Feedback 与路由结果在 State 中的流转

这些字段使用默认覆盖行为即可。新一轮笑话应该替换上一轮笑话，新一轮反馈也应该替换旧反馈，因此不需要列表 Reducer。

评估模型返回的 Feedback 对象不会直接交给路由函数。evaluate_joke 先取出其中的 grade 和 feedback，以局部更新的形式写回 State；LangGraph 合并更新后，路由函数再从 State 读取 grade 和 attempts。路由函数只选择路径，不修改 State。

5. 先用确定性函数验证循环

接入模型之前，先让生成器和评估器返回固定结果：

  • 第一次生成普通陈述，固定评估为不通过。
  • 第二次根据反馈生成问答式笑话，固定评估为通过。
  • 当最大次数为 1 时，不允许进入第二次生成。

这样可以稳定验证条件路由，而不会受模型输出变化影响。

完整代码如下：

# 本示例不调用大模型，用固定结果演示评估器优化器的循环、反馈和停止条件。

from typing import Literal, NotRequired, TypedDict

from langgraph.graph import END, START, StateGraph


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

    topic: str
    joke: NotRequired[str]
    feedback: NotRequired[str]
    grade: NotRequired[Literal["funny", "not funny"]]
    attempts: NotRequired[int]
    max_attempts: int
    stop_reason: NotRequired[str]


def generate_joke(state: JokeState) -> dict:
    """第一次生成普通句子，收到反馈后生成带有反转的冷笑话。"""

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

    if state.get("feedback"):
        joke = "程序员去海边为什么带键盘？因为他想按一下浪（浪）。"
    else:
        joke = "程序员正在工作。"

    return {"joke": joke, "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:
    """写入稳定的停止原因，便于调用方判断工作流如何结束。"""

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


builder = StateGraph(JokeState)
builder.add_node("generate_joke", generate_joke)
builder.add_node("evaluate_joke", evaluate_joke)
builder.add_node("finish", finish)

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)

workflow = builder.compile()


def run_case(max_attempts: int) -> None:
    """运行一个案例，并打印每个节点返回的局部更新。"""

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

    for update in workflow.stream(final_state, stream_mode="updates"):
        for node_name, node_update in update.items():
            final_state.update(node_update)
            print(f"{node_name}: {node_update}")

    print(
        "最终结果：",
        {
            "attempts": final_state["attempts"],
            "grade": final_state["grade"],
            "stop_reason": final_state["stop_reason"],
        },
    )


run_case(max_attempts=3)
run_case(max_attempts=1)

运行代码：

source .venv_langgraph/bin/activate
python langgraph/p11_evaluator_optimizer/01_deterministic_evaluator_loop.py

真实输出如下：

最大生成次数：3
generate_joke: {'joke': '程序员正在工作。', 'attempts': 1}
evaluate_joke: {'grade': 'not funny', 'feedback': '这只是一句陈述，请增加问答结构和结尾反转。'}
generate_joke: {'joke': '程序员去海边为什么带键盘？因为他想按一下浪（浪）。', 'attempts': 2}
evaluate_joke: {'grade': 'funny', 'feedback': '笑话包含问答结构和反转，可以通过。'}
finish: {'stop_reason': 'accepted'}
最终结果： {'attempts': 2, 'grade': 'funny', 'stop_reason': 'accepted'}

最大生成次数：1
generate_joke: {'joke': '程序员正在工作。', 'attempts': 1}
evaluate_joke: {'grade': 'not funny', 'feedback': '这只是一句陈述，请增加问答结构和结尾反转。'}
finish: {'stop_reason': 'max_attempts'}
最终结果： {'attempts': 1, 'grade': 'not funny', 'stop_reason': 'max_attempts'}

第一组案例证明反馈进入了第二轮生成。第二组案例证明达到次数上限后，工作流不会继续循环。

6. 使用 Pydantic 定义评估结果

如果让评估器返回一段自然语言，路由函数就需要自行判断“通过”“很好”“可以”等不同说法。模型措辞稍有变化，条件判断就可能失效。

更稳定的做法是定义结构化结果：

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

Literal 把等级限制为两个值，feedback 保存判断理由。路由函数只需要读取 grade，不再解析自然语言。

本地 Qwen3 通过 MLX-LM 提供 OpenAI 兼容接口。本文使用已经验证的 function_calling 方法：

evaluator = model.with_structured_output(
    Feedback,
    method="function_calling",
)

OpenAI 兼容接口只说明请求和响应格式相似，并不代表推理服务支持每一种 Structured Output 方法。本地服务不应在没有测试的情况下直接改用默认 json_schema。

7. 单独验证结构化评估器

在完整工作流中同时排查生成、评估和路由会比较困难。因此先用一条固定笑话验证评估器。

完整代码如下：

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

from typing import Literal

import httpx
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


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

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


# 禁用系统代理，确保本机请求直接发送到 MLX-LM。
http_client = httpx.Client(trust_env=False)

model = ChatOpenAI(
    model="default_model",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=256,
    http_client=http_client,
)

# 本地 MLX-LM 已验证 function_calling，可以返回符合 Feedback 的参数。
evaluator = model.with_structured_output(Feedback, method="function_calling")

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

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

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

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

print("返回类型：", type(result).__name__)
print("评估等级：", result.grade)
print("评估反馈：", result.feedback)

http_client.close()

8. 启动本地 Qwen3 服务

进入 llm_learning 根目录，在独立终端启动模型服务：

source .venv_tool_server/bin/activate

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

运行模型代码之前先检查健康接口：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

再执行结构化评估脚本：

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

本次真实输出为：

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

返回类型是经过 Pydantic 校验的 Feedback，不是普通字典，也不是 AIMessage。

9. 编写生成节点

生成节点需要区分第一次生成和后续修改：

feedback = state.get("feedback")

if feedback:
    instruction = f"请根据反馈重新生成：{feedback}"
else:
    instruction = "请根据主题第一次生成"

如果节点每一轮都只读取 topic，而完全忽略 feedback，虽然图上形成了循环，内容却没有真正被优化。

另一个容易出错的地方是返回类型。ChatOpenAI.invoke() 返回 AIMessage，但 State 中的 joke 被定义为字符串，因此节点应该保存 response.content：

return {
    "joke": str(response.content).strip(),
    "attempts": attempts,
}

10. 编写评估节点

评估节点读取主题和当前笑话，再调用结构化模型：

result = structured_evaluator.invoke(...)

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

评估标准需要尽可能具体。本文检查以下三项：

  1. 是否与主题相关。
  2. 是否有清晰的双关、误解或结尾反转。
  3. 中文表达是否完整，读者能否理解笑点。

如果只要求模型判断“好不好”，评估结果会缺少稳定依据，反馈也很难指导下一轮修改。

11. 编写条件路由

路由函数先检查是否通过，再检查最大次数：

def route_after_evaluation(state: JokeState):
    if state["grade"] == "funny":
        return "accepted"
    if state["attempts"] >= MAX_ATTEMPTS:
        return "max_attempts"
    return "retry"

三个路由值通过 path_map 映射到真实节点：

builder.add_conditional_edges(
    "evaluate_joke",
    route_after_evaluation,
    {
        "accepted": "finish",
        "retry": "generate_joke",
        "max_attempts": "finish",
    },
)

path_map 让路由含义和图中的节点名称分离。accepted 是业务结果，finish 才是真实节点。

12. 为什么必须限制最大次数

评估器可能连续拒绝生成内容。如果只有“通过才结束”这一条规则，工作流可能无限循环。

本文使用两层保护：

12.1 业务停止条件

MAX_ATTEMPTS = 3

三次生成后仍未通过，就把 stop_reason 写成 max_attempts 并结束。这是程序能够向调用方解释的业务结果。

12.2 LangGraph 递归限制

config={"recursion_limit": 20}

recursion_limit 是运行时最后一道保护。如果图因为路由错误一直循环，LangGraph 会抛出 GraphRecursionError。

它不能替代业务停止条件。递归异常只能说明图没有按预期结束，无法告诉调用方是评估未通过还是程序存在错误。

13. 完整评估器优化器 Workflow

下面把生成器、评估器、条件路由和结束节点组合起来：

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

from typing import Literal, NotRequired, TypedDict

import httpx
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field


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="判断理由；不通过时给出具体修改建议")


# 两个模型客户端访问同一个 Qwen3 服务，但使用不同温度。
generator_http_client = httpx.Client(trust_env=False)
evaluator_http_client = httpx.Client(trust_env=False)

generator_model = ChatOpenAI(
    model="default_model",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0.7,
    max_tokens=256,
    http_client=generator_http_client,
)

evaluator_model = ChatOpenAI(
    model="default_model",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=256,
    http_client=evaluator_http_client,
)

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']}”写一个简短的中文冷笑话。
笑话需要有问答结构、双关或结尾反转，只输出笑话正文。"""

    response = generator_model.invoke(
        [
            SystemMessage(content="你是一名擅长写简短中文冷笑话的创作者。"),
            HumanMessage(content=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:
    """记录工作流的真实停止原因。"""

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


builder = StateGraph(JokeState)
builder.add_node("generate_joke", generate_joke)
builder.add_node("evaluate_joke", evaluate_joke)
builder.add_node("finish", finish)

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)

workflow = builder.compile()

initial_state: JokeState = {"topic": "程序员"}
final_state = initial_state.copy()

print("节点更新：")
for update in workflow.stream(
    initial_state,
    config={"recursion_limit": 20},
    stream_mode="updates",
):
    for node_name, node_update in update.items():
        final_state.update(node_update)

        if node_name == "generate_joke":
            print(f"- 第 {node_update['attempts']} 次生成：{node_update['joke']}")
        elif node_name == "evaluate_joke":
            print(f"- 评估：{node_update['grade']}")
            print(f"  反馈：{node_update['feedback']}")
        else:
            print(f"- 停止原因：{node_update['stop_reason']}")

print("\n最终状态：")
print("主题：", final_state["topic"])
print("笑话：", final_state["joke"])
print("生成次数：", final_state["attempts"])
print("评估等级：", final_state["grade"])
print("停止原因：", final_state["stop_reason"])

generator_http_client.close()
evaluator_http_client.close()

运行代码：

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

14. 观察真实运行结果

本次运行中，生成器连续修改了三次，评估器仍然认为结尾反转不够清晰，因此工作流按次数上限停止：

节点更新：
- 第 1 次生成：为什么程序员喜欢用黑暗模式？
因为亮着的灯容易让他们“代码”出错。
- 评估：not funny
  反馈：笑话中的双关不够清晰，可以修改为：为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们‘调试’出错。
- 第 2 次生成：为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们“调试”出错。
- 评估：not funny
  反馈：笑话的结尾反转不够清晰，建议修改为：为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们‘调试’出错。
- 第 3 次生成：为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们“调试”出错。
- 评估：not funny
  反馈：笑话的结尾反转不够清晰，建议修改为：为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们‘调试’出错。
- 停止原因：max_attempts

最终状态：
主题： 程序员
笑话： 为什么程序员喜欢用黑暗模式？因为亮着的灯容易让他们“调试”出错。
生成次数： 3
评估等级： not funny
停止原因： max_attempts

这不是程序执行失败，而是一个有效的业务结果：模型确实读取了反馈并重新生成，但三次之内没有达到评估器的标准。

模型输出存在非确定性。再次运行可能第一次就通过，也可能生成不同笑话。稳定的是 State 结构、路由规则、最大次数和停止原因。

15. 使用 stream 查看每轮更新

invoke() 适合直接取得最终 State：

result = workflow.invoke(
    {"topic": "程序员"},
    config={"recursion_limit": 20},
)

本文使用：

workflow.stream(..., stream_mode="updates")

updates 每次返回节点产生的局部更新，所以可以看到：

  • 生成节点写入了哪一版笑话。
  • 评估节点返回了什么等级和反馈。
  • 工作流最终因为什么停止。

它不会自动返回一份累计 State，因此示例使用 final_state.update(node_update) 将每次局部更新合并起来，仅用于最后打印完整结果。

16. 模型调用次数与成本

每一轮通常包含两次模型调用：

  1. 生成器调用一次。
  2. 评估器调用一次。

如果生成了 N 轮，通常约有 2N 次模型调用。本文最多生成 3 次，因此最多调用模型 6 次。

实际系统还需要关注：

  • 单轮生成和评估耗时。
  • 最大重试次数。
  • 每轮笑话与最新反馈的长度。
  • 本地推理资源或在线 API 成本。

本文的 joke 和 feedback 都使用默认覆盖行为，只保留最新一轮内容，不会把历次反馈自动累积到 State。如果业务需要保存完整修改历史，应单独增加列表字段并配置合适的 Reducer。

不要为了追求“评估通过”无限增加重试次数。次数上限应该由任务价值、延迟要求和资源预算共同决定。

17. 评估器的局限

17.1 标准不明确会导致反馈不稳定

“写得更好”不是可执行标准。应尽量拆成主题相关性、格式完整性、字段齐全等可以判断的条件。

17.2 同一个模型可能存在自我评估偏差

本文为了本地演示，生成和评估都使用 Qwen3，只是温度不同。在要求更高的系统中，可以使用独立模型、规则校验或人工检查共同评估。

17.3 通过不代表事实一定正确

评估器只能按照提示词和自身能力判断。结构化输出保证了结果格式，不保证判断本身绝对正确。

17.4 优化器可能没有真正采纳反馈

本次实测中，第二次与第三次生成基本相同。这说明把反馈放进提示词只能提高模型采纳建议的可能性，不能保证每次都产生有效改进。

18. 常见问题

18.1 把 AIMessage 直接写入字符串字段

ChatOpenAI.invoke() 返回 AIMessage。如果 joke 定义为字符串，应写入 response.content，否则后续格式化提示词时会混入消息对象表示。

18.2 Pydantic 字段与模型结果不匹配

路由依赖 grade，Schema 和提示词中就必须使用同一个字段名。字段缺失或 Literal 值不合法会产生验证错误。

18.3 路由值与 path_map 不一致

路由函数返回 retry，path_map 中也必须存在 retry。拼写不同会导致图找不到下一节点。

18.4 生成节点没有读取反馈

图可以循环不代表内容会改进。重试提示词必须明确包含 state[“feedback”]。

18.5 没有最大循环次数

评估器可能永远返回 not funny。必须使用业务次数上限，并保留 recursion_limit 作为最后保护。

18.6 本地服务不支持默认 json_schema

with_structured_output() 的具体方法依赖模型服务。本文的 MLX-LM 环境使用已经实测通过的 method=”function_calling”。

18.7 本机请求被代理转发

如果 VPN 或系统代理影响 127.0.0.1 请求，可以像本文一样传入：

httpx.Client(trust_env=False)

同时先用 /v1/models 健康接口区分“服务没有启动”和“模型调用失败”。

19. 总结

知识点 作用 本文验证结果
Evaluator-optimizer 生成后评估，不通过则携带反馈重试 冷笑话最多生成三次
JokeState 保存主题、内容、反馈、等级和次数 每个节点只返回局部更新
生成节点 第一次按主题生成，后续按反馈修改 实测连续生成三版内容
评估节点 按明确标准判断当前结果 返回 Pydantic Feedback
Literal 限制评估等级的合法值 只允许 funny 和 not funny
function_calling 让本地 Qwen3 返回结构化参数 MLX-LM 实测成功
条件边 根据评估结果选择结束或重试 accepted、retry、max_attempts
path_map 将业务路由值映射到真实节点 图结构包含完整分支
stream_mode=”updates” 查看每个节点的局部状态更新 可观察每轮生成与反馈
最大次数 防止评估器持续拒绝造成无限循环 三次后以 max_attempts 停止
recursion_limit 防止错误图无限执行 低上限测试触发 GraphRecursionError
调用次数 每轮一次生成和一次评估 N 轮通常约 2N 次模型调用
评估限制 结构正确不等于判断绝对正确 实测反馈未保证生成器有效改进

评估器优化器的价值不是保证模型一定写出完美结果，而是把“生成、检查、反馈和停止”变成一条可观察、可限制、可解释的 Workflow。

下一篇开始构建课程中的智能小秘书案例。首先处理模型一次生成多个 Tool Call 时的异步并发执行，再使用 ToolNode 简化工具节点。

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/b8e8f2d0.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
langgraph structured-output workflow evaluator
评论
  目录