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

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

在反思过程中，模型不再局限于传统的“输入-输出”模式，而是引入了“生成-评估-改进”的动态循环流程。这种机制能够帮助AI模型在初步生成内容后，主动回顾其输出，分析存在的不足，并不断优化，以生成更符合用户需求的内容。

然而，仅依靠模型自身的能力并不足以实现全面优化。在某些复杂或专业领域中，模型可能因训练数据的局限性或知识库的不完整性而面临瓶颈。

因此，为了让AI具备基本的反思能力，外部工具的引入显得尤为重要。这些工具可以为模型提供最新的知识和实时数据，帮助其更好地识别和补足初步生成内容中的缺陷。

通过反思与工具的结合，AI可以模拟人类完善答案的过程：先给出初步回答，然后检查是否存在疏漏，最终借助外部资源完善答案。

1. 基于外部知识的增强反思 

在面对问题时，人类会先根据已有知识给出一个答案，然后再检查自己的回答是否存在疏漏，并通过查阅资料或与他人讨论来完善答案。

将这一理念应用于生成式大模型的场景中，就是让模型在初步生成答案后主动评估其准确性和全面性，并通过调用工具补足可能的缺陷。

在这个过程中，工具扮演着关键角色。模型本身并非全知全能，工具的作用是弥补这些盲区。

例如，通过搜索引擎获取最新信息，或调用特定 API 来获取实时数据，工具可以为模型提供丰富的补充材料。而基于外部知识的增强反思机制则是促使模型主动发现需要工具帮助的地方，使得工具调用变得更加精准和高效。

2. 实现思路 

构建一个具备反思应答能力的生成式模型，核心在于合理设计其生成、反思、工具调用和答案修订的流程。

基于外部知识的增强反思涉及的关键节点包括初始响应（Responder）、工具（Execute Tools）、响应修订（Revisor）。基本流程为：

1. 用户请求：用户输入查询问题或任务需求
2. 初始生成：初步答案、初步反思结果、以及增强建议。
3. 执行工具：执行外部工具获得结果，这些结果用来优化输出
4. 响应修订：根据工具结果对输出进行修订，并再次给出反思与增强建议（经过3、4步的多轮修订、反思、外部验证的迭代，逐步优化输出）
5. 最终输出：返回给用户经过验证和改进的高质量结果

在这一过程中，模型会结合工具返回的数据对初步答案进行优化。优化不仅包括简单地添加工具返回的内容，还需要根据问题的上下文调整答案的结构，使其更加连贯、准确。

3. 代码实现 

代码已放入学习资料包

3.1 定义基本对象

通过环境变量配置实现 OpenAI、Tavily 等外部服务的无缝接入，构建语言模型实例作为核心处理引擎。

数据模型采用分层建模策略：Reflection 模型聚焦答案质量反思维度，AnswerQuestion 模型实现答案、反思、工具调用的三元数据封装，而继承扩展的 ReviseAnswer 模型通过新增文献引用字段强化验证依据。

# 导入操作系统模块
import os

# 设置OpenAI API密钥环境变量
os.environ['OpenAI_API_KEY'] = 'hk-iwtb427'# 设置OpenAI API密钥

# 设置Tavily搜索API密钥环境变量
os.environ['TAVILY_API_KEY'] = 'tvly-pd1zc'

# LangSmith 环境变量 (可选) ,如果需要使用 LangSmith 功能，请在环境变量中设置以下变量
os.environ['LANGCHAIN_TRACING_V2'] = "true"
os.environ['LANGCHAIN_ENDPOINT'] = "https://api.smith.langchain.com"
os.environ['LANGCHAIN_API_KEY'] = "lsv2_pt_03c513"
os.environ['LANGCHAIN_PROJECT'] = "reflexion"

##############################################################################################

# 导入LangChain的OpenAI模块，用于与OpenAI API交互
from langchain_openai import OpenAI, ChatOpenAI

llm = ChatOpenAI(  # 创建ChatOpenAI的实例
    model="gpt-4o-mini",  # 指定模型
    temperature=0,  # 设置温度
    base_url="https://api.openai-hk.com/v1"# 设置基础URL
)

##############################################################################################

# 导入Tavily搜索工具，用于网络搜索
from langchain_community.tools.tavily_search import TavilySearchResults
# 导入Tavily搜索API封装工具
from langchain_community.utilities.tavily_search import TavilySearchAPIWrapper

# 初始化Tavily搜索API封装实例
search = TavilySearchAPIWrapper()
# 创建Tavily搜索工具实例，设置最大返回结果数为5
tavily_tool = TavilySearchResults(api_wrapper=search, max_results=5)

##############################################################################################
# 导入LangChain核心消息类型
from langchain_core.messages import HumanMessage, ToolMessage
# 导入OpenAI工具的输出解析器
from langchain_core.output_parsers.openai_tools import PydanticToolsParser
# 导入聊天提示模板和消息占位符
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导入Pydantic验证错误类
from pydantic import ValidationError

# 导入Pydantic基础模型和字段定义
from pydantic import BaseModel, Field


# 定义反思内容，评估回答的质量信息
class Reflection(BaseModel):
    # 定义缺失内容评判字段
    missing: str = Field(description="对所缺失内容的评判。")
    # 定义多余内容评判字段
    superfluous: str = Field(description="对多余内容的评判。")


# 定义回答问题内容，包含答案、反思内容和搜索查询关键字
class AnswerQuestion(BaseModel):
    """回答问题。提供一个答案，进行思考，然后后续跟进搜索查询以改进答案。"""

    # 定义答案字段，要求约250词的详细回答
    answer: str = Field(description="针对问题的约250词的详细答案。")
    # 定义反思字段，包含对答案的评估
    reflection: Reflection = Field(description="你对最初答案的思考。")
    # 定义搜索查询字段，用于改进答案
    search_queries: list[str] = Field(
        description="1-3个搜索关键词，用于研究改进以回应对你当前答案的批评。"
    )


# 定义带重试机制的响应器，包装一个链，并对链的返回内容做校验
class ResponderWithRetries:
    """重新应答"""
    # 初始化响应器实例
    def __init__(self, runnable, validator):
        # 设置可运行对象
        self.runnable = runnable
        # 设置验证器
        self.validator = validator

    # 定义响应方法，最多重试3次
    def respond(self, state: dict):
        # 初始化响应
        response = []
        # 尝试最多3次
        for attempt in range(3):
            # 调用可运行对象生成响应
            response = self.runnable.invoke(
                # 传入消息状态
                {"messages": state["messages"]}, 
                # 添加尝试次数标签
                {"tags": [f"attempt:{attempt}"]}
            )
            # 尝试验证响应
            try:
                self.validator.invoke(response)
                # 如果验证成功，返回响应
                return {"messages": response}
            # 如果验证失败，记录错误并重试
            except ValidationError as e:
                print("碰到异常：")
                state = state + [
                    response,
                    # 创建工具消息包含错误信息
                    ToolMessage(
                        content=f"{repr(e)}\n\nPay close attention to the function schema.\n\n"
                        + self.validator.schema_json()
                        + " Respond by fixing all validation errors.",
                        tool_call_id=response.tool_calls[0]["id"],
                    ),
                ]
        return {"messages": response}

3.2 定义初始响应节点Responder

在反思应答的第一阶段，模型会生成初步答案并附带反思内容。与传统的单一答案生成不同，这一阶段的重点在于模型需要具备基本的自我评估能力。

初步生成的答案需要解决用户问题的核心需求，而反思内容则需要对答案进行质量分析，包括但不限于以下几点：

1. 是否遗漏了问题中的重要部分。
2. 是否包含冗余或无关信息。
3. 是否需要外部工具辅助验证或补充内容。

例如，当用户提出“如何解决全球变暖的问题”这一复杂问题时，模型在给出答案后，可能会附带以下反思内容：“答案中未详细提及具体政策的实施案例，可以通过搜索工具查找更多信息。”这一反思内容将直接指导后续工具的调用。

# 导入日期时间模块
import datetime

# 创建actor提示模板，用于生成研究员的响应
actor_prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """ 你是专家级研究员。
                当前时间: {time}

                1. {first_instruction}
                2. 反思并严格评判你的答案，以最大化改进答案。
                3. 推荐Search Query以研究信息并改进你的答案。""",
        ),
        # 使用消息占位符插入对话历史
        MessagesPlaceholder(variable_name="messages"),
        (
            "user",
            "\n\n<system>反思用户的原始问题以及到目前为止采取的行动。使用 {function_name} 函数进行响应</reminder>",
        ),
    ]
).partial(
    # 设置当前时间参数
    time=lambda: datetime.datetime.now().isoformat(),
)
# 创建初始答案生成链
initial_answer_chain = actor_prompt_template.partial(
    # 设置初始指令
    first_instruction="提供一个约250字的详细答案。",
    # 设置函数名称
    function_name=AnswerQuestion.__name__,
) | llm.bind_tools(tools=[AnswerQuestion])  # 绑定工具
# 创建验证器，用于验证答案格式
validator = PydanticToolsParser(tools=[AnswerQuestion])

# 创建带重试机制的初始响应器
first_responder = ResponderWithRetries(
    # 设置可运行对象
    runnable=initial_answer_chain, 
    # 设置验证器
    validator=validator
)

3.3 定义修订节点Revisor

在获取工具返回结果后，模型需要对初步答案进行修订。这一阶段的目标是结合工具提供的新信息，优化答案的完整性、准确性和逻辑性。

答案修订不仅是对内容的简单补充，还包括对答案结构的调整。例如，针对复杂问题的答案可能需要按逻辑分段，而非工具返回的零散信息。修订过程中，模型需要判断哪些工具返回的内容是有用的，哪些是不相关的，以确保修订后的答案更加聚焦。

在某些情况下，修订后的答案可能仍然存在不足。这时，可以再次触发反思和工具调用的循环，直至答案达到理想的质量或系统判断问题已经被充分解决。这种循环机制的设计确保了反思应答流程具有高度的适应性。

# 定义修订指令模板
revise_instructions = """根据新的信息修改你之前的答案
    - 你应该利用之前的评判来向你的答案中添加重要信息。
        - 你"必须"在修订答案中包含数字引用以确保其可验证性。
        - 在答案的底部添加一个“References”部分（不计入字数限制），形式如下:
            - [1] https://example.com
            - [2] https://example.com
    - 你应该利用之前的评判来移除答案中多余的信息，并"确保"它不超过250个字。.
"""


# 扩展初始答案的结构以包括参考文献
# 要求模型强制引用信息来源有助于生成更有依据的回答。
class ReviseAnswer(AnswerQuestion):
    """修订你原本对问题的回答。提供一个答案和反思,

    用参考资料来支持你的反思, 最后添加 search queries 以改进答案。"""

    references: list[str] = Field(
        description="使用参考资料来更新你的答案"
    )


# 创建修订链
revision_chain = actor_prompt_template.partial(
    # 设置修订指令
    first_instruction=revise_instructions,
    # 设置函数名称
    function_name=ReviseAnswer.__name__,
) | llm.bind_tools(tools=[ReviseAnswer])  # 绑定修订工具
# 创建修订验证器
revision_validator = PydanticToolsParser(tools=[ReviseAnswer])

# 创建带重试机制的修订器
revisor = ResponderWithRetries(
    # 设置可运行对象
    runnable=revision_chain, 
    # 设置验证器
    validator=revision_validator
)

3.4 定义工具节点Tools

在反思生成的基础上，工具调用是反思应答机制的核心环节之一。模型需要根据反思内容生成具体的工具调用请求，并通过调用工具获取补充信息。

工具的设计需要明确其输入和输出规则。例如，一个搜索工具可以接收查询关键词，并返回与关键词相关的简要信息或链接；一个计算工具则可以接收特定公式或参数，并返回计算结果。模型生成的工具调用请求需要符合这些规则，以确保调用能够顺利执行。

# 导入LangChain结构化工具
from langchain_core.tools import StructuredTool

# 导入LangGraph预构建工具节点
from langgraph.prebuilt import ToolNode


# 定义运行查询的函数
def run_queries(search_queries: list[str], **kwargs):
    """Run the generated queries."""
    return tavily_tool.batch([{"query": query} for query in search_queries])


# 创建工具节点
tool_node = ToolNode(
    [
        StructuredTool.from_function(run_queries, name=AnswerQuestion.__name__),
        StructuredTool.from_function(run_queries, name=ReviseAnswer.__name__),
    ]
)

3.5 定义图

完成整个状态图的定义：

节点拓扑架构：构建 StateGraph 状态图，定义 Responder（初稿生成）、execute_tools（工具执行）、Revisor（答案修订）三类节点及流转关系

循环控制机制：

• 采用 event_loop 函数跟踪消息类型实现最大迭代次数控制（MAX_ITERATIONS=5）
• 通过条件边动态切换 execute_tools 与终止状态

状态管理设计：基于 TypedDict 定义包含消息序列的状态结构，通过 add_messages 注解实现消息自动追加

# 导入LangGraph图结构相关类
from langgraph.graph import END, StateGraph, START
# 导入消息添加功能
from langgraph.graph.message import add_messages
# 导入注解类型
from typing import Annotated
# 导入类型字典扩展
from typing_extensions import TypedDict


# 定义状态类型
class State(TypedDict):
    # 定义消息字段
    messages: Annotated[list, add_messages]


# 设置最大迭代次数
MAX_ITERATIONS = 5
# 创建状态图构建器
builder = StateGraph(State)
# 添加草稿节点
builder.add_node("Responder", first_responder.respond)
# 添加工具执行节点
builder.add_node("execute_tools", tool_node)
# 添加修订节点
builder.add_node("Revisor", revisor.respond)
# Responder -> execute_tools
builder.add_edge("Responder", "execute_tools")
# execute_tools -> Revisor
builder.add_edge("execute_tools", "Revisor")

# Define looping logic:


# 定义获取迭代次数的函数
def _get_num_iterations(state: list):
    # 初始化计数器
    i = 0
    # 反向遍历消息列表
    for m in state[::-1]:
        # 如果消息类型不是工具或AI生成，则停止计数
        if m.type notin {"tool", "ai"}:
            break
        # 增加计数
        i += 1
    return i


# 定义事件循环函数
def event_loop(state: list):
    # in our case, we'll just stop after N plans
    # 获取当前迭代次数
    num_iterations = _get_num_iterations(state["messages"])
    # 如果超过最大迭代次数则结束
    if num_iterations > MAX_ITERATIONS:
        return END
    # 否则继续执行工具
    return"execute_tools"


# Revisor -> execute_tools OR end
# 添加条件边：根据事件循环结果决定下一步
builder.add_conditional_edges("Revisor", event_loop, ["execute_tools", END])
# 添加起始边：从START到Responder节点
builder.add_edge(START, "Responder")
# 编译状态图
graph = builder.compile()

3.6 运行图

采用流式处理模式逐步展开各环节执行,pretty_print() 方法提供直观的中间结果可视化。

# 创建事件流
events = graph.stream(
    # 初始化消息
    {"messages": [("user", "我们应当如何应对气候危机？")]},
    # 设置流模式为值
    stream_mode="values",
)
# 遍历事件流
for i, step in enumerate(events):
    # print(step)
    # 打印当前步骤
    print(f"Step {i+1}")
    # 格式化打印最后一条消息
    step["messages"][-1].pretty_print()
    print("\n\n")

4. 输出结果 

可以看到优化后的最终答案，包含对应接引用文章

Step 8
================================== Ai Message ==================================
Tool Calls:
  ReviseAnswer (call_2cMNg8dtTcahZo7qmFXeWMep)
 Call ID: call_2cMNg8dtTcahZo7qmFXeWMep
  Args:
    answer: 应对气候危机需要多方面的共同努力。首先，各国必须实施基于科学的数据驱动政策，遵循《巴黎协定》，设定具体的减排目标。例如，推广可再生能源（如太阳能和风能）可以显著减少温室气体排放[1]。

其次，增强能效和推动可持续交通是关键。城市应优化公共交通，增加自行车道等基础设施，以降低碳排放并改善空气质量[2]。此外，保护生态系统，特别是森林和湿地，可以加强碳汇作用，帮助减缓气候变化[3]。

公众参与和教育也至关重要，可以通过提高环保意识，促使个人和企业采取绿色行动。同时，国际间的合作和共享技术也是应对气候危机的必要条件，只有全球合作才能有效解决这一问题[4]。
    references: ['[1] https://www.cbc.ca/news/science/effective-climate-policies-1.7301998', '[2] https://www.epa.gov/arc-x/case-studies-climate-change-adaptation', '[3] https://www.weforum.org/stories/2020/01/how-cross-sector-collaboration-is-driving-the-global-climate-agenda/', '[4] https://unfccc.int/UN-Climate-Change-partnerships']
    reflection: {'missing': '缺乏具体成功案例或实施细节，例如特定国家的成功经验。', 'superfluous': '内容虽然全面，但可以更加简洁，避免信息过于重复。'}
    search_queries: ['climate action global examples', 'effective climate policies case studies', 'international cooperation climate change']

5. 使用场景 

反思机制可以应用于多个场景。例如，在教育领域，教师可以利用这一机制生成结构清晰、内容全面的教学材料；在知识管理领域，企业可以用其自动生成准确且精炼的知识库内容；在客户服务领域，这一机制则可以帮助客服系统生成高质量的答案，解决用户的复杂问题。

例如，当一个客户提问“如何解决账户被盗的问题”时，模型可能会生成一个初步答案，并反思“答案中缺少具体的解决流程，可以通过工具查询最新的账户恢复政策。”工具调用后，模型获取了最新的恢复流程信息，并在答案中加入了具体的步骤，从而大幅提升了答案的实用性。

基于外部知识的增强反思机制，不仅提升了模型回答复杂问题的能力，还使其生成内容的过程更加透明和可控。这一机制展示了生成式AI与工具结合的巨大潜力,进一步提高模型的灵活性和适用性。