参考自：all-agentic-architectures

1. 什么是反思架构 (Reflection Architecture)？

反思架构是指 Agent 在向用户返回最终答案之前，先对其自身的输出进行批评（Critique）和修订（Revise）的一种工作模式。

• 核心机制：它摒弃了单次生成的模式，转而采用“生成 -> 评估 -> 改进”的多步内部独白机制。
• 类比：这模仿了人类在处理复杂任务时的“起草、审查、编辑”过程，旨在捕捉错误并提高最终输出的质量。

---

2. 宏观工作流

该过程通常分为三个连续的阶段：

1. 生成 (Generate)
   Agent 根据用户的 Prompt 产出初稿或初步解决方案。
2. 批评 (Critique)
   Agent 切换角色成为“批评者”或“审查员”。它会进行自我反问，例如：

• • “这个答案有什么问题？”
  • “遗漏了什么关键信息？”
  • “是否存在逻辑漏洞或代码 Bug？”

1. 优化 (Refine)
   利用在批评阶段获得的见解，Agent 对初稿进行修改，生成最终的改进版本。

---

3. 应用场景与优缺点

3.1 典型场景

• 代码生成：充当 Code Reviewer，修复 Bug 并优化代码风格。
• 复杂摘要：查漏补缺，确保摘要的全面性和准确性。
• 内容创作：优化邮件或博客的语气、清晰度和感染力。

3.2 优缺点分析

维度	描述
优点	质量提升：通过自我修正，产出更准确、逻辑更严密的回答。 低门槛：只需单个 LLM 即可实现，无需复杂外部工具。
缺点	自我局限：模型无法修复超出其知识边界的错误。 延迟与成本：涉及多次 LLM 调用，响应更慢，成本更高。

---

4. 环境配置与依赖安装

4.1 安装依赖

在 Jupyter 环境中运行：

%pip install langchain-openai langchain langgraph rich python-dotenv pygraphviz

4.2 初始化配置

这里使用兼容 OpenAI 接口的模型（如阿里百炼 Qwen）。

import os
import json
from typing import List, TypedDict, Optional
from dotenv import load_dotenv

# LangChain 组件
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END

# 美化输出
from rich.console import Console
from rich.markdown import Markdown
from rich.syntax import Syntax

# 加载环境变量
load_dotenv()

# 配置 LangSmith 追踪 (可选)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Agentic Architecture - Reflection"

# 初始化 LLM
llm = ChatOpenAI(
    model="qwen-flash", # 这里使用 Qwen 模型
    openai_api_key=os.environ["DASHSCOPE_API_KEY"],
    openai_api_base="[https://dashscope.aliyuncs.com/compatible-mode/v1](https://dashscope.aliyuncs.com/compatible-mode/v1)",
    temperature=0.1
)

console = Console()
print("环境初始化完成。")

---

5. 构建核心组件

Reflection 架构依赖结构化数据来保证流程的稳定性。我们需要定义三个核心 Pydantic 模型。

5.1 定义数据结构 (Pydantic Models)

# 1. 初稿结构
class DraftCode(BaseModel):
    """生成的初始代码草稿"""
    code: str = Field(description="解决用户请求的 Python 代码")
    explanation: str = Field(description="代码工作原理的简要解释")

# 2. 评价结构
class Critique(BaseModel):
    """代码自我评价"""
    has_errors: bool = Field(description="代码是否有潜在 Bug 或逻辑错误？")
    is_efficient: bool = Field(description="代码是否高效？")
    suggested_improvements: List[str] = Field(description="具体的改进建议列表")
    critique_summary: str = Field(description="评价总结")

# 3. 终稿结构
class RefinedCode(BaseModel):
    """优化后的最终代码"""
    refined_code: str = Field(description="最终的、改进后的 Python 代码")
    refinement_summary: str = Field(description="基于批评意见所做的更改总结")

5.2 定义处理节点 (Nodes)

节点 1：生成器 (Generator)

def generate_node(state):
    console.print("--- 1. 生成初始代码 ---")
    generator_llm = llm.with_structured_output(DraftCode)
    
    prompt = f"""You are an expert Python programmer. Write a Python function to solve the following request.
    Provide a simple, clear implementation and an explanation.
    
    Request: {state['user_request']}
    """
    draft = generator_llm.invoke(prompt)
    return {"draft": draft.model_dump()}

节点 2：批评者 (Critic)

def critic_node(state):
    console.print("--- 2. 评价初始代码 ---")
    critic_llm = llm.with_structured_output(Critique)
    
    code_to_critique = state["draft"]["code"]
    
    prompt = f"""You are an expert code reviewer. Analyze the code for:
    1. Bugs and Errors
    2. Efficiency and Best Practices
    
    Code to Review:
    ```python
    {code_to_critique}
    ```
    """
    critique = critic_llm.invoke(prompt)
    return {"critique": critique.model_dump()}

节点 3：优化器 (Refiner)

def refine_node(state):
    console.print("--- 3. 完善代码 ---")
    refiner_llm = llm.with_structured_output(RefinedCode)
    
    draft_code = state["draft"]["code"]
    critique = json.dumps(state["critique"], indent=2)
    
    prompt = f"""Rewrite the original code implementing all suggested improvements.
    
    Original Code:
    ```python
    {draft_code}
    ```
    
    Critique:
    {critique}
    """
    refined_code = refiner_llm.invoke(prompt)
    return {"refined_code": refined_code.model_dump()}

---

6. 使用 LangGraph 编排工作流

我们将节点连接成一个线性的图结构：Start -> Generator -> Critic -> Refiner -> End。

# 1. 定义图状态
class ReflectionState(TypedDict):
    user_request: str
    draft: Optional[dict]
    critique: Optional[dict]
    refined_code: Optional[dict]

# 2. 构建图
graph_builder = StateGraph(ReflectionState)

# 添加节点
graph_builder.add_node("generator", generate_node)
graph_builder.add_node("critic", critic_node)
graph_builder.add_node("refiner", refine_node)

# 定义边（流程走向）
graph_builder.set_entry_point("generator")
graph_builder.add_edge("generator", "critic")
graph_builder.add_edge("critic", "refiner")
graph_builder.add_edge("refiner", END)

# 编译图
reflection_graph = graph_builder.compile()

# 可视化（需要 pygraphviz）
try:
    from IPython.display import display, Image
    display(Image(reflection_graph.get_graph().draw_png()))
except Exception:
    print("可视化依赖未安装，跳过绘图。")

---

7. 执行与评估

7.1 运行工作流

user_request = "Write a fast sorting function and a test case for it."
initial_input = {"user_request": user_request}

final_state = None

print(f"🚀 开始处理任务: {user_request}")

# 流式运行以捕获最终状态
for state in reflection_graph.stream(initial_input, stream_mode="values"):
    final_state = state

print("✅ 工作流结束")

7.2 量化评估 (LLM-as-a-Judge)

使用另一个 LLM 调用作为“裁判”，对比初稿和终稿的质量。

class CodeEvaluation(BaseModel):
    correctness_score: int = Field(description="逻辑正确性评分 (1-10)")
    efficiency_score: int = Field(description="算法效率评分 (1-10)")
    justification: str = Field(description="评分理由")

judge_llm = llm.with_structured_output(CodeEvaluation)

def evaluate_code(code: str):
    return judge_llm.invoke(f"Evaluate this Python code logic and efficiency (1-10):\n```python\n{code}\n```")

# 执行评估
if final_state and 'draft' in final_state and 'refined_code' in final_state:
    print("--- 初稿评分 ---")
    print(evaluate_code(final_state['draft']['code']).model_dump())
    
    print("\n--- 终稿评分 ---")
    print(evaluate_code(final_state['refined_code']['refined_code']).model_dump())

---

8. 总结

本教程通过 Reflection Pattern 展示了如何将一个普通的 LLM 转化为具备自我纠错能力的智能体。核心在于：

1. 结构化输出：使用 Pydantic 强制模型按格式思考。
2. 角色分离：生成与评价解耦，模拟 Code Review 流程。
3. 图编排：使用 LangGraph 清晰地定义状态流转。