各位同仁，下午好！
今天，我们齐聚一堂，探讨一个在人工智能领域日益受到关注的话题：如何让我们的智能体（Agent）不仅仅是机械地执行任务，而能够真正从错误中学习，特别是从人类的负面反馈中汲取教训，实现自我进化。这不仅仅是简单的迭代优化，更是一种深层次的“Reflection-on-Feedback”机制，旨在让Agent自动生成一份“教训总结”，并据此改写未来的Prompt，从而避免重蹈覆辙。作为一名编程专家，我将从技术实现的角度，为大家剖析这一复杂而又充满潜力的过程。
一、引言：超越简单的迭代 – 迈向智能自省的 Agent
在当今AI模型，特别是大型语言模型（LLM）驱动的Agent系统中，我们常常面临一个挑战：尽管模型拥有强大的生成能力和知识储备，但在复杂或特定场景下，其输出仍然可能偏离预期，甚至出现“幻觉”或逻辑错误。传统的Agent优化方法通常依赖于大量的数据标注、模型微调或者人工干预。然而，这些方法往往成本高昂、周期漫长，且难以适应快速变化的需求。
人类反馈，尤其是负面反馈，是Agent改进的宝贵资源。它直指Agent的不足，揭示了其在理解任务、执行逻辑或生成内容上的盲点。但简单地收集反馈并不能解决问题。我们所追求的，是一种更高级的机制：让Agent能够“理解”反馈的深层含义，自我分析错误原因，并主动将这些“教训”内化到其未来的行为模式中。这里的“行为模式”对于基于LLM的Agent而言，最直接和高效的体现就是其接收的初始指令，即Prompt。
“Reflection-on-Feedback”机制的核心思想是，Agent不应仅仅是根据Prompt执行，然后等待新的Prompt。它应该能够形成一个自省的循环：
- 执行任务：Agent根据当前Prompt生成输出。
- 接收反馈：用户对Agent的输出提供负面反馈。
- 反思学习：Agent分析反馈，识别问题，推断根源，并生成“教训总结”。
- 自我调整：Agent根据“教训总结”自动修改其未来的Prompt。
- 重新执行：Agent使用新的Prompt执行任务，期望表现更好。
今天，我们的目标就是深入探讨如何构建这样一个系统，让Agent能够自动化地完成这个从反馈到教训，再到Prompt优化的全链路过程。
二、负面反馈的本质与挑战
负面反馈是推动任何系统进步的关键动力，但它也带来了独特的挑战。与明确的成功信号不同，负面反馈往往具有以下特点：
- 模糊性与非结构化：用户通常会用自然语言表达不满，例如“这个答案不对”、“代码有bug”、“文不对题”，这些表述缺乏统一的结构，难以直接用于机器处理。
- 主观性与情绪化：反馈可能夹杂用户的个人偏好、情绪，甚至带有讽刺或抱怨，这使得从中提取客观事实变得复杂。
- 多维度性：一个简单的负面反馈可能指向Agent在事实准确性、逻辑推理、格式遵循、完整性、相关性等多个维度的不足。
- 隐含性与因果链：用户通常只描述结果（“代码无法运行”），而不会直接指出根本原因（“没有导入正确的库”或“API调用参数错误”）。Agent需要推断导致错误的深层原因。
- 稀疏性与延迟性：用户不一定会对所有不满都提供反馈，且反馈可能在Agent输出后的一段时间才出现。
传统上，处理负面反馈的方式往往是人工进行分析和归类，然后由Prompt工程师手动修改Prompt。这种方法虽然精确，但效率低下，难以扩展到大规模应用场景，也无法实现Agent的实时、持续学习。因此，如何让Agent自主、高效、准确地从这些复杂的负面反馈中提取有价值的洞察，是实现“Reflection-on-Feedback”机制的首要挑战。
三、系统架构：构建一个自我改进的反馈循环
要实现Agent的智能自省，我们需要设计一个包含多个模块的健壮系统。这个系统将围绕Agent的执行流构建一个反馈驱动的优化闭环。
以下是该系统的核心架构概览：
+-------------------+ +--------------------+ +-------------------+
| Agent 任务执行层 | | 用户界面/API | | 反馈存储与管理 |
| (Agent Execution) |-----> | (User Interface/API)|-----> | (Feedback Storage) |
| - 接收Prompt | | - 展示Agent输出 | | - 原始反馈日志 |
| - 生成输出 | | - 收集用户反馈 | | - 任务上下文 |
+-------------------+ +--------------------+ +-------------------+
^ |
| |
| v
+-------------------+ +--------------------+ +-------------------+
| Prompt 版本管理 |<-----| Prompt 改写模块 |<-----| 教训总结生成模块 |
| (Prompt Version | | (Prompt Rewriter) | | (Lesson Learned |
| Manager) | | - 修改当前Prompt | | Generator) |
| - 历史版本追踪 | | - 结合教训总结 | | - 识别核心教训 |
| - 激活/回滚 | | - 输出新Prompt | | - 泛化改进建议 |
+-------------------+ +--------------------+ +-------------------+
^ |
| |
| v
+-------------------+ +--------------------+ +-------------------+
| 效果评估模块 |<-----| 反馈分析模块 | | LLM 基础模型 |
| (Performance | | (Feedback Analyzer)| | (Base LLM) |
| Evaluator) | | - 情感分析 | | - 文本生成 |
| - 负反馈率 | | - 关键词提取 | | - 推理与理解 |
| - 用户满意度 | | - 问题分类 | | |
| - A/B 测试 | | - 根因推断 | | |
+-------------------+ +--------------------+ +-------------------+各模块职责：
- Agent 任务执行层 (Agent Execution)：这是Agent实际工作的场所。它接收一个Prompt，然后利用其内置的LLM和其他工具生成一个输出。这个输出可能是文本、代码、图像或其他形式。
- 用户界面/API (User Interface/API)：负责向用户展示Agent的输出，并提供收集用户反馈的机制。反馈可以是结构化的（如评分、预设选项）或非结构化的（如自由文本评论）。
- 反馈存储与管理 (Feedback Storage)：将收集到的原始用户反馈、Agent的输出、原始任务上下文（包括Agent使用的Prompt）以及时间戳等信息持久化存储。这是后续分析的基础数据源。
- 反馈分析模块 (Feedback Analyzer)：这是整个系统的核心智能之一。它负责解析和理解非结构化的用户反馈，从中提取出结构化的、可操作的洞察。
- 情感分析器 (Sentiment Analyzer)：判断反馈的整体情感倾向和强度。
- 问题识别器 (Issue Identifier)：从反馈文本中识别出具体的错误或不满之处。
- 关键词与主题提取器 (Keyword & Topic Extractor)：发现反馈中提及的关键概念和主题。
- 根因推断器 (Root Cause Inferencer)：结合Agent的输出和任务上下文，尝试推断导致问题发生的深层原因。
- 教训总结生成模块 (Lesson Learned Generator)：接收反馈分析模块输出的结构化洞察，利用LLM生成一份简洁、明确且具有指导意义的“教训总结”。这份总结应明确指出Agent犯了什么错误、为什么会犯错，以及未来应该如何避免。
- Prompt 改写模块 (Prompt Rewriter)：根据“教训总结”和原始Prompt，利用LLM自动生成一个新的、优化过的Prompt。新的Prompt旨在通过增加约束、引入示例、调整角色或提示自检机制等方式，来解决先前发现的问题。
- Prompt 版本管理 (Prompt Version Manager)：负责追踪所有Prompt的变更历史。它允许系统激活最新的、经过优化的Prompt，也支持回滚到之前的版本，并可以用于A/B测试不同版本的Prompt。
- 效果评估模块 (Performance Evaluator)：监控Agent在使用新Prompt后的表现，例如负面反馈率、用户满意度等指标，从而验证Prompt优化的有效性，形成完整的闭环。
- LLM 基础模型 (Base LLM)：作为所有涉及文本理解、生成和推理的核心引擎，贯穿于反馈分析、教训总结和Prompt改写等多个智能模块中。
整个流程形成一个持续学习的闭环：Agent执行任务，获得反馈，分析学习，优化自身（通过Prompt），然后用优化后的Prompt执行新任务，如此往复。
四、技术深潜：核心组件的实现细节与代码示例
现在，我们将深入到各个核心组件的实现细节，并通过Python代码示例来展示具体的技术方案。
A. 反馈的捕获与结构化
无论是来自用户界面的评分、错误类型选择，还是自由文本评论，我们需要一个统一的数据模型来存储和管理这些反馈。这有助于后续模块的标准化处理。
1. 数据模型定义
我们定义一个 FeedbackEntry 类来封装反馈信息：
import datetime
import json
from typing import Dict, Any, Optional
class FeedbackEntry:
"""
表示一个用户反馈的统一数据模型。
"""
def __init__(self,
agent_output: str,
user_feedback: str,
timestamp: Optional[str] = None,
task_context: Optional[Dict[str, Any]] = None,
feedback_type: str = "text", # "text", "structured", "rating"
feedback_id: Optional[str] = None):
self.feedback_id = feedback_id if feedback_id else self._generate_id()
self.agent_output = agent_output
self.user_feedback = user_feedback
self.timestamp = timestamp if timestamp else datetime.datetime.now().isoformat()
self.task_context = task_context if task_context is not None else {}
self.feedback_type = feedback_type
def _generate_id(self) -> str:
"""生成一个简单的唯一ID，实际生产环境应使用更健壮的方案。"""
return f"fb_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}"
def to_dict(self) -> Dict[str, Any]:
"""将反馈对象转换为字典格式。"""
return {
"feedback_id": self.feedback_id,
"agent_output": self.agent_output,
"user_feedback": self.user_feedback,
"timestamp": self.timestamp,
"task_context": self.task_context,
"feedback_type": self.feedback_type
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'FeedbackEntry':
"""从字典格式创建反馈对象。"""
return cls(
agent_output=data.get("agent_output"),
user_feedback=data.get("user_feedback"),
timestamp=data.get("timestamp"),
task_context=data.get("task_context"),
feedback_type=data.get("feedback_type"),
feedback_id=data.get("feedback_id")
)
def __repr__(self):
return f"FeedbackEntry(id={self.feedback_id}, type={self.feedback_type}, timestamp={self.timestamp[:19]}, feedback='{self.user_feedback[:50]}...')"
# 示例：从日志或API捕获反馈
def capture_feedback(agent_output: str, user_feedback_raw: str, original_prompt: str, task_id: str) -> FeedbackEntry:
"""模拟捕获用户反馈的函数。"""
task_context = {
"original_prompt": original_prompt,
"task_id": task_id
}
return FeedbackEntry(
agent_output=agent_output,
user_feedback=user_feedback_raw,
task_context=task_context,
feedback_type="text" # 假设大部分反馈是自由文本
)
# 实际使用示例
if __name__ == '__main__':
agent_output_example = "Python是一种静态类型语言，广泛应用于Web开发。"
user_feedback_example = "这个说法不对，Python是动态类型语言。Agent的回答有事实性错误。"
original_prompt_example = "请介绍Python的特点和应用场景。"
task_id_example = "task_001"
feedback_entry = capture_feedback(
agent_output=agent_output_example,
user_feedback_raw=user_feedback_example,
original_prompt=original_prompt_example,
task_id=task_id_example
)
print(feedback_entry.to_dict())
# {'feedback_id': 'fb_...', 'agent_output': 'Python是一种静态类型语言，...', 'user_feedback': '这个说法不对，Python是动态类型语言。...', 'timestamp': '...', 'task_context': {'original_prompt': '请介绍Python的特点和应用场景。', 'task_id': 'task_001'}, 'feedback_type': 'text'}B. 反馈分析模块：提取关键洞察
这个模块是“Reflection”过程的核心，它将非结构化的反馈转化为Agent可理解和学习的结构化信息。我们将利用LLM和一些NLP工具来完成这项工作。
1. 情感分析 (Sentiment Analysis)
识别用户反馈的情感极性（正、负、中性）和强度，有助于我们过滤掉中性或无关的反馈，并优先处理负面反馈。
from transformers import pipeline
class SentimentAnalyzer:
"""
使用Hugging Face Transformers进行情感分析。
"""
def __init__(self, model_name: str = "distilbert-base-uncased-finetuned-sst-2-english"):
# 对于中文，可以考虑使用 'uer/roberta-base-finetuned-jd-binary-chinese' 或其他中文情感分析模型
# 这里为了演示方便，先使用英文模型
self.analyzer = pipeline("sentiment-analysis", model=model_name)
def analyze(self, text: str) -> Dict[str, Any]:
"""
对给定文本进行情感分析。
返回示例: {'label': 'NEGATIVE', 'score': 0.999}
"""
if not text:
return {"label": "NEUTRAL", "score": 0.5} # 空文本或无效文本视为中性
results = self.analyzer(text)
return results[0]
# 示例使用
if __name__ == '__main__':
sentiment_analyzer = SentimentAnalyzer()
feedback_texts = [
"这个回答简直是驴唇不对马嘴，完全不是我想要的！",
"It's absolutely wrong and not what I expected!",
"Well done, very helpful.",
"The answer is okay, nothing special."
]
for text in feedback_texts:
sentiment = sentiment_analyzer.analyze(text)
print(f"Text: '{text[:30]}...' -> Sentiment: {sentiment}")
# 注意：如果使用英文模型处理中文，结果可能不准确。实际应根据语言选择模型。
# Text: '这个回答简直是驴唇不对马嘴，...' -> Sentiment: {'label': 'NEGATIVE', 'score': 0.9997...} (这里假设模型能处理中文，或实际使用时替换为中文模型)
# Text: 'It's absolutely wrong and not...' -> Sentiment: {'label': 'NEGATIVE', 'score': 0.9998...}
# Text: 'Well done, very helpful.' -> Sentiment: {'label': 'POSITIVE', 'score': 0.9998...}
# Text: 'The answer is okay, nothing s...' -> Sentiment: {'label': 'POSITIVE', 'score': 0.9959...} (因为'okay'被判为正面)2. 关键词与主题提取 (Keyword & Topic Extraction)
提取反馈中的关键词和关键短语，能够帮助我们快速了解用户关注的焦点和具体问题所在。
from keybert import KeyBERT
class KeywordExtractor:
"""
使用KeyBERT提取文本中的关键词和关键短语。
"""
def __init__(self, model: str = "paraphrase-multilingual-MiniLM-L12-v2"):
# 使用多语言模型以更好地支持中文
self.kw_model = KeyBERT(model=model)
def extract(self, text: str, top_n: int = 5, keyphrase_ngram_range: tuple = (1, 2)) -> list[tuple[str, float]]:
"""
提取文本中的关键词或关键短语。
返回示例: [('语法错误', 0.78), ('边界条件处理', 0.75)]
"""
if not text:
return []
keywords = self.kw_model.extract_keywords(
text,
keyphrase_ngram_range=keyphrase_ngram_range,
stop_words='english', # 停用词可以根据语言调整或禁用
top_n=top_n,
use_mmr=True, # 使用Maximal Marginal Relevance确保多样性
diversity=0.7
)
return keywords
# 示例使用
if __name__ == '__main__':
keyword_extractor = KeywordExtractor()
feedback_text = "生成的代码有语法错误，并且逻辑上也不符合需求，应该更注重边界条件的处理。"
keywords = keyword_extractor.extract(feedback_text)
print(f"Text: '{feedback_text}' -> Keywords: {keywords}")
# Text: '生成的代码有语法错误，并且逻辑上也不符合需求，应该更注重边界条件的处理。' -> Keywords: [('语法错误', 0.81), ('边界条件处理', 0.77), ('逻辑不符合需求', 0.74), ('生成代码', 0.69), ('不符合需求', 0.65)]3. 问题分类与根因推断 (Issue Classification & Root Cause Inference)
这是最关键的一步，我们需要利用LLM的强大推理能力，将非结构化反馈映射到预定义的问题类型，并尝试推断Agent出错的深层原因。这需要精心设计的Prompt来引导LLM。
import os
import json
from openai import OpenAI # 假设使用OpenAI API，也可以替换为其他LLM客户端
from dotenv import load_dotenv
load_dotenv() # 加载.env文件中的环境变量，如OPENAI_API_KEY
class IssueAnalyzer:
"""
使用LLM对用户反馈进行问题分类和根因推断。
"""
def __init__(self, llm_client: Any, model_name: str = "gpt-4o"):
self.client = llm_client
self.model_name = model_name
self.issue_types = [
"事实错误", "格式错误", "理解偏差", "不完整", "不相关",
"逻辑错误", "代码错误", "语气/风格问题", "重复冗余", "其他"
]
def analyze(self, feedback_entry: FeedbackEntry) -> Dict[str, Any]:
"""
分析用户反馈，输出问题类型、详细描述和根因推断。
"""
original_prompt = feedback_entry.task_context.get('original_prompt', 'N/A')
agent_output = feedback_entry.agent_output
user_feedback = feedback_entry.user_feedback
system_prompt = f"""
你是一个严谨的AI错误分析专家。你的任务是深入分析Agent的输出和用户的负面反馈，
识别出Agent的核心问题类型，并推断导致这些问题发生的可能根源。
请严格按照以下JSON格式输出结果。
可用的问题类型列表: {', '.join(self.issue_types)}
输出格式要求：
{{
"issue_type": "从可用问题类型中选择最符合的一项",
"detailed_issue_description": "对Agent具体错误的详细描述，引用Agent输出和用户反馈中的关键信息",
"root_cause_inference": "推断Agent出错的深层原因，例如：对知识的理解偏差、Prompt指令理解不准确、忽略了某些约束、内部工具调用错误等。",
"suggested_improvement_area": "基于根因，提出未来改进Agent行为的广义方向，例如：增强事实核查、明确Prompt中的约束、细化逻辑步骤等。"
}}
"""
user_message = f"""
Agent的原始Prompt (任务上下文):
{original_prompt}
Agent的输出:
{agent_output}
用户的负面反馈:
{user_feedback}
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
response_format={"type": "json_object"},
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
temperature=0.2 # 降低温度以获得更确定性的结果
)
analysis_result = json.loads(response.choices[0].message.content)
return analysis_result
except Exception as e:
print(f"LLM调用失败: {e}")
return {
"issue_type": "其他",
"detailed_issue_description": f"LLM分析失败: {e}",
"root_cause_inference": "系统内部错误或LLM无法处理此反馈。",
"suggested_improvement_area": "检查LLM调用或反馈内容。"
}
# 示例使用
if __name__ == '__main__':
# 确保 OPENAI_API_KEY 环境变量已设置
llm_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
issue_analyzer = IssueAnalyzer(llm_client=llm_client)
agent_output_example = "Python是一种静态类型语言，广泛应用于Web开发。"
user_feedback_example = "这个说法不对，Python是动态类型语言。Agent的回答有事实性错误。"
original_prompt_example = "请介绍Python的特点和应用场景。"
task_id_example = "task_001"
feedback_entry = capture_feedback(
agent_output=agent_output_example,
user_feedback_raw=user_feedback_example,
original_prompt=original_prompt_example,
task_id=task_id_example
)
analysis_result = issue_analyzer.analyze(feedback_entry)
print("Issue Analysis Result:")
print(json.dumps(analysis_result, indent=2, ensure_ascii=False))
# 预期输出示例 (可能因模型而异，但结构相似)
# {
# "issue_type": "事实错误",
# "detailed_issue_description": "Agent错误地将Python描述为静态类型语言。用户明确指出Python是动态类型语言。",
# "root_cause_inference": "Agent对Python核心语言特性的知识存在偏差或记忆错误。",
# "suggested_improvement_area": "增强事实核查，特别是针对编程语言基础知识的准确性。"
# }
# 另一个例子：代码错误
agent_output_code = """
def add(a, b):
return a - b # 错误：应该是加法
print(add(1, 2))
"""
user_feedback_code = "这段代码的add函数有bug，1+2应该等于3，但它输出了-1。逻辑错了。"
original_prompt_code = "请用Python编写一个简单的加法函数，并提供一个测试示例。"
feedback_entry_code = capture_feedback(
agent_output=agent_output_code,
user_feedback_raw=user_feedback_code,
original_prompt=original_prompt_code,
task_id="task_002"
)
analysis_result_code = issue_analyzer.analyze(feedback_entry_code)
print("nIssue Analysis Result (Code Error):")
print(json.dumps(analysis_result_code, indent=2, ensure_ascii=False))
# 预期输出示例
# {
# "issue_type": "逻辑错误",
# "detailed_issue_description": "Agent在编写加法函数时，错误地使用了减法操作符，导致计算结果不正确。",
# "root_cause_inference": "Agent在将自然语言的“加法”概念转换为具体编程实现时，映射错误。",
# "suggested_improvement_area": "增强代码逻辑的准确性，特别是在基本算术运算的实现上；引入自检机制验证代码的意图与实现是否一致。"
# }C. 教训总结生成模块 (Lesson Learned Generator)
有了结构化的分析结果，我们现在可以引导LLM生成一份“教训总结”。这份总结不仅仅是错误的复述，更重要的是要提炼出普遍性的经验和可操作的改进方向。
class LessonLearnedGenerator:
"""
根据错误分析结果，生成一份可操作的“教训总结”。
"""
def __init__(self, llm_client: Any, model_name: str = "gpt-4o"):
self.client = llm_client
self.model_name = model_name
def generate(self, analysis_result: Dict[str, Any], original_prompt: str) -> str:
"""
生成教训总结。
"""
issue_type = analysis_result.get('issue_type', '未知错误')
detailed_issue = analysis_result.get('detailed_issue_description', '未提供详细描述')
root_cause = analysis_result.get('root_cause_inference', '未推断出根源')
suggested_improvement = analysis_result.get('suggested_improvement_area', '无具体建议')
system_prompt = """
你是一名经验丰富的导师，擅长从具体的错误案例中提炼出深刻而具有指导意义的教训。
你的任务是根据提供的Agent错误分析，生成一份简洁、清晰且可操作的“教训总结”。
这份总结应该包含：
1. 明确指出Agent在此次任务中犯了什么具体错误。
2. 深入分析导致此错误可能存在的深层原因（即根源）。
3. 最重要的是，为Agent在未来处理类似任务时提供具体的、可实践的指导，说明如何避免同类错误，并提升其性能。
请直接输出教训总结内容，不要包含任何前缀、标题或额外的说明文字。
"""
user_message = f"""
Agent的原始Prompt:
{original_prompt}
错误类型: {issue_type}
详细问题描述: {detailed_issue}
根源推断: {root_cause}
初步改进方向: {suggested_improvement}
请根据以上信息，生成一份教训总结。
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
temperature=0.5 # 适当提高温度以增加生成多样性，但仍保持相关性
)
return response.choices[0].message.content.strip()
except Exception as e:
print(f"LLM调用失败，无法生成教训总结: {e}")
return f"未能生成教训总结，错误: {e}"
# 示例使用 (继续上一个IssueAnalyzer的例子)
if __name__ == '__main__':
# 假设 analysis_result_code 已经从 IssueAnalyzer 获得
# analysis_result_code = {
# "issue_type": "逻辑错误",
# "detailed_issue_description": "Agent在编写加法函数时，错误地使用了减法操作符，导致计算结果不正确。",
# "root_cause_inference": "Agent在将自然语言的“加法”概念转换为具体编程实现时，映射错误。",
# "suggested_improvement_area": "增强代码逻辑的准确性，特别是在基本算术运算的实现上；引入自检机制验证代码的意图与实现是否一致。"
# }
# original_prompt_code = "请用Python编写一个简单的加法函数，并提供一个测试示例。"
lesson_generator = LessonLearnedGenerator(llm_client=llm_client)
lesson_learned = lesson_generator.generate(analysis_result_code, original_prompt_code)
print("nLesson Learned Summary:")
print(lesson_learned)
# 预期输出示例:
# 教训总结：Agent在实现“加法函数”时，将加法误实现为减法，这是一个明显的逻辑错误。
# 根本原因在于Agent未能准确地将人类语言中的数学概念精确映射到编程语言的具体操作符上，
# 可能缺乏对基础运算符号的严格检查机制，或是对任务意图的深层理解不足。
# 未来在处理编程任务，特别是涉及基本数学运算时，Agent必须：
# 1. 在代码生成后进行自我审查，确保核心逻辑与任务描述完全一致。
# 2. 对于简单的数学运算，优先使用明确的测试用例进行验证。
# 3. 增加对编程语言基础操作符的严格对照和确认，避免低级逻辑错误。D. Prompt 改写模块 (Prompt Rewriter)
Prompt改写是最终将学习到的教训内化到Agent行为中的步骤。这个模块将根据“教训总结”来修改原始Prompt，使其更具指导性、约束性和鲁棒性。
Prompt 改写策略：
- 增加明确约束：直接在Prompt中加入避免特定错误的指令。
- 引入负面示例 (Negative Examples)：通过“不要做X”或“这不是Y”的示例来指导Agent。
- 强化角色与语气：要求Agent以更严谨、更细致的角色来执行任务。
- 加入自检机制：鼓励Agent在生成输出前进行自我评估和验证。
- 明确边界条件：针对之前忽略的边界情况进行补充说明。
class PromptRewriter:
"""
根据教训总结，自动改写Agent的原始Prompt。
"""
def __init__(self, llm_client: Any, model_name: str = "gpt-4o"):
self.client = llm_client
self.model_name = model_name
def revise(self, original_prompt: str, lesson_learned: str) -> str:
"""
根据教训总结修订Prompt。
"""
system_prompt = """
你是一名顶级的Prompt工程师，擅长将复杂的教训总结转化为清晰、高效、具有指导性的新Prompt。
你的任务是根据Agent在过去任务中获得的“教训总结”，对原始Prompt进行修订。
修订后的Prompt应该能够帮助Agent避免再次犯同样的错误，并提升其在未来类似任务中的性能。
修订应侧重于增加必要的约束、明确要求、引入自检机制或调整Agent的角色，使其更健壮。
请直接输出修订后的完整Prompt内容，不要包含任何前缀、标题或额外的说明文字。
确保修订后的Prompt依然自然流畅，易于理解，并且能够直接用于Agent。
"""
user_message = f"""
原始Prompt:
{original_prompt}
教训总结:
{lesson_learned}
请根据上述信息，生成一个修订后的Prompt。
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
temperature=0.3 # 保持较低温度以确保Prompt的精准性
)
return response.choices[0].message.content.strip()
except Exception as e:
print(f"LLM调用失败，无法改写Prompt: {e}")
return original_prompt # 失败时返回原始Prompt
# 示例使用 (继续上一个LessonLearnedGenerator的例子)
if __name__ == '__main__':
# 假设 lesson_learned 已经从 LessonLearnedGenerator 获得
# lesson_learned = "教训总结：Agent在实现“加法函数”时，将加法误实现为减法，这是一个明显的逻辑错误。..."
# original_prompt_code = "请用Python编写一个简单的加法函数，并提供一个测试示例。"
prompt_rewriter = PromptRewriter(llm_client=llm_client)
revised_prompt = prompt_rewriter.revise(original_prompt_code, lesson_learned)
print("nRevised Prompt:")
print(revised_prompt)
# 预期输出示例:
# 请用Python编写一个简单的加法函数，并提供一个测试示例。
# 务必确保函数逻辑的准确性，特别是核心算术操作符（如加法应使用'+'，而非'-'）。
# 在生成代码后，请进行自我审查，确保函数行为与“加法”的数学定义完全一致，并通过提供的测试示例验证其正确性。E. Prompt 版本管理
Prompt作为Agent的“基因”，其演变过程需要被妥善管理。版本管理系统可以追踪Prompt的历史，支持回滚，并为A/B测试提供基础。
import sqlite3
import datetime
class PromptVersionManager:
"""
用于管理Agent Prompt版本的工具类。
支持保存、激活、获取Prompt，并追踪其来源。
"""
def __init__(self, db_path: str = "prompt_versions.db"):
self.conn = sqlite3.connect(db_path)
self._create_table()
def _create_table(self):
"""创建Prompt版本数据库表。"""
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS prompt_versions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
version_tag TEXT NOT NULL UNIQUE, -- 版本标签，例如 v1.0, v1.1-feedback-001
prompt_content TEXT NOT NULL, -- Prompt的具体内容
original_prompt_id INTEGER, -- 如果是修订版本，指向其父Prompt的ID
lesson_learned_source TEXT, -- 导致此修订的教训总结
timestamp TEXT NOT NULL, -- 版本创建时间
is_active INTEGER DEFAULT 0, -- 是否为当前活跃的Prompt (0:否, 1:是)
FOREIGN KEY (original_prompt_id) REFERENCES prompt_versions(id)
)
""")
self.conn.commit()
def save_prompt(self,
prompt_content: str,
version_tag: Optional[str] = None,
original_prompt_id: Optional[int] = None,
lesson_learned: Optional[str] = None,
make_active: bool = False) -> int:
"""
保存一个新的Prompt版本。
Args:
prompt_content: Prompt的文本内容。
version_tag: 可选的版本标签，如果未提供则自动生成。
original_prompt_id: 如果是修订版本，指向其所基于的原始Prompt的ID。
lesson_learned: 导致此修订的教训总结文本。
make_active: 是否将此新版本设置为活跃Prompt。
Returns:
新Prompt版本的ID。
"""
timestamp = datetime.datetime.now().isoformat()
if not version_tag:
version_tag = f"v{timestamp.replace(':', '-').replace('.', '-')}"
cursor = self.conn.cursor()
cursor.execute("""
INSERT INTO prompt_versions (version_tag, prompt_content, original_prompt_id, lesson_learned_source, timestamp, is_active)
VALUES (?, ?, ?, ?, ?, ?)
""", (version_tag, prompt_content, original_prompt_id, lesson_learned, timestamp, 1 if make_active else 0))
self.conn.commit()
new_id = cursor.lastrowid
if make_active:
self.set_active_prompt(new_id) # 确保只有一个活跃Prompt
return new_id
def get_prompt_by_id(self, prompt_id: int) -> Optional[Dict[str, Any]]:
"""根据ID获取Prompt内容和元数据。"""
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM prompt_versions WHERE id = ?", (prompt_id,))
row = cursor.fetchone()
if row:
columns = [description[0] for description in cursor.description]
return dict(zip(columns, row))
return None
def get_active_prompt(self) -> Optional[str]:
"""获取当前活跃的Prompt内容。"""
cursor = self.conn.cursor()
cursor.execute("SELECT prompt_content FROM prompt_versions WHERE is_active = 1 ORDER BY timestamp DESC LIMIT 1")
result = cursor.fetchone()
return result[0] if result else None
def get_active_prompt_id(self) -> Optional[int]:
"""获取当前活跃的Prompt ID。"""
cursor = self.conn.cursor()
cursor.execute("SELECT id FROM prompt_versions WHERE is_active = 1 ORDER BY timestamp DESC LIMIT 1")
result = cursor.fetchone()
return result[0] if result else None
def set_active_prompt(self, prompt_id: int):
"""
设置指定ID的Prompt为活跃Prompt，并停用其他所有Prompt。
"""
cursor = self.conn.cursor()
cursor.execute("UPDATE prompt_versions SET is_active = 0") # 先停用所有Prompt
cursor.execute("UPDATE prompt_versions SET is_active = 1 WHERE id = ?", (prompt_id,)) # 激活指定Prompt
self.conn.commit()
print(f"Prompt ID {prompt_id} 已设置为活跃。")
def get_all_prompts(self) -> list[Dict[str, Any]]:
"""获取所有Prompt版本及其元数据。"""
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM prompt_versions ORDER BY timestamp DESC")
rows = cursor.fetchall()
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in rows]
def close(self):
"""关闭数据库连接。"""
self.conn.close()
# 示例使用
if __name__ == '__main__':
pvm = PromptVersionManager(db_path="my_agent_prompts.db")
# 1. 保存初始Prompt
initial_prompt = "请介绍Python的特点和应用场景。"
initial_prompt_id = pvm.save_prompt(initial_prompt, version_tag="v1.0", make_active=True)
print(f"初始Prompt (ID: {initial_prompt_id}) 已保存并激活。")
print(f"当前活跃Prompt: {pvm.get_active_prompt()}")
# 模拟经过反馈和修订
# 假设 revised_prompt 和 lesson_learned 来自上文的 PromptRewriter 和 LessonLearnedGenerator
revised_prompt_from_lesson = revised_prompt # 使用上文生成的修订Prompt
lesson_summary_from_generator = lesson_learned # 使用上文生成的教训总结
# 2. 保存修订后的Prompt
revised_prompt_id = pvm.save_prompt(
prompt_content=revised_prompt_from_lesson,
version_tag="v1.1-fact-correction",
original_prompt_id=initial_prompt_id,
lesson_learned=lesson_summary_from_generator,
make_active=True # 激活新版本
)
print(f"n修订后Prompt (ID: {revised_prompt_id}) 已保存并激活。")
print(f"当前活跃Prompt: {pvm.get_active_prompt()}")
# 3. 查看所有版本
print("n所有Prompt版本:")
for prompt_info in pvm.get_all_prompts():
print(f" ID: {prompt_info['id']}, Tag: {prompt_info['version_tag']}, Active: {bool(prompt_info['is_active'])}, Timestamp: {prompt_info['timestamp']}")
# 4. 回滚到初始版本 (如果需要)
# pvm.set_active_prompt(initial_prompt_id)
# print(f"n已回滚到Prompt ID {initial_prompt_id}。")
# print(f"当前活跃Prompt: {pvm.get_active_prompt()}")
pvm.close()F. 效果评估模块
仅仅修改Prompt是不够的，我们还需要验证这些修改是否真正带来了Agent性能的提升。效果评估模块负责收集数据并衡量改进。
评估指标：
- 负面反馈率 (Negative Feedback Rate)：这是最直接的指标。新Prompt上线后，在相同或类似任务上，负面反馈的比例是否显著下降？
- 特定错误类型发生频率：例如，如果之前频繁出现“事实错误”，新Prompt是否能有效减少这类错误？
- 用户满意度评分：通过用户调研、满意度问卷或隐式信号（如用户停留时间、重复使用率）来衡量。
- A/B 测试：将一部分用户流量引导至使用旧Prompt的Agent，另一部分引导至使用新Prompt的Agent，通过对比两组的性能指标来评估。
- 人工评审：定期对Agent使用新Prompt生成的输出进行人工质量检查。
挑战：评估本身也需要数据和反馈，这形成了一个更大的反馈循环。有时，改进一个方面可能会在另一个方面引入新的问题，需要权衡。
示例：一个简化的负面反馈率监控
class PerformanceEvaluator:
"""
一个简化的性能评估器，用于监控负面反馈率。
在实际系统中，这将是一个更复杂的指标收集和分析系统。
"""
def __init__(self, feedback_db_path: str = "feedback.db"):
self.conn = sqlite3.connect(feedback_db_path)
self._create_table()
def _create_table(self):
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS evaluation_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt_id INTEGER NOT NULL,
task_id TEXT NOT NULL,
is_negative_feedback INTEGER NOT NULL, -- 1 for negative, 0 for neutral/positive
timestamp TEXT NOT NULL,
FOREIGN KEY (prompt_id) REFERENCES prompt_versions(id)
)
""")
self.conn.commit()
def log_feedback_for_evaluation(self, prompt_id: int, task_id: str, is_negative: bool):
"""记录每次任务执行后的反馈情况。"""
timestamp = datetime.datetime.now().isoformat()
cursor = self.conn.cursor()
cursor.execute("""
INSERT INTO evaluation_metrics (prompt_id, task_id, is_negative_feedback, timestamp)
VALUES (?, ?, ?, ?)
""", (prompt_id, task_id, 1 if is_negative else 0, timestamp))
self.conn.commit()
def calculate_negative_feedback_rate(self, prompt_id: int = None, time_window_days: int = 7) -> float:
"""
计算指定Prompt或所有Prompt在给定时间窗口内的负面反馈率。
"""
cutoff_time = datetime.datetime.now() - datetime.timedelta(days=time_window_days)
cursor = self.conn.cursor()
query = "SELECT COUNT(*) FROM evaluation_metrics WHERE timestamp >= ?"
params = [cutoff_time.isoformat()]
if prompt_id:
query += " AND prompt_id = ?"
params.append(prompt_id)
total_feedback_count = cursor.execute(query, params).fetchone()[0]
query_negative = query + " AND is_negative_feedback = 1"
negative_feedback_count = cursor.execute(query_negative, params).fetchone()[0]
if total_feedback_count == 0:
return 0.0
return (negative_feedback_count / total_feedback_count) * 100
def close(self):
self.conn.close()
# 示例使用
if __name__ == '__main__':
# 假设 pvm 实例和 initial_prompt_id, revised_prompt_id 已经设置
evaluator = PerformanceEvaluator(feedback_db_path="my_agent_feedback.db")
# 模拟使用初始Prompt的几次任务执行和反馈
evaluator.log_feedback_for_evaluation(initial_prompt_id, "task_A1", is_negative=True)
evaluator.log_feedback_for_evaluation(initial_prompt_id, "task_A2", is_negative=False)
evaluator.log_feedback_for_evaluation(initial_prompt_id, "task_A3", is_negative=True)
# 模拟使用修订后Prompt的几次任务执行和反馈
evaluator.log_feedback_for_evaluation(revised_prompt_id, "task_B1", is_negative=False)
evaluator.log_feedback_for_evaluation(revised_prompt_id, "task_B2", is_negative=False)
evaluator.log_feedback_for_evaluation(revised_prompt_id, "task_B3", is_negative=True) # 仍然可能有负面反馈
# 计算负面反馈率
rate_initial = evaluator.calculate_negative_feedback_rate(initial_prompt_id)
rate_revised = evaluator.calculate_negative_feedback_rate(revised_prompt_id)
print(f"n初始Prompt (ID: {initial_prompt_id}) 的负面反馈率: {rate_initial:.2f}%")
print(f"修订后Prompt (ID: {revised_prompt_id}) 的负面反馈率: {rate_revised:.2f}%")
evaluator.close()五、挑战与未来展望
尽管“Reflection-on-Feedback”机制展现了巨大的潜力，但在实际落地过程中，我们仍需面对诸多挑战：
- 反馈质量与量化：如何有效处理低质量、模糊、甚至有偏见的反馈？如何将定性反馈转化为Agent可理解的定量信号？
- LLM的推理局限性：LLM在根因推断、泛化教训和精确改写Prompt时，仍可能出现“幻觉”或理解偏差。如何确保其分析和修改的准确性和可靠性？
- 过拟合风险：Agent可能过度优化针对某个特定负面反馈的Prompt，导致在其他更广泛的任务场景中表现下降。如何平衡特定改进与通用鲁棒性？
- 多维度冲突：一个Prompt可能同时影响Agent在多个维度（如准确性、创造力、简洁性）的表现。改进一个维度可能牺牲另一个维度，如何进行权衡和优化？
- 可解释性与透明度：当Prompt被自动修改后，我们如何理解其背后的原因和Agent的“学习路径”？这对于调试和人工干预至关重要。
- 复杂任务的反馈：对于涉及多步骤、多工具调用的复杂Agent任务，如何定位具体哪个子步骤或哪个工具调用导致了问题，从而生成更精确的教训？
- 自主实验与探索：Agent能否超越被动响应反馈，主动通过小规模的自我实验或模拟来发现Prompt的潜在不足，并提前进行优化？
- 多模态反馈：未来Agent可能接收语音、图片、视频等多模态反馈，这需要更高级的感知和理解能力。
展望未来，我们期待Agent能够：
- 更智能的根因分析：结合更丰富的上下文信息（如Agent内部思考链、工具调用日志），更精准地定位问题根源。
- 自适应的Prompt策略：根据任务类型、用户偏好和历史表现，智能选择和组合不同的Prompt优化策略。
- 主动式学习：Agent不仅从负面反馈中学习，也能从正面反馈中提取成功模式，并将其泛化到新的Prompt中。
- 人机协作的Prompt工程：在关键决策点（例如重大Prompt修改前），能够征求人类专家的意见，形成高效的人机协作循环。
六、持续学习与智能体的进化路径
今天我们探讨的“Reflection-on-Feedback”机制，不仅仅是技术层面的创新，它更代表了Agent从被动执行到主动学习、从工具属性到智能伙伴的进化方向。通过构建一个能够自动解析负面反馈、总结教训并迭代优化自身Prompt的系统，我们正在赋予Agent一种核心的自我改进能力。
这不仅仅是简单的迭代，而是一种深刻的自省机制。Agent将不再只是执行者，它将成为一个能够理解自身错误、从中汲取经验、并不断调整自身“认知模型”的智能体。这条持续学习与进化的路径，将最终引领我们走向更健壮、更智能、更值得信赖的通用人工智能。这条路充满挑战，但也孕育着无限的可能。我坚信，在座各位的共同努力下，我们将能够攻克这些技术难关，共同推动AI技术迈向新的高峰。
谢谢大家！