评估器案例

案例概述

在 AI 应用中，经常需要让模型生成内容后进行评估，根据评估结果决定是否需要重新生成。这种"生成-评估-优化"的循环机制广泛应用于：

• 代码生成与审查
• 文档润色与改进
• 对话内容质量控制
• 创意内容筛选

本文以"冷笑话生成器"为例，展示如何使用 LangGraph 实现带评估功能的工作流。

---

案例效果

用户输入：主题 "程序员"
    │
    ▼
┌─────────────────┐
│   generator     │ ──生成笑话──▶
│   (生成器节点)  │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│   avaluator     │ ──评估幽默程度──▶
│   (评估器节点)  │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│   route_func    │ ──判断是否通过──▶
│   (路由函数)    │
└─────────────────┘
    │
    ├── "not funny" ──▶ 返回 generator（循环）
    │
    └── "funny" ──▶ END（结束）

---

完整代码实现

1. 定义状态和输出模型

from typing import TypedDict, Literal, Optional
from pydantic import BaseModel, Field

# 执行过程中需要保存的数据
class State(TypedDict):
    joke: Optional[str]           # 生成的冷笑话内容
    topic: str                   # 用户指定的主题
    feedback: Optional[str]      # 改进建议
    funny_or_not: Optional[str] # 幽默评级

# 结构化输出模型（用于 LLM 评估反馈）
class Feedback(BaseModel):
    """使用此工具来结构化你的响应"""
    grade: Literal['funny', 'not funny'] = Field(
        description='判断笑话是否幽默',
        examples=['funny', 'not funny']
    )
    feedback: str = Field(
        description='若不幽默，提供改进建议',
        example='可以加入双关语或意外结局'
    )

2. 定义节点函数

from langchain_core.output_parsers import StrOutputParser

def generator_func(state: State):
    """大模型生成冷笑话的节点"""
    # 构建 prompt
    has_feedback = state.get('feedback') and state.get('feedback') != 'None'
    if has_feedback:
        prompt = f'根据反馈改进笑话：{state["feedback"]}\n主题：{state["topic"]}'
    else:
        prompt = f'创作一个关于{state["topic"]}的笑话'

    # 调用 LLM
    chain = llm | StrOutputParser()
    resp = chain.invoke(prompt)

    return {'joke': resp}


def avaluator_func(state: State):
    """评估状态中的冷笑话"""
    # 检查 joke 是否存在
    if not state.get("joke"):
        return {"feedback": "无法评估，笑话生成失败", "funny_or_not": "not funny"}

    # 使用结构化输出
    structured_llm = llm.with_structured_output(Feedback)

    eval_prompt = f'评估此笑话的幽默程度：\n{state["joke"]}\n注意：幽默应包含意外性或巧妙措辞'
    resp = structured_llm.invoke(eval_prompt)

    return {'feedback': resp.feedback, 'funny_or_not': resp.grade}

3. 定义路由函数

def route_func(state: State) -> str:
    """动态路由决策函数"""
    funny_or_not = state.get("funny_or_not", "unknown")

    if funny_or_not == 'not funny':
        return 'Rejected'  # 回到生成器重新生成
    else:
        return 'Accepted' # 结束流程

4. 构建工作流图

from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 构建工作流
builder = StateGraph(State)
builder.add_node('generator', generator_func)
builder.add_node('avaluator', avaluator_func)

# 添加边
builder.add_edge(START, 'generator')
builder.add_edge('generator', 'avaluator')

# 条件边：根据评估结果决定走向
builder.add_conditional_edges(
    'avaluator',
    route_func,
    {
        'Accepted': END,
        'Rejected': 'generator'  # 循环回生成器
    }
)

graph = builder.compile()

5. 执行工作流

# 运行工作流
result = graph.invoke({"topic": "程序员"})
print(result["joke"])

---

核心要点解析

结构化输出

使用 with_structured_output() 方法让 LLM 返回结构化数据：

from pydantic import BaseModel, Field
from typing import Literal

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

# 创建结构化 LLM
structured_llm = llm.with_structured_output(Feedback)
result = structured_llm.invoke("评估这个笑话...")

优势：

• 输出格式可控，避免解析错误
• 可定义字段描述和示例
• 支持复杂嵌套结构

条件边实现循环

builder.add_conditional_edges(
    'avaluator',
    route_func,
    {
        'Accepted': END,
        'Rejected': 'generator'  # 关键：回到生成器形成循环
    }
)

循环机制：

• 评估不通过 → 返回 generator 节点重新生成
• 评估通过 → 流向 END 结束流程
• 可通过状态中的计数器限制循环次数

状态累积

class State(TypedDict):
    joke: Optional[str]
    feedback: Optional[str]
    funny_or_not: Optional[str]

每次循环迭代：

• joke 会被新生成的笑话覆盖
• feedback 携带改进建议传给生成器
• 状态在整个工作流中持续传递

---

扩展优化

添加循环次数限制

from typing import TypedDict, Optional, Literal

class State(TypedDict):
    joke: Optional[str]
    topic: str
    feedback: Optional[str]
    funny_or_not: Optional[str]
    attempts: int  # 添加尝试次数

def route_func(state: State) -> str:
    # 限制最多尝试 3 次
    if state.get("attempts", 0) >= 3:
        return "Accepted"  # 强制结束

    if state.get("funny_or_not") == 'not funny':
        return 'Rejected'
    return 'Accepted'

def generator_func(state: State):
    # 更新尝试次数
    attempts = state.get("attempts", 0) + 1
    return {"attempts": attempts, "joke": "..."}

添加人类审核节点

def human_review_node(state: State):
    """人类审核节点"""
    # 可集成外部审核系统
    return {"funny_or_not": "funny"}  # 假设审核通过

---

总结

组件	作用	关键代码
State	定义工作流数据结构	TypedDict + 可选字段
generator	生成内容	LLM + StrOutputParser
avaluator	评估内容质量	with_structured_output()
route_func	控制流程走向	返回节点名称
条件边	实现循环逻辑	add_conditional_edges()

这个案例展示了 LangGraph 实现"生成-评估-优化"循环的标准范式，可广泛应用于各类 AI 应用场景。