作为人类，当我们进行自我反思时，我们会反思自己的行为，然后制定我们的下一步和行动。我们会继续对下一步行动进行自我反思，直到我们感觉自己已尽最大努力完成任务。

AI Agent 作为一种能够感知环境、做出决策并采取行动的系统，反思（Reflection）是最基本的一个能力

1. 反思（Reflection）是什么？ 

在AI Agent的领域中，Reflection（反思）是一个重要的概念，它是一种提示策略，用于提高Agent和类似的人工智能系统的运行质量和成功率。包括提示大型语言模型对过往行为进行反思和评判，有时还会融合工具观测作为外部信息的来源，帮助模型更好地回顾和评估自己的行为。

人们经常提到“系统一”和“系统二”这两种思维方式，其中系统一是反应性或本能的，系统二则更加系统化和反思性。

2002年诺贝尔经济学奖得主丹尼尔·卡尼曼于2011年出版的畅销书《思考，快与慢》，将人类的思维归纳为两大思考模式：系统一快速、直觉且情绪化。系统二较慢、较具计划性且更仰赖逻辑。

反思可以帮助LLM系统跳出纯粹的系统一“思维”模式，朝着更接近系统二行为的方向发展

2. 反思（Reflection）的核心原理 

Reflection 的基本原理是通过模型的自我审查和自我反馈来改进初步生成的内容。

这个过程通常分为两个阶段：生成和反思。

生成阶段，AI 模型会根据用户提供的输入或任务要求，生成一个初步的结果（例如文章、代码或解决方案）。

反思阶段，是让 AI 模型对这个生成的结果进行评估，找出其中的不足之处并进行自我纠正或改进。

这两个阶段的交替进行，构成了 Reflection 的循环流程。

在每一轮反思中，AI 模型会回顾生成内容，评估其质量，并提出修改的建议。这些修改建议可以是细节的调整，也可以是对生成结构的重新组织，目的是提升内容的连贯性、逻辑性、表达清晰度，或是加强对用户需求的适应性。

最重要的是，反思不单是对内容的审视，也是对内容背后生成策略的审视。AI 在反思阶段不仅仅在细节上做出改变，还可能重新评估生成的逻辑架构或推理过程，以确保最终结果更加符合实际需求。因此，Reflection 在 AI 系统中扮演了“审稿人”或“教师”的角色，具备改进、优化、并完善生成输出的能力。

3. 为什么需要反思（Reflection） 

在生成模型中，AI 往往在初步生成的内容中存在一定的不足，可能是逻辑不清晰、细节不充分、结构松散或者与用户需求不完全匹配。

通过 Reflection，AI 可以对这些问题进行自我审查，识别出哪些部分可以改进，哪些表达可以更清晰，从而生成更加符合要求的内容。

Reflection 提高了 AI 对用户需求的灵活响应能力，并使得模型输出具有更高的个性化和相关性。

并且，Reflection 有助于优化生成内容的多样性和创新性。通过不断反思和修改，AI 系统能够探索不同的生成路径，从而生成更多样化的内容。

在一些创意性强的任务中，反思机制能够激发 AI 从不同的角度审视问题并产生更具创新性或深度的输出，避免单一、僵化的生成模式。

5. 代码实现 

LangGraph 是一个用于构建有状态的多参与者应用程序的库。状态可以是聊天消息列表。而参与者是一个LLM链。因此，LangGraph 能够在多个计算步骤中以循环的方式协调多个LLMs。

循环对于类似Agent的行为很重要，在这种情况下，你会在一个循环中调用LLM，询问它接下来该采取什么行动。这正是自我反思所需的。

定义模型：

import os  # 导入操作系统模块

os.environ['OpenAI_API_KEY'] = 'sk-3c5e'

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain_openai import OpenAI, ChatOpenAI
llm = ChatOpenAI(  # 创建ChatOpenAI的实例
    model="deepseek-chat",  # 指定模型
    temperature=0,  # 设置温度
    base_url="https://api.deepseek.com"  # 设置基础URL
)

创建生成链:

# 创建生成链
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是一名论文助手，负责撰写出色的五段式文章。"
            " 为用户请求生成最佳文章。"
            " 如果用户提供批评意见，请根据之前的版本进行修改并回应",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

generate = prompt | llm

创建反思链:

# 创建反思链
reflection_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是一名教师，负责批改论文。生成对该用户提交的评语及建议."
            " 给出详尽的建议，并提出对长度、行文、段落结构、公式化表达、文本的衔接、深度、风格等方面的要求。",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)
reflect = reflection_prompt | llm

创建状态图:

# 创建状态图
# 导入类型注解工具，Annotated 用于标注字段类型和额外的元数据，List 和 Sequence 表示容器类型
from typing import Annotated, List, Sequence
# 导入 StateGraph、START 和 END，用于构建状态图和标记起始与结束节点
from langgraph.graph import END, StateGraph, START
# 导入用于操作消息的工具函数 add_messages
from langgraph.graph.message import add_messages
# 导入用于保存状态的内存检查点类 MemorySaver
from langgraph.checkpoint.memory import MemorySaver
# 导入 TypedDict，用于创建严格类型的字典
from typing_extensions import TypedDict


# 定义 State 类型，包含一个字段 messages，其类型是一个被 Annotated 修饰的列表
class State(TypedDict):
    messages: Annotated[list, add_messages]


# 异步函数 generation_node，接收当前状态 state，返回新的状态
asyncdef generation_node(state: State) -> State:
    # 使用异步生成器生成消息并返回状态
    return {"messages": [await generate.ainvoke(state["messages"])]}


# 异步函数 reflection_node，用于处理和反思消息，接收当前状态并返回更新后的状态
asyncdef reflection_node(state: State) -> State:
    # 定义一个类型映射字典，将消息类型从 AI 转换为 Human 或反之
    cls_map = {"ai": HumanMessage, "human": AIMessage}
    # 保持用户的第一个请求消息不变，并转换后续消息的类型
    translated = [state["messages"][0]] + [
        cls_map[msg.type](content=msg.content) for msg in state["messages"][1:]
    ]
    # 调用反思工具 reflect 的异步方法处理转换后的消息
    res = await reflect.ainvoke(translated)
    # 将反思结果视为用户反馈消息并返回新的状态
    return {"messages": [HumanMessage(content=res.content)]}


# 创建一个状态图实例，并指定节点的输入和输出类型为 State
builder = StateGraph(State)
# 向状态图中添加生成节点，节点函数为 generation_node
builder.add_node("generate", generation_node)
# 向状态图中添加反思节点，节点函数为 reflection_node
builder.add_node("reflect", reflection_node)
# 在状态图中添加起点到生成节点的边
builder.add_edge(START, "generate")


# 定义函数 should_continue，根据状态判断是否继续执行或结束
def should_continue(state: State):
    # 如果消息长度超过 6，则终止流程
    if len(state["messages"]) > 6:
        return END  # 返回结束标记
    return"reflect"# 否则返回反思节点


# 在状态图中为生成节点添加条件边，根据 should_continue 的结果决定下一步
builder.add_conditional_edges("generate", should_continue)
# 在状态图中添加反思节点到生成节点的边，形成循环
builder.add_edge("reflect", "generate")
# 创建一个内存保存器实例，用于保存和恢复状态图的检查点
memory = MemorySaver()

# 编译状态图，传入检查点保存器作为参数
graph = builder.compile(checkpointer=memory)

运行状态图:

# 创建初始请求
initial_request = HumanMessage(
    content="写一篇关于为什么《小王子》在现代童年中具有相关性的文章。"
)
config = {"configurable": {"thread_id": "1"}}
# 运行状态图
asyncdef run_graph():
    # 创建初始状态
    initial_state = {"messages": [initial_request]}
    
    # 运行状态图
    asyncfor event in graph.astream(initial_state,config):
        for key, value in event.items():
            print(f"output from node '{key}':")
            print("---")
            print(value)
        print("\n===\n")

# 运行主程序
import asyncio
asyncio.run(run_graph())

6. 反思过程日志 

7. 应用场景 

反思（Reflection）在很多领域发挥着重大作用，如文章写作、代码生成等。

例如，在文章写作任务中，AI 首先会根据题目生成一篇文章，反思器会审查文章的结构、逻辑性及细节，再次提供改进意见，生成器根据这些反馈调整文章内容，直到达到最佳效果。

在编程任务中，生成器生成初步代码，反思器分析代码的可读性、效率和可维护性，然后提出优化建议，帮助提升代码质量。

反思（Reflection）在 AI Agent 场景中的作用不可忽视。它不仅提升了生成内容的质量，也增强了 AI 系统的灵活性和适应性。通过不断的反思和改进，AI 系统能够生成更符合用户需求的内容，实现更高效、更智能的任务处理。随着 AI 技术的进步，Reflection 将成为推动 AI 系统发展的重要工具，带来更多创新的应用场景。