LangGraph Reflection Agent：让 AI 学会自我审视，改到一半崩了还能接着改

目录

一、AI 为什么需要一个“内置 Reviewer”

二、为什么 Reflection 往往有效

（一）「批判能力」补「生成能力」

（二）它具体解决什么问题

三、核心原理：把生成流程改造成有状态闭环

四、源码说明与演示：状态、评审、改写与终止条件

（一）评审：结构化 prompt 强制 JSON 输出

（二）改进：重写也要防「越改越差」

（三）实战代码：把节点连接成真正的闭环

（四）运行效果：四组演示分别验证什么

1. 演示 1：基本反思改进——完整链路 + 执行轨迹

2. 演示 2：高质量阈值——9 分压线达标

3. 演示 3：无限反思防护——反思闭环真正转起来

4. 演示 4：中断恢复——断点续跑不浪费已消耗的调用

五、常见问题与线上说明

（一）常见坑与排查

坑 1：阈值设太高，Agent 无限自我否定

坑 2：反馈太笼统，改进无从下手

坑 3：LLM 改写退化，越改越差

坑 4：评审 JSON 解析失败，流程直接崩

坑 5：中断后从头重跑，白烧 token

参数调优时，建议至少记录这四组数据

（二）工程化检查表

（三）生产级方案：从“能运行”到“可治理”

六、总结：Reflection 是 Agent 的内置质量系统

干货分享，感谢您的阅读！

这是「LangGraph Agent Engineering Mastery」系列 Stage 4 推理 Agent · 第 3 篇。

本次我们用 LangGraph 搭建一条可运行、可评审、可恢复的 Reflection 链路——真实 LLM 先规划大纲，再逐节起草；评审节点给出结构化评分和可执行反馈；未达标则重写，达标或达到 max_rounds 即停止；任一步骤中断后，都能依靠 Checkpointer 从最近进度继续。

一、AI 为什么需要一个“内置 Reviewer”

写过文章的人都知道：初稿通常只是把想法写出来，修改才真正决定质量。 你会检查论证是否完整、例子是否足够、段落是否顺畅；写代码时也一样，提交前总要再看一遍变量命名、边界条件、异常分支和可维护性。

大模型的问题并不是“完全不会写”，而是一次生成很难同时兼顾内容、结构、准确性和表达。于是，一个自然的工程化思路出现了：把“写完再审”的人类工作方式拆成多个节点，让 Agent 自己完成“写 → 审 → 改”。

Reflection Agent 就是这套机制的系统化实现。它不是让模型在同一个提示词里含糊地“再想一遍”，而是把不同职责拆开：

上一篇的 Planning Agent 解决的是“如何按步骤做完”；Reflection 在它之上继续追问：做完以后，质量够不够？如果不够，具体改哪里？如果改到一半系统崩了，能不能接着改？

一分钟建立心智模型：

核心判断：Reflection 的价值不在于“模型多生成几次”，而在于把质量控制变成一个有状态、可观测、有退出条件的闭环。

二、为什么 Reflection 往往有效

（一）「批判能力」补「生成能力」

一次生成常见的问题并不神秘：内容可能浅、结构可能散、示例可能少、表达可能绕。真正值得利用的是模型能力之间的不对称。

LLM 一次生成的内容，常常有这些问题：内容不够深入、结构松散、缺少示例、表述啰嗦。但有意思的是——

LLM 的「评审能力」往往强于「一次生成能力」。 让它发现问题，比让它一次写完美，要容易得多。

这跟人是一样的：你可能写不出完美的初稿，但你一眼就能看出别人文章哪里不对。Reflection 正是利用了这个不对称——用模型更强的「批判能力」去补它较弱的「生成能力」。

从工程角度看，Reflection 能发挥作用，主要依赖下面三个条件：

（二）它具体解决什么问题

把 Reflection 与一次性生成放在一起对比，就能看清它不是单纯“更长的提示词”，而是增加了质量控制、终止保障和失败恢复能力。

三、核心原理：把生成流程改造成有状态闭环

本 Demo 设计的流程是「LLM 规划大纲 → 逐节起草 → 反思打分 → 改进 → 再反思……直到达标或到顶」，五类节点全部由真实 LLM 驱动：

• Plan 节点：LLM 输出 {"sections": […]} JSON，把「生成初稿」拆成可分步执行的小节——这正是上一篇 Planning 思想在 Reflection 里的复用。

• Draft 节点：每次只起草一节，把「任务 + 完整大纲 + 已完成小节」喂给 LLM，通过自循环边逐节推进。

• Reflect 节点：LLM 按结构化 prompt 评审草稿，输出总分、四个维度分（准确性/完整性/可读性/结构性）和可执行的改进建议。

• Improve 节点：LLM 拿着评审反馈重写草稿，改完回到 Reflect 再评。

关键问题是什么时候停，这里有两个出口：

• 质量达标（score >= quality_threshold）→ 停。

• 改了太多轮还没达标（current_round >= max_rounds）→ 也得停，否则无限循环。

第二个出口至关重要——否则一个“追求完美”的 AI 会永远觉得“还能再改改”，最终把质量优化变成没有边界的成本黑洞。

按照这个思路整体设计代码教学如下：

"""Demo 03: Reflection Agent — LLM 规划、分步起草、自我评审、断点续跑。

演示 Reflection 推理模式（全链路真实 LLM 调用）：
1. Plan 节点：真实 LLM 规划写作大纲（结构化 JSON），把"生成"拆成可分步执行的小节
2. Draft 节点：真实 LLM 按大纲逐节起草（每次只写一节，便于 Checkpoint）
3. Reflect 节点：真实 LLM 评审草稿，输出结构化评分（总分 + 多维度）与改进建议
4. Improve 节点：真实 LLM 根据评审反馈重写草稿
5. 循环控制：质量达标或达到 max_rounds 时结束，防止无限否定循环
6. 全程可追踪：execution_trace 记录每个节点的执行轨迹，输出时完整回放
7. 断点续跑：Checkpointer 持久化每个超步的进度，真实调用中断（网络异常、
进程崩溃）后，同一 thread_id 传入 None 即可续跑，已完成的规划/起草
不会重复执行（不浪费已消耗的 LLM 调用）

本 Demo 通过 shared.get_llm() 调用 .env 配置的真实在线模型
（fallback_to_mock=False，需联网）。LLM 输出无法解析时回退到
启发式保底逻辑（也用于 mock/离线环境，保证流程可运行）。

运行方式：
python stages/stage4_reasoning/03_reflection/main.py
"""

from __future__ import annotations

import json
import operator
import re
import sys
import time
from pathlib import Path
from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent.parent))

from shared import get_llm, get_logger, log_step, log_success, log_warning

logger = get_logger("demo.04_03_reflection")

MAX_REFLECTION_ROUNDS = 3
QUALITY_THRESHOLD = 7
MAX_OUTLINE_SECTIONS = 4

# 成文存档：真实运行（python main.py）时把每个演示的各轮草稿、评审记录、
# 最终成文写入 runs/ 目录；测试环境不写文件（仅 __main__ 入口开启）
_SAVE_ARTIFACTS = False
RUNS_DIR = Path(__file__).resolve().parent / "runs"

# 故障注入开关（仅供"中断恢复"演示使用）：
# Reflect 节点执行到指定轮次时抛出异常，模拟真实 LLM 调用中断（网络故障/进程崩溃）
_FAIL_AT_REFLECT_ROUND: int | None = None

# ============================================================
# 真实 LLM（通过 shared.get_llm 获取 .env 配置的在线模型）
# ============================================================
_LLM = None

def _get_reflection_llm():
"""获取真实在线 LLM 实例（模块内复用，fallback_to_mock=False 确保真实调用）。"""
global _LLM
if _LLM is None:
_LLM = get_llm(fallback_to_mock=False)
return _LLM

def _parse_llm_json(text: str) -> dict | None:
"""从 LLM 输出中解析 JSON（容忍 Markdown 代码块包裹等格式噪音）。"""
text = text.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\\s*|\\s*```$", "", text, flags=re.S).strip()
try:
parsed = json.loads(text)
return parsed if isinstance(parsed, dict) else None
except json.JSONDecodeError:
match = re.search(r"\\{.*\\}", text, re.S)
if match:
try:
parsed = json.loads(match.group())
return parsed if isinstance(parsed, dict) else None
except json.JSONDecodeError:
return None
return None

def _trace_event(node: str, detail: str) -> dict:
"""构造一条执行轨迹事件（全程可追踪的基础）。"""
return {"node": node, "detail": detail, "ts": time.strftime("%H:%M:%S")}

# ============================================================
# Reflection State
# ============================================================
class ReflectionState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
task: str
outline: list[str] # LLM 规划的写作大纲（分步执行的依据）
section_drafts: list[str] # 各小节草稿（逐节生成）
current_section_idx: int # 起草进度指针
current_draft: str # 当前完整草稿
draft_versions: Annotated[list[str], operator.add] # 各轮草稿版本（初稿 + 每轮改进稿）
reflection_history: list[dict] # [{round, score, feedback, improvements, dimensions, source}]
current_round: int
max_rounds: int
quality_threshold: int
is_satisfied: bool
execution_trace: Annotated[list[dict], operator.add] # 全程执行轨迹

# ============================================================
# 启发式保底评估（LLM 评审输出无法解析时使用，也用于 mock/离线环境）
# ============================================================
def evaluate_quality(draft: str, round_num: int) -> dict:
"""启发式质量评估（保底方案）。

评估维度：准确性、完整性、可读性、结构性。
随着改进轮次增加，质量分数递增（保证保底路径也能收敛）。
"""
base_score = min(5 + round_num * 2, 9)

length_bonus = min(len(draft) // 50, 2)
score = min(base_score + length_bonus, 10)

if round_num == 0:
feedback = "初稿存在以下问题：1) 内容不够深入，缺少具体示例；2) 结构可以更清晰；3) 缺少总结段落。"
improvements = ["增加具体代码示例", "添加小标题分段", "补充总结段落"]
elif round_num == 1:
feedback = "第二稿有明显改进：1) 示例已添加；2) 结构更清晰。但仍可优化：表述可以更精练。"
improvements = ["精简冗余表述", "突出核心观点"]
else:
feedback = "当前版本质量良好，内容完整、结构清晰、表述精练。"
improvements = []

return {
"score": score,
"feedback": feedback,
"improvements": improvements,
"dimensions": {
"accuracy": min(6 + round_num, 9),
"completeness": min(5 + round_num * 2, 9),
"readability": min(6 + round_num, 9),
"structure": min(5 + round_num * 2, 10),
},
}

def _fallback_improve(draft: str, improvements: list[str], round_num: int) -> str:
"""LLM 改写输出不可用时的保底改进（按建议追加/精简内容，保证草稿演进）。"""
improved_parts = [draft]
if "增加具体代码示例" in improvements or round_num == 0:
improved_parts.append(
"\\n\\n## 代码示例\\n"
"```python\\n"
"from langgraph.graph import StateGraph\\n"
"graph = StateGraph(AgentState)\\n"
"graph.add_node('reflect', reflect_node)\\n"
"graph.add_node('improve', improve_node)\\n"
"```"
)
if "添加小标题分段" in improvements or round_num == 0:
improved_parts.append("\\n\\n## 核心优势\\n多步推理、自我评审、迭代改进、可观测性。")
if "补充总结段落" in improvements or round_num == 0:
improved_parts.append("\\n\\n## 总结\\nReflection 让 Agent 从一次性输出走向持续自我改进。")
if "精简冗余表述" in improvements:
improved_parts = [p.replace("能够自主感知环境、做出决策并执行行动的", "自主决策的") for p in improved_parts]
if "突出核心观点" in improvements:
improved_parts.append("\\n\\n> **核心观点**: Reflection = 生成 + 评审 + 改进 的闭环")
return "".join(improved_parts)

# ============================================================
# LLM 规划：生成写作大纲（分步执行的依据）
# ============================================================
_PLAN_SYSTEM_PROMPT = """你是一个内容规划专家，负责为写作任务规划大纲。

只输出一个 JSON 对象（不要输出任何其他文字）：
{"sections": ["小节标题1", "小节标题2", …]}

规划要求：
1. 小节按逻辑顺序排列，覆盖任务的核心要点
2. 小节标题具体、可独立成文
3. 小节数量控制在给定上限以内"""

_FALLBACK_OUTLINE = [
"核心概念与背景",
"关键机制与技术要点",
"实践示例与应用场景",
"总结与展望",
]

def generate_outline(task: str, max_sections: int) -> list[str]:
"""调用真实 LLM 规划写作大纲；解析失败时回退到保底大纲。"""
llm = _get_reflection_llm()
response = llm.invoke([
SystemMessage(content=_PLAN_SYSTEM_PROMPT),
HumanMessage(content=f"任务: {task}\\n小节数上限: {max_sections}\\n请输出大纲 JSON。"),
])
decision = _parse_llm_json(str(response.content))

sections: list[str] = []
if decision:
sections = [str(s).strip() for s in decision.get("sections", []) if str(s).strip()]

if not sections:
log_warning(logger, "LLM 大纲输出无法解析为 JSON，使用保底通用大纲")
sections = list(_FALLBACK_OUTLINE)

return sections[:max_sections]

# ============================================================
# LLM 分步起草：每次只写一节
# ============================================================
def draft_section(task: str, outline: list[str], drafted: list[str], idx: int) -> str:
"""调用真实 LLM 起草大纲中的一个小节。

注意：这里不捕获 LLM 调用异常——异常向上传播会让本次 invoke 中断，
但已完成小节的进度已由 Checkpointer 保存，重跑时可从当前小节继续。
"""
outline_text = "\\n".join(f"{i + 1}. {s}" for i, s in enumerate(outline))
drafted_text = (
"\\n".join(f"### {outline[i]}\\n{d}" for i, d in enumerate(drafted))
if drafted
else "（这是第一节，暂无已完成内容）"
)
prompt = (
f"写作任务: {task}\\n"
f"完整大纲:\\n{outline_text}\\n\\n"
f"已完成的小节:\\n{drafted_text}\\n\\n"
f"现在请撰写小节「{outline[idx]}」的正文，要求：\\n"
f"1. 用中文，2~4 句话，内容具体、承接前文\\n"
f"2. 只输出该小节正文，不要输出小节标题，不要写其他小节"
)
llm = _get_reflection_llm()
response = llm.invoke([HumanMessage(content=prompt)])
return str(response.content).strip()

def _assemble_draft(outline: list[str], section_drafts: list[str]) -> str:
"""把已完成的小节草稿组装为完整草稿。"""
return "\\n\\n".join(
f"## {outline[i]}\\n{body}" for i, body in enumerate(section_drafts)
)

# ============================================================
# LLM 评审：结构化打分 + 改进建议
# ============================================================
_REFLECT_SYSTEM_PROMPT = """你是一位严格的内容评审专家，负责评审草稿质量。

只输出一个 JSON 对象（不要输出任何其他文字）：
{"score": <0-10 整数总分>,
"feedback": "<总体评价与主要问题>",
"improvements": ["<具体可执行的改进建议1>", …],
"dimensions": {"accuracy": <0-10>, "completeness": <0-10>, "readability": <0-10>, "structure": <0-10>}}

评审要求：
1. score 为综合评分：7 分表示合格，9 分以上表示优秀，评分严格客观
2. 质量不足时给出具体、可落实的 improvements；已达标时 improvements 返回空数组
3. dimensions 分别评估准确性、完整性、可读性、结构性"""

def _sanitize_evaluation(decision: dict) -> dict | None:
"""校验并规整 LLM 评审 JSON；关键字段缺失/非法时返回 None（触发保底）。"""
try:
score = int(round(float(decision["score"])))
except (KeyError, TypeError, ValueError):
return None
score = max(0, min(score, 10))

feedback = str(decision.get("feedback", "")).strip()
if not feedback:
return None

improvements = [str(i).strip() for i in decision.get("improvements", []) if str(i).strip()]

raw_dims = decision.get("dimensions", {})
dimensions = {}
for key in ("accuracy", "completeness", "readability", "structure"):
try:
dimensions[key] = max(0, min(int(round(float(raw_dims[key]))), 10))
except (KeyError, TypeError, ValueError):
dimensions[key] = score

return {
"score": score,
"feedback": feedback,
"improvements": improvements,
"dimensions": dimensions,
}

def critique_draft(task: str, draft: str, round_num: int) -> tuple[dict, str]:
"""调用真实 LLM 评审草稿，返回 (评估结果, 来源标记 llm/fallback)。"""
llm = _get_reflection_llm()
response = llm.invoke([
SystemMessage(content=_REFLECT_SYSTEM_PROMPT),
HumanMessage(content=(
f"写作任务: {task}\\n"
f"当前是第 {round_num + 1} 轮评审。\\n"
f"待评审草稿:\\n{draft}\\n\\n"
f"请输出评审 JSON。"
)),
])
decision = _parse_llm_json(str(response.content))
evaluation = _sanitize_evaluation(decision) if decision else None

if evaluation is None:
log_warning(logger, "LLM 评审输出无法解析为 JSON，使用启发式保底评估")
return evaluate_quality(draft, round_num), "fallback"
return evaluation, "llm"

# ============================================================
# LLM 改进：根据评审反馈重写草稿
# ============================================================
_IMPROVE_SYSTEM_PROMPT = """你是一位资深内容编辑，负责根据评审反馈修改草稿。

修改要求：
1. 保留草稿中已合格的内容与整体结构（Markdown 小节）
2. 逐条落实评审给出的改进建议
3. 只输出修改后的完整草稿正文（Markdown），不要输出解释或额外说明"""

def rewrite_draft(task: str, draft: str, feedback: str, improvements: list[str]) -> str:
"""调用真实 LLM 根据评审反馈重写草稿。"""
improvements_text = (
"\\n".join(f"- {i}" for i in improvements) if improvements else "（无具体建议，整体润色）"
)
llm = _get_reflection_llm()
response = llm.invoke([
SystemMessage(content=_IMPROVE_SYSTEM_PROMPT),
HumanMessage(content=(
f"写作任务: {task}\\n"
f"评审反馈: {feedback}\\n"
f"改进建议:\\n{improvements_text}\\n\\n"
f"当前草稿:\\n{draft}\\n\\n"
f"请输出修改后的完整草稿。"
)),
])
return str(response.content).strip()

# ============================================================
# 节点实现
# ============================================================
def plan_node(state: ReflectionState) -> dict:
"""Plan 节点：真实 LLM 规划写作大纲，把生成任务拆成可分步执行的小节。"""
task = state["task"]

log_step(logger, "Plan", f"规划写作大纲（真实 LLM）: '{task[:40]}…'")
print(f"\\n [规划] 任务: {task}")

outline = generate_outline(task, MAX_OUTLINE_SECTIONS)

print(f" [规划] 生成了 {len(outline)} 节大纲:")
for i, section in enumerate(outline):
print(f" 第 {i + 1} 节: {section}")

return {
"outline": outline,
"section_drafts": [],
"current_section_idx": 0,
"current_draft": "",
"current_round": 0,
"is_satisfied": False,
"execution_trace": [_trace_event("plan", f"生成 {len(outline)} 节大纲")],
}

def draft_node(state: ReflectionState) -> dict:
"""Draft 节点：真实 LLM 按大纲逐节起草（每次只写一节，便于 Checkpoint）。"""
task = state["task"]
outline = state.get("outline", [])
drafted = list(state.get("section_drafts", []))
idx = state.get("current_section_idx", 0)

log_step(logger, "Draft", f"起草第 {idx + 1}/{len(outline)} 节: {outline[idx]}")
print(f"\\n [起草] 第 {idx + 1}/{len(outline)} 节「{outline[idx]}」…")

body = draft_section(task, outline, drafted, idx)
drafted.append(body)

print(f" [成稿] {body[:80]}{'…' if len(body) > 80 else ''}")
print(f" [进度] 起草 {len(drafted)}/{len(outline)} 节完成")

assembled = _assemble_draft(outline, drafted)
update = {
"section_drafts": drafted,
"current_section_idx": idx + 1,
"current_draft": assembled,
"execution_trace": [_trace_event("draft", f"完成第 {idx + 1} 节「{outline[idx]}」")],
}
if len(drafted) == len(outline):
# 全部小节起草完毕，记录初稿版本（成文过程的 v1）
update["draft_versions"] = [assembled]
return update

def reflect_node(state: ReflectionState) -> dict:
"""Reflect 节点：真实 LLM 评审当前草稿，输出结构化评分与改进建议。"""
task = state["task"]
draft = state["current_draft"]
round_num = state["current_round"]
max_rounds = state["max_rounds"]
threshold = state["quality_threshold"]
history = list(state.get("reflection_history", []))

# 故障注入（仅"中断恢复"演示）：模拟真实 LLM 调用中途失败
if _FAIL_AT_REFLECT_ROUND is not None and round_num == _FAIL_AT_REFLECT_ROUND:
raise RuntimeError(
f"模拟中断：第 {round_num + 1} 轮评审时 LLM 调用失败（网络异常/进程崩溃）"
)

log_step(logger, "Reflect", f"第 {round_num + 1}/{max_rounds} 轮评审（真实 LLM）")
print(f"\\n [反思] 第 {round_num + 1} 轮评审…")

evaluation, source = critique_draft(task, draft, round_num)
score = evaluation["score"]
feedback = evaluation["feedback"]
improvements = evaluation["improvements"]
dimensions = evaluation["dimensions"]

print(f" [评分] 总分: {score}/10 (阈值: {threshold}, 评审来源: {source})")
print(
f" [维度] 准确性 {dimensions['accuracy']} | 完整性 {dimensions['completeness']}"
f" | 可读性 {dimensions['readability']} | 结构性 {dimensions['structure']}"
)
print(f" [反馈] {feedback[:100]}{'…' if len(feedback) > 100 else ''}")
if improvements:
print(f" [建议] {'; '.join(improvements[:3])}")

history.append({
"round": round_num + 1,
"score": score,
"feedback": feedback,
"improvements": improvements,
"dimensions": dimensions,
"source": source,
})

is_satisfied = score >= threshold

if is_satisfied:
print(f" [判定] 质量达标 ✓ (分数 {score} >= 阈值 {threshold})")
else:
print(f" [判定] 质量未达标 ✗ (分数 {score} < 阈值 {threshold})")

return {
"reflection_history": history,
"is_satisfied": is_satisfied,
"execution_trace": [
_trace_event("reflect", f"第 {round_num + 1} 轮评审: {score}/10 ({source})")
],
}

def improve_node(state: ReflectionState) -> dict:
"""Improve 节点：真实 LLM 根据评审反馈重写草稿。"""
task = state["task"]
draft = state["current_draft"]
history = state["reflection_history"]
round_num = state["current_round"]

log_step(logger, "Improve", f"根据评审反馈改进草稿（真实 LLM，第 {round_num + 1} 轮）")
print("\\n [改进] 根据评审反馈重写草稿…")

latest = history[-1] if history else {}
feedback = latest.get("feedback", "")
improvements = latest.get("improvements", [])

improved_draft = rewrite_draft(task, draft, feedback, improvements)

# 保底防护：LLM 改写输出为空、严重缩水（丢失内容）或异常膨胀（退化输出，
# 如无意义重复文本）时，回退到追加式改进，避免劣质稿污染后续评审并浪费 Token
degenerate = len(improved_draft) > max(len(draft) * 6, 8000)
if not improved_draft or len(improved_draft) < len(draft) // 2 or degenerate:
log_warning(logger, "LLM 改写输出不可用（为空/严重缩水/异常膨胀），使用保底追加式改进")
improved_draft = _fallback_improve(draft, improvements, round_num)

print(f" [改进后] 字数: {len(draft)} → {len(improved_draft)}")

return {
"current_draft": improved_draft,
"current_round": round_num + 1,
"draft_versions": [improved_draft], # 记录本轮改进稿（成文过程的 v2/v3/…）
"execution_trace": [
_trace_event("improve", f"第 {round_num + 1} 轮改进: {len(draft)} → {len(improved_draft)} 字")
],
}

def output_node(state: ReflectionState) -> dict:
"""Output 节点：输出最终结果与完整反思历程。"""
draft = state["current_draft"]
history = state["reflection_history"]
round_num = state["current_round"]

log_success(logger, f"反思完成，共 {round_num} 轮改进")
print(f"\\n [完成] 最终版本已生成（共 {round_num} 轮反思改进）")

trace = "\\n\\n— 反思历程 —\\n"
for h in history:
trace += f"第 {h['round']} 轮: 分数 {h['score']}/10 – {h['feedback'][:50]}…\\n"

return {
"messages": [AIMessage(content=draft + trace)],
"execution_trace": [_trace_event("output", f"输出最终稿（{len(draft)} 字）")],
}

# ============================================================
# 条件边
# ============================================================
def should_continue_drafting(state: ReflectionState) -> str:
"""判断是否还有小节需要起草。"""
if state.get("current_section_idx", 0) < len(state.get("outline", [])):
return "draft"
return "reflect"

def should_continue_reflection(state: ReflectionState) -> str:
"""判断是否继续反思改进。"""
if state["is_satisfied"]:
return "output"
if state["current_round"] >= state["max_rounds"]:
log_warning(logger, f"达到最大反思轮次 {state['max_rounds']}，强制输出")
return "output"
return "improve"

# ============================================================
# 构建 Reflection Graph
# ============================================================
def build_reflection_graph(
max_rounds: int = MAX_REFLECTION_ROUNDS,
quality_threshold: int = QUALITY_THRESHOLD,
):
"""构建 Reflection Agent 图。

图结构：
START → plan → draft → [draft …] → reflect → [improve → reflect …] → output → END

draft 节点每次只起草一节、improve/reflect 每轮一个超步（super-step），
每个超步结束后 Checkpointer 都会保存一次状态——这正是断点续跑的基础。
"""
graph = StateGraph(ReflectionState)

graph.add_node("plan", plan_node)
graph.add_node("draft", draft_node)
graph.add_node("reflect", reflect_node)
graph.add_node("improve", improve_node)
graph.add_node("output", output_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "draft")
graph.add_conditional_edges(
"draft",
should_continue_drafting,
{"draft": "draft", "reflect": "reflect"},
)
graph.add_conditional_edges(
"reflect",
should_continue_reflection,
{"improve": "improve", "output": "output"},
)
graph.add_edge("improve", "reflect")
graph.add_edge("output", END)

return graph

def _initial_state(task: str, max_rounds: int, quality_threshold: int) -> dict:
"""构造初始状态。"""
return {
"messages": [HumanMessage(content=task)],
"task": task,
"outline": [],
"section_drafts": [],
"current_section_idx": 0,
"current_draft": "",
"reflection_history": [],
"current_round": 0,
"max_rounds": max_rounds,
"quality_threshold": quality_threshold,
"is_satisfied": False,
}

def _print_execution_trace(trace: list[dict]) -> None:
"""打印全程执行轨迹（可追踪性演示）。"""
print("\\n — 执行轨迹（全程可追踪） —")
for i, event in enumerate(trace, 1):
print(f" {i:02d}. [{event['ts']}] {event['node']:<8} {event['detail']}")

def _save_run_artifacts(demo_name: str, title: str, result: dict) -> None:
"""把一次演示的真实成文情况存档为 Markdown（仅脚本直跑时开启）。

存档内容：任务与参数、各轮草稿版本（成文过程）、完整评审记录、
执行轨迹、最终成文全文。
"""
if not _SAVE_ARTIFACTS:
return

RUNS_DIR.mkdir(exist_ok=True)
path = RUNS_DIR / f"{demo_name}.md"

lines: list[str] = [
f"# {title}",
"",
f"- 运行时间: {time.strftime('%Y-%m-%d %H:%M:%S')}",
f"- 任务: {result.get('task', '')}",
f"- 质量阈值: {result.get('quality_threshold')} | 最大轮次: {result.get('max_rounds')}",
f"- 反思轮次: {result.get('current_round')} | 质量达标: {'是' if result.get('is_satisfied') else '否'}",
"",
"## 写作大纲（LLM 规划）",
"",
]
for i, section in enumerate(result.get("outline", []), 1):
lines.append(f"{i}. {section}")

lines += ["", "## 成文过程（各轮草稿版本）", ""]
versions = result.get("draft_versions", [])
for i, version in enumerate(versions):
label = "初稿（分节起草组装）" if i == 0 else f"第 {i} 轮改进稿"
lines += [f"### v{i + 1} · {label}（{len(version)} 字）", "", version, ""]

lines += ["## 评审记录（完整）", ""]
for h in result.get("reflection_history", []):
dims = h.get("dimensions", {})
lines += [
f"### 第 {h['round']} 轮评审 — {h['score']}/10（来源: {h.get('source', '?')}）",
"",
f"- 维度: 准确性 {dims.get('accuracy', '-')} | 完整性 {dims.get('completeness', '-')}"
f" | 可读性 {dims.get('readability', '-')} | 结构性 {dims.get('structure', '-')}",
f"- 反馈: {h['feedback']}",
]
improvements = h.get("improvements", [])
if improvements:
lines.append("- 改进建议:")
lines += [f" – {item}" for item in improvements]
lines.append("")

lines += ["## 执行轨迹", ""]
for i, event in enumerate(result.get("execution_trace", []), 1):
lines.append(f"{i:02d}. [{event['ts']}] {event['node']} — {event['detail']}")

final_draft = result.get("current_draft", "")
lines += ["", f"## 最终成文（{len(final_draft)} 字）", "", final_draft, ""]

path.write_text("\\n".join(lines), encoding="utf-8")
print(f"\\n [存档] 本次真实成文与评审记录已保存: {path}")

# ============================================================
# 运行演示
# ============================================================
def demo_basic_reflection():
"""演示 1：完整链路 — LLM 规划 → 分步起草 → 评审 → 改进 → 输出。"""
print("\\n— 演示 1: 基本反思改进（规划 + 分步起草 + 评审 + 改进） —\\n")

graph = build_reflection_graph(max_rounds=3, quality_threshold=7)
app = graph.compile()

task = "分析 AI Agent 的技术架构和发展趋势"
result = app.invoke(_initial_state(task, max_rounds=3, quality_threshold=7))

print(f"\\n {'─' * 50}")
print(f" 大纲节数: {len(result['outline'])}")
print(f" 反思轮次: {result['current_round']}")
print(f" 质量达标: {'是' if result['is_satisfied'] else '否(强制结束)'}")
_print_execution_trace(result["execution_trace"])
_save_run_artifacts("demo1_basic_reflection", "演示 1: 基本反思改进", result)
return result

def demo_high_threshold():
"""演示 2：高质量阈值（需要更多反思轮次）。"""
print("\\n— 演示 2: 高质量阈值 —\\n")

graph = build_reflection_graph(max_rounds=3, quality_threshold=9)
app = graph.compile()

task = "撰写 LangGraph 框架的技术入门指南"
print(f" 任务: {task}")
print(" quality_threshold = 9 (高要求)")

result = app.invoke(_initial_state(task, max_rounds=3, quality_threshold=9))

print(f"\\n 反思轮次: {result['current_round']}")
print(f" 最终是否达标: {'是' if result['is_satisfied'] else '否(达到最大轮次)'}")
scores = [h["score"] for h in result["reflection_history"]]
print(f" 评分演进: {' → '.join(str(s) for s in scores)}")
_save_run_artifacts("demo2_high_threshold", "演示 2: 高质量阈值", result)
return result

def demo_infinite_reflection_guard():
"""演示 3：无限反思防护。"""
print("\\n— 演示 3: 无限反思防护 —\\n")

graph = build_reflection_graph(max_rounds=2, quality_threshold=10)
app = graph.compile()

task = "写一份完美的技术文档（质量阈值设为满分 10）"
print(f" 任务: {task}")
print(" quality_threshold = 10 (几乎不可能满足)")
print(" max_rounds = 2 (防护限制)")

result = app.invoke(_initial_state(task, max_rounds=2, quality_threshold=10))

print(f"\\n 实际反思轮次: {result['current_round']}")
print(f" 防护触发: {'是' if not result['is_satisfied'] else '否'}")
print(" (max_rounds=2 防止了无限反思循环)")
_save_run_artifacts("demo3_infinite_guard", "演示 3: 无限反思防护", result)
return result

def demo_interrupt_resume():
"""演示 4：中断恢复（Checkpoint 断点续跑）。

模拟真实场景：大纲规划和分节起草全部完成后，首轮评审时 LLM 调用中断
（网络故障/进程崩溃）。由于 Checkpointer 已保存规划和起草进度，恢复时
对同一 thread_id 传入 None 即可从评审继续，规划/起草的 LLM 调用不会重复。
"""
global _FAIL_AT_REFLECT_ROUND

print("\\n— 演示 4: 中断恢复（断点续跑） —\\n")

checkpointer = MemorySaver()
graph = build_reflection_graph(max_rounds=3, quality_threshold=8)
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "reflection-resume-demo"}}

task = "写一篇关于 Reflection 推理模式的分析短文"
print(f" 任务: {task}")
print(" （故障注入: 第 1 轮评审时模拟 LLM 调用中断）")

# 阶段 1：起草全部完成后，首轮评审时人为中断
log_step(logger, "Resume-阶段1", "开始执行，第 1 轮评审将触发模拟中断")
_FAIL_AT_REFLECT_ROUND = 0
try:
app.invoke(_initial_state(task, max_rounds=3, quality_threshold=8), config)
log_warning(logger, "故障未触发（未进入评审），跳过恢复演示")
return app.get_state(config).values
except RuntimeError as e:
log_warning(logger, f"执行中断: {e}")
print(f"\\n [中断] {e}")
finally:
_FAIL_AT_REFLECT_ROUND = None # 清除故障（模拟网络恢复/进程重启）

# 阶段 2：查看 Checkpoint 中保存的进度
saved_state = app.get_state(config)
saved_outline = saved_state.values.get("outline", [])
saved_sections = saved_state.values.get("section_drafts", [])
print(f"\\n [Checkpoint] 已保存进度: 大纲 {len(saved_outline)} 节，起草完成 {len(saved_sections)} 节")
for i, section in enumerate(saved_outline):
icon = "✓" if i < len(saved_sections) else "○"
print(f" {icon} 第 {i + 1} 节: {section}")

# 阶段 3：传入 None 从最近一次 Checkpoint 恢复执行
log_step(logger, "Resume-阶段2", "从 Checkpoint 恢复，从评审环节继续")
print("\\n [恢复] 重新运行（输入传 None，同一 thread_id）→ 从评审继续:")
result = app.invoke(None, config)

log_success(
logger,
f"断点续跑完成：规划 + {len(saved_sections)} 节起草未重复执行，"
f"恢复后完成 {result['current_round']} 轮反思改进",
)
print(f"\\n {'─' * 50}")
print(f" 中断前完成: 规划 + {len(saved_sections)} 节起草（已持久化，恢复后未重复调用 LLM）")
print(f" 恢复后完成: {len(result['reflection_history'])} 轮评审、{result['current_round']} 轮改进")
_print_execution_trace(result["execution_trace"])
_save_run_artifacts("demo4_interrupt_resume", "演示 4: 中断恢复（断点续跑）", result)
return result

def run_demo() -> dict:
"""运行 Reflection Agent 全部演示。"""
print("=" * 60)
print(" Demo 03: Reflection Agent — 自我评审与输出改进（真实 LLM）")
print("=" * 60)

basic_result = demo_basic_reflection()
high_result = demo_high_threshold()
guard_result = demo_infinite_reflection_guard()
resume_result = demo_interrupt_resume()

print()
print("=" * 60)
print(" 关键概念回顾")
print("=" * 60)
print(" 1. Plan 节点 : 真实 LLM 规划写作大纲，把生成拆成可分步执行的小节")
print(" 2. Draft 节点 : 真实 LLM 逐节起草，每节一个超步便于 Checkpoint")
print(" 3. Reflect 节点 : 真实 LLM 评审打分（总分 + 多维度）+ 改进建议")
print(" 4. Improve 节点 : 真实 LLM 根据反馈重写草稿")
print(" 5. 质量阈值 : 达到阈值即停止反思；max_rounds 防止无限否定循环")
print(" 6. 全程可追踪 : execution_trace 记录每个节点的执行轨迹")
print(" 7. 断点续跑 : Checkpointer 保存进度，中断后传 None 续跑不重复调用")
print()

return {
"basic_result": basic_result,
"high_result": high_result,
"guard_result": guard_result,
"resume_result": resume_result,
}

if __name__ == "__main__":
_SAVE_ARTIFACTS = True
run_demo()

四、源码说明与演示：状态、评审、改写与终止条件

下面不按文件顺序逐行解释，而是沿着“状态如何流动、结果如何被校验、循环如何停止”三条主线来说我们写的源码。

先看状态。相比单纯的「生成→反思」，多了大纲、分节草稿和执行轨迹三组字段：

class ReflectionState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
task: str
outline: list[str] # LLM 规划的写作大纲（分步执行的依据）
section_drafts: list[str] # 各小节草稿（逐节生成）
current_section_idx: int # 起草进度指针
current_draft: str # 当前完整草稿
draft_versions: Annotated[list[str], operator.add] # 各轮草稿版本（初稿 + 每轮改进稿）
reflection_history: list[dict] # [{round, score, feedback, improvements, dimensions, source}]
current_round: int
max_rounds: int
quality_threshold: int
is_satisfied: bool
execution_trace: Annotated[list[dict], operator.add] # 全程执行轨迹

注意 execution_trace 用了 operator.add reducer——每个节点只负责追加自己的事件，LangGraph 自动合并，最终形成一条可回放的执行轨迹。它不仅是演示效果，更是排查“为什么改了 3 轮还不达标”“中断发生在哪个超步”“恢复后有没有重复执行”的第一手证据。

从这组状态字段还能读出一个重要设计原则：不要只保存最终结果，要保存过程中的可恢复中间态。 outline、section_drafts、current_section_idx 让 Draft 可以分步恢复；draft_versions 和 reflection_history 让质量演进可审计；current_round、max_rounds 与 quality_threshold 则共同决定循环边界。

（一）评审：结构化 prompt 强制 JSON 输出

Reflect 是核心。评审标准直接写进系统提示词——不是问「好不好」，而是要总分、四个维度分、可执行建议：

_REFLECT_SYSTEM_PROMPT = """你是一位严格的内容评审专家，负责评审草稿质量。

只输出一个 JSON 对象（不要输出任何其他文字）：
{"score": <0-10 整数总分>,
"feedback": "<总体评价与主要问题>",
"improvements": ["<具体可执行的改进建议1>", …],
"dimensions": {"accuracy": <0-10>, "completeness": <0-10>, "readability": <0-10>, "structure": <0-10>}}

评审要求：
1. score 为综合评分：7 分表示合格，9 分以上表示优秀，评分严格客观
2. 质量不足时给出具体、可落实的 improvements；已达标时 improvements 返回空数组
3. dimensions 分别评估准确性、完整性、可读性、结构性"""

但永远不要假设 LLM 会乖乖输出合法 JSON。评审结果要过两道关：先容错解析（剥 Markdown 代码块、正则兜底），再字段校验（分数钳到 0-10、feedback 非空、维度缺失时用总分补齐）。两道关都过不了，就回退到启发式保底评估，保证流程不断：

def critique_draft(task: str, draft: str, round_num: int) -> tuple[dict, str]:
"""调用真实 LLM 评审草稿，返回 (评估结果, 来源标记 llm/fallback)。"""
llm = _get_reflection_llm()
response = llm.invoke([
SystemMessage(content=_REFLECT_SYSTEM_PROMPT),
HumanMessage(content=(
f"写作任务: {task}\\n"
f"当前是第 {round_num + 1} 轮评审。\\n"
f"待评审草稿:\\n{draft}\\n\\n"
f"请输出评审 JSON。"
)),
])
decision = _parse_llm_json(str(response.content))
evaluation = _sanitize_evaluation(decision) if decision else None

if evaluation is None:
log_warning(logger, "LLM 评审输出无法解析为 JSON，使用启发式保底评估")
return evaluate_quality(draft, round_num), "fallback"
return evaluation, "llm"

每条评审记录都带 source 标记（llm / fallback），事后能分清哪些分数来自模型、哪些来自兜底逻辑。这个字段很小，却让监控和复盘有了依据：如果 fallback 比例突然升高，通常意味着模型输出格式、提示词稳定性或网关响应出现了变化。

（二）改进：重写也要防「越改越差」

Improve 节点让 LLM 拿着反馈重写全文。但真实模型的改写输出并不总是可用的——实测中就遇到过模型在高压阈值下产生退化输出（无意义重复文本把草稿从 777 字撑到 3.5 万字，评分反而从 9 掉到 5）。所以改写结果要做健全性检查：

improved_draft = rewrite_draft(task, draft, feedback, improvements)

# 保底防护：LLM 改写输出为空、严重缩水（丢失内容）或异常膨胀（退化输出，
# 如无意义重复文本）时，回退到追加式改进，避免劣质稿污染后续评审并浪费 Token
degenerate = len(improved_draft) > max(len(draft) * 6, 8000)
if not improved_draft or len(improved_draft) < len(draft) // 2 or degenerate:
log_warning(logger, "LLM 改写输出不可用（为空/严重缩水/异常膨胀），使用保底追加式改进")
improved_draft = _fallback_improve(draft, improvements, round_num)

而防无限反思的命门在 Reflect 之后的条件边——只要撞上 max_rounds，无论质量是否达标都强制输出。阈值决定“什么叫足够好”，最大轮次决定“最多愿意为更好付出多少成本”：

def should_continue_reflection(state: ReflectionState) -> str:
"""判断是否继续反思改进。"""
if state["is_satisfied"]:
return "output"
if state["current_round"] >= state["max_rounds"]:
log_warning(logger, f"达到最大反思轮次 {state['max_rounds']}，强制输出")
return "output"
return "improve"

这一段代码真正保护了什么：

• 解析防线保护的是控制流：评审 JSON 失败时，流程不能直接崩。

• 健全性检查保护的是内容状态：新版草稿异常时，不能让劣质结果覆盖上一版。

• max_rounds 保护的是成本上界：模型再挑剔，也必须在有限轮次内交付。

（三）实战代码：把节点连接成真正的闭环

这里关注图结构本身：draft 有自循环，reflect 有条件分支，improve 会回到 reflect，最终由 output 结束。

图结构：plan 进 draft，draft 自循环逐节起草，起草完进 reflect；reflect 按条件边决定「改进」还是「输出」，improve 改完回到 reflect 再评：

graph.add_edge(START, "plan")
graph.add_edge("plan", "draft")
graph.add_conditional_edges(
"draft",
should_continue_drafting,
{"draft": "draft", "reflect": "reflect"},
)
graph.add_conditional_edges(
"reflect",
should_continue_reflection,
{"improve": "improve", "output": "output"},
)
graph.add_edge("improve", "reflect")
graph.add_edge("output", END)

draft 每次只写一节、reflect/improve 每轮一个超步（super-step）——这意味着每次 LLM 调用后 Checkpointer 都会落一次盘，这正是断点续跑的基础。

运行方式：

source .venv/bin/activate
python stages/stage4_reasoning/03_reflection/main.py

（四）运行效果：四组演示分别验证什么

四个 Demo 不是重复展示日志，而是在验证四种不同的工程行为：

真实执行 main.py（zyftopia 网关 · qwen3-next-80b-a3b-instruct），以下四个演示的输出全部截取自同一次真实运行，只对过长的成稿内容做了省略。

📄 完整成文存档：脚本直跑时会把每个演示的真实成文情况自动存档到 runs/ 目录——每个文件包含 LLM 规划的大纲、各轮草稿版本全文（初稿 + 每轮改进稿）、完整评审记录（总分/维度分/反馈/建议）、执行轨迹和最终成文。想看「一篇文章是怎么被反思改出来的」，直接对比 runs/demo3_infinite_guard.md 里的 v1 和 v2 两版全文。

1. 演示 1：基本反思改进——完整链路 + 执行轨迹

— 演示 1: 基本反思改进（规划 + 分步起草 + 评审 + 改进） —

[规划] 任务: 分析 AI Agent 的技术架构和发展趋势
[规划] 生成了 4 节大纲:
第 1 节: AI Agent 的核心技术架构组成
第 2 节: 关键能力模块：感知、决策与行动
第 3 节: 当前主流架构模式与技术对比
第 4 节: 未来发展趋势：多模态、自主性与分布式协同

[起草] 第 1/4 节「AI Agent 的核心技术架构组成」…
[成稿] AI Agent 的核心技术架构由感知模块、决策引擎、记忆系统、行动执行器和交互接口五大核心组件构成……
[进度] 起草 1/4 节完成
……（第 2~4 节逐节起草，略）

[反思] 第 1 轮评审…
[评分] 总分: 9/10 (阈值: 7, 评审来源: llm)
[维度] 准确性 10 | 完整性 9 | 可读性 10 | 结构性 10
[反馈] 草稿结构清晰、技术表述准确……仅在部分细节上可进一步深化实例支撑。
[建议] 在'当前主流架构模式'部分，可补充一个简要对比表格……; '未来发展趋势'中提到的'分布式协同机制'可举例说明……
[判定] 质量达标 ✓ (分数 9 >= 阈值 7)

[完成] 最终版本已生成（共 0 轮反思改进）

大纲节数: 4
反思轮次: 0
质量达标: 是

— 执行轨迹（全程可追踪） —
01. [17:19:51] plan 生成 4 节大纲
02. [17:19:52] draft 完成第 1 节「AI Agent 的核心技术架构组成」
03. [17:19:54] draft 完成第 2 节「关键能力模块：感知、决策与行动」
04. [17:19:54] draft 完成第 3 节「当前主流架构模式与技术对比」
05. [17:19:55] draft 完成第 4 节「未来发展趋势：多模态、自主性与分布式协同」
06. [17:19:57] reflect 第 1 轮评审: 9/10 (llm)
07. [17:19:57] output 输出最终稿（678 字）

[存档] 本次真实成文与评审记录已保存: …/03_reflection/runs/demo1_basic_reflection.md

演示 1 的工程结论：首轮高分后立即输出，说明“少改一轮”与“多改几轮”一样，都是正确的控制流结果。

这个演示揭示了一个真实运行才能看到的事实：模型起草质量高时，反思循环的正确行为就是「评审一次、确认达标、立即输出」——首轮 9 分 ≥ 阈值 7，0 轮改进，不多烧一次 token。这不是流程没生效，而是终止条件在正确工作。末尾的执行轨迹把 7 个事件（1 次规划 + 4 次起草 + 1 次评审 + 输出）完整回放，每一步耗时多久、发生了什么一目了然。

运行输出稿如下：

# 演示 1: 基本反思改进

– 运行时间: 2026-08-09 17:19:57
– 任务: 分析 AI Agent 的技术架构和发展趋势
– 质量阈值: 7 | 最大轮次: 3
– 反思轮次: 0 | 质量达标: 是

## 写作大纲（LLM 规划）

1. AI Agent 的核心技术架构组成
2. 关键能力模块：感知、决策与行动
3. 当前主流架构模式与技术对比
4. 未来发展趋势：多模态、自主性与分布式协同

## 成文过程（各轮草稿版本）

### v1 · 初稿（分节起草组装）（678 字）

## AI Agent 的核心技术架构组成
AI Agent 的核心技术架构由感知模块、决策引擎、记忆系统、行动执行器和交互接口五大核心组件构成，各模块通过标准化协议协同工作，实现从环境信息输入到自主任务完成的闭环。其中，感知模块负责多源数据采集，决策引擎基于大模型与规划算法进行推理，记忆系统支撑上下文持久化，行动执行器对接外部工具或物理世界，交互接口则保障人机协同的自然流畅。

## 关键能力模块：感知、决策与行动
感知模块通过多模态传感器与自然语言理解技术，实时采集并解析环境中的视觉、语音、文本等异构数据；决策引擎基于大语言模型与强化学习算法，结合记忆系统中的历史上下文，动态生成任务规划与动作序列；行动执行器则将决策结果转化为具体操作，调用API、控制机器人或交互终端，完成闭环执行。

## 当前主流架构模式与技术对比
当前主流架构模式包括基于大语言模型的反应式架构、规划-执行式架构与记忆增强型架构，前者依赖LLM即时推理，响应快但缺乏长期规划；后者通过外挂记忆库与工具调用机制，显著提升复杂任务的持久性与准确性，代表系统如AutoGPT与LangChain在灵活性与可控性上形成鲜明对比。

## 未来发展趋势：多模态、自主性与分布式协同
未来AI Agent将深度融合多模态感知与生成能力，实现对视觉、语音、文本乃至传感器数据的统一理解与跨模态推理，显著提升环境适应力；同时，通过自主目标设定、长期记忆驱动的计划迭代与分布式Agent间动态协同机制，系统将从工具响应型转向目标导向型自治体，在开放环境中实现群体智能与任务分发的高效协同。

## 评审记录（完整）

### 第 1 轮评审 — 9/10（来源: llm）

– 维度: 准确性 10 | 完整性 9 | 可读性 10 | 结构性 10
– 反馈: 草稿结构清晰、技术表述准确，全面覆盖了AI Agent的核心架构、关键模块、主流模式与未来趋势，内容专业且逻辑严密。语言精炼，术语使用恰当，具备较高学术与工程参考价值。仅在部分细节上可进一步深化实例支撑与对比维度。
– 改进建议:
– 在‘当前主流架构模式’部分，可补充一个简要对比表格，明确各架构在响应速度、规划能力、记忆利用、工具调用支持等维度的差异，增强可比性。
– ‘未来发展趋势’中提到的‘分布式协同机制’可举例说明（如Multi-Agent System中的角色分配或通信协议），以提升具体性与说服力。
– 建议在感知模块中提及‘噪声鲁棒性’或‘实时性优化’等工程挑战，使技术分析更立体。

## 执行轨迹

01. [17:19:51] plan — 生成 4 节大纲
02. [17:19:52] draft — 完成第 1 节「AI Agent 的核心技术架构组成」
03. [17:19:54] draft — 完成第 2 节「关键能力模块：感知、决策与行动」
04. [17:19:54] draft — 完成第 3 节「当前主流架构模式与技术对比」
05. [17:19:55] draft — 完成第 4 节「未来发展趋势：多模态、自主性与分布式协同」
06. [17:19:57] reflect — 第 1 轮评审: 9/10 (llm)
07. [17:19:57] output — 输出最终稿（678 字）

## 最终成文（678 字）

## AI Agent 的核心技术架构组成
AI Agent 的核心技术架构由感知模块、决策引擎、记忆系统、行动执行器和交互接口五大核心组件构成，各模块通过标准化协议协同工作，实现从环境信息输入到自主任务完成的闭环。其中，感知模块负责多源数据采集，决策引擎基于大模型与规划算法进行推理，记忆系统支撑上下文持久化，行动执行器对接外部工具或物理世界，交互接口则保障人机协同的自然流畅。

## 关键能力模块：感知、决策与行动
感知模块通过多模态传感器与自然语言理解技术，实时采集并解析环境中的视觉、语音、文本等异构数据；决策引擎基于大语言模型与强化学习算法，结合记忆系统中的历史上下文，动态生成任务规划与动作序列；行动执行器则将决策结果转化为具体操作，调用API、控制机器人或交互终端，完成闭环执行。

## 当前主流架构模式与技术对比
当前主流架构模式包括基于大语言模型的反应式架构、规划-执行式架构与记忆增强型架构，前者依赖LLM即时推理，响应快但缺乏长期规划；后者通过外挂记忆库与工具调用机制，显著提升复杂任务的持久性与准确性，代表系统如AutoGPT与LangChain在灵活性与可控性上形成鲜明对比。

## 未来发展趋势：多模态、自主性与分布式协同
未来AI Agent将深度融合多模态感知与生成能力，实现对视觉、语音、文本乃至传感器数据的统一理解与跨模态推理，显著提升环境适应力；同时，通过自主目标设定、长期记忆驱动的计划迭代与分布式Agent间动态协同机制，系统将从工具响应型转向目标导向型自治体，在开放环境中实现群体智能与任务分发的高效协同。

2. 演示 2：高质量阈值——9 分压线达标

— 演示 2: 高质量阈值 —
任务: 撰写 LangGraph 框架的技术入门指南
quality_threshold = 9 (高要求)

[规划] 生成了 4 节大纲:
第 1 节: LangGraph 框架概述：基于图的有状态工作流设计
第 2 节: 核心组件解析：节点、边、状态与循环控制机制
第 3 节: 构建第一个 LangGraph 应用：从配置到运行的完整示例
第 4 节: 最佳实践与常见陷阱：优化性能与调试有状态流程
……（4 节逐节起草，略）

[反思] 第 1 轮评审…
[评分] 总分: 9/10 (阈值: 9, 评审来源: llm)
[维度] 准确性 10 | 完整性 9 | 可读性 10 | 结构性 10
[建议] 在'构建第一个应用'部分增加具体代码片段（如StateGraph的导入、add_node/add_edge的调用示例），增强可操作性;
……在'循环控制机制'中明确提及'recursion_limit'参数作为防无限循环的官方机制，以提升实用性
[判定] 质量达标 ✓ (分数 9 >= 阈值 9)

反思轮次: 0
最终是否达标: 是
评分演进: 9

演示 2 的工程结论：阈值是可配置的质量门槛，而不是固定写死的“及格线”。

阈值提到 9，首轮评审恰好 9 分压线达标。值得注意的是：评审在达标的同时仍给出了 3 条具体建议（补 add_node/add_edge 代码片段、提及 recursion_limit 防循环机制）——达标后不再消耗改进调用，但这些建议会留在 reflection_history 里，可供事后分析或人工采纳。这一轮如果模型给 8 分，循环就会转起来——阈值和模型能力之间的关系，只有真实运行才能标定。

运行输出稿如下：

# 演示 2: 高质量阈值

– 运行时间: 2026-08-04 17:20:03
– 任务: 撰写 LangGraph 框架的技术入门指南
– 质量阈值: 9 | 最大轮次: 3
– 反思轮次: 0 | 质量达标: 是

## 写作大纲（LLM 规划）

1. LangGraph 框架概述：基于图的有状态工作流设计
2. 核心组件解析：节点、边、状态与循环控制机制
3. 构建第一个 LangGraph 应用：从配置到运行的完整示例
4. 最佳实践与常见陷阱：优化性能与调试有状态流程

## 成文过程（各轮草稿版本）

### v1 · 初稿（分节起草组装）（792 字）

## LangGraph 框架概述：基于图的有状态工作流设计
LangGraph 是一个基于有向图的有状态工作流框架，专为构建复杂、可恢复的 LLM 应用而设计，它通过图结构显式定义任务间的依赖与流转逻辑，突破了传统线性链式调用的局限。与普通流程引擎不同，LangGraph 在每个节点执行后自动持久化状态，并支持条件分支与循环回溯，使对话代理、多轮推理等场景得以高效实现。

## 核心组件解析：节点、边、状态与循环控制机制
在 LangGraph 中，节点代表单个可执行的函数或 LLM 调用，边则定义节点间的条件流转规则，基于状态值动态决定下一步走向；状态以共享字典形式贯穿整个图执行过程，支持跨节点持久化与修改，结合循环边（如 `__start__` 或自定义条件边）可实现多轮对话、重试机制等复杂控制流。

## 构建第一个 LangGraph 应用：从配置到运行的完整示例
我们通过定义一个简单的“问答助手”图：节点`ask_llm`调用LLM生成回答，节点`check_response`根据状态中`"needs_clarification"`字段决定是否重问或结束；使用`StateGraph`构建图，添加节点与条件边（如`check_response -> ask_llm`当条件为真），最终编译并调用`app.invoke({"question": "什么是LangGraph？"})`即可启动有状态多轮交互流程。

## 最佳实践与常见陷阱：优化性能与调试有状态流程
为优化性能，应避免在节点中执行阻塞IO或重复调用LLM，建议使用缓存机制（如`@lru_cache`）或异步执行；调试时可通过启用`LangGraph`的`debug=True`模式或在状态中注入自定义日志字段，追踪状态变更路径，尤其注意循环边可能导致的无限递归，需设定最大迭代次数或退出条件。

## 评审记录（完整）

### 第 1 轮评审 — 9/10（来源: llm）

– 维度: 准确性 10 | 完整性 9 | 可读性 10 | 结构性 10
– 反馈: 草稿结构清晰、技术表述准确，完整覆盖了LangGraph的核心概念、组件、应用示例与最佳实践，语言专业且逻辑严密，适用于技术入门指南。仅在示例细节和术语一致性上略有提升空间。
– 改进建议:
– 在‘构建第一个应用’部分增加具体代码片段（如StateGraph的导入、add_node/add_edge的调用示例），增强可操作性
– 统一术语使用，如‘LangGraph’在全文应保持首字母大写且无空格（当前有‘LangGraph’与‘LangGraph’混用）
– 在‘循环控制机制’中明确提及‘recursion_limit’参数作为防无限循环的官方机制，以提升实用性

## 执行轨迹

01. [17:19:58] plan — 生成 4 节大纲
02. [17:19:59] draft — 完成第 1 节「LangGraph 框架概述：基于图的有状态工作流设计」
03. [17:19:59] draft — 完成第 2 节「核心组件解析：节点、边、状态与循环控制机制」
04. [17:20:01] draft — 完成第 3 节「构建第一个 LangGraph 应用：从配置到运行的完整示例」
05. [17:20:02] draft — 完成第 4 节「最佳实践与常见陷阱：优化性能与调试有状态流程」
06. [17:20:03] reflect — 第 1 轮评审: 9/10 (llm)
07. [17:20:03] output — 输出最终稿（792 字）

## 最终成文（792 字）

## LangGraph 框架概述：基于图的有状态工作流设计
LangGraph 是一个基于有向图的有状态工作流框架，专为构建复杂、可恢复的 LLM 应用而设计，它通过图结构显式定义任务间的依赖与流转逻辑，突破了传统线性链式调用的局限。与普通流程引擎不同，LangGraph 在每个节点执行后自动持久化状态，并支持条件分支与循环回溯，使对话代理、多轮推理等场景得以高效实现。

## 核心组件解析：节点、边、状态与循环控制机制
在 LangGraph 中，节点代表单个可执行的函数或 LLM 调用，边则定义节点间的条件流转规则，基于状态值动态决定下一步走向；状态以共享字典形式贯穿整个图执行过程，支持跨节点持久化与修改，结合循环边（如 `__start__` 或自定义条件边）可实现多轮对话、重试机制等复杂控制流。

## 构建第一个 LangGraph 应用：从配置到运行的完整示例
我们通过定义一个简单的“问答助手”图：节点`ask_llm`调用LLM生成回答，节点`check_response`根据状态中`"needs_clarification"`字段决定是否重问或结束；使用`StateGraph`构建图，添加节点与条件边（如`check_response -> ask_llm`当条件为真），最终编译并调用`app.invoke({"question": "什么是LangGraph？"})`即可启动有状态多轮交互流程。

## 最佳实践与常见陷阱：优化性能与调试有状态流程
为优化性能，应避免在节点中执行阻塞IO或重复调用LLM，建议使用缓存机制（如`@lru_cache`）或异步执行；调试时可通过启用`LangGraph`的`debug=True`模式或在状态中注入自定义日志字段，追踪状态变更路径，尤其注意循环边可能导致的无限递归，需设定最大迭代次数或退出条件。

3. 演示 3：无限反思防护——反思闭环真正转起来

阈值故意设成满分 10：

— 演示 3: 无限反思防护 —
quality_threshold = 10 (几乎不可能满足)
max_rounds = 2 (防护限制)

[反思] 第 1 轮评审…
[评分] 总分: 9/10 (阈值: 10, 评审来源: llm)
[维度] 准确性 10 | 完整性 8 | 可读性 10 | 结构性 10
[反馈] 文档结构清晰、内容专业且高度实用……唯一不足是缺少版本说明、依赖版本约束和扩展性说明，轻微影响完整性。
[建议] 在'系统架构与核心组件'中明确各组件的版本要求（如FastAPI>=0.100, Redis>=7.0）;
增加'版本与兼容性'小节，说明文档支持的系统版本与软件兼容性矩阵;
补充'扩展性说明'，简述如何添加新服务或替换组件的架构设计思路
[判定] 质量未达标 ✗ (分数 9 < 阈值 10)

[改进] 根据评审反馈重写草稿…
[改进后] 字数: 660 → 1772

[反思] 第 2 轮评审…
[评分] 总分: 10/10 (阈值: 10, 评审来源: llm)
[维度] 准确性 10 | 完整性 10 | 可读性 10 | 结构性 10
[判定] 质量达标 ✓ (分数 10 >= 阈值 10)

[完成] 最终版本已生成（共 1 轮反思改进）

实际反思轮次: 1
防护触发: 否
(max_rounds=2 防止了无限反思循环)

演示 3 的工程结论：循环必须既能启动，也能被刹住；只有改进能力、没有终止条件，同样不是生产系统。

解读这段“状态演进”：

注意真实 LLM 的评分不保证单调递增——上一次实测中同一场景模型第 1 轮给 9 分、改写退化后第 2 轮掉到 5 分（于是有了上面那道「异常膨胀」防护）。这次运行第 2 轮拿到满分提前达标，所以 max_rounds=2 防护没有触发；如果模型永远给不到满分，它就是那道防止 AI 钻牛角尖的刹车。

运行输出稿如下：

# 演示 3: 无限反思防护

– 运行时间: 2026-08-04 17:20:14
– 任务: 写一份完美的技术文档（质量阈值设为满分 10）
– 质量阈值: 10 | 最大轮次: 2
– 反思轮次: 1 | 质量达标: 是

## 写作大纲（LLM 规划）

1. 目标与适用范围：明确文档用途和读者群体
2. 系统架构与核心组件：详述技术栈与模块交互
3. 安装、配置与使用指南：分步操作与示例代码
4. 故障排查与最佳实践：常见问题解决方案与性能优化

## 成文过程（各轮草稿版本）

### v1 · 初稿（分节起草组装）（660 字）

## 目标与适用范围：明确文档用途和读者群体
本文档旨在为开发人员和系统运维人员提供清晰、可操作的技术指南，帮助其快速部署、配置并高效使用本系统。适用于具备基础Linux命令行与Python编程经验的技术用户，不适用于无技术背景的普通终端用户。

## 系统架构与核心组件：详述技术栈与模块交互
本系统采用微服务架构，核心组件包括基于FastAPI构建的API网关、使用Redis缓存的会话管理模块、以及由Celery调度的异步任务队列，各模块通过gRPC与AMQP协议实现低延迟通信，确保高并发场景下的稳定协同。

## 安装、配置与使用指南：分步操作与示例代码
请通过Docker Compose一键部署所有服务：`docker-compose up -d`，并确保Redis与RabbitMQ容器正常运行；随后在`config/.env`中配置API密钥与gRPC端口，运行`python init_db.py`初始化数据库，即可通过`curl -X GET http://localhost:8000/health`验证API网关就绪。

## 故障排查与最佳实践：常见问题解决方案与性能优化
若API网关返回502错误，检查RabbitMQ队列是否积压，可通过`docker logs rabbitmq`查看连接异常，并适当调高`CELERY_WORKER_CONCURRENCY`至CPU核心数的1.5倍以提升异步处理能力；建议为Redis启用持久化（AOF）并配置连接池，避免高并发下连接耗尽导致服务雪崩。

### v2 · 第 1 轮改进稿（1772 字）

## 目标与适用范围：明确文档用途和读者群体
本文档旨在为开发人员和系统运维人员提供清晰、可操作的技术指南，帮助其快速部署、配置并高效使用本系统。适用于具备基础Linux命令行与Python编程经验的技术用户，不适用于无技术背景的普通终端用户。

## 系统架构与核心组件：详述技术栈与模块交互
本系统采用微服务架构，核心组件包括：
– **API网关**：基于 FastAPI (>=0.100) 构建，提供RESTful接口与OpenAPI文档支持；
– **会话管理**：使用 Redis (>=7.0) 实现分布式缓存与会话状态存储，支持过期自动清理；
– **异步任务队列**：由 Celery (>=5.3) 驱动，依赖 RabbitMQ (>=3.9) 作为消息代理，通过AMQP协议实现任务分发；
– **服务通信**：各微服务间通过 gRPC (>=1.50) 实现高效、强类型远程调用。

各模块通过标准化接口与协议解耦，确保高并发场景下的稳定协同与低延迟通信。

## 版本与兼容性：支持的系统与软件兼容性矩阵
为确保系统稳定运行，建议使用以下正式支持的版本组合：

| 组件 | 推荐版本 | 最低兼容版本 | 操作系统支持 |
|——|———-|—————-|—————-|
| Python | 3.10.x – 3.11.x | 3.10.0 | Ubuntu 20.04+, CentOS 8+, macOS 11+ |
| Docker | 20.10.25+ | 20.10.0 | 所有主流 Linux 发行版、Docker Desktop |
| Docker Compose | v2.20.0+ | v2.0.0 | 同上 |
| Redis | 7.2.x | 7.0.0 | 同上 |
| RabbitMQ | 3.11.x | 3.9.0 | 同上 |
| FastAPI | 0.109.x | 0.100.0 | Python 3.10+ |
| Celery | 5.3.6 | 5.3.0 | 同上 |
| gRPC | 1.59.x | 1.50.0 | 同上 |

> ⚠️ 不兼容版本：Python <3.10、Redis <7.0、FastAPI <0.100。使用非推荐版本可能导致功能异常或安全漏洞。

## 安装、配置与使用指南：分步操作与示例代码
请通过Docker Compose一键部署所有服务：`docker-compose up -d`，并确保Redis与RabbitMQ容器正常运行；随后在`config/.env`中配置API密钥与gRPC端口，运行`python init_db.py`初始化数据库，即可通过`curl -X GET http://localhost:8000/health`验证API网关就绪。

## 扩展性说明：架构设计与组件替换策略
本系统采用松耦合架构设计，支持灵活扩展与组件替换：
– **添加新服务**：只需遵循gRPC服务契约定义新.proto文件，编写独立容器镜像，注册至服务发现模块（如Consul或内置服务注册表），并更新API网关路由配置，无需修改核心模块。
– **替换核心组件**：如需替换Redis为Memcached，只需实现统一的`CacheInterface`抽象层，替换具体实现类，其余业务逻辑无需重构；同理，AMQP可替换为Kafka，只需适配Celery的Broker后端配置即可。
– **水平扩展**：所有无状态服务（API网关、Celery Worker）均可通过增加容器实例实现负载均衡，配合Kubernetes或Docker Swarm可实现自动伸缩。

## 故障排查与最佳实践：常见问题解决方案与性能优化
若API网关返回502错误，检查RabbitMQ队列是否积压，可通过`docker logs rabbitmq`查看连接异常，并适当调高`CELERY_WORKER_CONCURRENCY`至CPU核心数的1.5倍以提升异步处理能力；建议为Redis启用持久化（AOF）并配置连接池，避免高并发下连接耗尽导致服务雪崩。

## 评审记录（完整）

### 第 1 轮评审 — 9/10（来源: llm）

– 维度: 准确性 10 | 完整性 8 | 可读性 10 | 结构性 10
– 反馈: 文档结构清晰、内容专业且高度实用，覆盖了目标用户所需的关键技术环节，语言精确，示例具体。唯一不足是缺少版本说明、依赖版本约束和扩展性说明，轻微影响完整性。
– 改进建议:
– 在‘系统架构与核心组件’中明确各组件的版本要求（如FastAPI>=0.100, Redis>=7.0）
– 增加‘版本与兼容性’小节，说明文档支持的系统版本与软件兼容性矩阵
– 补充‘扩展性说明’，简述如何添加新服务或替换组件的架构设计思路

### 第 2 轮评审 — 10/10（来源: llm）

– 维度: 准确性 10 | 完整性 10 | 可读性 10 | 结构性 10
– 反馈: 文档结构严谨、内容完整、技术细节精准，完全达到技术文档的满分标准。目标清晰、架构描述专业、兼容性矩阵详尽、部署步骤可操作、扩展性说明具备工程前瞻性，故障排查提供具体诊断与调优建议，无冗余或模糊表述。

## 执行轨迹

01. [17:20:04] plan — 生成 4 节大纲
02. [17:20:05] draft — 完成第 1 节「目标与适用范围：明确文档用途和读者群体」
03. [17:20:05] draft — 完成第 2 节「系统架构与核心组件：详述技术栈与模块交互」
04. [17:20:06] draft — 完成第 3 节「安装、配置与使用指南：分步操作与示例代码」
05. [17:20:07] draft — 完成第 4 节「故障排查与最佳实践：常见问题解决方案与性能优化」
06. [17:20:08] reflect — 第 1 轮评审: 9/10 (llm)
07. [17:20:13] improve — 第 1 轮改进: 660 → 1772 字
08. [17:20:14] reflect — 第 2 轮评审: 10/10 (llm)
09. [17:20:14] output — 输出最终稿（1772 字）

## 最终成文（1772 字）

## 目标与适用范围：明确文档用途和读者群体
本文档旨在为开发人员和系统运维人员提供清晰、可操作的技术指南，帮助其快速部署、配置并高效使用本系统。适用于具备基础Linux命令行与Python编程经验的技术用户，不适用于无技术背景的普通终端用户。

## 系统架构与核心组件：详述技术栈与模块交互
本系统采用微服务架构，核心组件包括：
– **API网关**：基于 FastAPI (>=0.100) 构建，提供RESTful接口与OpenAPI文档支持；
– **会话管理**：使用 Redis (>=7.0) 实现分布式缓存与会话状态存储，支持过期自动清理；
– **异步任务队列**：由 Celery (>=5.3) 驱动，依赖 RabbitMQ (>=3.9) 作为消息代理，通过AMQP协议实现任务分发；
– **服务通信**：各微服务间通过 gRPC (>=1.50) 实现高效、强类型远程调用。

各模块通过标准化接口与协议解耦，确保高并发场景下的稳定协同与低延迟通信。

## 版本与兼容性：支持的系统与软件兼容性矩阵
为确保系统稳定运行，建议使用以下正式支持的版本组合：

| 组件 | 推荐版本 | 最低兼容版本 | 操作系统支持 |
|——|———-|—————-|—————-|
| Python | 3.10.x – 3.11.x | 3.10.0 | Ubuntu 20.04+, CentOS 8+, macOS 11+ |
| Docker | 20.10.25+ | 20.10.0 | 所有主流 Linux 发行版、Docker Desktop |
| Docker Compose | v2.20.0+ | v2.0.0 | 同上 |
| Redis | 7.2.x | 7.0.0 | 同上 |
| RabbitMQ | 3.11.x | 3.9.0 | 同上 |
| FastAPI | 0.109.x | 0.100.0 | Python 3.10+ |
| Celery | 5.3.6 | 5.3.0 | 同上 |
| gRPC | 1.59.x | 1.50.0 | 同上 |

> ⚠️ 不兼容版本：Python <3.10、Redis <7.0、FastAPI <0.100。使用非推荐版本可能导致功能异常或安全漏洞。

## 安装、配置与使用指南：分步操作与示例代码
请通过Docker Compose一键部署所有服务：`docker-compose up -d`，并确保Redis与RabbitMQ容器正常运行；随后在`config/.env`中配置API密钥与gRPC端口，运行`python init_db.py`初始化数据库，即可通过`curl -X GET http://localhost:8000/health`验证API网关就绪。

## 扩展性说明：架构设计与组件替换策略
本系统采用松耦合架构设计，支持灵活扩展与组件替换：
– **添加新服务**：只需遵循gRPC服务契约定义新.proto文件，编写独立容器镜像，注册至服务发现模块（如Consul或内置服务注册表），并更新API网关路由配置，无需修改核心模块。
– **替换核心组件**：如需替换Redis为Memcached，只需实现统一的`CacheInterface`抽象层，替换具体实现类，其余业务逻辑无需重构；同理，AMQP可替换为Kafka，只需适配Celery的Broker后端配置即可。
– **水平扩展**：所有无状态服务（API网关、Celery Worker）均可通过增加容器实例实现负载均衡，配合Kubernetes或Docker Swarm可实现自动伸缩。

## 故障排查与最佳实践：常见问题解决方案与性能优化
若API网关返回502错误，检查RabbitMQ队列是否积压，可通过`docker logs rabbitmq`查看连接异常，并适当调高`CELERY_WORKER_CONCURRENCY`至CPU核心数的1.5倍以提升异步处理能力；建议为Redis启用持久化（AOF）并配置连接池，避免高并发下连接耗尽导致服务雪崩。

4. 演示 4：中断恢复——断点续跑不浪费已消耗的调用

本篇的重头戏。规划 + 4 节起草（共 5 次真实 LLM 调用）完成后，首轮评审时故障注入模拟 LLM 调用中断：

— 演示 4: 中断恢复（断点续跑） —
（故障注入: 第 1 轮评审时模拟 LLM 调用中断）

[起草] 第 4/4 节「Reflection 推理模式的优势、局限与未来方向」…
[进度] 起草 4/4 节完成

[中断] 模拟中断：第 1 轮评审时 LLM 调用失败（网络异常/进程崩溃）

[Checkpoint] 已保存进度: 大纲 4 节，起草完成 4 节
✓ 第 1 节: Reflection 推理模式的基本定义与核心机制
✓ 第 2 节: Reflection 与传统推理方法的对比分析
✓ 第 3 节: Reflection 在大型语言模型中的实际应用案例
✓ 第 4 节: Reflection 推理模式的优势、局限与未来方向

[恢复] 重新运行（输入传 None，同一 thread_id）→ 从评审继续:

[反思] 第 1 轮评审…
[评分] 总分: 7/10 (阈值: 8, 评审来源: fallback) ← LLM 评审输出解析失败，保底评估接管
[判定] 质量未达标 ✗ (分数 7 < 阈值 8)

[改进] 根据评审反馈重写草稿…
[改进后] 字数: 631 → 2078

[反思] 第 2 轮评审…
[评分] 总分: 9/10 (阈值: 8, 评审来源: llm)
[反馈] 草稿结构严谨、内容详实、论证有力……仅在个别表述上存在轻微冗余，如结尾'更更审慎'为笔误……
[判定] 质量达标 ✓ (分数 9 >= 阈值 8)

中断前完成: 规划 + 4 节起草（已持久化，恢复后未重复调用 LLM）
恢复后完成: 2 轮评审、1 轮改进

— 执行轨迹（全程可追踪） —
01. [17:20:15] plan 生成 4 节大纲
02. [17:20:16] draft 完成第 1 节「Reflection 推理模式的基本定义与核心机制」
03. [17:20:16] draft 完成第 2 节「Reflection 与传统推理方法的对比分析」
04. [17:20:17] draft 完成第 3 节「Reflection 在大型语言模型中的实际应用案例」
05. [17:20:18] draft 完成第 4 节「Reflection 推理模式的优势、局限与未来方向」
06. [17:20:20] reflect 第 1 轮评审: 7/10 (fallback)
07. [17:20:29] improve 第 1 轮改进: 631 → 2078 字
08. [17:20:31] reflect 第 2 轮评审: 9/10 (llm)
09. [17:20:31] output 输出最终稿（2078 字）

演示 4 的工程结论：真正节省成本的不是“失败后自动重试”，而是把任务拆成足够小的可持久化超步。

恢复的姿势只有一行：对同一 thread_id 传入 None 重新 invoke。LangGraph 从最近的 Checkpoint 取回状态，规划和 4 节起草的成果原样复用，直接从评审继续——5 次已消耗的 LLM 调用一次都没浪费。

这段真实输出还附赠了一个彩蛋：恢复后的第 1 轮评审，模型输出的 JSON 恰好解析失败，source 显示 fallback——启发式保底评估无缝接管给出 7 分，流程没有中断；改进后的第 2 轮评审 LLM 恢复正常，给出 9 分达标，还顺手抓出了改进稿里「更更审慎」这个真实笔误。三层兜底（容错解析 → 字段校验 → 启发式评估）不是防御性摆设，真实运行中真的会用到。

运行输出稿如下：

# 演示 4: 中断恢复（断点续跑）

– 运行时间: 2026-08-04 17:20:31
– 任务: 写一篇关于 Reflection 推理模式的分析短文
– 质量阈值: 8 | 最大轮次: 3
– 反思轮次: 1 | 质量达标: 是

## 写作大纲（LLM 规划）

1. Reflection 推理模式的基本定义与核心机制
2. Reflection 与传统推理方法的对比分析
3. Reflection 在大型语言模型中的实际应用案例
4. Reflection 推理模式的优势、局限与未来方向

## 成文过程（各轮草稿版本）

### v1 · 初稿（分节起草组装）（631 字）

## Reflection 推理模式的基本定义与核心机制
Reflection 推理模式是一种使大型语言模型能够通过自我审视、逐步校验与修正内部推理过程来提升输出准确性的机制，其核心在于引入“生成-评估-修正”的闭环循环，允许模型在输出前对自身推理路径进行多轮反思与调整。

## Reflection 与传统推理方法的对比分析
与传统推理方法依赖单次线性生成不同，Reflection 模式通过多次自我评估与迭代修正，显著降低了因初始假设偏差导致的错误累积；传统方法如链式思维（Chain-of-Thought）虽能结构化推理，但缺乏对中间步骤的动态校验能力，而 Reflection 则赋予模型“自我纠错”的元认知机制。

## Reflection 在大型语言模型中的实际应用案例
在GPT-4和Claude 3等模型中，Reflection机制被用于数学推理任务，模型在生成解答后自动回溯每一步推导，识别逻辑矛盾并重写错误步骤，使准确率提升15%以上；在代码生成场景中，模型通过反思生成的代码是否符合函数规范，主动修复类型错误与边界条件遗漏，显著提升可执行性。

## Reflection 推理模式的优势、局限与未来方向
Reflection模式虽能显著提升推理准确性，但其多轮迭代机制带来计算开销增大与响应延迟问题，尤其在实时交互场景中受限；未来方向可聚焦于轻量化反思模块设计、结合外部工具验证以降低自检偏差，并探索人类反馈与自动反思的协同优化策略。

### v2 · 第 1 轮改进稿（2078 字）

## Reflection 推理模式：机制、实践与演进

### 基本定义与核心机制
Reflection 推理模式是一种使大型语言模型（LLM）通过“生成-评估-修正”闭环实现自我改进的元推理机制。其本质是赋予模型“元认知”能力——即在生成答案后，主动回溯并批判性审查自身的推理路径，识别矛盾、漏洞或不合理假设，并进行迭代修正。这与传统单次线性推理形成鲜明对比，推动模型从“输出即终点”转向“输出即中间产物”。

### 与传统推理方法的对比分析
传统方法如链式思维（Chain-of-Thought, CoT）依赖线性生成：“解题步骤1 → 步骤2 → 步骤3 → 答案”。该方式虽结构清晰，但一旦某步出现偏差，错误将不可逆地传递至最终结果。

而 Reflection 模式引入动态校验环节。例如，在数学推理任务中，模型不仅输出解答，还会自问：“这一步是否合理？是否有更简洁的推导？”并生成反思语句，如：
> “第3步假设 x > 0，但原始条件仅说明 x ≠ 0，可能存在负解。需重审。”

随后模型根据反思内容重写推理路径，形成多轮迭代。这种“自我质疑+修正”的能力，使模型具备类似人类的审慎思维习惯。

### 代码生成中的具体实践示例
在代码生成场景中，Reflection 模式的有效性尤为突出。以使用 GPT-4 生成 Python 函数为例：

**初始生成（未反思）：**
```python
def find_max(arr):
max_val = 0
for x in arr:
if x > max_val:
max_val = x
return max_val
```
该代码在输入为负数数组（如 `[-5, -2, -10]`）时返回错误结果 `0`，因初始值设定错误。

**经 Reflection 修正后：**
```python
def find_max(arr):
# 反思：初始值设为0会导致全负数数组出错
# 修正：应使用第一个元素作为初始值
if len(arr) == 0:
return None
max_val = arr[0] # 修正点：使用实际首元素
for x in arr[1:]:
if x > max_val:
max_val = x
return max_val
```

模型通过反思语句：“初始值0不合理，应取数组首元素”识别逻辑缺陷，并主动修正边界条件与初始化逻辑。实验表明，经 Reflection 优化后，生成代码的单元测试通过率提升达 23%（Anthropic, 2024）。

### 实际应用与性能提升
在真实系统中，Reflection 已广泛应用于 GPT-4、Claude 3 和 Gemini 1.5 的高阶推理任务：
– **数学推理**（GSM8K 数据集）：Reflection 使准确率从 74.1% 提升至 89.6%（+15.5%）
– **逻辑谜题**（LogiQA）：通过多轮自我质疑，模型错误率下降 31%
– **代码生成**（HumanEval）：在要求“包含注释、边界检查、异常处理”的复杂任务中，可执行代码比例从 58% 上升至 81%

这些提升均源于模型在每轮反思中生成结构化评估（如：“该推论是否与前提矛盾？”“是否覆盖所有边界情况？”），并据此调整后续步骤。

### 优势、局限与未来方向
Reflection 模式显著提升了推理准确性和鲁棒性，尤其在复杂、多跳任务中优势明显。然而，其代价不容忽视：
– **计算开销**：每轮反思需额外生成与评估，推理延迟增加 40–70%
– **自检偏差**：模型可能因训练数据偏差而“错误地相信错误推理”
– **资源消耗**：不适合高并发、低延迟的实时场景（如客服对话）

未来研究可聚焦于三大方向：
1. **轻量化反思模块**：训练专用小型反思评估器（如 Reflection-LLM），降低主模型负担
2. **外部工具协同**：结合符号引擎（如 Wolfram Alpha）、代码解释器等进行交叉验证，减少纯内部反思的偏差
3. **人机协同反思**：允许人类对模型反思过程进行标注反馈（如“你忽略了这个约束”），构建可学习的反思知识库

### 总结
Reflection 推理模式标志着语言模型从“被动应答”向“主动思考”的关键跃迁。它通过构建自我审查的闭环机制，有效缓解了传统推理中的错误累积问题。实证表明，在数学、代码、逻辑等高精度场景中，Reflection 可带来显著性能增益。尽管存在延迟与资源成本的挑战，但随着轻量化反思架构与外部验证工具的发展，Reflection 有望成为下一代智能系统的核心推理范式——不仅更准确，更更“审慎”。

## 评审记录（完整）

### 第 1 轮评审 — 7/10（来源: fallback）

– 维度: 准确性 6 | 完整性 5 | 可读性 6 | 结构性 5
– 反馈: 初稿存在以下问题：1) 内容不够深入，缺少具体示例；2) 结构可以更清晰；3) 缺少总结段落。
– 改进建议:
– 增加具体代码示例
– 添加小标题分段
– 补充总结段落

### 第 2 轮评审 — 9/10（来源: llm）

– 维度: 准确性 10 | 完整性 9 | 可读性 10 | 结构性 10
– 反馈: 草稿结构严谨、内容详实、论证有力，精准把握了Reflection推理模式的核心机制与实践价值。案例具体、数据支撑充分，语言专业流畅，已达到优秀水平。仅在个别表述上存在轻微冗余，如结尾‘更更审慎’为笔误，且外部引用（如Anthropic, 2024）缺乏正式文献出处，可进一步规范化。
– 改进建议:
– 修正结尾处重复副词‘更更审慎’为‘更审慎’
– 为Anthropic, 2024等引用补充正式文献或报告来源，提升学术严谨性
– 可考虑在‘自检偏差’部分简要提及‘过自信偏差’（overconfidence bias）术语，增强理论深度

## 执行轨迹

01. [17:20:15] plan — 生成 4 节大纲
02. [17:20:16] draft — 完成第 1 节「Reflection 推理模式的基本定义与核心机制」
03. [17:20:16] draft — 完成第 2 节「Reflection 与传统推理方法的对比分析」
04. [17:20:17] draft — 完成第 3 节「Reflection 在大型语言模型中的实际应用案例」
05. [17:20:18] draft — 完成第 4 节「Reflection 推理模式的优势、局限与未来方向」
06. [17:20:20] reflect — 第 1 轮评审: 7/10 (fallback)
07. [17:20:29] improve — 第 1 轮改进: 631 → 2078 字
08. [17:20:31] reflect — 第 2 轮评审: 9/10 (llm)
09. [17:20:31] output — 输出最终稿（2078 字）

## 最终成文（2078 字）

## Reflection 推理模式：机制、实践与演进

### 基本定义与核心机制
Reflection 推理模式是一种使大型语言模型（LLM）通过“生成-评估-修正”闭环实现自我改进的元推理机制。其本质是赋予模型“元认知”能力——即在生成答案后，主动回溯并批判性审查自身的推理路径，识别矛盾、漏洞或不合理假设，并进行迭代修正。这与传统单次线性推理形成鲜明对比，推动模型从“输出即终点”转向“输出即中间产物”。

### 与传统推理方法的对比分析
传统方法如链式思维（Chain-of-Thought, CoT）依赖线性生成：“解题步骤1 → 步骤2 → 步骤3 → 答案”。该方式虽结构清晰，但一旦某步出现偏差，错误将不可逆地传递至最终结果。

而 Reflection 模式引入动态校验环节。例如，在数学推理任务中，模型不仅输出解答，还会自问：“这一步是否合理？是否有更简洁的推导？”并生成反思语句，如：
> “第3步假设 x > 0，但原始条件仅说明 x ≠ 0，可能存在负解。需重审。”

随后模型根据反思内容重写推理路径，形成多轮迭代。这种“自我质疑+修正”的能力，使模型具备类似人类的审慎思维习惯。

### 代码生成中的具体实践示例
在代码生成场景中，Reflection 模式的有效性尤为突出。以使用 GPT-4 生成 Python 函数为例：

**初始生成（未反思）：**
```python
def find_max(arr):
max_val = 0
for x in arr:
if x > max_val:
max_val = x
return max_val
```
该代码在输入为负数数组（如 `[-5, -2, -10]`）时返回错误结果 `0`，因初始值设定错误。

**经 Reflection 修正后：**
```python
def find_max(arr):
# 反思：初始值设为0会导致全负数数组出错
# 修正：应使用第一个元素作为初始值
if len(arr) == 0:
return None
max_val = arr[0] # 修正点：使用实际首元素
for x in arr[1:]:
if x > max_val:
max_val = x
return max_val
```

模型通过反思语句：“初始值0不合理，应取数组首元素”识别逻辑缺陷，并主动修正边界条件与初始化逻辑。实验表明，经 Reflection 优化后，生成代码的单元测试通过率提升达 23%（Anthropic, 2024）。

### 实际应用与性能提升
在真实系统中，Reflection 已广泛应用于 GPT-4、Claude 3 和 Gemini 1.5 的高阶推理任务：
– **数学推理**（GSM8K 数据集）：Reflection 使准确率从 74.1% 提升至 89.6%（+15.5%）
– **逻辑谜题**（LogiQA）：通过多轮自我质疑，模型错误率下降 31%
– **代码生成**（HumanEval）：在要求“包含注释、边界检查、异常处理”的复杂任务中，可执行代码比例从 58% 上升至 81%

这些提升均源于模型在每轮反思中生成结构化评估（如：“该推论是否与前提矛盾？”“是否覆盖所有边界情况？”），并据此调整后续步骤。

### 优势、局限与未来方向
Reflection 模式显著提升了推理准确性和鲁棒性，尤其在复杂、多跳任务中优势明显。然而，其代价不容忽视：
– **计算开销**：每轮反思需额外生成与评估，推理延迟增加 40–70%
– **自检偏差**：模型可能因训练数据偏差而“错误地相信错误推理”
– **资源消耗**：不适合高并发、低延迟的实时场景（如客服对话）

未来研究可聚焦于三大方向：
1. **轻量化反思模块**：训练专用小型反思评估器（如 Reflection-LLM），降低主模型负担
2. **外部工具协同**：结合符号引擎（如 Wolfram Alpha）、代码解释器等进行交叉验证，减少纯内部反思的偏差
3. **人机协同反思**：允许人类对模型反思过程进行标注反馈（如“你忽略了这个约束”），构建可学习的反思知识库

### 总结
Reflection 推理模式标志着语言模型从“被动应答”向“主动思考”的关键跃迁。它通过构建自我审查的闭环机制，有效缓解了传统推理中的错误累积问题。实证表明，在数学、代码、逻辑等高精度场景中，Reflection 可带来显著性能增益。尽管存在延迟与资源成本的挑战，但随着轻量化反思架构与外部验证工具的发展，Reflection 有望成为下一代智能系统的核心推理范式——不仅更准确，更更“审慎”。

五、常见问题与线上说明

（一）常见坑与排查

坑 1：阈值设太高，Agent 无限自我否定

• 现象：反思循环停不下来，每轮都「还能更好」，token 持续燃烧，迟迟不输出。

• 原因：quality_threshold 设成了模型几乎不可能达到的值（比如要求满分），又没有 max_rounds 兜底。

• 解决：必须设 max_rounds（见 should_continue_reflection 的第二个判断），并把阈值定在「足够好」而非「完美」。这跟你设接口重试次数一个道理——不能无限重试，得有上限。

坑 2：反馈太笼统，改进无从下手

• 现象：反思说「质量不够好」，改进节点不知道改什么，改完分数原地踏步甚至来回震荡。

• 原因：评审只输出一个分数和一句空话，没有可执行的改进项。

• 解决：评审 prompt 强制结构化输出（本 Demo 的 _REFLECT_SYSTEM_PROMPT）——总分 + 维度分 + improvements 列表，Improve 节点按清单逐条改。实测模型给出的建议能具体到「明确各组件版本要求（如 FastAPI>=0.100, Redis>=7.0）」这个粒度。

坑 3：LLM 改写退化，越改越差

• 现象：第 2 稿分数比第 1 稿还低，甚至草稿被无意义重复文本撑爆。

• 原因：改写是整篇重写，模型可能丢失上一版优点，高压阈值下还可能产生退化输出（实测：777 字被撑到 3.5 万字，9 分掉到 5 分）。

• 解决：对改写结果做健全性检查——为空、严重缩水、异常膨胀都回退到追加式保底改进。生产环境更进一步：保留历史最优版本，新版更差就回退。

坑 4：评审 JSON 解析失败，流程直接崩

• 现象：模型偶尔在 JSON 外面包一段解释文字，或者字段类型不对（"score": "九分"），json.loads 直接抛异常。

• 原因：把「LLM 输出合法 JSON」当成了必然。

• 解决：容错解析（剥代码块 + 正则提取）→ 字段校验规整（_sanitize_evaluation）→ 启发式保底（evaluate_quality），三层兜底。并用 source 字段标记评分来源，方便事后审计。

坑 5：中断后从头重跑，白烧 token

• 现象：多轮反思执行到一半网络抖动，重跑时规划、起草全部重来。

• 原因：没有 Checkpointer，或者把「起草全文」做成了一个节点——一个超步内的进度不会被保存。

• 解决：把大动作拆成小超步（本 Demo 的 draft 每次只写一节），配 Checkpointer；恢复时同一 thread_id 传 None。

参数调优时，建议至少记录这四组数据

不要只记录“最后成功了没有”。真正能帮助你优化系统的是：它在哪一轮成功、为此调用了多少次模型、哪些反馈最有效、哪些防线被触发过。

（二）工程化检查表

当这套 Demo 要进入真实业务，建议用下面五个问题做上线前检查。

• 评审维度要明确：不能只问「好不好」，要拆成准确性、完整性、可读性、结构性等维度分别打分（本 Demo 的 dimensions 已实现）。

• 改进建议要可执行：「写好点」没用，「增加代码示例」才有用——靠结构化 prompt 约束。

• 阈值与轮次要平衡：阈值太高会无限反思，太低会输出低质量内容；轮次太少改不到位，太多浪费 token。

• 改写结果要校验：真实 LLM 的重写可能缩水或退化膨胀，必须有健全性检查和回退路径。

• 进度要可恢复：每轮反思都是真金白银的 LLM 调用，Checkpoint + 小超步让中断不再意味着从头再来。

（三）生产级方案：从“能运行”到“可治理”

六、总结：Reflection 是 Agent 的内置质量系统

Reflection 模式的本质是内置 code review——Agent 自己当自己的 reviewer，让“生成 → 打分 → 改进”收敛到“足够好”，而不是盲目追求“永远还能更好”。本篇把它做成了完整的生产雏形：LLM 规划大纲、分步起草、结构化评审、按反馈重写，全程轨迹可追踪，配合 max_rounds 防无限否定、多层保底防 LLM 输出不可用、Checkpoint 断点续跑防 token 白烧。

Reflection 关注的是「质量好不好」这种主观评价。但有一类问题更硬核——对错有明确标准：代码能不能跑通、JSON 格式合不合法、单元测试过不过。这时候不需要「打分」，需要的是「执行 → 验证 → 修复」的确定性闭环。

👉 下一篇：《Self-Correction：让 AI 自己改 Bug》，我们会看到 Agent 如何执行任务、自动验证结果、对错误分类并升级修复策略，以及如何用 max_retries 防止它反复修复同一个错误。

本文最终带走的五句话：

 系列导航：LangGraph从零构建生产级 AI Agent 平台的递进式学习项目-CSDN博客