本文是《从零开始构建AI智能体》系列的第二部分，继续探讨如何实现AI智能体的反思模式和短期记忆机制。
在第一部分中，我们在不使用任何LLM编排框架的情况下实现了工具使用模式（Tool Use Pattern）。今天，我们将在此基础上进一步构建，探讨反思模式如何提升智能体的表现。本文后续将详细解释实现这一过程的具体方法。
本文内容概览
通过阅读本文，你将掌握以下知识点：
- 什么是AI智能体系统中的反思模式。
- 反思模式与短期记忆之间的关联。
- 实现反思模式的优势与局限。
- 如何在不依赖编排框架的情况下，构建一个能够结合记忆实现反思模式的智能体类。
- 解决我们在上一部分项目中遇到的一些“幻觉”（hallucination）问题，即模型生成的虚假或不准确信息。
相关代码示例和其他项目资源可以在我的GitHub仓库中找到：
AI智能体中的反思模式解析
在当前AI智能体领域，反思模式的定义尚未完全统一。从高层次来看，反思模式可以描述为智能体系统审视自身输出的能力。进一步地，它还能提出改进建议，并根据反馈优化后续行为。
在解释智能体相关概念时，我倾向于抛开“智能体”这一抽象概念，转而以“智能体工作流程”（Agentic Flow）为视角进行思考。通过流程图分析会更加直观，因为智能体本质上是一系列步骤通过不同拓扑结构连接而成的系统。反思模式可以在智能体工作流程的不同步骤中应用，下面我们来看几个具体示例。
反思：最简案例
上述智能体工作流程包含以下步骤：
1. 用户向大语言模型（LLM）提出问题。
2. 模型生成的答案被再次传递给LLM，要求其对先前生成的答案进行反馈，并提出改进建议（如果有的话）。
3. 改进后的答案返回给用户。
这一流程虽然简单，但在许多情况下，它能显著提升答案的准确性。当然，通过精心设计系统提示词（System Prompt）也可以达到类似效果，但反思模式通常更强大且灵活。
反思：循环机制
上述智能体工作流程的步骤如下：
1. 用户的问题被传递给LLM。
2. 模型生成的答案再次传递给LLM，要求其对答案进行反馈并提出改进建议。
3. 在应用改进后，优化后的答案会再次传递给LLM，请求进一步反馈和改进建议。这一循环会重复预设次数，或者直到LLM无法再提出新建议并返回一个终止信号（通常是一个预定义字符串，如“END”）。
4. 最终答案返回给用户。
这种系统非常强大，但适用场景较为特定。例如，在文献中，这种方法常用于代码生成领域。原因显而易见——生成的代码可以在多次迭代中不断优化，系统总能找到可以“过度工程化”的地方。吴恩达（Andrew Ng）在一篇文章中详细描述了这种代码生成中的反思模式，感兴趣的读者可以参考：Agentic设计模式 - 第2部分：反思。
然而，在大多数现实场景中，我认为反思循环的商业价值较难体现。因为它往往会显著增加成本，包括额外的LLM调用和延迟，这可能不符合商业需求。
反思：执行计划的验证
我发现反思模式的一个实用场景是验证智能体工作流程中的执行计划。以下是一个可能的流程：
1. 用户的问题和系统提示词传递给LLM，生成执行计划。这里是流程可能中断的关键点。例如：
- 智能体工作流程判断是直接回答用户，还是使用预定义工具生成额外上下文。
- 决定使用工具。
- 由于幻觉，传递给工具的参数不正确。
- 工具返回错误。
2. 通过精心设计的反思步骤，我们可以让智能体尝试修复前一步中产生的幻觉。
3. 执行计划可能直接返回答案给用户。
4. 或者提示使用工具，通过另一次LLM调用丰富答案内容，然后返回给用户。
在本文后续的实践部分，我们将实现这种流程。
反思：复杂的智能体流程
如前所述，反思步骤没有固定的应用位置。在复杂的智能体工作流程中，它可以多次使用，用于验证中间答案、计划或其他拓扑结构中的任何部分。
在实际应用中，为了自动化组织中的复杂流程，你可能需要构建多步骤拓扑结构。这些结构包含多个概率路由器，用于连接执行节点。有些路由器可能通过LLM实现，有些基于规则，有些可能使用常规机器学习模型。执行节点中，有些会使用工具，有些仅调用LLM，有些则是非概率性执行。反思步骤可以遍布各处，以提高非确定性路由器和执行的准确性。
智能体记忆与反思模式的关系
通常，为了让反思模式发挥最佳效果，需要实现某种形式的短期（工作）记忆。让我们通过之前定义的第三个智能体工作流程示例来分析原因。
在调用反思步骤之前的智能体工作流程包括：
- A) 系统提示词和用户问题，将传递给LLM以生成初始执行计划。
- B) 初始步骤生成的执行计划。
- C) 反思步骤需要所有这些信息。因为系统提示词很可能包含有关可用工具等有用信息，用户问题提供了用户意图的上下文，而计划则是我们试图改进的内容。
在基础生成实现中，我们不会将这些信息保存在记忆中，因此需要实现工作记忆。
上图展示了工作记忆的最简单实现方式——每次与智能体的交互都简单地存储在一个列表中，并在每次调用生成时作为额外上下文传递给系统提示词。这就是聊天机器人（如ChatGPT）中实现的最简单记忆类型。
如果我们要实现一个具备规划能力的智能体，并希望包含反思功能，流程可能如下：
如你所见，这与之前非常相似。唯一的区别在于智能体的初始响应是执行计划，而用户的下一个问题是基于之前交互对计划进行反思的提示。
使用反思模式的优缺点
如同任何能够改进系统的技术，反思模式既有优势也有局限。
优势
- 显著提升系统最终输出的准确性：有时，这是唯一能满足极高准确性要求、使应用可行的方式。
- 相比修改初始系统提示词更灵活：例如，可以针对流程的不同部分使用不同的反思方法甚至不同的智能体。
- 在应用反思模式时，小模型也能实现更多功能：当然，需要综合分析所有缺点，但在项目中总有一个阶段的目标是优化AI系统成本。
缺点
- 增加应用复杂性。
- 增加端到端流程的延迟：因为会调用额外的LLM。
- 增加额外成本：因为每次反思步骤至少会多提示一次LLM（通常不止一次）。
构建反思智能体
正如本文开头所述，我们将通过一个示例来实现反思模式，用于修正由LLM生成的行动计划。这一选择是有针对性的。
如果你关注了系列的第一部分，可能记得我们实现了一个按需转换两种货币的工具，并提示智能体仅在需要转换时使用该工具。虽然整体运行良好，但在示例中我们请求从塞尔维亚货币转换为日本货币。
然而，这并非我的初衷。我来自立陶宛，本想展示从立陶宛到日本旅行的场景，所以最初我向模型提出了以下问题：
我从立陶宛前往日本，带有1500单位当地货币，能兑换多少日本货币？
想象一下我的惊讶，当智能体返回以下响应时：
思考：我需要使用货币转换工具将1500立陶宛立特（LTL）转换为日元（JPY）。
计划：使用convert_currency工具将1500 LTL转换为JPY。返回转换结果。
结果：错误：无法获取汇率
显然，工具调用导致了错误——立陶宛自2015年起已采用欧元（EUR）作为官方货币！工具自然无法找到LTL到JPY的转换率。
这次我决定认真对待这个问题，在本篇中，我的目标是利用反思模式修复这个计划。
相关代码可以在GitHub仓库中找到：代码仓库。你也可以通过Jupyter笔记本进行教程学习：反思笔记本。
实现工作记忆
我们将使用数据类（Dataclass）来记录与智能体的单次交互：
@dataclass
class Interaction:
"""记录与智能体的单次交互"""
timestamp: datetime
query: str
plan: Dict[str, Any]
该数据类将存储用户的问题和智能体生成的计划。需要注意的是，反思步骤还需要系统提示词，但我们会单独实现这一点。
为了更好地专注于反思计划的生成，这次我们将简化智能体类，剥离与工具相关的功能。
初始系统提示词
我们将沿用系列第一部分的系统提示词，具体解释可以参考之前的文章：系统提示词设计。有两个不同之处：
- 我们现在模拟可用工具，而不是实际实现，因此工具部分如下：
"tools": [\
{\
"name": "convert_currency",\
"description": "使用最新汇率转换货币。",\
"parameters": {\
"amount": {\
"type": "float",\
"description": "转换金额"\
},\
"from_currency": {\
"type": "str",\
"description": "源货币代码（例如USD）"\
},\
"to_currency": {\
"type": "str",\
"description": "目标货币代码（例如EUR）"\
}\
}\
}\
]
- 我们通过在“能力”和“指令”中增加第四项，扩展了智能体的能力和指令：
"capabilities": [\
"在必要时使用提供的工具帮助用户",\
"对于不需要工具的问题直接回答",\
"规划高效的工具使用顺序",\
"如果用户要求，反思计划并在需要时提出修改建议"\
],
"instructions": [\
"仅在任务需要时使用工具",\
"如果问题可以直接回答，则用简单消息回复而非使用工具",\
"当需要工具时，高效规划使用以减少工具调用",\
"如果用户要求，反思计划并在需要时提出修改建议"\
]
实现智能体类
智能体类将使用空的交互历史记录进行初始化。交互历史记录在本例中即为工作记忆。
class Agent:
def __init__(self, model: str = "gpt-4o-mini"):
"""以空的交互历史记录初始化智能体。"""
self.client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.interactions: List[Interaction] = [] # 工作记忆
self.model = model
规划
plan方法通过LLM生成初始执行计划，并将用户问题和生成的计划更新到工作记忆中。
def plan(self, user_query: str) -> Dict:
"""使用LLM创建计划并存储到记忆中。"""
messages = [\
{"role": "system", "content": self.create_system_prompt()},\
{"role": "user", "content": user_query}\
]response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0
)
try:
plan = json.loads(response.choices[0].message.content)
# 在规划后立即存储交互记录
interaction = Interaction(
timestamp=datetime.now(),
query=user_query,
plan=plan
)
self.interactions.append(interaction)
return plan
except json.JSONDecodeError:
raise ValueError("无法将LLM响应解析为JSON")
反思计划
reflect_on_plan方法是本文的核心内容。
def reflect_on_plan(self) -> Dict[str, Any]:
"""使用交互历史记录反思最新的计划。"""
if not self.interactions:
return {"reflection": "没有可反思的计划", "requires_changes": False}latest_interaction = self.interactions[-1]
reflection_prompt = {
"task": "reflection",
"context": {
"user_query": latest_interaction.query,
"generated_plan": latest_interaction.plan
},
"instructions": [\
"审查生成的计划，寻找可能的改进点",\
"考虑所选工具是否合适",\
"验证工具参数是否正确",\
"检查计划是否高效",\
"判断是否真的需要工具"\
],
"response_format": {
"type": "json",
"schema": {
"requires_changes": {
"type": "boolean",
"description": "计划是否需要修改"
},
"reflection": {
"type": "string",
"description": "解释需要哪些更改或为何不需要更改"
},
"suggestions": {
"type": "array",
"items": {"type": "string"},
"description": "具体的改进建议",
"optional": True
}
}
}
}messages = [\
{"role": "system", "content": self.create_system_prompt()},\
{"role": "user", "content": json.dumps(reflection_prompt, indent=2)}\
]response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0
)
try:
return json.loads(response.choices[0].message.content)
except json.JSONDecodeError:
return {"reflection": response.choices[0].message.content}
我们创建了一个包含具体指令和期望输出格式的新提示词。注意，该提示词通过context键将用户问题和初始计划作为额外上下文传递。然后，我们将该提示词与初始系统提示词一起传递，以生成对初始计划的反思，包括任何改进建议。
执行流程与生成修订计划
我们通过execute方法将整个流程整合。
def execute(self, user_query: str) -> str:
"""执行完整流程：规划、反思并可能重新规划。"""
try:
# 创建初始计划（同时存储到记忆中）
initial_plan = self.plan(user_query)# 使用记忆反思计划
reflection = self.reflect_on_plan()# 检查反思是否建议更改
if reflection.get("requires_changes", False):
# 根据反思生成新计划
messages = [\
{"role": "system", "content": self.create_system_prompt()},\
{"role": "user", "content": user_query},\
{"role": "assistant", "content": json.dumps(initial_plan)},\
{"role": "user", "content": f"请根据以下反馈修订计划：{json.dumps(reflection)}"}\
]response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0
)try:
final_plan = json.loads(response.choices[0].message.content)
except json.JSONDecodeError:
final_plan = initial_plan # 如果解析失败，回退到初始计划
else:
final_plan = initial_plan# 更新存储的交互记录，包含所有信息
self.interactions[-1].plan = {
"initial_plan": initial_plan,
"reflection": reflection,
"final_plan": final_plan
}# 返回适当的响应
if final_plan.get("requires_tools", True):
return f"""初始思考：{initial_plan['thought']}
初始计划：{'. '.join(initial_plan['plan'])}
反思：{reflection.get('reflection', '未建议改进')}
最终计划：{'. '.join(final_plan['plan'])}"""
else:
return f"""响应：{final_plan['direct_response']}
反思：{reflection.get('reflection', '未建议改进')}"""
except Exception as e:
return f"执行计划时出错：{str(e)}"
请注意，该方法仅在反思生成请求时应用反思改进。如果不需要，我们保留旧计划。
执行智能体
接下来，我们验证计划是否已成功修复。通过运行以下代码执行智能体：
query_list = ["我从立陶宛前往日本，带有1500单位当地货币，能兑换多少日本货币？"]
for query in query_list:
print(f"\n问题：{query}")
result = agent.execute(query)
print(result)
如果你使用与我相同的模型，应该会得到类似的结果：
问题：我从立陶宛前往日本，带有1500单位当地货币，能兑换多少日本货币？
初始思考：我需要使用货币转换工具将1500立陶宛立特（LTL）转换为日元（JPY）。
初始计划：使用convert_currency工具将1500 LTL转换为JPY。返回转换结果
反思：计划需要修改，因为立陶宛立特（LTL）自2015年立陶宛采用欧元（EUR）后已不再使用。因此，转换应从EUR到JPY，而非LTL。
最终计划：使用convert_currency工具将1500 EUR转换为JPY。返回转换结果
大获成功！我们修复了计划，并能够成功执行工具。
今日总结
通过本篇文章，我们学习了：
- 如何实现一种简单的工作记忆。
- 如何构建反思步骤，反思智能体生成的执行计划。
- 如何实现反思步骤生成的建议并改进计划。
希望本文对您有所帮助。如果有任何问题或想法，欢迎在评论区交流。