补充my_reflection_agent
·
1. 原目录
hello-agents/docs/chapter7/第七章 构建你的Agent框架.md at main · datawhalechina/hello-agents · GitHub
2. 补充代码
# my_reflection_agent.py
"""重写的 Reflection Agent - 执行-反思-优化循环"""
import string
from typing import Optional, Dict
from hello_agents import ReflectionAgent, HelloAgentsLLM, Config, Message
from hello_agents.agents.reflection_agent import Memory
class MyReflectionAgent(ReflectionAgent):
"""
重写的 Reflection Agent
基于框架 ReflectionAgent,实现「初始执行 → 反思 → 优化」迭代循环
"""
def __init__(
self,
name: str,
llm: HelloAgentsLLM,
system_prompt: Optional[str] = None,
config: Optional[Config] = None,
max_iterations: int = 3,
custom_prompts: Optional[Dict[str, str]] = None,
):
super().__init__(
name=name,
llm=llm,
system_prompt=system_prompt,
config=config,
max_iterations=max_iterations,
custom_prompts=custom_prompts,
)
prompt_type = "自定义" if custom_prompts else "默认"
print(f"✅ {name} 初始化完成,最大迭代: {max_iterations},提示词: {prompt_type}")
def run(self, input_text: str, **kwargs) -> str:
"""运行 Reflection Agent:初始执行 → 反思 → 优化"""
print(f"\n🤖 {self.name} 开始处理任务: {input_text}")
self.memory = Memory()
# 1. 初始执行
print("\n--- 正在进行初始尝试 ---")
initial_prompt = self._format_prompt("initial", task=input_text)
initial_result = self._get_llm_response(initial_prompt, **kwargs)
self.memory.add_record("execution", initial_result)
# 2. 迭代:反思 → 优化
for i in range(self.max_iterations):
print(f"\n--- 第 {i + 1}/{self.max_iterations} 轮迭代 ---")
print("\n-> 正在进行反思...")
last_result = self.memory.get_last_execution()
reflect_prompt = self._format_prompt(
"reflect",
task=input_text,
content=last_result,
code=last_result,
)
feedback = self._get_llm_response(reflect_prompt, **kwargs)
self.memory.add_record("reflection", feedback)
if self._should_stop(feedback):
print("\n✅ 反思认为结果已无需改进,任务完成。")
break
print("\n-> 正在进行优化...")
refine_prompt = self._format_prompt(
"refine",
task=input_text,
last_attempt=last_result,
last_code_attempt=last_result,
feedback=feedback,
content=last_result,
)
refined_result = self._get_llm_response(refine_prompt, **kwargs)
self.memory.add_record("execution", refined_result)
final_result = self.memory.get_last_execution()
print(f"\n--- 任务完成 ---\n最终结果:\n{final_result}")
self.add_message(Message(input_text, "user"))
self.add_message(Message(final_result, "assistant"))
return final_result
def _format_prompt(self, stage: str, **kwargs) -> str:
"""格式化提示词,仅传入模板中实际使用的占位符"""
template = self.prompts[stage]
formatter = string.Formatter()
field_names = {
field_name
for _, field_name, _, _ in formatter.parse(template)
if field_name
}
filtered = {k: v for k, v in kwargs.items() if k in field_names}
return template.format(**filtered)
def _should_stop(self, feedback: str) -> bool:
"""判断反思反馈是否表示无需继续优化"""
stop_phrases = ("无需改进", "不需要改进", "无须改进", "已经很好")
feedback_lower = feedback.lower()
return any(p in feedback for p in stop_phrases) or "no need for improvement" in feedback_lower
def _get_llm_response(self, prompt: str, **kwargs) -> str:
"""调用 LLM,可选注入 system_prompt"""
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
messages.append({"role": "user", "content": prompt})
return self.llm.invoke(messages, **kwargs) or ""
def get_trajectory(self) -> str:
"""获取完整的执行-反思轨迹"""
return self.memory.get_trajectory()
3. 运行结果
# python test_reflection_agent.py
✅ 我的反思助手 初始化完成,最大迭代: 3,提示词: 默认
✅ 我的代码生成助手 初始化完成,最大迭代: 3,提示词: 自定义
🤖 我的反思助手 开始处理任务: 写一篇关于人工智能发展历程的简短文章
--- 正在进行初始尝试 ---
📝 记忆已更新,新增一条 'execution' 记录。
--- 第 1/3 轮迭代 ---
-> 正在进行反思...
📝 记忆已更新,新增一条 'reflection' 记录。
-> 正在进行优化...
📝 记忆已更新,新增一条 'execution' 记录。
--- 第 2/3 轮迭代 ---
-> 正在进行反思...
📝 记忆已更新,新增一条 'reflection' 记录。
-> 正在进行优化...
📝 记忆已更新,新增一条 'execution' 记录。
--- 第 3/3 轮迭代 ---
-> 正在进行反思...
📝 记忆已更新,新增一条 'reflection' 记录。
-> 正在进行优化...
📝 记忆已更新,新增一条 'execution' 记录。
--- 任务完成 ---
最终结果:
当然可以!根据反馈意见,我对文章进行了结构优化、内容补充和语言简化,添加了开头引入段落,细化了时间节点划分,补充了1980年代神经网络发展及第二次AI寒冬,均衡介绍了专业术语,并在未来展望中加入了具体案例说明。以下是改进后的版本:
---
# 人工智能发展历程简述
更多推荐


所有评论(0)