核心理念:Agent 不是"想好再做"或"做了再想",而是想一步、做一步、观察结果、再想下一步。
核心理念:在给出最终答案前,先把推理过程"说出来"。
我们在第二阶段的任务规划模块已经实践了 CoT 的核心思想。高阶的 CoT 增加了自一致性(Self-Consistency):让模型多次推理,取多数结果。
核心理念:Agent 生成输出 → 自己评价输出质量 → 根据评价改进 → 再评价 → 直到满意。
这就像写作文先打草稿,自己读一遍,修改不满意的地方,再读,再改。
核心理念:给 Agent 一个终极目标,它自己不断拆解子任务、调用工具、执行、验证,不需要每一步都问用户。
| 架构 | 核心特点 | 适用场景 | 自主程度 | 风险 |
|---|---|---|---|---|
| ReAct | 思考行动交替,每步观察 | 需要与外部环境交互的任务 | ⭐⭐⭐ | 低(每步可控) |
| CoT | 先推理后回答 | 复杂推理、数学、逻辑问题 | ⭐⭐ | 低 |
| Self-Refine | 自我评价+迭代改进 | 写作、代码生成、创意任务 | ⭐⭐⭐⭐ | 中(可能"改坏") |
| AutoGPT | 全自主目标驱动 | 开放探索、研究任务 | ⭐⭐⭐⭐⭐ | 高(失控风险) |
#!/usr/bin/env python3
"""SelfRefineAgent — 自我反思与迭代优化 Agent"""
import os, json
from dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
class SelfRefineAgent:
"""具备自我反思和迭代优化能力的 Agent"""
def __init__(self):
self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-sonnet-4-6"
self.max_iterations = 3 # 最多优化3轮
def generate_with_refinement(self, task: str, quality_criteria: str = "") -> str:
"""生成内容并自我优化"""
print(f"\n🎯 任务: {task}")
# 第1步:生成初始版本
print("📝 生成初始版本...")
draft = self._generate(task)
print(f" 初稿: {draft[:200]}...")
# 第2步:自我评价 + 迭代优化
for iteration in range(self.max_iterations):
print(f"\n🔍 第{iteration+1}轮自我评价...")
# 自我评价
feedback = self._evaluate(draft, task, quality_criteria)
score = feedback.get("score", 0)
issues = feedback.get("issues", [])
print(f" 自评分数: {score}/10")
print(f" 发现问题: {issues}")
# 如果满意或没有发现问题,停止
if score >= 8 or not issues:
print(f" ✅ 质量达标,停止优化")
break
# 根据反馈改进
print(f" 🔄 根据反馈改进...")
draft = self._refine(draft, feedback, task)
print(f" 改进稿: {draft[:200]}...")
return draft
def _generate(self, task: str) -> str:
resp = self.client.messages.create(
model=self.model, max_tokens=1000,
messages=[{"role":"user","content":f"完成以下任务(第一次尝试,尽你最大努力):\n{task}"}],
temperature=0.5)
return resp.content[0].text.strip()
def _evaluate(self, content: str, task: str, criteria: str) -> dict:
prompt = f"""评价以下内容的质量。任务要求: {task}
质量标准: {criteria or '准确性、完整性、清晰度、实用性'}
内容:
{content[:2000]}
请以JSON格式输出:
{{"score": 1-10, "issues": ["问题1","问题2"], "strengths": ["优点1"], "suggestions": ["建议1"]}}"""
resp = self.client.messages.create(
model=self.model, max_tokens=500,
messages=[{"role":"user","content":prompt}], temperature=0.2)
try:
text = resp.content[0].text.strip()
if "```" in text: text = text.split("```")[1].split("```")[0]
return json.loads(text)
except:
return {"score": 7, "issues": [], "suggestions": []}
def _refine(self, content: str, feedback: dict, task: str) -> str:
issues = "\n".join([f"- {i}" for i in feedback.get("issues", [])])
suggestions = "\n".join([f"- {s}" for s in feedback.get("suggestions", [])])
prompt = f"""改进以下内容。原始任务: {task}
当前内容: {content[:2000]}
发现的问题: {issues}
改进建议: {suggestions}
请输出改进后的完整内容。"""
resp = self.client.messages.create(
model=self.model, max_tokens=1000,
messages=[{"role":"user","content":prompt}], temperature=0.4)
return resp.content[0].text.strip()
# 测试
if __name__ == "__main__":
agent = SelfRefineAgent()
result = agent.generate_with_refinement(
"写一份200字的智能手机选购指南",
"简洁实用,适合小白,包含关键选购指标"
)
print(f"\n{'='*50}")
print(f"✅ 最终结果:\n{result}")
多Agent协作的三种通信模式:
| 层次 | 职责 | 技术实现 |
|---|---|---|
| 感知层 | 接收输入、理解意图、提取信息 | Claude API + 意图识别器 |
| 规划层 | 任务拆解、步骤排序、依赖管理 | TaskPlanner + CoT |
| 工具层 | 工具注册、发现、调用、错误恢复 | ToolRegistry + ToolExecutor |
| 记忆层 | 短期/长期/会话记忆管理 | MemorySystem + 向量数据库 |
| 输出层 | 结果格式化、質量校驗、呈现 | Self-Refine + 模板引擎 |
每一层独立开发、独立测试、独立替换。比如你想换一个更好的记忆系统,只需要修改记忆层的实现,其他层不需要动。这就是"高内聚、低耦合"的体现。