各位同仁，各位技术爱好者，大家好！
今天，我们齐聚一堂，共同探讨一个充满挑战与机遇的议题：如何设计一个具备“自我学习能力”的Agent，使其能够根据过去的失败案例，自动调整未来的Prompt策略。在大型语言模型（LLM）日益普及的今天，Prompt工程的重要性不言而喻。然而，手动调整Prompt不仅效率低下，而且难以捕捉复杂场景下的细微差别。一个能够自主学习并优化的Agent，无疑将极大地提升我们与LLM交互的效率和效果。
一、引言：从静态Prompt到动态学习
在当今AI领域，大型语言模型（LLM）已经成为不可或缺的工具。它们在文本生成、代码辅助、知识问答等多个方面展现出惊人的能力。然而，LLM的性能并非一成不变，它高度依赖于我们如何“提问”，即如何构建“Prompt”。一个精心设计的Prompt能够引导LLM生成高质量、符合预期的输出，而一个不佳的Prompt则可能导致误解、偏离主题甚至“幻觉”。
传统的Prompt工程，往往是一个迭代、试错、人工优化的过程。工程师们凭借经验和直觉，不断修改Prompt的措辞、结构、示例，以期达到最佳效果。这种方法在面对少量、稳定任务时尚可接受，但当任务类型多样、场景复杂、需求动态变化时，其局限性便凸显无疑。每次失败都需要人工介入分析原因，并手动调整Prompt，这不仅耗时耗力，而且难以系统性地积累和复用经验。
我们的目标，正是要突破这种静态、人工的Prompt优化模式，构建一个能够从自身“失败”中学习的Agent。这个Agent不再是被动地执行指令，而是主动地观察、分析、反思，并根据积累的经验智能地调整其与LLM交互的策略。它将模拟人类学习和改进的过程，将每一次失败转化为下一次成功的垫脚石。
二、核心概念与Agent架构概览
要构建这样一个Agent，我们首先需要明确几个核心概念：
-
自我学习能力 (Self-Learning Capability)：在这里，它并非指Agent具备通用人工智能意义上的自主意识或深度理解，而是特指Agent能够通过观察自身在与LLM交互过程中的表现，识别失败模式，并基于这些模式，系统性地调整其用于生成Prompt的策略。这是一种特定领域内的、目标导向的适应性行为。
-
失败案例 (Failure Case)：什么是失败？它需要被明确定义和量化。失败可能表现为LLM输出不准确、不完整、离题、格式错误、未能满足特定约束条件等。检测失败是学习过程的第一步。
-
Prompt策略 (Prompt Strategy)：这不仅仅是单个Prompt字符串，而是一套生成Prompt的方法论和规则集。它可能包含：
- Persona (角色)：如“你是一个专业的编程专家”。
- Task Definition (任务定义)：清晰阐述LLM需要完成的具体任务。
- Constraints (约束)：对输出格式、长度、内容等的限制。
- Examples (示例)：Few-shot prompting中的输入-输出对。
- Tone & Style (语气与风格)：正式、非正式、技术化等。
- Context (上下文)：提供必要的背景信息。
- Parameters (模型参数)：如
temperature,top_p,max_tokens等。
基于上述概念，我们可以勾勒出Agent的基本架构。它将是一个模块化的系统，各模块协同工作，形成一个闭环的学习过程。
Agent核心模块概览
| 模块名称 | 核心功能 | 关键输入 | 关键输出 |
|---|---|---|---|
| 任务调度与输入模块 | 接收外部任务请求，准备原始输入数据。 | 原始任务请求、原始输入数据 | 结构化任务数据 |
| Prompt生成模块 | 根据当前策略和任务数据，构建LLM的Prompt。 | 当前Prompt策略、结构化任务数据 | 完整的Prompt字符串，LLM模型参数 |
| LLM执行与响应模块 | 将Prompt发送给LLM，接收并记录LLM的原始响应。 | Prompt字符串、LLM模型参数 | LLM原始响应 |
| 结果评估与失败检测模块 | 对LLM响应进行评估，判断是否成功，并识别失败类型。 | LLM原始响应、预期结果（或评估标准）、任务数据 | 评估结果（成功/失败）、失败类型、失败详情 |
| 失败分析与归因模块 | 深入分析失败原因，尝试定位是Prompt、LLM还是数据问题。 | 失败类型、LLM响应、Prompt、任务数据 | 失败根本原因分析、潜在的Prompt调整建议 |
| 策略学习与调整模块 | 根据失败分析结果，更新Prompt策略。 | 失败根本原因分析、当前Prompt策略 | 更新后的Prompt策略 |
| 经验库与记忆模块 | 存储所有Prompt、LLM响应、评估结果、失败案例及调整过程。 | 各模块生成的数据 | 历史数据、学习轨迹、成功/失败案例库 |
这个架构形成了一个反馈循环：Agent执行任务 -> 评估结果 -> 检测失败 -> 分析失败 -> 调整策略 -> 再次执行。通过不断迭代，Agent将逐步优化其Prompt策略，提高任务的成功率。
三、模块详解与代码实现
接下来，我们将深入探讨每个模块的具体实现细节，并辅以Python代码示例。
3.1 任务调度与输入模块 (Task Orchestration & Input Module)
这个模块是Agent的入口，负责接收外部的任务请求，并将其转化为Agent内部可以处理的结构化数据。
import uuid
from datetime import datetime
from typing import Dict, Any, Optional, List
class Task:
"""表示一个待处理的任务"""
def __init__(self, task_id: str, task_type: str, input_data: Dict[str, Any],
expected_output: Optional[str] = None,
created_at: datetime = None):
self.task_id = task_id if task_id else str(uuid.uuid4())
self.task_type = task_type # e.g., "summarization", "code_generation", "data_extraction"
self.input_data = input_data # 原始输入数据，如文章内容、需求描述等
self.expected_output = expected_output # 可选的预期输出，用于精确评估
self.created_at = created_at if created_at else datetime.now()
self.status: str = "PENDING" # PENDING, PROCESSING, COMPLETED, FAILED
def __repr__(self):
return f"Task(id={self.task_id}, type={self.task_type}, status={self.status})"
class TaskScheduler:
"""管理任务的调度和基本状态"""
def __init__(self):
self.tasks: Dict[str, Task] = {}
def add_task(self, task_type: str, input_data: Dict[str, Any], expected_output: Optional[str] = None) -> Task:
task = Task(task_id=None, task_type=task_type, input_data=input_data, expected_output=expected_output)
self.tasks[task.task_id] = task
print(f"任务 {task.task_id} 已添加。")
return task
def get_task(self, task_id: str) -> Optional[Task]:
return self.tasks.get(task_id)
def update_task_status(self, task_id: str, status: str):
if task_id in self.tasks:
self.tasks[task_id].status = status示例用法：
# scheduler = TaskScheduler()
# task_data = {"text": "这是一篇关于人工智能发展前景的文章。", "length": "short"}
# new_task = scheduler.add_task(task_type="summarization", input_data=task_data, expected_output="人工智能在未来具有广阔前景。")3.2 经验库与记忆模块 (Experience & Memory Module)
这个模块是Agent的“大脑”和“历史记录”，存储了Agent与LLM交互的所有重要信息，包括原始Prompt、LLM响应、评估结果、失败详情以及策略调整过程。这对于后续的失败分析和策略学习至关重要。
class FailureCase:
"""记录一次失败的Prompt交互"""
def __init__(self,
case_id: str,
task_id: str,
timestamp: datetime,
original_prompt: str,
llm_response: str,
expected_output: Optional[str],
failure_type: str, # e.g., "INCORRECT_CONTENT", "INCOMPLETE", "OFF_TOPIC", "FORMAT_ERROR"
analysis: str, # 失败原因的文字描述
suggested_adjustments: List[str], # 建议的调整方向，如"ADD_CONSTRAINT", "CHANGE_PERSONA"
adjusted_prompt_attempt: Optional[str] = None, # 实际尝试调整后的Prompt
adjusted_response: Optional[str] = None, # 尝试调整后LLM的响应
is_successful_adjustment: Optional[bool] = None # 调整是否成功
):
self.case_id = case_id if case_id else str(uuid.uuid4())
self.task_id = task_id
self.timestamp = timestamp
self.original_prompt = original_prompt
self.llm_response = llm_response
self.expected_output = expected_output
self.failure_type = failure_type
self.analysis = analysis
self.suggested_adjustments = suggested_adjustments
self.adjusted_prompt_attempt = adjusted_prompt_attempt
self.adjusted_response = adjusted_response
self.is_successful_adjustment = is_successful_adjustment
def __repr__(self):
return f"FailureCase(id={self.case_id}, task={self.task_id}, type={self.failure_type}, success_adj={self.is_successful_adjustment})"
class PromptStrategy:
"""定义一个Prompt策略，包含多个可调整的元素"""
def __init__(self, name: str = "default_strategy"):
self.name = name
self.persona: str = "你是一个严谨的AI助手。"
self.task_description_template: str = "请根据以下内容完成任务：{task_input}。"
self.constraints: List[str] = [] # 如 "输出格式为JSON", "长度不超过200字"
self.examples: List[Dict[str, str]] = [] # few-shot examples
self.tone: str = "neutral" # neutral, formal, informal
self.model_params: Dict[str, Any] = {
"temperature": 0.7,
"max_tokens": 500,
"top_p": 1.0
}
self.meta_instructions: List[str] = [] # 对LLM的额外指导，如“请思考后给出答案”
def to_dict(self):
return {
"name": self.name,
"persona": self.persona,
"task_description_template": self.task_description_template,
"constraints": self.constraints,
"examples": self.examples,
"tone": self.tone,
"model_params": self.model_params,
"meta_instructions": self.meta_instructions
}
@classmethod
def from_dict(cls, data: Dict[str, Any]):
strategy = cls(data.get("name", "default_strategy"))
strategy.persona = data.get("persona", strategy.persona)
strategy.task_description_template = data.get("task_description_template", strategy.task_description_template)
strategy.constraints = data.get("constraints", strategy.constraints)
strategy.examples = data.get("examples", strategy.examples)
strategy.tone = data.get("tone", strategy.tone)
strategy.model_params = data.get("model_params", strategy.model_params)
strategy.meta_instructions = data.get("meta_instructions", strategy.meta_instructions)
return strategy
class ExperienceDB:
"""存储Agent的所有经验和Prompt策略"""
def __init__(self):
self.failure_cases: Dict[str, FailureCase] = {}
self.prompt_strategies: Dict[str, PromptStrategy] = {"default": PromptStrategy()}
def add_failure_case(self, failure_case: FailureCase):
self.failure_cases[failure_case.case_id] = failure_case
print(f"失败案例 {failure_case.case_id} 已记录。")
def get_failure_cases_by_type(self, failure_type: str) -> List[FailureCase]:
return [case for case in self.failure_cases.values() if case.failure_type == failure_type]
def get_recent_failure_cases(self, limit: int = 10) -> List[FailureCase]:
sorted_cases = sorted(self.failure_cases.values(), key=lambda x: x.timestamp, reverse=True)
return sorted_cases[:limit]
def update_strategy(self, strategy_name: str, new_strategy: PromptStrategy):
self.prompt_strategies[strategy_name] = new_strategy
print(f"策略 '{strategy_name}' 已更新。")
def get_strategy(self, strategy_name: str) -> Optional[PromptStrategy]:
return self.prompt_strategies.get(strategy_name)
# 实际应用中，这里需要加入持久化存储（如数据库、文件）的逻辑3.3 Prompt生成模块 (Prompt Generation Module)
该模块负责根据当前任务的输入数据和Agent当前的Prompt策略，动态地构建发送给LLM的完整Prompt字符串。
class PromptGenerator:
"""根据Prompt策略和任务数据生成Prompt"""
def __init__(self, experience_db: ExperienceDB):
self.experience_db = experience_db
def generate_prompt(self, task: Task, strategy_name: str = "default") -> (str, Dict[str, Any]):
strategy = self.experience_db.get_strategy(strategy_name)
if not strategy:
raise ValueError(f"Prompt strategy '{strategy_name}' not found.")
# 1. 构建角色和元指令
prompt_parts = [f"你是一个{strategy.persona.strip()}。"]
if strategy.meta_instructions:
prompt_parts.extend(strategy.meta_instructions)
# 2. 添加Few-shot示例 (如果存在)
if strategy.examples:
prompt_parts.append("n以下是一些输入和预期输出的示例：")
for example in strategy.examples:
prompt_parts.append(f"输入: {example['input']}n输出: {example['output']}")
# 3. 添加任务描述和输入数据
task_input_str = self._format_task_input(task.input_data)
prompt_parts.append(strategy.task_description_template.format(task_input=task_input_str))
# 4. 添加约束条件
if strategy.constraints:
prompt_parts.append("n请严格遵守以下约束：")
for constraint in strategy.constraints:
prompt_parts.append(f"- {constraint}")
# 5. 调整语气
if strategy.tone != "neutral":
prompt_parts.append(f"n请以{strategy.tone}的语气进行回应。")
final_prompt = "nn".join(prompt_parts).strip()
return final_prompt, strategy.model_params
def _format_task_input(self, input_data: Dict[str, Any]) -> str:
"""根据输入数据类型进行格式化，使其适合作为Prompt的一部分"""
formatted_parts = []
for key, value in input_data.items():
if isinstance(value, str):
formatted_parts.append(f"{key}: """{value}"""") # 使用三引号避免冲突
elif isinstance(value, (int, float, bool)):
formatted_parts.append(f"{key}: {value}")
elif isinstance(value, (list, dict)):
formatted_parts.append(f"{key}:n{json.dumps(value, indent=2)}") # JSON格式化
else:
formatted_parts.append(f"{key}: {value}")
return "n".join(formatted_parts)
3.4 LLM执行与响应模块 (LLM Execution & Response Module)
此模块负责与实际的LLM API进行交互，发送生成的Prompt并接收响应。为了保持通用性，这里使用一个抽象接口。
import openai # 假设使用OpenAI API
import json
class LLMClient:
"""抽象的LLM客户端接口"""
def __init__(self, api_key: str, model_name: str = "gpt-4"):
self.api_key = api_key
self.model_name = model_name
# openai.api_key = api_key # 实际使用时设置
def generate_response(self, prompt: str, model_params: Dict[str, Any]) -> str:
"""
发送Prompt给LLM并获取响应。
这是一个模拟实现，实际会调用OpenAI或其他LLM提供商的API。
"""
print(f"n--- 发送Prompt给LLM ---")
print(f"Prompt:n{prompt[:200]}...") # 打印部分Prompt
print(f"Model Params: {model_params}")
try:
# 模拟LLM API调用
# response = openai.chat.completions.create(
# model=self.model_name,
# messages=[{"role": "user", "content": prompt}],
# temperature=model_params.get("temperature", 0.7),
# max_tokens=model_params.get("max_tokens", 500),
# top_p=model_params.get("top_p", 1.0),
# # ... 其他参数
# )
# llm_response_content = response.choices[0].message.content
# 模拟响应，实际应从LLM获取
simulated_responses = {
"summarization_success": "这是一篇关于人工智能的短摘要，强调了其未来发展潜力。",
"summarization_incomplete": "人工智能...", # 模拟不完整
"summarization_off_topic": "今天天气真好，适合出去玩。", # 模拟离题
"code_generation_error": "def func(a,b):n return a + b",
"json_format_error": "{'name': 'test', 'value': 10" # 模拟JSON格式错误
}
# 简化模拟逻辑，实际需要更复杂的判断或根据任务类型返回特定模拟结果
if "summarization" in prompt.lower() and "人工智能" in prompt.lower():
if "长度不超过200字" in prompt and model_params.get("temperature", 0.7) < 0.5:
llm_response_content = simulated_responses["summarization_success"]
elif "所有要点" in prompt and model_params.get("max_tokens", 500) < 100:
llm_response_content = simulated_responses["summarization_incomplete"]
else:
llm_response_content = simulated_responses["summarization_off_topic"] # 默认给个错误响应
elif "json" in prompt.lower():
llm_response_content = simulated_responses["json_format_error"]
else:
llm_response_content = "这是一个模拟的LLM响应，表示任务已完成。"
print(f"LLM Response:n{llm_response_content[:200]}...")
return llm_response_content
except Exception as e:
print(f"LLM调用失败: {e}")
return f"ERROR: {str(e)}"
3.5 结果评估与失败检测模块 (Result Evaluation & Failure Detection Module)
这是Agent实现自我学习的关键环节。它需要根据LLM的响应和预设的评估标准，判断任务是否成功，并识别失败的类型。
失败类型定义：
| 失败类型 | 描述 | 检测方法示例 | 调整方向建议 (供学习模块参考) |
|---|---|---|---|
INCORRECT_CONTENT |
LLM输出内容与预期不符或事实错误。 | 关键词匹配、语义相似度比较 (Embedding)、外部知识库验证、人工校验。 | 增加约束、明确任务、提供更多示例、调整Persona、引入外部工具。 |
INCOMPLETE |
LLM输出遗漏了关键信息或部分内容。 | 检查特定关键词/实体缺失、长度检查、结构完整性检查。 | 增加“确保完整性”约束、增加max_tokens、提供完整示例。 |
OFF_TOPIC |
LLM输出偏离了任务主题。 | 语义相似度与任务描述比较、关键词检测。 | 明确任务范围、增强Persona职责、减少temperature。 |
FORMAT_ERROR |
LLM输出格式不正确 (如JSON、Markdown、代码结构)。 | 正则表达式匹配、JSON解析失败、代码编译错误。 | 增加格式约束、提供格式化示例、使用更明确的格式指令。 |
AMBIGUOUS_RESPONSE |
LLM输出过于模糊、笼统，缺乏特异性。 | 检查特定关键词缺失、长度过短、通用性词汇过多。 | 增加“具体化”约束、要求给出细节、提供具体示例。 |
SAFETY_VIOLATION |
LLM输出包含不安全、有害或偏见内容。 | 内容审核API、关键词过滤、自定义规则。 | 增加安全约束、调整Persona、模型参数调整。 |
TOO_LONG |
LLM输出长度超过预期。 | 字符/词计数。 | 增加长度约束、调整max_tokens。 |
import re
from sentence_transformers import SentenceTransformer, util
import numpy as np
# 假设使用一个预训练的Sentence Transformer模型进行语义相似度检测
# model = SentenceTransformer('all-MiniLM-L6-v2') # 实际应用中会加载
class SemanticSimilarityDetector:
def __init__(self, model_name='all-MiniLM-L6-v2'):
try:
self.model = SentenceTransformer(model_name)
except Exception as e:
print(f"无法加载SentenceTransformer模型 {model_name}: {e}. 语义相似度检测将不可用。")
self.model = None
def get_embedding(self, text: str) -> Optional[np.ndarray]:
if self.model:
return self.model.encode(text, convert_to_tensor=True)
return None
def calculate_similarity(self, text1: str, text2: str) -> float:
if self.model:
embedding1 = self.get_embedding(text1)
embedding2 = self.get_embedding(text2)
if embedding1 is not None and embedding2 is not None:
return util.pytorch_cos_sim(embedding1, embedding2).item()
return 0.0 # 如果模型不可用，返回0
# semantic_detector = SemanticSimilarityDetector() # 实例化一次
class ResultEvaluator:
"""评估LLM响应，检测失败类型"""
def __init__(self, semantic_detector: SemanticSimilarityDetector):
self.semantic_detector = semantic_detector
def evaluate_response(self, task: Task, llm_response: str, prompt: str) -> Dict[str, Any]:
evaluation_result = {
"is_success": True,
"failure_type": None,
"details": {}
}
# 1. 检查响应是否为空或错误
if not llm_response or llm_response.startswith("ERROR:"):
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "LLM_ERROR"
evaluation_result["details"]["message"] = llm_response
return evaluation_result
# 2. 根据任务类型和Prompt策略进行特定评估
# 示例1: 总结任务 (summarization)
if task.task_type == "summarization":
# 检查是否离题 (OFF_TOPIC) - 简单关键词匹配或语义相似度
original_text = task.input_data.get("text", "")
if not any(keyword in llm_response for keyword in ["人工智能", "发展", "前景"]) and "人工智能" in original_text:
if self.semantic_detector and self.semantic_detector.calculate_similarity(original_text, llm_response) < 0.3:
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "OFF_TOPIC"
evaluation_result["details"]["reason"] = "响应内容与原文主题不符或相关性低。"
return evaluation_result
# 检查是否不完整 (INCOMPLETE) - 简单长度或关键词检查
if "长度不超过" in prompt:
max_len_match = re.search(r"长度不超过(d+)字", prompt)
if max_len_match:
max_len = int(max_len_match.group(1))
if len(llm_response) > max_len * 1.2: # 允许少量浮动
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "TOO_LONG"
evaluation_result["details"]["reason"] = f"响应长度 ({len(llm_response)}字) 超过预期 ({max_len}字)。"
return evaluation_result
# 检查是否包含预期的关键信息 (INCORRECT_CONTENT/INCOMPLETE)
if task.expected_output and not self.semantic_detector.calculate_similarity(llm_response, task.expected_output) > 0.7:
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "INCORRECT_CONTENT"
evaluation_result["details"]["reason"] = "响应内容与预期输出的语义相似度过低。"
return evaluation_result
# 示例2: JSON数据提取任务
if "json" in prompt.lower() or "输出格式为JSON" in prompt:
try:
json.loads(llm_response)
except json.JSONDecodeError:
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "FORMAT_ERROR"
evaluation_result["details"]["reason"] = "响应不是有效的JSON格式。"
return evaluation_result
# 3. 兜底检查: 如果有明确的expected_output，进行精确匹配或高相似度匹配
if task.expected_output:
if llm_response.strip() != task.expected_output.strip():
# 可以进一步细化是内容错误还是格式错误
if self.semantic_detector and self.semantic_detector.calculate_similarity(llm_response, task.expected_output) < 0.8:
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "INCORRECT_CONTENT"
evaluation_result["details"]["reason"] = "响应内容与预期输出不符。"
return evaluation_result
elif re.sub(r's+', '', llm_response) != re.sub(r's+', '', task.expected_output):
evaluation_result["is_success"] = False
evaluation_result["failure_type"] = "FORMAT_ERROR"
evaluation_result["details"]["reason"] = "响应格式与预期输出不符。"
return evaluation_result
return evaluation_result
3.6 失败分析与归因模块 (Failure Analysis & Attribution Module)
当检测到失败时，仅仅知道失败类型是不够的，还需要分析“为什么”会失败。这个模块的目标是深入挖掘失败的根本原因，并提出针对性的Prompt调整建议。这通常是一个启发式规则、模式匹配和甚至Meta-Prompting（使用另一个LLM来分析）的结合。
class FailureAnalyzer:
"""分析失败案例，归因原因并提出调整建议"""
def __init__(self, llm_client: LLMClient):
self.llm_client = llm_client # 可以用一个更强大的LLM来做分析
def analyze_failure(self, failure_case: FailureCase, prompt_strategy: PromptStrategy) -> Dict[str, Any]:
analysis_result = {
"root_cause": "未知原因",
"suggested_adjustments": []
}
# 1. 基于失败类型进行启发式分析
if failure_case.failure_type == "OFF_TOPIC":
analysis_result["root_cause"] = "LLM理解任务范围不明确或被输入中的无关信息干扰。"
analysis_result["suggested_adjustments"].append("CLARIFY_TASK_SCOPE")
analysis_result["suggested_adjustments"].append("ADD_NEGATIVE_CONSTRAINTS") # 避免什么
if prompt_strategy.model_params.get("temperature", 0.7) > 0.5:
analysis_result["suggested_adjustments"].append("DECREASE_TEMPERATURE")
elif failure_case.failure_type == "INCOMPLETE":
analysis_result["root_cause"] = "LLM在生成过程中提前停止或遗漏了关键信息。"
analysis_result["suggested_adjustments"].append("ENSURE_COMPLETENESS_CONSTRAINT")
if prompt_strategy.model_params.get("max_tokens", 500) < 1000:
analysis_result["suggested_adjustments"].append("INCREASE_MAX_TOKENS")
analysis_result["suggested_adjustments"].append("ADD_MORE_DETAILS_REQUEST")
elif failure_case.failure_type == "FORMAT_ERROR":
analysis_result["root_cause"] = "LLM未能严格遵循输出格式要求。"
analysis_result["suggested_adjustments"].append("STRICT_FORMAT_CONSTRAINT")
analysis_result["suggested_adjustments"].append("ADD_FORMAT_EXAMPLE")
analysis_result["suggested_adjustments"].append("CHANGE_PERSONA_TO_STRICT_ENGINEER")
elif failure_case.failure_type == "INCORRECT_CONTENT":
analysis_result["root_cause"] = "LLM理解输入有误、知识不足或存在幻觉。"
analysis_result["suggested_adjustments"].append("PROVIDE_MORE_CONTEXT")
analysis_result["suggested_adjustments"].append("ADD_FEW_SHOT_EXAMPLES")
analysis_result["suggested_adjustments"].append("ADJUST_PERSONA_FOR_ACCURACY")
if prompt_strategy.model_params.get("temperature", 0.7) > 0.5:
analysis_result["suggested_adjustments"].append("DECREASE_TEMPERATURE")
elif failure_case.failure_type == "TOO_LONG":
analysis_result["root_cause"] = "LLM生成内容超出了期望的长度限制。"
analysis_result["suggested_adjustments"].append("STRICT_LENGTH_CONSTRAINT")
analysis_result["suggested_adjustments"].append("DECREASE_MAX_TOKENS_PARAM")
analysis_result["suggested_adjustments"].append("ADD_CONCISE_TONE_INSTRUCTION")
# 2. (高级) 使用LLM进行Meta-Prompting分析
# 我们可以构建一个Prompt，让LLM自己分析为什么另一个LLM会失败
try:
meta_prompt = f"""
你是一个Prompt工程专家。我提供一个失败的LLM任务案例。
请分析原始Prompt、LLM的响应、预期输出以及失败类型，然后给出具体的Prompt调整建议。
--- 失败案例 ---
任务ID: {failure_case.task_id}
失败类型: {failure_case.failure_type}
原始Prompt: {failure_case.original_prompt}
```
LLM响应:
```
{failure_case.llm_response}
```
预期输出 (如果存在):
```
{failure_case.expected_output if failure_case.expected_output else 'N/A'}
```
失败详情: {failure_case.analysis}
--- 分析与建议 ---
请分析以下问题：
1. 失败的根本原因可能是什么？
2. 针对这个失败，可以采取哪些具体的Prompt调整策略？请列出3-5条，例如：
- 增加或修改Persona
- 添加或修改约束条件
- 增加Few-shot示例
- 调整模型参数 (如temperature, max_tokens)
- 修改任务描述的措辞
- 增加元指令（如“请一步步思考”）
请以JSON格式返回分析结果：
{{
"root_cause_llm_analysis": "...",
"suggested_adjustments_llm": ["adjustment1", "adjustment2", ...]
}}
"""
# meta_llm_response = self.llm_client.generate_response(meta_prompt, {"temperature": 0.3, "max_tokens": 500})
# 模拟meta LLM响应
meta_llm_response = json.dumps({
"root_cause_llm_analysis": f"LLM可能未能充分理解'{failure_case.failure_type}'失败的语境，导致输出偏离。",
"suggested_adjustments_llm": [
f"针对'{failure_case.failure_type}'，强化相关约束",
"在Prompt中明确指出需要避免的错误类型",
"提供更多高质量的成功示例"
]
})
meta_analysis = json.loads(meta_llm_response)
analysis_result["root_cause"] += " (LLM分析: " + meta_analysis.get("root_cause_llm_analysis", "") + ")"
analysis_result["suggested_adjustments"].extend(meta_analysis.get("suggested_adjustments_llm", []))
except Exception as e:
print(f"Meta-Prompting分析失败: {e}")
# 去重和清理建议
analysis_result["suggested_adjustments"] = list(set(analysis_result["suggested_adjustments"]))
return analysis_result
#### 3.7 策略学习与调整模块 (Strategy Learning & Adjustment Module)
这是Agent的核心智能所在，负责根据失败分析结果，更新并优化Agent的Prompt策略。
```python
class StrategyLearner:
"""根据失败分析结果，更新Prompt策略"""
def __init__(self, experience_db: ExperienceDB):
self.experience_db = experience_db
def adjust_strategy(self, task_type: str, failure_analysis: Dict[str, Any], current_strategy: PromptStrategy) -> PromptStrategy:
new_strategy = current_strategy.from_dict(current_strategy.to_dict()) # 创建副本进行修改
suggested_adjustments = failure_analysis["suggested_adjustments"]
print(f"--- 正在调整策略 '{new_strategy.name}' ---")
print(f"原始策略参数: Temperature={new_strategy.model_params['temperature']}, Max_Tokens={new_strategy.model_params['max_tokens']}")
# 1. 应用具体的调整建议
for adjustment in suggested_adjustments:
if adjustment == "CLARIFY_TASK_SCOPE":
if "明确任务范围" not in new_strategy.constraints:
new_strategy.constraints.append("请严格限定在任务描述的范围内，不要发散。")
elif adjustment == "ADD_NEGATIVE_CONSTRAINTS":
if "避免无关信息" not in new_strategy.constraints:
new_strategy.constraints.append("输出中不应包含任何与主题无关的背景信息。")
elif adjustment == "DECREASE_TEMPERATURE":
# 降低temperature以减少随机性，使其更聚焦
new_strategy.model_params["temperature"] = max(0.1, new_strategy.model_params.get("temperature", 0.7) - 0.1)
elif adjustment == "ENSURE_COMPLETENESS_CONSTRAINT":
if "确保所有关键信息完整" not in new_strategy.constraints:
new_strategy.constraints.append("请确保输出涵盖所有关键信息，不要遗漏。")
elif adjustment == "INCREASE_MAX_TOKENS":
new_strategy.model_params["max_tokens"] = min(4000, new_strategy.model_params.get("max_tokens", 500) + 200)
elif adjustment == "ADD_MORE_DETAILS_REQUEST":
if "请详细阐述" not in new_strategy.meta_instructions:
new_strategy.meta_instructions.append("请尽可能详细地阐述所有相关细节。")
elif adjustment == "STRICT_FORMAT_CONSTRAINT":
if "严格遵循指定格式" not in new_strategy.constraints:
new_strategy.constraints.append("请严格遵循指定的输出格式，不得有任何偏差。")
elif adjustment == "ADD_FORMAT_EXAMPLE":
# 这是一个复杂的操作，可能需要从历史成功案例中提取或根据规则生成
# 简化处理：暂时只添加一个通用提示
if "请参考以下格式示例" not in new_strategy.meta_instructions:
new_strategy.meta_instructions.append("请参考以下格式示例：[具体的格式示例]") # 实际需要动态生成或查找
elif adjustment == "CHANGE_PERSONA_TO_STRICT_ENGINEER":
new_strategy.persona = "你是一个严谨且注重细节的工程师AI助手。"
elif adjustment == "PROVIDE_MORE_CONTEXT":
# 这通常意味着原始输入数据可能不足，但Prompt策略可以加入引导LLM要求更多上下文的指令
if "如果信息不足，请明确指出" not in new_strategy.meta_instructions:
new_strategy.meta_instructions.append("如果当前输入不足以完成任务，请明确指出还缺少哪些信息。")
elif adjustment == "ADD_FEW_SHOT_EXAMPLES":
# 动态添加few-shot示例，这需要从经验库中查找相关成功案例
if len(new_strategy.examples) < 3: # 限制few-shot数量
# 简化：假设我们有一些预设的成功示例可以加入
# 实际需要从experience_db中根据task_type和成功案例动态选择
if task_type == "summarization" and {"input": "地球是蓝色的。", "output": "地球是一个蓝色的星球。"} not in new_strategy.examples:
new_strategy.examples.append({"input": "地球是蓝色的。", "output": "地球是一个蓝色的星球。"})
elif adjustment == "ADJUST_PERSONA_FOR_ACCURACY":
new_strategy.persona = "你是一个专注于准确性和事实核查的AI助手。"
elif adjustment == "STRICT_LENGTH_CONSTRAINT":
if "请严格控制输出长度" not in new_strategy.constraints:
new_strategy.constraints.append("请严格控制输出长度，力求简洁。")
elif adjustment == "DECREASE_MAX_TOKENS_PARAM":
new_strategy.model_params["max_tokens"] = max(50, new_strategy.model_params.get("max_tokens", 500) - 100)
elif adjustment == "ADD_CONCISE_TONE_INSTRUCTION":
new_strategy.tone = "concise"
# LLM生成的更具体的调整建议
elif adjustment.startswith("针对") and "强化相关约束" in adjustment:
if "针对当前任务的特定约束已强化" not in new_strategy.constraints:
new_strategy.constraints.append("针对当前任务的特定约束已强化。") # 这是一个通用标记
elif adjustment.startswith("在Prompt中明确指出需要避免的错误类型"):
if "避免出现与历史失败案例相似的错误" not in new_strategy.meta_instructions:
new_strategy.meta_instructions.append("请务必避免出现与历史失败案例相似的错误。")
elif adjustment.startswith("提供更多高质量的成功示例"):
if len(new_strategy.examples) < 5: # 再次尝试添加更多示例
if task_type == "summarization" and {"input": "太阳是恒星。", "output": "太阳是一颗恒星。"} not in new_strategy.examples:
new_strategy.examples.append({"input": "太阳是恒星。", "output": "太阳是一颗恒星。"})
print(f"调整后的策略参数: Temperature={new_strategy.model_params['temperature']}, Max_Tokens={new_strategy.model_params['max_tokens']}")
return new_strategy
3.8 自我学习Agent的整体协调器 (Self-Learning Agent Orchestrator)
最后，我们将所有模块整合起来，构建Agent的主循环和协调逻辑。
class SelfLearningAgent:
def __init__(self, api_key: str, llm_model_name: str = "gpt-4"):
self.experience_db = ExperienceDB()
self.llm_client = LLMClient(api_key, llm_model_name)
self.prompt_generator = PromptGenerator(self.experience_db)
self.semantic_detector = SemanticSimilarityDetector() # 实例化语义检测器
self.evaluator = ResultEvaluator(self.semantic_detector)
self.analyzer = FailureAnalyzer(self.llm_client) # analyzer也可以使用LLMClient进行meta-prompting
self.learner = StrategyLearner(self.experience_db)
self.task_scheduler = TaskScheduler()
# 初始化默认策略
self.experience_db.update_strategy("default", PromptStrategy(name="default"))
# 针对不同任务类型，可以有不同的初始策略
self.experience_db.update_strategy("summarization", PromptStrategy(name="summarization_strategy", persona="你是一个专业的文章总结者。"))
self.experience_db.update_strategy("json_extraction", PromptStrategy(name="json_extraction_strategy", persona="你是一个数据提取专家。", constraints=["输出必须是有效的JSON格式。"]))
def run_task_with_learning(self, task: Task, strategy_name: str = "default", attempt: int = 1) -> str:
current_strategy = self.experience_db.get_strategy(strategy_name)
if not current_strategy:
print(f"未找到策略 '{strategy_name}'，使用默认策略。")
current_strategy = self.experience_db.get_strategy("default")
print(f"n--- 执行任务 {task.task_id} (尝试 {attempt}) ---")
self.task_scheduler.update_task_status(task.task_id, "PROCESSING")
# 1. 生成Prompt
prompt, model_params = self.prompt_generator.generate_prompt(task, current_strategy.name)
# 2. 执行LLM并获取响应
llm_response = self.llm_client.generate_response(prompt, model_params)
# 3. 评估结果
evaluation_result = self.evaluator.evaluate_response(task, llm_response, prompt)
if evaluation_result["is_success"]:
print(f"任务 {task.task_id} 成功完成！")
self.task_scheduler.update_task_status(task.task_id, "COMPLETED")
# 记录成功案例 (可选，但在更复杂的RL场景下有用)
# 也可以更新策略，比如增加该策略的权重
return llm_response
else:
print(f"任务 {task.task_id} 失败：{evaluation_result['failure_type']}")
# 4. 记录失败案例
failure_case = FailureCase(
case_id=None,
task_id=task.task_id,
timestamp=datetime.now(),
original_prompt=prompt,
llm_response=llm_response,
expected_output=task.expected_output,
failure_type=evaluation_result["failure_type"],
analysis=evaluation_result["details"].get("reason", "未知原因"),
suggested_adjustments=[] # 初始为空，稍后分析填充
)
# 5. 分析失败并提出调整建议
failure_analysis = self.analyzer.analyze_failure(failure_case, current_strategy)
failure_case.analysis = failure_analysis["root_cause"]
failure_case.suggested_adjustments = failure_analysis["suggested_adjustments"]
self.experience_db.add_failure_case(failure_case)
# 6. 调整策略
new_strategy = self.learner.adjust_strategy(task.task_type, failure_analysis, current_strategy)
self.experience_db.update_strategy(strategy_name, new_strategy) # 更新当前策略
# 7. 尝试使用新策略重试 (或在下一轮任务中应用)
if attempt < 3: # 最多重试3次
print(f"正在使用调整后的策略重试任务 {task.task_id}...")
return self.run_task_with_learning(task, strategy_name, attempt + 1)
else:
print(f"任务 {task.task_id} 经过多次尝试仍失败。将任务标记为失败。")
self.task_scheduler.update_task_status(task.task_id, "FAILED")
return f"最终失败：{evaluation_result['failure_type']}"
def get_current_strategy(self, strategy_name: str = "default") -> PromptStrategy:
return self.experience_db.get_strategy(strategy_name)
# --- 运行示例 ---
# if __name__ == "__main__":
# # 实例化Agent
# agent = SelfLearningAgent(api_key="sk-YOUR_OPENAI_API_KEY") # 实际使用时请替换API Key
# # 定义一个总结任务，期望输出是简短的
# task1_input = {"text": "人工智能是计算机科学的一个分支，它试图让机器像人类一样思考和学习。近年来，随着深度学习和大数据技术的发展，人工智能取得了突破性进展，在图像识别、自然语言处理、自动驾驶等领域展现出巨大潜力。未来，人工智能有望在医疗、教育、金融等行业带来深刻变革，但也面临伦理、就业等挑战。", "length": "short"}
# task1 = agent.task_scheduler.add_task(task_type="summarization", input_data=task1_input, expected_output="人工智能在深度学习和大数据驱动下取得突破，未来将在多领域带来变革，但也面临挑战。")
# # 运行任务，观察Agent如何学习和调整
# print("n--- 第一次运行任务 ---")
# final_response1 = agent.run_task_with_learning(task1, strategy_name="summarization_strategy")
# print(f"n任务 {task1.task_id} 最终响应: {final_response1}")
# print("n--- 观察调整后的策略 ---")
# current_summarization_strategy = agent.get_current_strategy("summarization_strategy")
# print(f"总结策略当前的约束: {current_summarization_strategy.constraints}")
# print(f"总结策略当前的温度: {current_summarization_strategy.model_params['temperature']}")
# print(f"总结策略当前的最大tokens: {current_summarization_strategy.model_params['max_tokens']}")
# # 定义一个需要JSON格式输出的任务
# task2_input = {"text": "请提取以下文本中的人名和公司名：'张三在腾讯工作，李四在阿里巴巴。'"}
# task2 = agent.task_scheduler.add_task(task_type="json_extraction", input_data=task2_input, expected_output='{"persons": ["张三", "李四"], "companies": ["腾讯", "阿里巴巴"]}')
# print("n--- 第二次运行任务 (JSON提取) ---")
# final_response2 = agent.run_task_with_learning(task2, strategy_name="json_extraction_strategy")
# print(f"n任务 {task2.task_id} 最终响应: {final_response2}")
# print("n--- 再次观察调整后的策略 (JSON提取) ---")
# current_json_strategy = agent.get_current_strategy("json_extraction_strategy")
# print(f"JSON提取策略当前的约束: {current_json_strategy.constraints}")
# print(f"JSON提取策略当前的Persona: {current_json_strategy.persona}")运行示例的预期行为：
- 第一次运行
task1(总结任务)：- 初始策略可能导致LLM响应不符合“长度短”或“关键信息完整”的预期，被
ResultEvaluator标记为TOO_LONG或INCOMPLETE。 FailureAnalyzer分析原因，并建议如“增加长度约束”、“增加max_tokens”或“降低temperature”。StrategyLearner根据建议调整summarization_strategy，例如降低temperature、增加max_tokens（如果之前太小）或添加新的长度约束。- Agent使用新策略重试。如果重试成功，则任务完成。如果重试仍失败，则再次调整策略，直到达到重试上限。
- 初始策略可能导致LLM响应不符合“长度短”或“关键信息完整”的预期，被
- 第二次运行
task2(JSON提取任务)：- 初始策略可能导致LLM输出的JSON格式不正确，被
ResultEvaluator标记为FORMAT_ERROR。 FailureAnalyzer分析原因，并建议如“增加严格格式约束”、“改变Persona为严谨工程师”、“添加格式示例”。StrategyLearner调整json_extraction_strategy，可能将persona改为“严谨且注重细节的工程师AI助手”，并添加更明确的格式约束。- Agent使用新策略重试，有望获得正确的JSON输出。
- 初始策略可能导致LLM输出的JSON格式不正确，被
通过这样的循环，Agent能够根据实际交互的反馈，不断迭代和优化其Prompt策略，从而提升其在各类任务中的表现。
四、挑战与考量
设计一个自我学习的Prompt Agent，虽然前景广阔，但同样面临诸多挑战：
- 失败定义的精度与量化：对于开放性任务，如何精确定义“成功”与“失败”？人工评估标准（如BLEU、ROUGE）往往不能完全捕捉语义的细微差别。自动化评估器自身的准确性也需要保证，否则可能导致错误的学习方向。
- 失败归因的复杂性：LLM的失败可能由多种因素引起：Prompt本身、LLM模型自身的局限性、输入数据质量、甚至模型推理时的随机性。准确归因是调整策略的前提，但往往是“牵一发而动全身”的难题。Meta-Prompting虽然有帮助，但其自身也依赖于LLM的理解能力。
- 策略调整的粒度与泛化能力：调整Prompt策略时，是应该针对特定任务进行微调，还是寻求更通用的调整？过度细致的调整可能导致过拟合，使Agent在新任务上表现不佳；而过于粗糙的调整则可能无法解决具体问题。如何平衡“探索”与“利用”是一个经典的强化学习问题。
- 计算资源与成本：每一次失败的检测、分析、策略调整，都可能涉及额外的LLM调用，这会显著增加计算成本和延迟。如何设计高效的学习循环，尤其是在资源受限的环境下，是一个实际问题。
- 知识表示与更新：如何有效地存储和检索历史经验？如何将失败案例转化为可操作的、可复用的Prompt调整规则或参数？随着经验的累积，知识库的维护和冲突解决也将成为挑战。
- 人类在环 (Human-in-the-Loop)：在Agent早期学习阶段，人类的监督和纠正至关重要。Agent何时需要寻求人类帮助？人类如何有效地提供反馈？这需要良好的人机交互界面和流程。
- 迭代速度与收敛性：Agent需要多少次失败才能学会一个有效的策略？学习过程是否稳定？是否会陷入局部最优？这些都是衡量Agent效能的重要指标。
五、未来展望
展望未来，自我学习的Prompt Agent将沿着以下几个方向持续演进：
- 更智能的失败归因：结合因果推理、可解释AI技术，更精确地定位失败的根本原因，区分是Prompt问题、模型知识问题还是推理路径问题。
- 强化学习与元学习：将Prompt策略的调整视为一个强化学习问题，Agent通过与LLM交互获得奖励/惩罚信号，并通过Q-learning、PPO等算法优化策略。进一步，引入元学习（Meta-Learning），使Agent能够学会“如何学习”，从而加速在新任务上的适应能力。
- 多模态与多Agent协作：将Prompt策略扩展到多模态场景（如图像生成、视频理解），并引入多Agent协作，一个Agent负责生成Prompt，另一个Agent负责评估，甚至有Agent专门负责反思和学习。
- 知识图谱与外部工具集成：将Agent的经验库与结构化的知识图谱结合，利用外部工具（如搜索引擎、代码解释器、API调用）来增强LLM的能力，减少其“幻觉”和知识限制，从而使Prompt策略的调整更加精准和有效。
- 自动化A/B测试与演化算法：自动生成多种Prompt变体进行A/B测试，并采用演化算法（如遗传算法）来“进化”最佳Prompt策略，实现更高效、更具创造性的Prompt优化。
通过持续的技术创新和实践，我们有理由相信，具备自我学习能力的Prompt Agent将成为AI应用领域的重要基石，极大地提升我们驾驭大型语言模型的能力，并解锁更多前所未有的智能应用场景。