各位同仁，各位技术爱好者，大家好！
今天，我们齐聚一堂，探讨一个令人兴奋且极具挑战性的话题：如何构建一个具备“自我学习能力”的Agent。更具体地说，这个Agent能够根据其日常的工作日志，自动分析、识别并优化自身的Prompt模板。
在当今这个由大型语言模型（LLM）驱动的时代，Prompt工程的重要性不言而喻。一个优秀的Prompt能够让LLM的性能事半功倍，而一个模糊或低效的Prompt则可能导致结果南辕北辙。然而，人工优化Prompt模板不仅耗时耗力，而且往往受限于人类的经验和认知偏差。我们是否能让AI自己来做这件事？答案是肯定的，这就是我们今天要深入探讨的核心。
想象一下，一个Agent在执行任务时，不仅完成了工作，还默默记录下每次任务的输入、它使用的Prompt、LLM的输出，以及最重要的——这项任务的成功与否，甚至人类对结果的反馈。日复一日，这些数据积累成了一份详尽的“工作日志”。我们的目标，就是赋予Agent解析这份日志的能力，从中发现规律，理解哪些Prompt结构或措辞导致了成功，哪些导致了失败，并最终，利用这些洞察力，生成或修改出更高效、更精准的Prompt模板。这是一个从“经验”中学习，并不断“进化”的过程。
今天，我将带领大家一步步解构这个复杂系统，从核心组件的设计，到数据流的构建，再到关键算法的实现。我们将用代码说话，力求逻辑严谨，让大家对如何从零开始构建这样一个具备“自我学习能力”的Agent有一个清晰而深入的理解。
第一章：自我学习Agent的架构蓝图
要构建一个具备自我学习能力的Agent，我们首先需要一个清晰的架构。这个Agent不是一个单一的程序，而是一个由多个协同工作的模块组成的系统。其核心思想是一个闭环反馈机制：Agent执行任务 -> 记录日志 -> 分析日志 -> 优化Prompt -> 使用优化后的Prompt执行任务。
1.1 核心组件概述
我们的Agent将由以下几个关键组件构成：
- Agent核心 (AgentCore): 负责接收外部任务请求，选择合适的Prompt模板，调用LLM，并返回结果。它是Agent与外部世界交互的接口。
- Prompt模板管理器 (PromptTemplateManager): 存储、检索、管理所有的Prompt模板。它需要支持模板的版本控制，并能追踪每个模板的使用情况和历史性能。
- 工作日志记录器 (WorkLogRecorder): 负责详细记录Agent每次任务执行的所有相关信息，包括输入、使用的Prompt、LLM输出、任务结果及反馈。
- 工作日志分析器 (WorkLogAnalyzer): 从积累的工作日志中提取有价值的信息，识别当前Prompt模板的优点和缺点，生成一份“性能报告”或“改进建议”。
- Prompt优化器 (PromptOptimizer): 这是Agent的“大脑”，它接收来自工作日志分析器的报告，并利用这些信息，结合自身的“学习能力”，生成新的或改进现有的Prompt模板。
- 评估与验证模块 (Evaluation & Validation): 新生成的Prompt模板不能立即投入生产。它需要一个机制来验证其有效性，例如通过A/B测试或在沙盒环境中进行回放测试。
1.2 数据流与学习闭环
为了更好地理解这些组件如何协同工作，我们来看一下数据流图（概念性描述）：
- 任务执行: 外部系统向
AgentCore发送任务请求。 - 模板选择:
AgentCore从PromptTemplateManager获取当前最适合（或默认）的Prompt模板。 - LLM调用:
AgentCore使用选定的Prompt和任务输入调用底层LLM。 - 结果返回与日志记录: LLM返回结果，
AgentCore将结果返回给外部系统，同时通过WorkLogRecorder详细记录本次任务的所有信息。 - 定期分析:
WorkLogAnalyzer定期（例如每日、每周）从WorkLogRecorder收集日志数据，进行统计分析，生成一份包含改进点的报告。 - Prompt优化:
PromptOptimizer接收WorkLogAnalyzer的报告，并根据报告内容，利用LLM（或启发式规则）生成新的或改进的Prompt模板。 - 模板入库与评估: 新生成的Prompt模板被提交给
PromptTemplateManager进行存储，并标记为“待验证”。Evaluation & Validation模块开始对其进行测试。 - 模板激活: 一旦新模板通过验证，其性能优于旧模板，
PromptTemplateManager会将其标记为“激活”，供AgentCore在未来的任务中优先使用。旧模板则可能被降级或归档。
这个循环确保了Agent能够不断地从自身经验中学习，并迭代优化其核心工具——Prompt模板。
第二章：工作日志 – 学习的基石
一切学习都源于经验。对于我们的Agent来说，这些“经验”就是其日常任务执行的详细记录——工作日志。设计一个结构良好、信息完备的工作日志是至关重要的。
2.1 工作日志的关键信息
一个有效的工作日志条目应该包含以下核心字段：
| 字段名称 | 数据类型 | 描述 | 示例值 |
|---|---|---|---|
timestamp |
datetime | 任务执行的时间戳 | 2023-10-27T10:30:00Z |
task_id |
string | 唯一标识一个任务的ID | task_abc_12345 |
task_type |
string | 任务的类型（例如：代码生成、文本摘要、问题回答） | code_generation |
input_data |
string | 任务的原始输入数据（例如：用户需求描述、待摘要文本） | "根据以下需求生成一个Python函数：计算斐波那契数列的第n项。" |
prompt_id |
string | 使用的Prompt模板的ID | prompt_v1_code_gen |
prompt_template |
string | 实际使用的Prompt模板内容（包括变量替换后的最终Prompt） | "你是一个专业的Python编程助手。请根据以下用户需求生成一个Python函数。确保代码可读性高，并包含必要的注释。需求：{user_requirement}" (这里是替换后的完整Prompt) |
llm_response |
string | LLM返回的原始响应 | def fibonacci(n):n if n <= 0: return 0n elif n == 1: return 1n else: return fibonacci(n-1) + fibonacci(n-2) |
parsed_output |
string | 从LLM响应中提取并结构化后的输出（如果需要） | {"function_name": "fibonacci", "code": "def fibonacci(n):..."} |
success_flag |
boolean | 任务是否成功完成（根据预设规则或后续处理判断） | True |
failure_reason |
string | 如果失败，失败的原因（例如：hallucination, off_topic, incomplete_response, syntax_error） |
None 或 syntax_error |
human_feedback |
string | 如果有，人类对结果的评价或修正意见 | "代码逻辑正确，但效率不高，建议使用迭代法。" 或 None |
rating |
int | 人类对结果的评分（例如：1-5星） | 4 |
cost |
float | 本次LLM调用的成本（token数量或实际费用） | 0.0012 |
latency |
float | 本次LLM调用的延迟（秒） | 2.5 |
2.2 结构化日志记录
为了便于后续的分析，我们强烈推荐使用结构化格式，如JSON，来记录日志。这使得日志的解析和查询变得非常简单。
import json
import datetime
import uuid
import os
class WorkLogEntry:
"""
代表一个工作日志条目。
"""
def __init__(self,
task_id: str,
task_type: str,
input_data: str,
prompt_id: str,
prompt_template: str,
llm_response: str,
parsed_output: str = None,
success_flag: bool = False,
failure_reason: str = None,
human_feedback: str = None,
rating: int = None,
cost: float = None,
latency: float = None):
self.timestamp = datetime.datetime.utcnow().isoformat() + 'Z'
self.task_id = task_id
self.task_type = task_type
self.input_data = input_data
self.prompt_id = prompt_id
self.prompt_template = prompt_template
self.llm_response = llm_response
self.parsed_output = parsed_output
self.success_flag = success_flag
self.failure_reason = failure_reason
self.human_feedback = human_feedback
self.rating = rating
self.cost = cost
self.latency = latency
def to_dict(self):
return self.__dict__
@classmethod
def from_dict(cls, data: dict):
# 注意：这里需要手动处理datetime的转换，或者在存储时就保持ISO格式
# 为了简化，我们假设timestamp在load时保持string
obj = cls(
task_id=data['task_id'],
task_type=data['task_type'],
input_data=data['input_data'],
prompt_id=data['prompt_id'],
prompt_template=data['prompt_template'],
llm_response=data['llm_response'],
parsed_output=data.get('parsed_output'),
success_flag=data.get('success_flag', False),
failure_reason=data.get('failure_reason'),
human_feedback=data.get('human_feedback'),
rating=data.get('rating'),
cost=data.get('cost'),
latency=data.get('latency')
)
obj.timestamp = data['timestamp'] # 恢复原始时间戳字符串
return obj
class WorkLogRecorder:
"""
负责将工作日志条目写入文件或数据库。
这里使用文件系统作为示例。
"""
def __init__(self, log_dir: str = "work_logs"):
self.log_dir = log_dir
os.makedirs(self.log_dir, exist_ok=True)
def record_log(self, entry: WorkLogEntry):
"""
记录单个日志条目到JSON文件。
为了避免单个文件过大，可以按日期或任务类型分文件。
这里简化为每天一个文件。
"""
today_str = datetime.datetime.utcnow().strftime("%Y-%m-%d")
log_file_path = os.path.join(self.log_dir, f"log_{today_str}.jsonl") # JSON Lines format
with open(log_file_path, 'a', encoding='utf-8') as f:
f.write(json.dumps(entry.to_dict(), ensure_ascii=False) + 'n')
print(f"Log recorded for task {entry.task_id} to {log_file_path}")
def load_logs_for_date(self, date_str: str) -> list[WorkLogEntry]:
"""
加载指定日期的所有日志条目。
"""
log_file_path = os.path.join(self.log_dir, f"log_{date_str}.jsonl")
logs = []
if not os.path.exists(log_file_path):
return logs
with open(log_file_path, 'r', encoding='utf-8') as f:
for line in f:
try:
data = json.loads(line)
logs.append(WorkLogEntry.from_dict(data))
except json.JSONDecodeError as e:
print(f"Error decoding JSON from log file {log_file_path}: {e}")
return logs
def load_all_logs(self) -> list[WorkLogEntry]:
"""
加载所有可用的日志条目。
在生产环境中，这可能需要分页或指定日期范围。
"""
all_logs = []
for filename in os.listdir(self.log_dir):
if filename.startswith("log_") and filename.endswith(".jsonl"):
date_str = filename[4:-5] # Extract YYYY-MM-DD
all_logs.extend(self.load_logs_for_date(date_str))
return all_logs
# 示例使用
if __name__ == "__main__":
recorder = WorkLogRecorder()
# 模拟一个日志条目
mock_log_entry = WorkLogEntry(
task_id=str(uuid.uuid4()),
task_type="text_summarization",
input_data="LLM is great. It can do many things. This is a text to be summarized.",
prompt_id="summary_v1",
prompt_template="请总结以下文本：{text}",
llm_response="LLM功能强大，可处理多种任务。",
success_flag=True,
rating=5,
cost=0.0001,
latency=0.5
)
recorder.record_log(mock_log_entry)
mock_log_entry_fail = WorkLogEntry(
task_id=str(uuid.uuid4()),
task_type="code_generation",
input_data="Generate a Python function to sort a list, but make it very inefficient.",
prompt_id="code_gen_v2",
prompt_template="生成一个Python函数，需求：{requirement}",
llm_response="def sort_list(l): return sorted(l)",
success_flag=False,
failure_reason="efficiency_issue",
human_feedback="代码正确但未满足'非常低效'的要求",
rating=2,
cost=0.0002,
latency=0.8
)
recorder.record_log(mock_log_entry_fail)
# 加载今天的所有日志
today = datetime.datetime.utcnow().strftime("%Y-%m-%d")
loaded_logs = recorder.load_logs_for_date(today)
print(f"nLoaded {len(loaded_logs)} logs for today:")
for log in loaded_logs:
print(log.to_dict())
第三章：Prompt模板管理器 – Agent的记忆与工具箱
Prompt模板管理器是Agent的“记忆”，它存储了Agent能够使用的所有Prompt模板。它不仅要管理模板本身，还要支持版本控制和性能追踪，以便Agent核心能够选择最佳模板，同时优化器也能对其进行迭代。
3.1 模板的结构与管理
每个Prompt模板都应该被赋予一个唯一的ID，并包含详细的元数据：
| 字段名称 | 数据类型 | 描述 | 示例值 |
|---|---|---|---|
prompt_id |
string | 模板的唯一标识符（包含版本信息） | summary_v1_0 |
name |
string | 模板的友好名称 | Standard Summarization |
description |
string | 模板的用途和设计思路 | 用于通用文本摘要，要求简洁明了。 |
template_string |
string | 实际的Prompt字符串，包含占位符 | "你是一个专业的文本摘要助手。请将以下文本总结为不超过100字的简洁摘要，并确保保留核心信息。nn文本：{text}" |
task_type |
string | 此模板适用于的任务类型 | text_summarization |
version |
string | 模板的版本号（例如：1.0, 1.1, 2.0） | 1.0 |
status |
string | 模板的状态（active, inactive, testing, deprecated） |
active |
created_at |
datetime | 模板创建时间 | 2023-01-15T09:00:00Z |
last_modified |
datetime | 模板最后修改时间 | 2023-10-20T14:30:00Z |
performance_metrics |
dict | 历史性能数据（平均成功率、平均评分、平均成本等） | {"success_rate": 0.85, "avg_rating": 4.2, "avg_cost": 0.00015} (这些数据会由日志分析器更新) |
parameters |
list[str] | 模板中需要填充的参数名列表 | ["text"] |
3.2 Code Example: PromptTemplateManager Class
import json
import datetime
import os
from typing import Dict, Any, Optional, List
class PromptTemplate:
"""
Prompt模板的数据结构。
"""
def __init__(self,
prompt_id: str,
name: str,
description: str,
template_string: str,
task_type: str,
version: str,
status: str = "active",
created_at: str = None,
last_modified: str = None,
performance_metrics: Dict[str, Any] = None,
parameters: List[str] = None):
self.prompt_id = prompt_id
self.name = name
self.description = description
self.template_string = template_string
self.task_type = task_type
self.version = version
self.status = status # active, inactive, testing, deprecated
self.created_at = created_at if created_at else datetime.datetime.utcnow().isoformat() + 'Z'
self.last_modified = last_modified if last_modified else self.created_at
self.performance_metrics = performance_metrics if performance_metrics is not None else {}
self.parameters = parameters if parameters is not None else []
def to_dict(self):
return self.__dict__
@classmethod
def from_dict(cls, data: Dict[str, Any]):
return cls(**data)
def fill_template(self, **kwargs) -> str:
"""
用提供的参数填充模板字符串。
"""
try:
return self.template_string.format(**kwargs)
except KeyError as e:
raise ValueError(f"Missing parameter for prompt '{self.name}': {e}")
class PromptTemplateManager:
"""
管理Prompt模板的存储、检索和更新。
这里使用JSON文件作为持久化存储。
"""
def __init__(self, storage_path: str = "prompt_templates.json"):
self.storage_path = storage_path
self._templates: Dict[str, PromptTemplate] = {}
self._load_templates()
def _load_templates(self):
"""从文件加载所有模板。"""
if not os.path.exists(self.storage_path):
self._templates = {}
return
with open(self.storage_path, 'r', encoding='utf-8') as f:
data = json.load(f)
self._templates = {item['prompt_id']: PromptTemplate.from_dict(item) for item in data}
print(f"Loaded {len(self._templates)} prompt templates from {self.storage_path}")
def _save_templates(self):
"""将所有模板保存到文件。"""
with open(self.storage_path, 'w', encoding='utf-8') as f:
json.dump([template.to_dict() for template in self._templates.values()], f, indent=4, ensure_ascii=False)
print(f"Saved {len(self._templates)} prompt templates to {self.storage_path}")
def add_template(self, template: PromptTemplate):
"""添加一个新模板。"""
if template.prompt_id in self._templates:
raise ValueError(f"Prompt with ID '{template.prompt_id}' already exists.")
self._templates[template.prompt_id] = template
self._save_templates()
print(f"Added new prompt template: {template.prompt_id}")
def get_template(self, prompt_id: str) -> Optional[PromptTemplate]:
"""根据ID获取模板。"""
return self._templates.get(prompt_id)
def get_active_templates_by_task_type(self, task_type: str) -> List[PromptTemplate]:
"""获取某个任务类型的所有活跃模板。"""
return [t for t in self._templates.values() if t.task_type == task_type and t.status == "active"]
def update_template(self, prompt_id: str, **kwargs):
"""更新现有模板的属性。"""
template = self.get_template(prompt_id)
if not template:
raise ValueError(f"Prompt with ID '{prompt_id}' not found.")
for key, value in kwargs.items():
if hasattr(template, key):
setattr(template, key, value)
else:
print(f"Warning: Attempted to set unknown attribute '{key}' for prompt '{prompt_id}'.")
template.last_modified = datetime.datetime.utcnow().isoformat() + 'Z'
self._save_templates()
print(f"Updated prompt template: {prompt_id}")
def deactivate_template(self, prompt_id: str):
"""将模板设置为非活跃状态。"""
self.update_template(prompt_id, status="inactive")
print(f"Deactivated prompt template: {prompt_id}")
def activate_template(self, prompt_id: str):
"""将模板设置为活跃状态。"""
self.update_template(prompt_id, status="active")
print(f"Activated prompt template: {prompt_id}")
def get_all_templates(self) -> List[PromptTemplate]:
"""获取所有模板。"""
return list(self._templates.values())
# 示例使用
if __name__ == "__main__":
manager = PromptTemplateManager()
# 添加一些初始模板
template1 = PromptTemplate(
prompt_id="summary_v1_0",
name="Standard Summarization",
description="用于通用文本摘要，要求简洁明了。",
template_string="你是一个专业的文本摘要助手。请将以下文本总结为不超过100字的简洁摘要，并确保保留核心信息。nn文本：{text}",
task_type="text_summarization",
version="1.0",
parameters=["text"]
)
manager.add_template(template1)
template2 = PromptTemplate(
prompt_id="code_gen_python_v1_0",
name="Python Code Generation",
description="根据用户需求生成Python函数，注重可读性和注释。",
template_string="你是一个专业的Python编程助手。请根据以下用户需求生成一个Python函数。确保代码可读性高，并包含必要的注释。nn需求：{user_requirement}",
task_type="code_generation",
version="1.0",
parameters=["user_requirement"]
)
manager.add_template(template2)
# 获取模板
retrieved_template = manager.get_template("summary_v1_0")
if retrieved_template:
print(f"nRetrieved template: {retrieved_template.name}")
filled_prompt = retrieved_template.fill_template(text="LLM is great. It can do many things.")
print(f"Filled Prompt: {filled_prompt}")
# 更新模板状态
manager.update_template("summary_v1_0", status="testing")
updated_template = manager.get_template("summary_v1_0")
if updated_template:
print(f"Updated template status: {updated_template.status}")
# 尝试添加一个重复ID的模板
try:
manager.add_template(template1)
except ValueError as e:
print(f"nError as expected: {e}")
# 获取所有活跃的文本摘要模板
active_summary_templates = manager.get_active_templates_by_task_type("text_summarization")
print(f"nActive summarization templates: {[t.prompt_id for t in active_summary_templates]}") # 此时应为空，因为summary_v1_0是testing状态
第四章：工作日志分析器 – 发现规律的眼睛
工作日志分析器是Agent的“眼睛”，它负责从海量的日志数据中找出模式、识别问题，并量化不同Prompt模板的表现。这是自我学习过程中的关键一步，因为没有准确的分析，后续的优化将无从谈起。
4.1 分析目标与策略
日志分析器的主要目标包括：
- 性能趋势分析: 追踪不同Prompt模板随时间变化的成功率、平均评分、成本和延迟。
- 问题模式识别: 找出哪些Prompt模板更容易导致失败，以及失败的具体原因（例如，幻觉、偏离主题、响应不完整）。
- 人类反馈洞察: 分析人类反馈文本，提取常见的问题点和改进建议。
- 参数敏感度: 探索Prompt模板中不同参数（例如，字数限制、角色设定）对结果的影响。
为了实现这些目标，分析器可能需要使用以下技术：
- 统计分析: 计算平均值、中位数、标准差、成功率等。
- 文本分析: 对
human_feedback和failure_reason进行词频统计、情感分析或主题建模。 - 关联规则挖掘: 发现特定Prompt结构与成功/失败之间的关联。
4.2 Code Example: WorkLogAnalyzer Class
import pandas as pd
import datetime
from collections import defaultdict
from typing import List, Dict, Any
# 假设 WorkLogEntry 和 WorkLogRecorder 已经定义并可用
from prompt_optimization_agent.work_log import WorkLogEntry, WorkLogRecorder
from prompt_optimization_agent.prompt_manager import PromptTemplate, PromptTemplateManager
class WorkLogAnalyzer:
"""
负责分析工作日志，生成Prompt模板的性能报告。
"""
def __init__(self, log_recorder: WorkLogRecorder, prompt_manager: PromptTemplateManager):
self.log_recorder = log_recorder
self.prompt_manager = prompt_manager
def _load_logs_as_dataframe(self, start_date_str: str = None, end_date_str: str = None) -> pd.DataFrame:
"""
加载指定日期范围内的日志，并转换为Pandas DataFrame。
"""
all_logs_data = []
if start_date_str and end_date_str:
start_date = datetime.datetime.strptime(start_date_str, "%Y-%m-%d").date()
end_date = datetime.datetime.strptime(end_date_str, "%Y-%m-%d").date()
current_date = start_date
while current_date <= end_date:
logs = self.log_recorder.load_logs_for_date(current_date.strftime("%Y-%m-%d"))
all_logs_data.extend([log.to_dict() for log in logs])
current_date += datetime.timedelta(days=1)
else:
all_logs_data = [log.to_dict() for log in self.log_recorder.load_all_logs()]
if not all_logs_data:
return pd.DataFrame()
df = pd.DataFrame(all_logs_data)
# 确保时间戳是datetime对象
df['timestamp'] = pd.to_datetime(df['timestamp'])
return df
def analyze_prompt_performance(self,
start_date: str = None,
end_date: str = None) -> Dict[str, Dict[str, Any]]:
"""
分析所有Prompt模板的性能指标。
返回一个字典，键是prompt_id，值是其性能指标。
"""
df = self._load_logs_as_dataframe(start_date, end_date)
if df.empty:
print("No logs available for analysis.")
return {}
performance_report = {}
for prompt_id, group in df.groupby('prompt_id'):
total_tasks = len(group)
if total_tasks == 0:
continue
success_rate = group['success_flag'].sum() / total_tasks
avg_rating = group['rating'].mean() if 'rating' in group and not group['rating'].isnull().all() else None
avg_cost = group['cost'].mean() if 'cost' in group and not group['cost'].isnull().all() else None
avg_latency = group['latency'].mean() if 'latency' in group and not group['latency'].isnull().all() else None
# 失败原因统计
failure_reasons_counts = group[group['success_flag'] == False]['failure_reason'].value_counts().to_dict()
# 人类反馈关键词提取（简化示例，实际可能用更复杂的NLP）
human_feedback_combined = " ".join(group['human_feedback'].dropna().tolist())
common_feedback_keywords = self._extract_common_keywords(human_feedback_combined)
performance_report[prompt_id] = {
"total_tasks": total_tasks,
"success_rate": success_rate,
"avg_rating": avg_rating,
"avg_cost": avg_cost,
"avg_latency": avg_latency,
"failure_reasons": failure_reasons_counts,
"common_feedback_keywords": common_feedback_keywords
}
return performance_report
def _extract_common_keywords(self, text: str, top_n: int = 5) -> List[str]:
"""
一个简单的关键词提取函数（仅作示例，实际需要NLP库）。
"""
if not text:
return []
words = text.lower().split()
# 移除常见停用词，这里只做非常简单的过滤
stop_words = {"的", "是", "了", "和", "有", "在", "一个", "这", "不", "很", "是", "可以", "需要"}
filtered_words = [word for word in words if word.isalnum() and word not in stop_words]
word_counts = pd.Series(filtered_words).value_counts()
return word_counts.head(top_n).index.tolist()
def identify_prompts_for_optimization(self,
report: Dict[str, Dict[str, Any]],
min_tasks: int = 10,
success_rate_threshold: float = 0.7,
rating_threshold: float = 3.5) -> List[str]:
"""
根据性能报告识别需要优化或替换的Prompt模板。
"""
prompts_to_optimize = []
for prompt_id, metrics in report.items():
if metrics["total_tasks"] < min_tasks:
continue # 数据不足，暂不优化
# 找出成功率低于阈值或平均评分低于阈值的模板
if metrics["success_rate"] < success_rate_threshold:
prompts_to_optimize.append(prompt_id)
print(f"Prompt '{prompt_id}' identified for optimization: Low success rate ({metrics['success_rate']:.2f})")
elif metrics["avg_rating"] is not None and metrics["avg_rating"] < rating_threshold:
prompts_to_optimize.append(prompt_id)
print(f"Prompt '{prompt_id}' identified for optimization: Low average rating ({metrics['avg_rating']:.1f})")
# 还可以根据常见的失败原因来识别
if "hallucination" in metrics["failure_reasons"] and metrics["failure_reasons"]["hallucination"] > 0.05 * metrics["total_tasks"]:
if prompt_id not in prompts_to_optimize:
prompts_to_optimize.append(prompt_id)
print(f"Prompt '{prompt_id}' identified for optimization: Frequent 'hallucination' failures.")
return list(set(prompts_to_optimize)) # 去重
# 示例使用
if __name__ == "__main__":
# 初始化管理器和记录器（假设已有一些数据）
recorder = WorkLogRecorder()
manager = PromptTemplateManager()
# 确保有一些日志数据用于分析
# 模拟更多日志数据
for i in range(15):
task_id_succ = str(uuid.uuid4())
task_id_fail = str(uuid.uuid4())
# 成功案例
recorder.record_log(WorkLogEntry(
task_id=task_id_succ,
task_type="text_summarization",
input_data=f"This is a text for summary {i}. It talks about AI.",
prompt_id="summary_v1_0", # 使用之前定义的模板
prompt_template=manager.get_template("summary_v1_0").template_string.format(text="..."),
llm_response=f"AI summary {i}.",
success_flag=True,
rating=5 if i % 3 != 0 else 4, # 偶尔给4星
cost=0.0001,
latency=0.5
))
# 失败案例，使用另一个Prompt ID
recorder.record_log(WorkLogEntry(
task_id=task_id_fail,
task_type="code_generation",
input_data=f"Generate a complex SQL query for database {i}.",
prompt_id="code_gen_python_v1_0", # 使用之前定义的模板，但用于SQL任务（模拟不匹配）
prompt_template=manager.get_template("code_gen_python_v1_0").template_string.format(user_requirement="..."),
llm_response=f"Python code for SQL query {i}.", # LLM可能生成Python代码而不是SQL
success_flag=False,
failure_reason="off_topic",
human_feedback="生成的不是SQL，而是Python代码",
rating=1,
cost=0.0002,
latency=0.8
))
# 模拟一个新模板，但表现不佳
template_bad = PromptTemplate(
prompt_id="summary_v2_0_bad",
name="Bad Summarization Attempt",
description="一个尝试改进摘要但效果不佳的模板。",
template_string="请用不超过5个词总结以下文本：{text}", # 太过于限制，可能导致失败
task_type="text_summarization",
version="2.0",
parameters=["text"]
)
manager.add_template(template_bad)
for i in range(10):
recorder.record_log(WorkLogEntry(
task_id=str(uuid.uuid4()),
task_type="text_summarization",
input_data=f"This is another text for summary {i}. It's quite long and complex.",
prompt_id="summary_v2_0_bad",
prompt_template=template_bad.template_string.format(text="..."),
llm_response=f"Short summary {i}.",
success_flag=False if i % 2 == 0 else True, # 50%成功率
failure_reason="incomplete_response" if i % 2 == 0 else None,
human_feedback="总结不完整，无法捕捉核心信息" if i % 2 == 0 else None,
rating=2 if i % 2 == 0 else 4,
cost=0.00015,
latency=0.6
))
analyzer = WorkLogAnalyzer(recorder, manager)
# 分析今天的日志
today_str = datetime.datetime.utcnow().strftime("%Y-%m-%d")
performance_report = analyzer.analyze_prompt_performance(start_date=today_str, end_date=today_str)
print("n--- Performance Report ---")
for prompt_id, metrics in performance_report.items():
print(f"nPrompt ID: {prompt_id}")
for key, value in metrics.items():
print(f" {key}: {value}")
# 识别需要优化的Prompt
prompts_for_optimization = analyzer.identify_prompts_for_optimization(performance_report)
print(f"n--- Prompts identified for optimization: {prompts_for_optimization} ---")
第五章：Prompt优化器 – 创造与进化的核心
Prompt优化器是Agent的“创造性大脑”。它接收日志分析器生成的性能报告，并利用这些洞察力来生成新的、更优的Prompt模板。这个模块是整个自学习机制的核心，因为它负责将“经验”转化为“智慧”。
5.1 优化策略：元Prompting
最直接且有效的方法是利用LLM本身来优化Prompt。我们称之为“元Prompting”（Meta-Prompting）。即，我们构建一个高级别的Prompt（元Prompt），将其发送给一个强大的LLM（可以是与Agent核心使用的LLM相同的模型，也可以是更强大的一个），并在这个元Prompt中描述：
- 当前Prompt模板的问题: 基于
WorkLogAnalyzer的报告，明确指出旧Prompt的缺点（例如，成功率低、常出现幻觉、用户反馈不佳等）。 - 旧Prompt的原文: 提供需要改进的Prompt模板的原始字符串。
- 期望的改进方向: 明确希望新Prompt达到的目标（例如，提高成功率、减少幻觉、更清晰地引导输出格式）。
- 新Prompt的要求: 明确新Prompt的结构、长度、包含的占位符等技术性要求。
- 任务类型和目标: 提醒LLM新Prompt将用于何种任务。
LLM接收到这个元Prompt后，将扮演“Prompt工程师”的角色，生成一个优化后的Prompt模板。
5.2 优化流程
- 接收分析报告:
PromptOptimizer接收WorkLogAnalyzer提供的performance_report和prompts_to_optimize列表。 - 选择待优化模板: 对于每一个被识别为需要优化的Prompt ID，获取其详细信息和性能指标。
- 构建元Prompt: 根据待优化模板的性能问题和任务类型，动态构建一个专门的元Prompt。
- 调用LLM生成新Prompt: 将元Prompt发送给LLM，请求其生成一个改进的Prompt模板。
- 解析LLM响应: LLM的响应可能包含新Prompt的建议、理由等。需要从中提取出纯粹的新Prompt模板字符串。
- 创建新模板对象: 将新生成的Prompt模板封装成
PromptTemplate对象，赋予新的版本号和“testing”状态。 - 提交给管理器: 将新模板提交给
PromptTemplateManager进行存储和后续评估。
5.3 Code Example: PromptOptimizer Class
为了简化，这里我们不直接调用真实的LLM API，而是用一个模拟函数 mock_llm_call 来替代。在实际应用中，您会替换为OpenAI、Anthropic等LLM的API调用。
import datetime
import re
from typing import Dict, Any, List
# 假设 WorkLogAnalyzer, PromptTemplate, PromptTemplateManager 已经定义并可用
from prompt_optimization_agent.work_log import WorkLogEntry, WorkLogRecorder
from prompt_optimization_agent.prompt_manager import PromptTemplate, PromptTemplateManager
from prompt_optimization_agent.work_log_analyzer import WorkLogAnalyzer
class MockLLM:
"""
模拟LLM的响应，用于测试PromptOptimizer。
在实际场景中，这里会集成真实的LLM API。
"""
def generate_response(self, prompt_text: str) -> str:
print(f"n--- Mock LLM Call ---")
print(f"Input Meta-Prompt:n{prompt_text[:500]}...") # 打印部分输入
# 简单模拟LLM根据元Prompt生成新Prompt
if "改进以下文本摘要的Prompt" in prompt_text:
if "减少不完整响应" in prompt_text:
new_prompt = (
"你是一个专业的文本摘要助手。请将以下文本总结为不超过150字的简洁摘要，"
"确保全面涵盖核心信息，并避免遗漏关键点。如果文本过长，请分点总结。nn文本：{text}"
)
elif "提高成功率" in prompt_text:
new_prompt = (
"你是一个专业的文本摘要助手。请提供一个精炼的摘要，总结以下文本的核心内容。字数限制在120字以内。"
"确保摘要清晰、准确，且不包含无关信息。nn文本：{text}"
)
else:
new_prompt = (
"你是一个专业的文本摘要助手。请将以下文本总结为不超过100字的简洁摘要，"
"确保保留核心信息，并使用清晰、专业的语言。nn文本：{text}"
)
return f"好的，这是我建议的新Prompt：n```promptn{new_prompt}n```n我认为这个Prompt更清晰。"
elif "改进以下代码生成的Prompt" in prompt_text:
if "解决生成错误的编程语言" in prompt_text:
new_prompt = (
"你是一个专业的{language}编程助手。请根据以下用户需求生成一个{language}函数。"
"确保代码可读性高，包含必要的注释，并严格遵循{language}的语法和最佳实践。nn需求：{user_requirement}"
)
else:
new_prompt = (
"你是一个专业的编程助手。请根据以下用户需求生成一个高质量的代码片段。确保代码功能正确，"
"可读性强，并附带简要说明和示例用法。nn需求：{user_requirement}"
)
return f"这是新Prompt：n```promptn{new_prompt}n```"
return "我无法生成新Prompt。请提供更详细的指导。"
class PromptOptimizer:
"""
负责根据日志分析结果，生成和管理新的Prompt模板。
"""
def __init__(self, prompt_manager: PromptTemplateManager, llm: MockLLM):
self.prompt_manager = prompt_manager
self.llm = llm
def _generate_meta_prompt(self,
original_prompt: PromptTemplate,
performance_metrics: Dict[str, Any],
improvement_suggestions: List[str]) -> str:
"""
根据原始Prompt和性能指标，构建一个元Prompt。
"""
meta_prompt_parts = [
f"你是一位经验丰富的Prompt工程师，你的任务是根据给定的性能数据和改进建议，优化现有的Prompt模板。",
f"目标任务类型：{original_prompt.task_type}",
f"原始Prompt ID：{original_prompt.prompt_id}",
f"原始Prompt名称：{original_prompt.name}",
f"原始Prompt描述：{original_prompt.description}",
f"原始Prompt模板：n```original_promptn{original_prompt.template_string}n```",
f"n以下是该Prompt模板的历史性能数据：",
f"- 总任务数：{performance_metrics.get('total_tasks', 0)}",
f"- 成功率：{performance_metrics.get('success_rate', 0.0):.2f}",
f"- 平均评分：{performance_metrics.get('avg_rating', 'N/A')}",
f"- 常见失败原因：{json.dumps(performance_metrics.get('failure_reasons', {}), ensure_ascii=False)}",
f"- 常见用户反馈关键词：{', '.join(performance_metrics.get('common_feedback_keywords', []))}",
f"n根据以上数据，我们发现该Prompt存在以下问题和改进建议：",
]
for suggestion in improvement_suggestions:
meta_prompt_parts.append(f"- {suggestion}")
meta_prompt_parts.append("n请你基于这些信息，设计一个全新的、更优化的Prompt模板。")
meta_prompt_parts.append("新Prompt应包含与旧Prompt相同的参数占位符，且输出格式应与旧Prompt保持一致。")
meta_prompt_parts.append("请将新Prompt模板放在一个```prompt\n...\n```的代码块中。")
meta_prompt_parts.append("例如：n```promptn新的Prompt模板内容{param1}n```")
return "n".join(meta_prompt_parts)
def optimize_prompt(self, prompt_id: str, performance_report: Dict[str, Any]) -> Optional[PromptTemplate]:
"""
根据分析报告优化指定的Prompt模板。
"""
original_template = self.prompt_manager.get_template(prompt_id)
if not original_template:
print(f"Error: Prompt template '{prompt_id}' not found for optimization.")
return None
metrics = performance_report.get(prompt_id)
if not metrics:
print(f"Error: No performance metrics found for prompt '{prompt_id}'. Cannot optimize.")
return None
# 根据metrics生成改进建议
suggestions = []
if metrics["success_rate"] < 0.8:
suggestions.append(f"成功率较低 ({metrics['success_rate']:.2f})，需要提高LLM的成功率。")
if metrics["avg_rating"] is not None and metrics["avg_rating"] < 4.0:
suggestions.append(f"平均评分较低 ({metrics['avg_rating']:.1f})，需要改进用户满意度。")
if "hallucination" in metrics["failure_reasons"]:
suggestions.append(f"频繁出现'幻觉'问题 ({metrics['failure_reasons']['hallucination']}次)，需要减少LLM的虚构内容。")
if "off_topic" in metrics["failure_reasons"]:
suggestions.append(f"频繁出现'偏离主题'问题 ({metrics['failure_reasons']['off_topic']}次)，需要更明确地限定LLM的响应范围。")
if "incomplete_response" in metrics["failure_reasons"]:
suggestions.append(f"频繁出现'不完整响应'问题 ({metrics['failure_reasons']['incomplete_response']}次)，需要确保LLM提供完整且详尽的输出。")
if metrics["common_feedback_keywords"]:
suggestions.append(f"用户反馈中常见的关键词有：{', '.join(metrics['common_feedback_keywords'])}。请考虑这些反馈。")
if not suggestions:
print(f"Prompt '{prompt_id}' seems to be performing well or lacks specific issues for optimization.")
return None
meta_prompt = self._generate_meta_prompt(original_template, metrics, suggestions)
llm_response = self.llm.generate_response(meta_prompt)
# 从LLM响应中提取新Prompt模板
match = re.search(r"```promptn(.*?)```", llm_response, re.DOTALL)
if not match:
print(f"Error: Could not extract new prompt from LLM response for '{prompt_id}'.")
return None
new_template_string = match.group(1).strip()
# 提取新模板的参数
# 简单的正则匹配，可能需要更健壮的解析
new_params = re.findall(r"{(w+)}", new_template_string)
# 生成新版本ID
current_version_parts = original_template.version.split('.')
new_version_major = int(current_version_parts[0])
new_version_minor = int(current_version_parts[1]) + 1
new_prompt_version = f"{new_version_major}.{new_version_minor}"
new_prompt_id = f"{original_template.name.replace(' ', '_').lower()}_v{new_prompt_version}"
new_prompt = PromptTemplate(
prompt_id=new_prompt_id,
name=f"{original_template.name} v{new_prompt_version}",
description=f"优化自 {original_template.prompt_id}。改进点：{'; '.join(suggestions)}",
template_string=new_template_string,
task_type=original_template.task_type,
version=new_prompt_version,
status="testing", # 新生成的模板需要先进行测试
parameters=list(set(new_params)) # 去重
)
self.prompt_manager.add_template(new_prompt)
print(f"Successfully generated and added new prompt '{new_prompt.prompt_id}' (status: testing).")
return new_prompt
# 示例使用
if __name__ == "__main__":
recorder = WorkLogRecorder()
manager = PromptTemplateManager()
analyzer = WorkLogAnalyzer(recorder, manager)
mock_llm = MockLLM()
optimizer = PromptOptimizer(manager, mock_llm)
# 确保管理器中有需要优化的模板
# 假设 'summary_v2_0_bad' 被识别为需要优化
# 我们需要先运行 WorkLogAnalyzer 来获取 performance_report
today_str = datetime.datetime.utcnow().strftime("%Y-%m-%d")
performance_report = analyzer.analyze_prompt_performance(start_date=today_str, end_date=today_str)
prompts_to_optimize = analyzer.identify_prompts_for_optimization(performance_report)
for prompt_id in prompts_to_optimize:
print(f"nAttempting to optimize prompt: {prompt_id}")
optimized_prompt = optimizer.optimize_prompt(prompt_id, performance_report)
if optimized_prompt:
print(f"Optimization successful. New prompt created: {optimized_prompt.prompt_id}")
print(f"New prompt string:n{optimized_prompt.template_string}")
else:
print(f"Optimization failed for prompt: {prompt_id}")
# 检查管理器中是否有了新模板
print("n--- All Templates after Optimization ---")
for template in manager.get_all_templates():
print(f"ID: {template.prompt_id}, Name: {template.name}, Status: {template.status}")
第六章：Agent核心与自学习循环的整合
现在，我们已经构建了所有核心组件。是时候将它们整合到 AgentCore 中，并展示如何驱动整个自学习循环。
6.1 AgentCore – 任务执行与日志记录
AgentCore 是Agent的入口点。它负责：
- 接收外部任务。
- 根据任务类型和当前活跃的Prompt模板选择策略，从
PromptTemplateManager获取模板。 - 填充模板并调用LLM。
- 处理LLM响应。
- 将任务执行的详细信息通过
WorkLogRecorder记录下来。
6.2 自学习调度器 – 驱动进化
自学习过程需要定期触发。一个简单的调度器可以每天或每周运行一次，执行以下步骤：
- 从
WorkLogRecorder加载最近的日志。 WorkLogAnalyzer对日志进行分析，生成性能报告并识别需要优化的Prompt。PromptOptimizer对识别出的Prompt进行优化，生成新的testing状态的Prompt模板。Evaluation & Validation模块（这里我们简化为手动或简单的模拟）验证这些testing模板。- 如果
testing模板表现优异，PromptTemplateManager将其状态更新为active，并可选地将旧模板降级。
6.3 Code Example: AgentCore and Main Loop
import uuid
import time
import random
from typing import Dict, Any, Optional
# 导入所有之前定义的模块
from prompt_optimization_agent.work_log import WorkLogEntry, WorkLogRecorder
from prompt_optimization_agent.prompt_manager import PromptTemplate, PromptTemplateManager
from prompt_optimization_agent.work_log_analyzer import WorkLogAnalyzer
from prompt_optimization_agent.prompt_optimizer import PromptOptimizer, MockLLM # 使用MockLLM
class RealLLMAdapter:
"""
模拟一个真实的LLM适配器，用于AgentCore。
在实际中，这里会集成OpenAI, Anthropic等的API。
"""
def __init__(self, api_key: str = "sk-mock-key"):
self.api_key = api_key # 实际中会用到
def call_llm(self, prompt: str, model: str = "gpt-3.5-turbo") -> Dict[str, Any]:
"""
模拟LLM调用。
"""
print(f"n--- Calling Real LLM Adapter (Mocked) ---")
print(f"Prompt (first 200 chars): {prompt[:200]}...")
time.sleep(random.uniform(0.1, 1.0)) # 模拟延迟
# 根据Prompt内容模拟不同的响应和成功/失败
if "生成一个Python函数" in prompt and "斐波那契" in prompt:
response_text = "```pythonndef fibonacci(n):n if n <= 0: return 0n elif n == 1: return 1n else: return fibonacci(n-1) + fibonacci(n-2)n```"
success = True
failure_reason = None
rating = 5
elif "总结以下文本" in prompt and "复杂" in prompt:
if "确保全面涵盖核心信息" in prompt: # 模拟优化后的Prompt效果更好
response_text = "文本总结：这是一篇关于复杂主题的文章，涵盖了多个关键概念，并提供了深入的分析。摘要旨在全面概括其核心论点和发现。"
success = True
rating = 4
failure_reason = None
else:
response_text = "文本总结：这是一篇复杂文章的摘要。"
success = False
failure_reason = "incomplete_response"
rating = 2
elif "生成一个SQL查询" in prompt:
response_text = "```sqlnSELECT * FROM users WHERE age > 18;n```"
success = True
failure_reason = None
rating = 5
else:
response_text = "我不太确定如何回答这个问题。"
success = False
failure_reason = "unclear_request"
rating = 1
cost = len(prompt) / 1000 * 0.002 # 模拟token成本
latency = random.uniform(0.5, 3.0) # 模拟延迟
return {
"response_text": response_text,
"success": success,
"failure_reason": failure_reason,
"rating": rating,
"cost": cost,
"latency": latency
}
class AgentCore:
"""
Agent的核心执行模块，处理任务、选择Prompt、调用LLM并记录日志。
"""
def __init__(self,
prompt_manager: PromptTemplateManager,
log_recorder: WorkLogRecorder,
llm_adapter: RealLLMAdapter):
self.prompt_manager = prompt_manager
self.log_recorder = log_recorder
self.llm_adapter = llm_adapter
def execute_task(self, task_type: str, input_data: Dict[str, Any], task_id: Optional[str] = None) -> Dict[str, Any]:
"""
执行一个任务，包括Prompt选择、LLM调用和日志记录。
"""
if task_id is None:
task_id = str(uuid.uuid4())
# 1. Prompt选择策略：
# 这里简化为获取指定任务类型下所有活跃模板的第一个（或随机选择一个）
# 在实际中，可以根据模板的性能指标（如成功率、成本）进行更智能的选择，
# 甚至可以实现A/B测试逻辑，轮流使用新旧模板。
active_templates = self.prompt_manager.get_active_templates_by_task_type(task_type)
if not active_templates:
print(f"No active prompt templates found for task type: {task_type}. Cannot execute task.")
return {"error": "No active templates"}
# 简单选择第一个活跃模板，或可以实现A/B测试逻辑
selected_template = active_templates[0]
# For A/B testing:
# if len(active_templates) > 1:
# selected_template = random.choice(active_templates)
# else:
# selected_template = active_templates[0]
try:
# 2. 填充Prompt模板
filled_prompt = selected_template.fill_template(**input_data)
except ValueError as e:
print(f"Error filling prompt for task {task_id}: {e}")
self.log_recorder.record_log(WorkLogEntry(
task_id=task_id,
task_type=task_type,
input_data=str(input_data),
prompt_id=selected_template.prompt_id,
prompt_template=selected_template.template_string,
llm_response="",
success_flag=False,
failure_reason=f"prompt_fill_error: {e}",
rating=1
))
return {"error": f"Prompt fill error: {e}"}
# 3. 调用LLM
llm_result = self.llm_adapter.call_llm(filled_prompt, model="gpt-3.5-turbo")
# 4. 记录日志
log_entry = WorkLogEntry(
task_id=task_id,
task_type=task_type,
input_data=str(input_data), # 记录原始输入数据
prompt_id=selected_template.prompt_id,
prompt_template=filled_prompt, # 记录实际发送给LLM的完整Prompt
llm_response=llm_result.get("response_text", ""),
success_flag=llm_result.get("success", False),
failure_reason=llm_result.get("failure_reason"),
human_feedback=None, # 人类反馈通常是异步或后续添加的
rating=llm_result.get("rating"),
cost=llm_result.get("cost"),
latency=llm_result.get("latency")
)
self.log_recorder.record_log(log_entry)
return {
"task_id": task_id,
"llm_response": llm_result.get("response_text", ""),
"success": llm_result.get("success", False),
"prompt_id_used": selected_template.prompt_id
}
class SelfLearningScheduler:
"""
调度自学习过程，定期分析日志并优化Prompt。
"""
def __init__(self,
agent_core: AgentCore,
analyzer: WorkLogAnalyzer,
optimizer: PromptOptimizer,
prompt_manager: PromptTemplateManager):
self.agent_core = agent_core
self.analyzer = analyzer
self.optimizer = optimizer
self.prompt_manager = prompt_manager
def run_learning_cycle(self, days_to_analyze: int = 7):
"""
执行一个完整的学习周期。
"""
print(f"n--- Starting Self-Learning Cycle (Analyzing last {days_to_analyze} days) ---")
end_date = datetime.datetime.utcnow().strftime("%Y-%m-%d")
start_date = (datetime.datetime.utcnow() - datetime.timedelta(days=days_to_analyze)).strftime("%Y-%m-%d")
# 1. 分析日志
performance_report = self.analyzer.analyze_prompt_performance(start_date, end_date)
prompts_to_optimize = self.analyzer.identify_prompts_for_optimization(performance_report)
if not prompts_to_optimize:
print("No prompts identified for optimization. Learning cycle complete.")
return
# 2. 优化Prompt
newly_optimized_prompts = []
for prompt_id in prompts_to_optimize:
optimized_prompt = self.optimizer.optimize_prompt(prompt_id, performance_report)
if optimized_prompt:
newly_optimized_prompts.append(optimized_prompt)
# 3. 评估与激活新Prompt (简化模拟)
print(f"n--- Evaluating and Activating New Prompts ---")
for new_prompt in newly_optimized_prompts:
print(f"Simulating evaluation for new prompt: {new_prompt.prompt_id}")
# 实际中这里会有A/B测试、沙盒测试等
# 假设我们模拟一个测试结果，如果新Prompt比旧的"理论上"更好，就激活它
original_template = self.prompt_manager.get_template(new_prompt.prompt_id.split('_v')[0] + '_' + new_prompt.version.split('.')[0] + '.0') # 粗略匹配旧版本ID
if original_template and new_prompt.task_type == original_template.task_type and new_prompt.status == "testing":
# 简单判断，如果新Prompt的描述包含"提高"或"减少"某个问题，就认为它可能更好
if "提高" in new_prompt.description or "减少" in new_prompt.description:
print(f" New prompt '{new_prompt.prompt_id}' passed simulated evaluation. Activating...")
self.prompt_manager.activate_template(new_prompt.prompt_id)
# 同时，将旧模板降级为 inactive，或者保留用于比较
# self.prompt_manager.deactivate_template(original_template.prompt_id)
else:
print(f" New prompt '{new_prompt.prompt_id}' did not pass simulated evaluation. Keeping as testing/inactive.")
else:
print(f" New prompt '{new_prompt.prompt_id}' either has no original to compare or is not in 'testing' status. Keeping as testing/inactive.")
print("Self-Learning Cycle Finished.")
# --- 主程序入口 ---
if __name__ == "__main__":
# 初始化所有组件
recorder = WorkLogRecorder()
manager = PromptTemplateManager()
llm_adapter = RealLLMAdapter() # 用于AgentCore的真实LLM模拟
mock_optimizer_llm = MockLLM() # 用于PromptOptimizer的LLM模拟
agent = AgentCore(manager, recorder, llm_adapter)
analyzer = WorkLogAnalyzer(recorder, manager)
optimizer = PromptOptimizer(manager, mock_optimizer_llm)
scheduler = SelfLearningScheduler(agent, analyzer, optimizer, manager)
# 初始状态：确保有一些Prompt模板
if not manager.get_all_templates():
print("No initial templates found. Adding some default templates.")
template_summary_v1 = PromptTemplate(
prompt_id="summary_v1_0",
name="Standard Summarization",
description="用于通用文本摘要，要求简洁明了。",
template_string="你是一个专业的文本摘要助手。请将以下文本总结为不超过100字的简洁摘要，并确保保留核心信息。nn文本：{text}",
task_type="text_summarization",
version="1.0",
parameters=["text"]
)
manager.add_template(template_summary_v1)
template_code_gen_v1 = PromptTemplate(
prompt_id="code_gen_v1_0",
name="Python Code Generation",
description="根据用户需求生成Python函数，注重可读性和注释。",
template_string="你是一个专业的Python编程助手。请根据以下用户需求生成一个Python函数。确保代码可读性高，并包含必要的注释。nn需求：{user_requirement}",
task_type="code_generation",
version="1.0",
parameters=["user_requirement"]
)
manager.add_template(template_code_gen_v1)
template_bad_sql = PromptTemplate(
prompt_id="sql_gen_v1_0_bad",
name="SQL Query Generation (Bad)",
description="用于SQL生成，但Prompt可能不够明确，导致LLM偏离主题。",
template_string="请生成一个SQL查询来获取数据：{query_description}",
task_type="sql_generation",
version="1.0",
parameters=["query_description"]
)
manager.add_template(template_bad_sql)
manager.deactivate_template("sql_gen_v1_0_bad") # 先停用，等优化后激活
print("n--- Initial Active Prompts ---")
for t in manager.get_all_templates():
if t.status == "active":
print(f" - {t.prompt_id} ({t.name})")
# --- 模拟Agent运行一段时间，产生日志 ---
print("n--- Simulating Agent Daily Operations (Generating Logs) ---")
for day in range(1, 4): # 模拟运行3天
print(f"n--- Day {day} ---")
for _ in range(5): # 每天执行5个任务
# 模拟文本摘要任务
agent.execute_task(
task_type="text_summarization",
input_data={"text": "This is a very complex article about quantum physics and its implications. It discusses various theories and experiments in detail."}
)
# 模拟代码生成任务
agent.execute_task(
task_type="code_generation",
input_data={"user_requirement": "Generate a Python function to calculate the nth Fibonacci number efficiently."}
)
# 模拟SQL生成任务，使用一个可能导致失败的Prompt (如果激活了的话)
if manager.get_template("sql_gen_v1_0_bad") and manager.get_template("sql_gen_v1_0_bad").status == "active":
agent.execute_task(
task_type="sql_generation",
input_data={"query_description": "Find all users older than 30 from the 'users' table."}
)
time.sleep(0.5) # 模拟一天结束
# --- 运行自学习循环 ---
print("n--- Running Self-Learning Cycle ---")
scheduler.run_learning_cycle(days_to_analyze=3) # 分析过去3天的日志
# --- 再次检查活跃的Prompt ---
print("n--- Active Prompts After Learning Cycle ---")
for t in manager.get_all_templates():
if t.status == "active":
print(f" - {t.prompt_id} ({t.name}) - Version: {t.version}")
if t.performance_metrics:
print(f" Performance: Success Rate={t.performance_metrics.get('success_rate', 'N/A'):.2f}, Avg Rating={t.performance_metrics.get('avg_rating', 'N/A'):.1f}")
在上面的 SelfLearningScheduler 中，Evaluation & Validation 模块被大大简化了。在实际生产环境中，这部分需要更精细的设计：
- A/B测试框架: 新生成的
testing状态的Prompt会与当前的active模板同时部署，Agent在执行任务时会随机选择一个进行测试。通过比较一段时间内两种模板的性能指标（成功率、评分、成本等），来决定是否激活新模板。 - 回放测试: 使用历史日志中的
input_data来重新执行任务，但使用新的testing状态的Prompt。然后将新Prompt的输出与旧Prompt的真实输出进行比较，通过自动化指标（例如，如果任务是代码生成，可以跑测试用例；如果是摘要，可以通过RougE分数）或人工评估来判断新Prompt的优劣。 - 人工审核: 对于关键任务，新Prompt在激活前可能需要经过人类专家的审查。
第七章：高级考量与未来展望
我们已经构建了一个功能强大的自我学习Agent的基础框架。但这个领域充满了无限的可能性。
7.1 上下文敏感的Prompt优化
目前的优化是基于任务类型和历史平均表现。未来可以引入更细粒度的上下文信息。例如，对于同样是“文本摘要”任务，如果文本是法律文件，Agent可能需要一个侧重“关键条款”的Prompt；如果文本是新闻报道，则可能需要一个侧重“事件经过”的Prompt。这意味着Prompt优化不仅是针对模板本身，还需要根据实时上下文动态调整Prompt的某些部分。
7.2 多Agent协作与Prompt共享
在一个复杂的系统中，可能存在多个Agent，每个负责不同类型的任务。它们之间可以共享Prompt模板，甚至可以互相学习和优化对方的Prompt。例如，一个负责“数据清洗”的Agent，其优化后的Prompt可能对“报告生成”Agent的数据预处理阶段有所启发。
7.3 可解释性与透明度
当Agent自动优化Prompt时，我们希望了解“为什么”它选择以这种方式修改Prompt。集成解释性AI（XAI）技术，让Agent能够解释其优化决策的理由（例如，基于哪些失败日志，为了解决什么问题），将极大地增强系统的可信度和可调试性。
7.4 持续学习与灾难性遗忘
Agent需要不断学习和适应新的任务类型和数据分布。然而，过度优化可能会导致“灾难性遗忘”，即新Prompt在解决旧问题时表现不佳。引入连续学习（Continual Learning）的策略，例如知识蒸馏、增量学习等，可以帮助Agent在不断进化的同时，保留重要的历史经验。
通过今天的讲座，我们深入探讨了如何从零开始构建一个具备“自我学习能力”的Agent，其核心在于一个基于工作日志的闭环优化机制。我们设计了WorkLogRecorder来捕捉Agent的“经验”，PromptTemplateManager来管理Agent的“工具箱”，WorkLogAnalyzer来发现“经验”中的“规律”，PromptOptimizer利用这些“规律”来“创造”更优秀的Prompt模板，而AgentCore则将这一切整合，并通过SelfLearningScheduler驱动 Agent 的持续进化。这个框架为构建更智能、更自主的AI系统奠定了坚实的基础，预示着一个由AI自我驱动的Prompt工程新时代的到来。