让 AI 学会“自我反思”：用 LangGraph 构建能不断改进的智能写作 Agent

一、问题的起源：一次生成真的够吗？

传统的 AI Agent 通常采用“一次性生成”的模式：用户输入指令 → 模型输出结果 → 结束。这种方式存在一个明显的缺陷：

AI 无法审视自己的输出，也不知道自己写得对不对。

想象一下，如果让一个人类写文章：

1. 先写初稿 ✍️
2. 再读一遍，找问题 🤔
3. 不满意就修改 🔧
4. 再读，再改，直到满意 ✅

这种写作-反思-修改-再反思的循环，是人类提升作品质量的本能方法。那么，能不能让 AI 也学会这套流程？

答案是肯定的。LangGraph 提供了构建这种循环工作流的能力。

二、什么是 LangGraph？

LangGraph 是 LangChain 生态中用于构建有状态、多步骤、循环式 Agent 的框架。它的核心概念非常简单：

概念	英文	说明	类比
状态	State	所有节点共享的“记事本”	白板
节点	Node	一个具体的操作/函数	工位上的一个员工
边	Edge	节点之间的连线	员工之间的传话路径
条件边	Conditional Edge	根据状态决定下一步走向	有决策权的经理

与传统 Agent 最大的区别：LangGraph 天然支持循环（Loop），这是实现“自我反思”的基础。

三、本次实现的工作流

我们将构建一个具有自我反思能力的写作 Agent，它的工作流程如下：

START → 写作节点 → 反思节点 → 质量评分 ≥8？ 
                              ↓
                         是 → END
                              ↓
                         否 → 修改节点 → (回到反思节点)

这是一个典型的反馈循环架构，广泛用于需要持续优化的场景。

四、完整代码深度解析

4.1 环境准备

pip install langgraph langchain-openai

4.2 定义全局状态（State）

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 对话历史，自动追加
    quality_score: int  # 质量评分（0-10分）

关键概念：add_messages

• 这是一个 reducer 函数，告诉 LangGraph 如何处理新消息
• 普通赋值会覆盖，但 add_messages 会自动追加到列表末尾
• 这使得我们可以保留完整的写作-修改历史

4.3 节点 1：写作节点（Writer）

def writer_node(state: State):
    print(" 正在撰写初稿...")
    prompt = "请帮我写一篇关于'人工智能未来发展趋势'的短文，200字左右。"
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [("assistant", response.content)], "quality_score": 0}

这个节点负责生成初稿。返回值中的 messages 会被自动追加到状态中。

4.4 节点 2：反思节点（Reflection）

def reflection_node(state: State):
    print(" 正在反思和评估文章质量...")
    last_message = state["messages"][-1].content
    
    prompt = f"""
    请作为一位严苛的编辑，对以下文章进行反思和评分（0-10分）：
    文章内容：{last_message}
    
    如果文章逻辑清晰、用词准确且没有废话，请打8分以上；否则请打低分。
    请严格按照以下格式输出：
    评分：[数字]
    """
    response = llm.invoke([{"role": "user", "content": prompt}])
    
    # 提取评分（实际生产建议用正则）
    score = 5
    if "评分：" in response.content:
        try:
            score = int(response.content.split("评分：")[1].strip()[0])
        except:
            pass
    
    print(f" 反思评分结果: {score}分")
    return {"quality_score": score}

这个节点是整个架构的灵魂：

• 让 AI 对自己的输出进行元认知（对自己的认知过程进行认知）
• 输出一个量化的质量分数，作为路由决策的依据

4.5 条件边：路由决策（Router）

def should_rewrite(state: State):
    if state["quality_score"] >= 8:
        print(" 质量达标，流程结束！")
        return "end"
    else:
        print(" 质量不达标，打回去重写！")
        return "rewrite"

这个函数就是“交通指挥员”，根据质量分数决定下一步走向。

4.6 节点 3：修改节点（Refiner）

def refiner_node(state: State):
    print(" 正在根据反思意见修改文章...")
    last_message = state["messages"][-1].content
    prompt = f"""
    上一版文章写得不够好，请重新修改并润色以下文章，使其更专业、更流畅：
    {last_message}
    """
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [("assistant", response.content)], "quality_score": 0}

修改节点会生成一个新版本，然后流程会再次进入反思节点，形成闭环。

4.7 组装图（Graph）

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)

# 添加节点
builder.add_node("writer", writer_node)
builder.add_node("reflect", reflection_node)
builder.add_node("refine", refiner_node)

# 添加边
builder.add_edge(START, "writer")
builder.add_edge("writer", "reflect")
builder.add_conditional_edges("reflect", should_rewrite, {
    "rewrite": "refine", 
    "end": END
})
builder.add_edge("refine", "reflect")

# 编译
reflection_agent = builder.compile()

这里的关键是第 4 条边：refine → reflect，它形成了循环，让 Agent 可以反复打磨直到满意。

4.8 运行 Agent

result = reflection_agent.invoke({"messages": [], "quality_score": 0})
print(f"最终定稿：{result['messages'][-1].content}")

五、运行效果演示

 启动带自我反思机制的写作 Agent...

 正在撰写初稿...
 正在反思和评估文章质量...
 反思评分结果: 6分
 质量不达标，打回去重写！

 正在根据反思意见修改文章...
 正在反思和评估文章质量...
 反思评分结果: 9分
 质量达标，流程结束！

 最终定稿文章：
人工智能正在以前所未有的速度重塑人类社会的方方面面...

六、架构可视化

┌─────────────────────────────────────────────────────┐
│                    全局状态 State                     │
│  ┌─────────────────────────────────────────────┐   │
│  │ messages: [初稿, 反思结果, 修改稿, 再反思...] │   │
│  │ quality_score: 动态更新的分数                 │   │
│  └─────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────┘
                          ↓↑
    ┌─────────┐    ┌─────────┐    ┌─────────┐
    │ Writer  │ → │ Reflect │ → │ Refiner │
    │  节点   │    │  节点   │    │  节点   │
    └─────────┘    └─────────┘    └─────────┘
         ↑              ↓              ↓
         └──────────────┴──────────────┘
                 (质量不达标时循环)

七、拓展应用场景

这个“生成-评估-改进”的循环架构可以迁移到多种场景：

7.1 代码审查 Agent

def coder_node(state):
    # 生成代码
    pass

def code_review_node(state):
    # 检查代码规范、潜在bug、性能问题
    # 输出评分和改进建议
    pass

def refactor_node(state):
    # 根据审查意见重构代码
    pass

7.2 翻译校对 Agent

def translator_node(state):
    # 初译
    pass

def quality_check_node(state):
    # 检查准确性、流畅度、术语一致性
    pass

def proofreader_node(state):
    # 润色修改
    pass

7.3 对话质量优化 Agent

def responder_node(state):
    # 生成回复
    pass

def safety_check_node(state):
    # 检查安全性、合规性、有用性
    pass

def improve_node(state):
    # 优化回复质量
    pass

八、进阶技巧

8.1 更精细的评分提取

import re

def extract_score(text: str) -> int:
    pattern = r"评分[:：]\s*(\d+)"
    match = re.search(pattern, text)
    return int(match.group(1)) if match else 5

8.2 添加最大循环次数防止死循环

class State(TypedDict):
    messages: Annotated[list, add_messages]
    quality_score: int
    rewrite_count: int  # 新增：记录重写次数

def should_rewrite(state: State):
    if state["rewrite_count"] >= 3:
        print(" 已达最大重写次数，强制结束")
        return "end"
    if state["quality_score"] >= 8:
        return "end"
    return "rewrite"

def refiner_node(state: State):
    # ... 修改文章逻辑 ...
    return {
        "messages": [("assistant", response.content)], 
        "quality_score": 0,
        "rewrite_count": state.get("rewrite_count", 0) + 1
    }

8.3 保存反思日志

def reflection_node(state: State):
    # ... 评分逻辑 ...
    criticism = extract_criticism(response.content)  # 提取具体意见
    print(f"💡 改进建议: {criticism}")
    # 可以将 criticism 也存入 state 供修改节点参考
    return {"quality_score": score, "feedback": criticism}

九、与传统 Agent 的对比

维度	传统 Agent	LangGraph 反思 Agent
执行模式	线性（一次过）	循环（可反复优化）
自我纠错	不具备	天然支持
输出质量	依赖单次生成质量	可通过多轮打磨提升
可观测性	黑盒	白盒（每个节点可追踪）
适用场景	简单任务	需要质量保证的复杂任务

十、总结

通过 LangGraph，我们用不到 100 行代码就构建了一个具备自我反思能力的智能 Agent。这种架构的核心价值在于：

1. 质量可控：通过评分阈值，可以控制输出质量
2. 过程透明：每个节点的输入输出都可追踪
3. 易于扩展：添加新节点或修改路由逻辑非常灵活
4. 通用性强：写作-反思-改进的模式适用于大量场景

思考延伸：

如果把这个架构和之前介绍的 Plan-and-Execute 结合起来，会碰撞出什么火花？一个能“先规划、再执行、同时自我反思”的 Agent，或许就是通往更智能 AI 系统的关键一步。

---

完整代码已附上，替换 API Key 即可运行体验。

如果这篇文章让你对 LangGraph 有了新的认识，欢迎点赞、收藏、转发！