AI-AutoGPT框架技术文档
AutoGPT是2025年先进的自主AI框架，最新v5.0版本通过自主决策引擎、任务执行系统等核心组件实现AI自主思考与执行。其分层架构包含用户交互层、认知能力层、工具集成层等七大模块，支持多智能体协作与安全约束机制。框架采用Python/JavaScript开发，具有任务自动分解、自我学习和风险控制等创新功能，为构建下一代自主AI系统提供完整基础设施。
·
AutoGPT 技术文档
框架概述
AutoGPT是2025年先进的自主AI智能体框架，专为构建完全自主的AI系统而设计。作为自主AI领域的先驱框架，AutoGPT通过赋予AI系统自主思考、规划、执行和自我改进的能力，实现了从传统AI工具向真正自主AI系统的重大跨越。2025年最新版本在自主决策、任务执行、自我学习和安全控制等方面实现了革命性突破，为构建下一代自主AI应用提供了完整的基础设施。
基本信息
- 开发团队: AutoGPT AI (原Significant Gravitas Ltd.)
- 最新版本: v5.0.0 (2025年9月)
- 框架类型: 自主AI智能体框架
- 主要语言: Python, JavaScript/TypeScript
- 架构模式: 自主决策、分层架构、插件化、微服务
- 核心创新: 自主决策引擎、任务自动分解、自我学习系统、安全约束机制、多智能体协作
架构设计
总体架构图
查看大图：鼠标右键 → “在新标签页打开图片” → 浏览器自带放大
graph TB
subgraph "用户交互层 User Interaction Layer"
UI1[Web界面 Web Interface]
UI2[CLI命令行 CLI Interface]
UI3[API接口 API Interface]
UI4[语音交互 Voice Interface]
AU1[用户认证 User Authentication]
AU2[权限管理 Permission Management]
AU3[会话管理 Session Management]
end
subgraph "AutoGPT核心架构 Core AutoGPT Architecture"
subgraph "自主决策引擎 Autonomous Decision Engine"
ADE1[目标理解 Goal Understanding]
ADE2[任务规划 Task Planning]
ADE3[策略制定 Strategy Formulation]
ADE4[决策优化 Decision Optimization]
ADE5[风险评估 Risk Assessment]
end
subgraph "任务执行系统 Task Execution System"
TES1[任务分解 Task Decomposition]
TES2[子任务调度 Subtask Scheduling]
TES3[执行监控 Execution Monitoring]
TES4[结果验证 Result Validation]
TES5[异常处理 Exception Handling]
end
subgraph "自我学习系统 Self-Learning System"
SLS1[经验积累 Experience Accumulation]
SLS2[模式识别 Pattern Recognition]
SLS3[性能优化 Performance Optimization]
SLS4[知识更新 Knowledge Update]
SLS5[策略调整 Strategy Adjustment]
end
subgraph "安全约束机制 Safety Constraint Mechanism"
SCM1[行为监控 Behavior Monitoring]
SCM2[约束检查 Constraint Checking]
SCM3[安全干预 Safety Intervention]
SCM4[审计日志 Audit Logging]
SCM5[合规验证 Compliance Verification]
end
subgraph "多智能体协作 Multi-Agent Collaboration"
MAC1[智能体协调 Agent Coordination]
MAC2[任务分配 Task Allocation]
MAC3[资源共享 Resource Sharing]
MAC4[冲突解决 Conflict Resolution]
MAC5[结果整合 Result Integration]
end
end
subgraph "认知能力层 Cognitive Capability Layer"
CCL1[自然语言处理 NLP Engine]
CCL2[推理引擎 Reasoning Engine]
CCL3[记忆系统 Memory System]
CCL4[注意力机制 Attention Mechanism]
CCL5[创造力生成 Creativity Generator]
end
subgraph "工具集成层 Tool Integration Layer"
TIL1[Web浏览 Web Browsing]
TIL2[文件操作 File Operations]
TIL3[代码执行 Code Execution]
TIL4[数据库操作 Database Operations]
TIL5[API调用 API Integration]
end
subgraph "模型服务层 Model Service Layer"
MSL1[OpenAI GPT-4o]
MSL2[Anthropic Claude-3.5]
MSL3[Google Gemini-1.5]
MSL4[Meta Llama-3.1]
MSL5[Custom Models]
end
subgraph "数据存储层 Data Storage Layer"
DSL1[向量数据库 Vector DB]
DSL2[图数据库 Graph DB]
DSL3[对象存储 Object Storage]
DSL4[缓存系统 Cache System]
DSL5[日志存储 Log Storage]
end
subgraph "监控与治理层 Monitoring & Governance Layer"
MGL1[性能监控 Performance Monitoring]
MGL2[行为分析 Behavior Analysis]
MGL3[成本追踪 Cost Tracking]
MGL4[安全审计 Security Audit]
MGL5[治理报告 Governance Reporting]
end
%% 用户交互层
UI1 --> AU1
UI2 --> AU2
UI3 --> AU3
%% 自主决策引擎
AU1 --> ADE1
AU2 --> ADE2
AU3 --> ADE3
ADE1 --> ADE2
ADE2 --> ADE3
ADE3 --> ADE4
ADE4 --> ADE5
%% 任务执行系统
ADE5 --> TES1
TES1 --> TES2
TES2 --> TES3
TES3 --> TES4
TES4 --> TES5
%% 自我学习系统
TES5 --> SLS1
SLS1 --> SLS2
SLS2 --> SLS3
SLS3 --> SLS4
SLS4 --> SLS5
%% 安全约束机制
TES3 --> SCM1
SCM1 --> SCM2
SCM2 --> SCM3
SCM3 --> SCM4
SCM4 --> SCM5
%% 多智能体协作
ADE4 --> MAC1
MAC1 --> MAC2
MAC2 --> MAC3
MAC3 --> MAC4
MAC4 --> MAC5
%% 认知能力层
SLS5 --> CCL1
CCL1 --> CCL2
CCL2 --> CCL3
CCL3 --> CCL4
CCL4 --> CCL5
%% 工具集成层
CCL5 --> TIL1
TIL1 --> TIL2
TIL2 --> TIL3
TIL3 --> TIL4
TIL4 --> TIL5
%% 模型服务层
TIL5 --> MSL1
MSL1 --> MSL2
MSL2 --> MSL3
MSL3 --> MSL4
MSL4 --> MSL5
%% 数据存储层
MSL5 --> DSL1
DSL1 --> DSL2
DSL2 --> DSL3
DSL3 --> DSL4
DSL4 --> DSL5
%% 监控与治理层
DSL5 --> MGL1
MGL1 --> MGL2
MGL2 --> MGL3
MGL3 --> MGL4
MGL4 --> MGL5
style UI1 fill:#1f2937
style ADE1 fill:#3b82f6
style TES1 fill:#10b981
style SLS1 fill:#f59e0b
style SCM1 fill:#8b5cf6
style MAC1 fill:#06b6d4
style CCL1 fill:#ef4444
style TIL1 fill:#84cc16
style MSL1 fill:#6b7280
核心组件详解
1. 自主决策引擎 (Autonomous Decision Engine)
- 目标理解: 使用最新的NLP技术深度理解用户设定的目标和约束条件
- 任务规划: 基于目标自动分解为可执行的子任务，支持多层次任务分解
- 策略制定: 根据环境和历史经验制定最优执行策略，支持动态策略调整
- 决策优化: 使用强化学习和贝叶斯优化持续改进决策质量
- 风险评估: 实时评估每个决策的风险和收益，确保安全可靠执行
2. 任务执行系统 (Task Execution System)
- 任务分解: 将复杂任务智能分解为原子级别的子任务，支持递归分解
- 子任务调度: 基于优先级、依赖关系和资源约束的智能调度算法
- 执行监控: 实时监控任务执行状态，支持异常检测和自动恢复
- 结果验证: 多维度验证任务执行结果的正确性和完整性
- 异常处理: 完善的异常捕获、分类和处理机制，支持自动重试和降级
3. 自我学习系统 (Self-Learning System)
- 经验积累: 自动收集和整理执行过程中的经验教训
- 模式识别: 使用机器学习识别成功和失败的模式
- 性能优化: 基于历史数据自动优化执行策略和参数
- 知识更新: 持续更新知识库，保持信息的时效性
- 策略调整: 根据学习结果动态调整决策策略
4. 安全约束机制 (Safety Constraint Mechanism)
- 行为监控: 实时监控AI系统的所有行为，识别潜在风险
- 约束检查: 多层次的安全约束检查，包括伦理、法律和技术约束
- 安全干预: 在检测到风险时自动进行安全干预和纠正
- 审计日志: 完整的操作审计日志，支持可追溯性
- 合规验证: 自动验证操作是否符合相关法规和标准
5. 多智能体协作 (Multi-Agent Collaboration)
- 智能体协调: 协调多个AI智能体的行为，避免冲突和重复
- 任务分配: 基于智能体能力和负载的智能任务分配
- 资源共享: 高效共享计算资源、数据资源和知识资源
- 冲突解决: 智能检测和解决智能体间的冲突
- 结果整合: 整合多个智能体的执行结果，形成统一输出
主要算法与技术
1. 自主决策算法
# AutoGPT自主决策算法
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import asyncio
import json
from datetime import datetime
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
class DecisionType(Enum):
"""决策类型枚举"""
TASK_PLANNING = "task_planning"
TOOL_SELECTION = "tool_selection"
STRATEGY_ADJUSTMENT = "strategy_adjustment"
RISK_ASSESSMENT = "risk_assessment"
GOAL_REFINEMENT = "goal_refinement"
class DecisionConfidence(Enum):
"""决策置信度枚举"""
HIGH = 0.9
MEDIUM = 0.7
LOW = 0.5
UNCERTAIN = 0.3
@dataclass
class DecisionContext:
"""决策上下文"""
goal: str
constraints: List[str]
available_resources: Dict[str, Any]
historical_data: List[Dict[str, Any]]
current_state: Dict[str, Any]
risk_tolerance: float = 0.5
time_limit: Optional[int] = None
@dataclass
class DecisionOutcome:
"""决策结果"""
decision_type: DecisionType
chosen_action: str
confidence: float
reasoning: str
alternatives: List[Dict[str, Any]]
risk_assessment: Dict[str, float]
expected_outcome: Dict[str, Any]
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
class AutoGPTDecisionEngine:
"""AutoGPT自主决策引擎"""
def __init__(self):
self.decision_models = {}
self.risk_assessor = RiskAssessmentEngine()
self.strategy_optimizer = StrategyOptimizationEngine()
self.experience_learner = ExperienceLearningEngine()
self.confidence_calculator = ConfidenceCalculationEngine()
self.setup_latest_models()
def setup_latest_models(self):
"""设置模型"""
# 任务规划模型
self.decision_models[DecisionType.TASK_PLANNING] = TaskPlanningModel(
model_type="transformer",
max_sequence_length=4096,
embedding_dim=768,
num_layers=12
)
# 工具选择模型
self.decision_models[DecisionType.TOOL_SELECTION] = ToolSelectionModel(
model_type="ensemble",
base_models=["random_forest", "xgboost", "neural_network"],
selection_criteria=["accuracy", "speed", "resource_usage"]
)
# 策略调整模型
self.decision_models[DecisionType.STRATEGY_ADJUSTMENT] = StrategyAdjustmentModel(
model_type="reinforcement_learning",
algorithm="PPO",
state_space="continuous",
action_space="discrete"
)
# 风险评估模型
self.decision_models[DecisionType.RISK_ASSESSMENT] = RiskAssessmentModel(
model_type="bayesian_network",
risk_factors=["technical", "ethical", "legal", "financial"],
confidence_threshold=0.8
)
# 目标优化模型
self.decision_models[DecisionType.GOAL_REFINEMENT] = GoalRefinementModel(
model_type="genetic_algorithm",
population_size=100,
mutation_rate=0.1,
crossover_rate=0.8
)
async def make_autonomous_decision(self, context: DecisionContext) -> DecisionOutcome:
"""做出自主决策"""
# 1. 分析当前情况
situation_analysis = await self._analyze_situation(context)
# 2. 生成候选方案
candidate_solutions = await self._generate_candidate_solutions(context, situation_analysis)
# 3. 评估每个方案
evaluated_solutions = await self._evaluate_solutions(candidate_solutions, context)
# 4. 风险评估
risk_assessments = await self._assess_risks(evaluated_solutions, context)
# 5. 选择最优方案
chosen_solution = await self._select_optimal_solution(evaluated_solutions, risk_assessments, context)
# 6. 计算置信度
confidence = await self._calculate_confidence(chosen_solution, context)
# 7. 生成决策结果
decision_outcome = DecisionOutcome(
decision_type=chosen_solution["type"],
chosen_action=chosen_solution["action"],
confidence=confidence,
reasoning=chosen_solution["reasoning"],
alternatives=evaluated_solutions,
risk_assessment=risk_assessments[chosen_solution["id"]],
expected_outcome=chosen_solution["expected_outcome"]
)
# 8. 学习经验
await self.experience_learner.learn_from_decision(decision_outcome, context)
return decision_outcome
async def _analyze_situation(self, context: DecisionContext) -> Dict[str, Any]:
"""分析当前情况"""
analysis = {
"goal_complexity": self._calculate_goal_complexity(context.goal),
"constraint_analysis": self._analyze_constraints(context.constraints),
"resource_availability": self._assess_resource_availability(context.available_resources),
"historical_patterns": await self._identify_historical_patterns(context.historical_data),
"current_state_evaluation": self._evaluate_current_state(context.current_state)
}
return analysis
def _calculate_goal_complexity(self, goal: str) -> float:
"""计算目标复杂度"""
# 基于目标的长度、词汇多样性和结构复杂度
words = goal.split()
unique_words = len(set(words))
total_words = len(words)
# 复杂度指标
lexical_diversity = unique_words / total_words if total_words > 0 else 0
length_factor = min(len(goal) / 1000, 1.0) # 归一化长度因子
structural_complexity = goal.count(" and ") + goal.count(" or ") + goal.count(",")
# 综合复杂度分数
complexity_score = (lexical_diversity * 0.4 + length_factor * 0.3 + structural_complexity * 0.3)
return min(complexity_score, 1.0)
def _analyze_constraints(self, constraints: List[str]) -> Dict[str, Any]:
"""分析约束条件"""
constraint_analysis = {
"total_constraints": len(constraints),
"constraint_types": {},
"constraint_severity": 0.0,
"feasibility_score": 1.0
}
for constraint in constraints:
# 分类约束类型
if "time" in constraint.lower():
constraint_analysis["constraint_types"]["time"] = constraint_analysis["constraint_types"].get("time", 0) + 1
elif "resource" in constraint.lower():
constraint_analysis["constraint_types"]["resource"] = constraint_analysis["constraint_types"].get("resource", 0) + 1
elif "ethical" in constraint.lower():
constraint_analysis["constraint_types"]["ethical"] = constraint_analysis["constraint_types"].get("ethical", 0) + 1
elif "legal" in constraint.lower():
constraint_analysis["constraint_types"]["legal"] = constraint_analysis["constraint_types"].get("legal", 0) + 1
# 评估约束严重性
if "must" in constraint.lower() or "required" in constraint.lower():
constraint_analysis["constraint_severity"] += 0.3
elif "should" in constraint.lower() or "preferred" in constraint.lower():
constraint_analysis["constraint_severity"] += 0.1
# 计算可行性分数
constraint_analysis["feasibility_score"] = max(0.0, 1.0 - constraint_analysis["constraint_severity"])
return constraint_analysis
async def _generate_candidate_solutions(self, context: DecisionContext, analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
"""生成候选解决方案"""
solutions = []
# 基于目标类型生成解决方案
goal_type = self._classify_goal(context.goal)
if goal_type == "task_execution":
solutions = await self._generate_task_solutions(context, analysis)
elif goal_type == "information_gathering":
solutions = await self._generate_research_solutions(context, analysis)
elif goal_type == "problem_solving":
solutions = await self._generate_problem_solving_solutions(context, analysis)
elif goal_type == "creative_generation":
solutions = await self._generate_creative_solutions(context, analysis)
# 添加元数据
for i, solution in enumerate(solutions):
solution["id"] = f"solution_{i}_{datetime.now().timestamp()}"
solution["type"] = goal_type
solution["generated_at"] = datetime.now().isoformat()
return solutions
async def _evaluate_solutions(self, solutions: List[Dict[str, Any]], context: DecisionContext) -> List[Dict[str, Any]]:
"""评估解决方案"""
evaluated_solutions = []
for solution in solutions:
# 多维度评估
evaluation = {
"feasibility": await self._evaluate_feasibility(solution, context),
"effectiveness": await self._evaluate_effectiveness(solution, context),
"efficiency": await self._evaluate_efficiency(solution, context),
"risk_level": await self._evaluate_risk_level(solution, context),
"resource_requirements": await self._evaluate_resource_requirements(solution, context)
}
# 计算综合评分
overall_score = (
evaluation["feasibility"] * 0.3 +
evaluation["effectiveness"] * 0.25 +
evaluation["efficiency"] * 0.2 +
(1 - evaluation["risk_level"]) * 0.15 +
evaluation["resource_requirements"] * 0.1
)
solution["evaluation"] = evaluation
solution["overall_score"] = overall_score
evaluated_solutions.append(solution)
# 按评分排序
evaluated_solutions.sort(key=lambda x: x["overall_score"], reverse=True)
return evaluated_solutions
async def _assess_risks(self, solutions: List[Dict[str, Any]], context: DecisionContext) -> Dict[str, Dict[str, float]]:
"""评估风险"""
risk_assessments = {}
for solution in solutions:
risk_assessment = await self.risk_assessor.assess_solution_risk(solution, context)
risk_assessments[solution["id"]] = risk_assessment
return risk_assessments
async def _select_optimal_solution(self,
evaluated_solutions: List[Dict[str, Any]],
risk_assessments: Dict[str, Dict[str, float]],
context: DecisionContext) -> Dict[str, Any]:
"""选择最优解决方案"""
# 应用风险过滤
filtered_solutions = []
for solution in evaluated_solutions:
risk_score = risk_assessments[solution["id"]]["overall_risk"]
if risk_score <= context.risk_tolerance:
filtered_solutions.append(solution)
if not filtered_solutions:
# 如果没有满足风险要求的方案，选择风险最低的
filtered_solutions = evaluated_solutions
# 选择评分最高的方案
optimal_solution = filtered_solutions[0]
# 生成决策推理
optimal_solution["reasoning"] = self._generate_decision_reasoning(optimal_solution, context)
return optimal_solution
async def _calculate_confidence(self, chosen_solution: Dict[str, Any], context: DecisionContext) -> float:
"""计算置信度"""
return await self.confidence_calculator.calculate_confidence(chosen_solution, context)
def _generate_decision_reasoning(self, solution: Dict[str, Any], context: DecisionContext) -> str:
"""生成决策推理"""
reasoning = f"""
基于以下分析选择此方案：
1. 目标复杂度: {self._calculate_goal_complexity(context.goal):.2f}
2. 方案综合评分: {solution['overall_score']:.2f}
3. 可行性评估: {solution['evaluation']['feasibility']:.2f}
4. 有效性评估: {solution['evaluation']['effectiveness']:.2f}
5. 效率评估: {solution['evaluation']['efficiency']:.2f}
选择理由: {solution['description']}
"""
return reasoning.strip()
class RiskAssessmentEngine:
"""风险评估引擎"""
async def assess_solution_risk(self, solution: Dict[str, Any], context: DecisionContext) -> Dict[str, float]:
"""评估解决方案风险"""
risk_assessment = {
"technical_risk": await self._assess_technical_risk(solution, context),
"ethical_risk": await self._assess_ethical_risk(solution, context),
"legal_risk": await self._assess_legal_risk(solution, context),
"financial_risk": await self._assess_financial_risk(solution, context),
"operational_risk": await self._assess_operational_risk(solution, context)
}
# 计算总体风险
risk_assessment["overall_risk"] = np.mean(list(risk_assessment.values()))
return risk_assessment
async def _assess_technical_risk(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""评估技术风险"""
# 实现技术风险评估逻辑
return 0.3 # 示例值
async def _assess_ethical_risk(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""评估伦理风险"""
# 实现伦理风险评估逻辑
return 0.2 # 示例值
async def _assess_legal_risk(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""评估法律风险"""
# 实现法律风险评估逻辑
return 0.1 # 示例值
async def _assess_financial_risk(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""评估财务风险"""
# 实现财务风险评估逻辑
return 0.4 # 示例值
async def _assess_operational_risk(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""评估操作风险"""
# 实现操作风险评估逻辑
return 0.25 # 示例值
class StrategyOptimizationEngine:
"""策略优化引擎"""
async def optimize_strategy(self, current_strategy: Dict[str, Any], feedback: Dict[str, Any]) -> Dict[str, Any]:
"""优化策略"""
# 实现策略优化逻辑
return current_strategy
class ExperienceLearningEngine:
"""经验学习引擎"""
async def learn_from_decision(self, decision_outcome: DecisionOutcome, context: DecisionContext) -> None:
"""从决策中学习"""
# 实现经验学习逻辑
pass
class ConfidenceCalculationEngine:
"""置信度计算引擎"""
async def calculate_confidence(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""计算置信度"""
# 实现置信度计算逻辑
base_confidence = solution["overall_score"]
# 考虑历史表现
historical_factor = await self._get_historical_factor(solution, context)
# 考虑不确定性
uncertainty_factor = await self._calculate_uncertainty(solution, context)
final_confidence = base_confidence * historical_factor * (1 - uncertainty_factor)
return min(final_confidence, 1.0)
async def _get_historical_factor(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""获取历史因子"""
# 实现历史因子计算
return 0.95
async def _calculate_uncertainty(self, solution: Dict[str, Any], context: DecisionContext) -> float:
"""计算不确定性"""
# 实现不确定性计算
return 0.1
# 使用示例
async def decision_engine_example():
"""决策引擎示例"""
# 创建决策引擎
decision_engine = AutoGPTDecisionEngine()
# 创建决策上下文
context = DecisionContext(
goal="Develop a comprehensive market analysis report for the AI industry",
constraints=[
"Must be completed within 24 hours",
"Budget limit of $1000",
"Must comply with data privacy regulations",
"Should include competitor analysis"
],
available_resources={
"compute_budget": 1000,
"api_calls_limit": 10000,
"storage_space": "10GB",
"team_members": ["analyst", "researcher", "writer"]
},
historical_data=[
{"task": "market_analysis", "success_rate": 0.85, "avg_time": 18},
{"task": "competitor_analysis", "success_rate": 0.92, "avg_time": 8}
],
current_state={
"progress": 0.1,
"completed_tasks": ["data_collection"],
"remaining_time": 20
},
risk_tolerance=0.3
)
# 做出自主决策
decision = await decision_engine.make_autonomous_decision(context)
print(f"Decision Type: {decision.decision_type}")
print(f"Chosen Action: {decision.chosen_action}")
print(f"Confidence: {decision.confidence}")
print(f"Reasoning: {decision.reasoning}")
print(f"Risk Assessment: {decision.risk_assessment}")
return decision
2. 任务执行算法
# AutoGPT任务执行算法
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import json
from datetime import datetime, timedelta
import uuid
from concurrent.futures import ThreadPoolExecutor
import traceback
class TaskStatus(Enum):
"""任务状态枚举"""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
RETRYING = "retrying"
TIMEOUT = "timeout"
class TaskPriority(Enum):
"""任务优先级枚举"""
CRITICAL = 1
HIGH = 2
MEDIUM = 3
LOW = 4
@dataclass
class Task:
"""任务数据结构"""
id: str
name: str
description: str
task_type: str
parameters: Dict[str, Any]
priority: TaskPriority
dependencies: List[str]
timeout: int
retry_config: Dict[str, Any]
created_at: datetime
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
status: TaskStatus = TaskStatus.PENDING
result: Optional[Any] = None
error: Optional[str] = None
execution_history: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class TaskExecutionContext:
"""任务执行上下文"""
task_id: str
global_state: Dict[str, Any]
available_tools: List[str]
resource_limits: Dict[str, Any]
safety_constraints: List[str]
execution_history: List[Dict[str, Any]]
class AutoGPTTaskExecutor:
"""AutoGPT任务执行器"""
def __init__(self):
self.task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self.active_tasks: Dict[str, Task] = {}
self.completed_tasks: Dict[str, Task] = {}
self.task_dependencies: Dict[str, List[str]] = {}
self.execution_pool = ThreadPoolExecutor(max_workers=10)
self.safety_monitor = SafetyMonitor()
self.performance_tracker = PerformanceTracker()
self.resource_manager = ResourceManager()
async def execute_task(self, task: Task, context: TaskExecutionContext) -> Any:
"""执行任务"""
try:
# 1. 任务预处理
await self._preprocess_task(task, context)
# 2. 安全检查
safety_check_passed = await self.safety_monitor.check_task_safety(task, context)
if not safety_check_passed:
raise ValueError(f"Task {task.id} failed safety check")
# 3. 资源分配
resource_allocation = await self.resource_manager.allocate_resources(task, context)
# 4. 执行核心任务
result = await self._execute_core_task(task, context, resource_allocation)
# 5. 结果验证
validated_result = await self._validate_task_result(task, result, context)
# 6. 后处理
final_result = await self._postprocess_task_result(task, validated_result, context)
return final_result
except Exception as e:
# 错误处理
await self._handle_task_error(task, e, context)
raise e
finally:
# 清理资源
await self.resource_manager.release_resources(task.id)
async def execute_task_graph(self, tasks: List[Task], global_context: Dict[str, Any]) -> Dict[str, Any]:
"""执行有依赖关系的任务图"""
# 1. 构建任务依赖图
task_graph = self._build_task_graph(tasks)
# 2. 拓扑排序
execution_order = self._topological_sort(task_graph)
# 3. 按顺序执行任务
results = {}
for task_id in execution_order:
task = next(t for t in tasks if t.id == task_id)
# 等待依赖任务完成
await self._wait_for_dependencies(task, results)
# 创建执行上下文
context = TaskExecutionContext(
task_id=task_id,
global_state=global_context,
available_tools=self._get_available_tools(task),
resource_limits=self._get_resource_limits(task),
safety_constraints=self._get_safety_constraints(task),
execution_history=[]
)
# 执行任务
try:
result = await self.execute_task(task, context)
results[task_id] = result
# 更新任务状态
task.status = TaskStatus.COMPLETED
task.result = result
task.completed_at = datetime.now()
except Exception as e:
# 处理任务失败
task.status = TaskStatus.FAILED
task.error = str(e)
# 根据配置决定是否继续执行其他任务
if not self._should_continue_on_failure(task):
break
return results
async def _preprocess_task(self, task: Task, context: TaskExecutionContext) -> None:
"""任务预处理"""
# 更新任务状态
task.status = TaskStatus.RUNNING
task.started_at = datetime.now()
# 记录执行历史
task.execution_history.append({
"event": "preprocessing_started",
"timestamp": datetime.now().isoformat(),
"details": {"task_name": task.name}
})
# 验证任务参数
await self._validate_task_parameters(task)
# 准备执行环境
await self._prepare_execution_environment(task, context)
# 记录预处理完成
task.execution_history.append({
"event": "preprocessing_completed",
"timestamp": datetime.now().isoformat()
})
async def _validate_task_parameters(self, task: Task) -> None:
"""验证任务参数"""
# 检查必需参数
required_params = self._get_required_parameters(task.task_type)
missing_params = set(required_params) - set(task.parameters.keys())
if missing_params:
raise ValueError(f"Missing required parameters: {missing_params}")
# 验证参数类型和范围
for param_name, param_value in task.parameters.items():
await self._validate_parameter(param_name, param_value, task.task_type)
async def _execute_core_task(self, task: Task, context: TaskExecutionContext, resources: Dict[str, Any]) -> Any:
"""执行核心任务"""
# 根据任务类型选择执行策略
if task.task_type == "web_research":
return await self._execute_web_research_task(task, context, resources)
elif task.task_type == "code_generation":
return await self._execute_code_generation_task(task, context, resources)
elif task.task_type == "data_analysis":
return await self._execute_data_analysis_task(task, context, resources)
elif task.task_type == "file_operation":
return await self._execute_file_operation_task(task, context, resources)
elif task.task_type == "api_call":
return await self._execute_api_call_task(task, context, resources)
else:
return await self._execute_generic_task(task, context, resources)
async def _execute_web_research_task(self, task: Task, context: TaskExecutionContext, resources: Dict[str, Any]) -> Any:
"""执行网络研究任务"""
# 获取研究参数
query = task.parameters.get("query", "")
max_results = task.parameters.get("max_results", 10)
search_engines = task.parameters.get("search_engines", ["google", "bing"])
# 执行搜索
search_results = []
for engine in search_engines:
try:
results = await self._search_web(engine, query, max_results // len(search_engines))
search_results.extend(results)
except Exception as e:
task.execution_history.append({
"event": "search_engine_error",
"timestamp": datetime.now().isoformat(),
"details": {"engine": engine, "error": str(e)}
})
# 分析和总结结果
analyzed_results = await self._analyze_search_results(search_results, query)
return {
"search_results": search_results,
"analysis": analyzed_results,
"sources": [result["source"] for result in search_results],
"summary": self._generate_research_summary(analyzed_results)
}
async def _execute_code_generation_task(self, task: Task, context: TaskExecutionContext, resources: Dict[str, Any]) -> Any:
"""执行代码生成任务"""
# 获取代码生成参数
requirements = task.parameters.get("requirements", "")
language = task.parameters.get("language", "python")
framework = task.parameters.get("framework", "")
constraints = task.parameters.get("constraints", [])
# 生成代码
generated_code = await self._generate_code(requirements, language, framework, constraints)
# 验证代码
validation_result = await self._validate_code(generated_code, language)
# 优化代码
if validation_result["is_valid"]:
optimized_code = await self._optimize_code(generated_code, language)
else:
optimized_code = generated_code
return {
"code": optimized_code,
"language": language,
"validation": validation_result,
"optimization_applied": validation_result["is_valid"]
}
async def _validate_task_result(self, task: Task, result: Any, context: TaskExecutionContext) -> Any:
"""验证任务结果"""
# 根据任务类型进行相应的验证
validation_rules = self._get_validation_rules(task.task_type)
for rule in validation_rules:
validation_result = await rule(result, task.parameters)
if not validation_result["is_valid"]:
raise ValueError(f"Task result validation failed: {validation_result['error']}")
# 记录验证通过
task.execution_history.append({
"event": "result_validation_passed",
"timestamp": datetime.now().isoformat()
})
return result
async def _postprocess_task_result(self, task: Task, result: Any, context: TaskExecutionContext) -> Any:
"""后处理任务结果"""
# 格式化结果
formatted_result = await self._format_result(result, task.task_type)
# 添加元数据
final_result = {
"task_id": task.id,
"task_name": task.name,
"result": formatted_result,
"execution_time": (datetime.now() - task.started_at).total_seconds(),
"metadata": {
"task_type": task.task_type,
"priority": task.priority.value,
"retry_count": len([h for h in task.execution_history if h["event"] == "retry"]),
"validation_passed": True
}
}
# 记录完成
task.execution_history.append({
"event": "task_completed",
"timestamp": datetime.now().isoformat(),
"details": {"execution_time": final_result["execution_time"]}
})
return final_result
async def _handle_task_error(self, task: Task, error: Exception, context: TaskExecutionContext) -> None:
"""处理任务错误"""
error_details = {
"error_type": type(error).__name__,
"error_message": str(error),
"stack_trace": traceback.format_exc(),
"timestamp": datetime.now().isoformat()
}
# 记录错误
task.error = json.dumps(error_details)
task.execution_history.append({
"event": "task_failed",
"timestamp": datetime.now().isoformat(),
"details": error_details
})
# 根据重试配置决定是否重试
if self._should_retry_task(task):
await self._retry_task(task, context)
else:
task.status = TaskStatus.FAILED
def _should_retry_task(self, task: Task) -> bool:
"""判断是否应该重试任务"""
retry_config = task.retry_config
max_retries = retry_config.get("max_retries", 3)
current_retries = len([h for h in task.execution_history if h["event"] == "retry"])
return current_retries < max_retries
async def _retry_task(self, task: Task, context: TaskExecutionContext) -> None:
"""重试任务"""
retry_delay = task.retry_config.get("retry_delay", 1)
exponential_backoff = task.retry_config.get("exponential_backoff", True)
# 计算重试延迟
retry_count = len([h for h in task.execution_history if h["event"] == "retry"])
if exponential_backoff:
delay = retry_delay * (2 ** retry_count)
else:
delay = retry_delay
# 记录重试
task.status = TaskStatus.RETRYING
task.execution_history.append({
"event": "retry",
"timestamp": datetime.now().isoformat(),
"details": {"retry_count": retry_count + 1, "delay": delay}
})
# 等待重试延迟
await asyncio.sleep(delay)
# 重置任务状态
task.status = TaskStatus.PENDING
task.started_at = None
# 重新执行任务
try:
result = await self.execute_task(task, context)
task.result = result
task.status = TaskStatus.COMPLETED
task.completed_at = datetime.now()
except Exception as e:
# 如果重试仍然失败，继续重试或标记为失败
await self._handle_task_error(task, e, context)
def _build_task_graph(self, tasks: List[Task]) -> Dict[str, List[str]]:
"""构建任务依赖图"""
graph = {}
for task in tasks:
graph[task.id] = task.dependencies.copy()
return graph
def _topological_sort(self, graph: Dict[str, List[str]]) -> List[str]:
"""拓扑排序"""
# 计算入度
in_degree = {node: 0 for node in graph}
for node, dependencies in graph.items():
for dep in dependencies:
if dep in in_degree:
in_degree[node] += 1
# 找到入度为0的节点
queue = [node for node, degree in in_degree.items() if degree == 0]
result = []
while queue:
current = queue.pop(0)
result.append(current)
# 减少依赖当前节点的节点的入度
for node, dependencies in graph.items():
if current in dependencies:
in_degree[node] -= 1
if in_degree[node] == 0:
queue.append(node)
# 检查是否有环
if len(result) != len(graph):
raise ValueError("Task dependency graph contains cycles")
return result
async def _wait_for_dependencies(self, task: Task, completed_results: Dict[str, Any]) -> None:
"""等待依赖任务完成"""
for dependency in task.dependencies:
while dependency not in completed_results:
await asyncio.sleep(0.1) # 简单的轮询等待
def _get_required_parameters(self, task_type: str) -> List[str]:
"""获取任务类型所需的参数"""
parameter_requirements = {
"web_research": ["query"],
"code_generation": ["requirements"],
"data_analysis": ["data_source"],
"file_operation": ["operation", "file_path"],
"api_call": ["endpoint", "method"]
}
return parameter_requirements.get(task_type, [])
async def _search_web(self, engine: str, query: str, max_results: int) -> List[Dict[str, Any]]:
"""执行网络搜索"""
# 这里集成实际的搜索API
# 模拟搜索结果
return [
{
"title": f"Result {i} for {query}",
"url": f"https://example{i}.com",
"snippet": f"Snippet {i} containing {query}",
"source": engine
}
for i in range(max_results)
]
async def _analyze_search_results(self, results: List[Dict[str, Any]], query: str) -> Dict[str, Any]:
"""分析搜索结果"""
# 实现搜索结果分析逻辑
return {
"total_results": len(results),
"unique_sources": len(set(r["source"] for r in results)),
"relevance_score": 0.85, # 模拟相关性评分
"key_themes": ["theme1", "theme2", "theme3"] # 模拟主题提取
}
def _generate_research_summary(self, analysis: Dict[str, Any]) -> str:
"""生成研究摘要"""
return f"Found {analysis['total_results']} results with relevance score {analysis['relevance_score']}"
class SafetyMonitor:
"""安全监控器"""
async def check_task_safety(self, task: Task, context: TaskExecutionContext) -> bool:
"""检查任务安全性"""
# 实现安全检查逻辑
return True
class PerformanceTracker:
"""性能跟踪器"""
def track_task_performance(self, task: Task) -> None:
"""跟踪任务性能"""
# 实现性能跟踪逻辑
pass
class ResourceManager:
"""资源管理器"""
async def allocate_resources(self, task: Task, context: TaskExecutionContext) -> Dict[str, Any]:
"""分配资源"""
# 实现资源分配逻辑
return {"cpu": 2, "memory": "4GB", "storage": "10GB"}
async def release_resources(self, task_id: str) -> None:
"""释放资源"""
# 实现资源释放逻辑
pass
# 使用示例
async def task_execution_example():
"""任务执行示例"""
# 创建任务执行器
executor = AutoGPTTaskExecutor()
# 创建任务
research_task = Task(
id="research_task_001",
name="AI Market Research",
description="Research the latest trends in AI market",
task_type="web_research",
parameters={
"query": "latest AI market trends 2025",
"max_results": 20,
"search_engines": ["google", "bing"]
},
priority=TaskPriority.HIGH,
dependencies=[],
timeout=300,
retry_config={
"max_retries": 3,
"retry_delay": 2,
"exponential_backoff": True
},
created_at=datetime.now()
)
# 创建执行上下文
context = TaskExecutionContext(
task_id=research_task.id,
global_state={},
available_tools=["web_search", "data_analysis"],
resource_limits={"cpu": 4, "memory": "8GB"},
safety_constraints=["no_sensitive_data", "rate_limit_compliance"],
execution_history=[]
)
# 执行任务
try:
result = await executor.execute_task(research_task, context)
print(f"Task completed successfully!")
print(f"Results: {result}")
except Exception as e:
print(f"Task failed: {e}")
return executor
3. 自我学习算法
# AutoGPT自我学习算法
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import json
import pickle
import numpy as np
from sklearn.cluster import DBSCAN
from sklearn.ensemble import IsolationForest
from sklearn.metrics import silhouette_score
import sqlite3
from pathlib import Path
@dataclass
class Experience:
"""经验数据结构"""
id: str
task_type: str
context: Dict[str, Any]
action: str
outcome: Dict[str, Any]
reward: float
timestamp: datetime
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"task_type": self.task_type,
"context": self.context,
"action": self.action,
"outcome": self.outcome,
"reward": self.reward,
"timestamp": self.timestamp.isoformat(),
"metadata": self.metadata
}
@dataclass
class LearningPattern:
"""学习模式"""
pattern_id: str
pattern_type: str
description: str
frequency: int
success_rate: float
average_reward: float
confidence: float
examples: List[Experience]
created_at: datetime
last_updated: datetime
class AutoGPTSelfLearningSystem:
"""AutoGPT自我学习系统"""
def __init__(self,
experience_db_path: str = "./experiences.db",
learning_rate: float = 0.01,
pattern_detection_threshold: float = 0.7,
min_experiences_for_pattern: int = 5):
self.experience_db_path = experience_db_path
self.learning_rate = learning_rate
self.pattern_detection_threshold = pattern_detection_threshold
self.min_experiences_for_pattern = min_experiences_for_pattern
self.experiences: List[Experience] = []
self.learning_patterns: Dict[str, LearningPattern] = {}
self.performance_metrics: Dict[str, List[float]] = {}
self.strategy_parameters: Dict[str, Any] = {}
self._initialize_database()
self._load_historical_experiences()
def _initialize_database(self):
"""初始化经验数据库"""
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
# 创建经验表
cursor.execute('''
CREATE TABLE IF NOT EXISTS experiences (
id TEXT PRIMARY KEY,
task_type TEXT,
context TEXT,
action TEXT,
outcome TEXT,
reward REAL,
timestamp TEXT,
metadata TEXT
)
''')
# 创建学习模式表
cursor.execute('''
CREATE TABLE IF NOT EXISTS learning_patterns (
pattern_id TEXT PRIMARY KEY,
pattern_type TEXT,
description TEXT,
frequency INTEGER,
success_rate REAL,
average_reward REAL,
confidence REAL,
examples TEXT,
created_at TEXT,
last_updated TEXT
)
''')
conn.commit()
conn.close()
def _load_historical_experiences(self):
"""加载历史经验"""
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
cursor.execute('SELECT * FROM experiences ORDER BY timestamp DESC LIMIT 10000')
rows = cursor.fetchall()
for row in rows:
experience = Experience(
id=row[0],
task_type=row[1],
context=json.loads(row[2]),
action=row[3],
outcome=json.loads(row[4]),
reward=row[5],
timestamp=datetime.fromisoformat(row[6]),
metadata=json.loads(row[7]) if row[7] else {}
)
self.experiences.append(experience)
conn.close()
# 加载学习模式
self._load_learning_patterns()
def _load_learning_patterns(self):
"""加载学习模式"""
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
cursor.execute('SELECT * FROM learning_patterns')
rows = cursor.fetchall()
for row in rows:
pattern = LearningPattern(
pattern_id=row[0],
pattern_type=row[1],
description=row[2],
frequency=row[3],
success_rate=row[4],
average_reward=row[5],
confidence=row[6],
examples=[], # 将在需要时加载
created_at=datetime.fromisoformat(row[7]),
last_updated=datetime.fromisoformat(row[8])
)
self.learning_patterns[pattern.pattern_id] = pattern
conn.close()
async def record_experience(self,
task_type: str,
context: Dict[str, Any],
action: str,
outcome: Dict[str, Any],
reward: float,
metadata: Dict[str, Any] = None) -> Experience:
"""记录经验"""
experience = Experience(
id=f"exp_{datetime.now().timestamp()}_{uuid.uuid4().hex[:8]}",
task_type=task_type,
context=context,
action=action,
outcome=outcome,
reward=reward,
timestamp=datetime.now(),
metadata=metadata or {}
)
# 添加到内存
self.experiences.append(experience)
# 保存到数据库
await self._save_experience_to_db(experience)
# 更新性能指标
await self._update_performance_metrics(task_type, reward)
# 检测新的学习模式
await self._detect_learning_patterns(task_type)
return experience
async def _save_experience_to_db(self, experience: Experience) -> None:
"""保存经验到数据库"""
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO experiences (id, task_type, context, action, outcome, reward, timestamp, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (
experience.id,
experience.task_type,
json.dumps(experience.context),
experience.action,
json.dumps(experience.outcome),
experience.reward,
experience.timestamp.isoformat(),
json.dumps(experience.metadata)
))
conn.commit()
conn.close()
async def _update_performance_metrics(self, task_type: str, reward: float) -> None:
"""更新性能指标"""
if task_type not in self.performance_metrics:
self.performance_metrics[task_type] = []
self.performance_metrics[task_type].append(reward)
# 保持最近100个指标
if len(self.performance_metrics[task_type]) > 100:
self.performance_metrics[task_type] = self.performance_metrics[task_type][-100:]
async def learn_from_experiences(self, task_type: Optional[str] = None, min_experiences: int = 10) -> Dict[str, Any]:
"""从经验中学习"""
# 筛选相关经验
relevant_experiences = [
exp for exp in self.experiences
if task_type is None or exp.task_type == task_type
]
if len(relevant_experiences) < min_experiences:
return {"error": f"Insufficient experiences. Need at least {min_experiences}"}
# 聚类分析
clusters = await self._cluster_experiences(relevant_experiences)
# 模式识别
patterns = await self._identify_patterns(clusters)
# 策略优化
optimized_strategies = await self._optimize_strategies(patterns)
# 更新学习模式
await self._update_learning_patterns(patterns, optimized_strategies)
learning_summary = {
"total_experiences_analyzed": len(relevant_experiences),
"clusters_found": len(clusters),
"patterns_identified": len(patterns),
"strategies_optimized": len(optimized_strategies),
"learning_timestamp": datetime.now().isoformat(),
"performance_improvement": await self._calculate_performance_improvement(task_type)
}
return learning_summary
async def _cluster_experiences(self, experiences: List[Experience]) -> List[List[Experience]]:
"""聚类经验"""
if len(experiences) < 5:
return [experiences] # 经验太少，不进行聚类
# 提取特征
features = []
for exp in experiences:
feature_vector = self._extract_features(exp)
features.append(feature_vector)
features = np.array(features)
# 标准化特征
scaler = StandardScaler()
features_scaled = scaler.fit_transform(features)
# DBSCAN聚类
clustering = DBSCAN(eps=0.5, min_samples=3)
labels = clustering.fit_predict(features_scaled)
# 组织聚类结果
clusters = [[] for _ in range(max(labels) + 1)]
for i, label in enumerate(labels):
if label != -1: # 忽略噪声点
clusters[label].append(experiences[i])
return clusters
def _extract_features(self, experience: Experience) -> List[float]:
"""提取经验特征"""
features = []
# 任务类型编码
task_type_encoding = hash(experience.task_type) % 100 / 100.0
features.append(task_type_encoding)
# 奖励值
features.append(experience.reward)
# 上下文复杂度
context_complexity = len(json.dumps(experience.context)) / 1000.0
features.append(min(context_complexity, 1.0))
# 动作复杂度
action_complexity = len(experience.action) / 1000.0
features.append(min(action_complexity, 1.0))
# 结果质量（基于奖励推断）
outcome_quality = max(0.0, min(1.0, experience.reward))
features.append(outcome_quality)
# 时间特征（小时）
hour_feature = experience.timestamp.hour / 24.0
features.append(hour_feature)
return features
async def _identify_patterns(self, clusters: List[List[Experience]]) -> List[Dict[str, Any]]:
"""识别模式"""
patterns = []
for i, cluster in enumerate(clusters):
if len(cluster) < self.min_experiences_for_pattern:
continue
# 分析聚类中的共同特征
pattern = await self._analyze_cluster_pattern(cluster, i)
if pattern:
patterns.append(pattern)
return patterns
async def _analyze_cluster_pattern(self, cluster: List[Experience], cluster_id: int) -> Optional[Dict[str, Any]]:
"""分析聚类模式"""
if not cluster:
return None
# 计算聚类统计信息
rewards = [exp.reward for exp in cluster]
avg_reward = np.mean(rewards)
success_rate = len([r for r in rewards if r > 0]) / len(rewards)
# 识别共同特征
common_features = self._find_common_features(cluster)
# 异常检测
outlier_detector = IsolationForest(contamination=0.1)
feature_matrix = np.array([self._extract_features(exp) for exp in cluster])
if len(feature_matrix) > 10: # 需要足够的数据进行异常检测
outlier_labels = outlier_detector.fit_predict(feature_matrix)
outlier_ratio = np.sum(outlier_labels == -1) / len(outlier_labels)
else:
outlier_ratio = 0.0
# 计算模式质量
pattern_quality = self._calculate_pattern_quality(cluster, avg_reward, success_rate, outlier_ratio)
if pattern_quality < self.pattern_detection_threshold:
return None
pattern = {
"pattern_id": f"pattern_{cluster_id}_{datetime.now().timestamp()}",
"cluster_id": cluster_id,
"frequency": len(cluster),
"success_rate": success_rate,
"average_reward": avg_reward,
"common_features": common_features,
"outlier_ratio": outlier_ratio,
"quality_score": pattern_quality,
"examples": cluster[:5], # 保存前5个示例
"confidence": min(1.0, len(cluster) / 20.0) # 基于样本量的置信度
}
return pattern
def _find_common_features(self, experiences: List[Experience]) -> Dict[str, Any]:
"""寻找共同特征"""
common_features = {}
if not experiences:
return common_features
# 任务类型
task_types = [exp.task_type for exp in experiences]
most_common_task = max(set(task_types), key=task_types.count)
if task_types.count(most_common_task) > len(experiences) * 0.7:
common_features["task_type"] = most_common_task
# 成功/失败模式
successful_experiences = [exp for exp in experiences if exp.reward > 0]
if len(successful_experiences) > len(experiences) * 0.8:
common_features["success_pattern"] = True
# 动作模式
actions = [exp.action for exp in experiences]
action_words = []
for action in actions:
action_words.extend(action.split())
common_words = set()
for word in set(action_words):
if action_words.count(word) > len(actions) * 0.6:
common_words.add(word)
if common_words:
common_features["common_action_words"] = list(common_words)
return common_features
def _calculate_pattern_quality(self, cluster: List[Experience], avg_reward: float, success_rate: float, outlier_ratio: float) -> float:
"""计算模式质量"""
# 基于多个因素计算模式质量
quality_factors = []
# 奖励因子
reward_factor = max(0.0, min(1.0, (avg_reward + 1) / 2)) # 归一化到[0,1]
quality_factors.append(reward_factor)
# 成功率因子
quality_factors.append(success_rate)
# 一致性因子（基于异常值比例）
consistency_factor = 1.0 - outlier_ratio
quality_factors.append(consistency_factor)
# 样本量因子
sample_size_factor = min(1.0, len(cluster) / 20.0)
quality_factors.append(sample_size_factor)
# 综合质量分数
quality_score = np.mean(quality_factors)
return quality_score
async def _optimize_strategies(self, patterns: List[Dict[str, Any]]) -> Dict[str, Any]:
"""优化策略"""
optimized_strategies = {}
for pattern in patterns:
strategy_optimization = await self._optimize_strategy_for_pattern(pattern)
optimized_strategies[pattern["pattern_id"]] = strategy_optimization
return optimized_strategies
async def _optimize_strategy_for_pattern(self, pattern: Dict[str, Any]) -> Dict[str, Any]:
"""为特定模式优化策略"""
# 基于模式特征优化策略参数
optimization = {
"original_pattern": pattern["pattern_id"],
"optimized_parameters": {},
"expected_improvement": 0.0,
"confidence": pattern["confidence"]
}
# 根据成功率调整策略
if pattern["success_rate"] > 0.8:
optimization["optimized_parameters"]["aggressiveness"] = "high"
optimization["expected_improvement"] = 0.15
elif pattern["success_rate"] > 0.6:
optimization["optimized_parameters"]["aggressiveness"] = "medium"
optimization["expected_improvement"] = 0.08
else:
optimization["optimized_parameters"]["aggressiveness"] = "low"
optimization["expected_improvement"] = 0.03
# 根据平均奖励调整参数
if pattern["average_reward"] > 0.5:
optimization["optimized_parameters"]["exploration_rate"] = 0.1
else:
optimization["optimized_parameters"]["exploration_rate"] = 0.3
return optimization
async def _update_learning_patterns(self, patterns: List[Dict[str, Any]], strategies: Dict[str, Any]) -> None:
"""更新学习模式"""
for pattern in patterns:
learning_pattern = LearningPattern(
pattern_id=pattern["pattern_id"],
pattern_type=pattern.get("common_features", {}).get("task_type", "general"),
description=f"Pattern with {pattern['frequency']} examples, {pattern['success_rate']:.2f} success rate",
frequency=pattern["frequency"],
success_rate=pattern["success_rate"],
average_reward=pattern["average_reward"],
confidence=pattern["confidence"],
examples=pattern["examples"],
created_at=datetime.now(),
last_updated=datetime.now()
)
self.learning_patterns[pattern["pattern_id"]] = learning_pattern
# 保存到数据库
await self._save_learning_pattern_to_db(learning_pattern)
async def _save_learning_pattern_to_db(self, pattern: LearningPattern) -> None:
"""保存学习模式到数据库"""
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO learning_patterns
(pattern_id, pattern_type, description, frequency, success_rate, average_reward, confidence, examples, created_at, last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
pattern.pattern_id,
pattern.pattern_type,
pattern.description,
pattern.frequency,
pattern.success_rate,
pattern.average_reward,
pattern.confidence,
json.dumps([exp.to_dict() for exp in pattern.examples]),
pattern.created_at.isoformat(),
pattern.last_updated.isoformat()
))
conn.commit()
conn.close()
async def _calculate_performance_improvement(self, task_type: Optional[str]) -> float:
"""计算性能改进"""
if not task_type or task_type not in self.performance_metrics:
return 0.0
metrics = self.performance_metrics[task_type]
if len(metrics) < 10:
return 0.0
# 比较最近和最早的性能
recent_avg = np.mean(metrics[-20:])
early_avg = np.mean(metrics[:20])
if early_avg == 0:
return 0.0
improvement = (recent_avg - early_avg) / abs(early_avg)
return improvement
def get_learning_recommendations(self, task_type: str, context: Dict[str, Any]) -> List[Dict[str, Any]]:
"""获取学习建议"""
recommendations = []
# 查找相关的学习模式
relevant_patterns = [
pattern for pattern in self.learning_patterns.values()
if pattern.pattern_type == task_type and pattern.confidence > 0.7
]
# 按成功率和置信度排序
relevant_patterns.sort(key=lambda p: (p.success_rate, p.confidence), reverse=True)
for pattern in relevant_patterns[:5]: # 返回前5个建议
recommendation = {
"pattern_id": pattern.pattern_id,
"description": pattern.description,
"success_rate": pattern.success_rate,
"confidence": pattern.confidence,
"recommended_actions": self._extract_recommended_actions(pattern),
"expected_reward": pattern.average_reward
}
recommendations.append(recommendation)
return recommendations
def _extract_recommended_actions(self, pattern: LearningPattern) -> List[str]:
"""提取推荐动作"""
actions = []
for example in pattern.examples[:3]: # 从前3个示例中提取
actions.append(example.action)
return actions
def export_learning_data(self, format: str = "json") -> str:
"""导出学习数据"""
learning_data = {
"total_experiences": len(self.experiences),
"learning_patterns": len(self.learning_patterns),
"performance_metrics": self.performance_metrics,
"export_timestamp": datetime.now().isoformat(),
"experiences_summary": self._generate_experiences_summary(),
"patterns_summary": self._generate_patterns_summary()
}
if format == "json":
return json.dumps(learning_data, indent=2, default=str)
elif format == "pickle":
return pickle.dumps(learning_data)
else:
raise ValueError(f"Unsupported format: {format}")
def _generate_experiences_summary(self) -> Dict[str, Any]:
"""生成经验摘要"""
if not self.experiences:
return {}
task_types = {}
for exp in self.experiences:
if exp.task_type not in task_types:
task_types[exp.task_type] = []
task_types[exp.task_type].append(exp.reward)
summary = {}
for task_type, rewards in task_types.items():
summary[task_type] = {
"count": len(rewards),
"avg_reward": np.mean(rewards),
"success_rate": len([r for r in rewards if r > 0]) / len(rewards)
}
return summary
def _generate_patterns_summary(self) -> Dict[str, Any]:
"""生成模式摘要"""
if not self.learning_patterns:
return {}
summary = {
"total_patterns": len(self.learning_patterns),
"high_confidence_patterns": len([p for p in self.learning_patterns.values() if p.confidence > 0.8]),
"avg_success_rate": np.mean([p.success_rate for p in self.learning_patterns.values()]),
"avg_frequency": np.mean([p.frequency for p in self.learning_patterns.values()])
}
return summary
def cleanup_old_data(self, max_age_days: int = 90) -> int:
"""清理旧数据"""
cutoff_date = datetime.now() - timedelta(days=max_age_days)
cleaned_count = 0
# 清理内存中的经验
original_count = len(self.experiences)
self.experiences = [
exp for exp in self.experiences
if exp.timestamp > cutoff_date
]
cleaned_count += original_count - len(self.experiences)
# 清理数据库
conn = sqlite3.connect(self.experience_db_path)
cursor = conn.cursor()
cursor.execute('DELETE FROM experiences WHERE timestamp < ?', (cutoff_date.isoformat(),))
deleted_experiences = cursor.rowcount
cursor.execute('DELETE FROM learning_patterns WHERE last_updated < ?', (cutoff_date.isoformat(),))
deleted_patterns = cursor.rowcount
conn.commit()
conn.close()
cleaned_count += deleted_experiences + deleted_patterns
return cleaned_count
# 使用示例
async def self_learning_example():
"""自我学习示例"""
# 创建自我学习系统
learning_system = AutoGPTSelfLearningSystem(
experience_db_path="./autogpt_learning.db",
learning_rate=0.01,
pattern_detection_threshold=0.7
)
# 记录一些经验
experiences = []
for i in range(20):
experience = await learning_system.record_experience(
task_type="web_research",
context={"query": f"AI trend {i}", "complexity": i % 5},
action=f"search_and_analyze_{i}",
outcome={"results_found": 10 + i, "quality_score": 0.8 + i * 0.01},
reward=0.7 + i * 0.01,
metadata={"iteration": i, "model_version": "5.0"}
)
experiences.append(experience)
print(f"Recorded {len(experiences)} experiences")
# 从经验中学习
learning_result = await learning_system.learn_from_experiences(
task_type="web_research",
min_experiences=10
)
print(f"Learning result: {learning_result}")
# 获取学习建议
recommendations = learning_system.get_learning_recommendations(
task_type="web_research",
context={"query": "latest AI trends"}
)
print(f"Learning recommendations: {recommendations}")
# 导出学习数据
learning_data = learning_system.export_learning_data()
print(f"Exported learning data length: {len(learning_data)}")
return learning_system
更多推荐
27
0- 0
已为社区贡献4条内容
所有评论(0)