构建AI智能体：深思熟虑智能体：从BDI架构到认知推理的完整流程体系

┌─────────────┐
│  环境感知    │
└──────┬──────┘
       ↓
┌─────────────┐
│  信念更新    │ ← 更新世界模型
└──────┬──────┘
       ↓
┌─────────────┐
│  愿望生成    │ ← 基于信念激活愿望
└──────┬──────┘
       ↓
┌─────────────┐
│  意图形成    │ ← 选择并承诺于特定愿望
└──────┬──────┘
       ↓
┌─────────────┐
│  计划生成    │ ← 制定实现意图的具体计划
└──────┬──────┘
       ↓
┌─────────────┐
│  行动执行    │ ← 执行计划中的行动
└──────┬──────┘
       ↓
┌─────────────┐
│  结果反思    │ ← 评估结果并学习
└──────┬──────┘
       ↓
┌─────────────┐
│  信念调整    │ ← 基于经验更新信念
└─────────────┘

from enum import Enum

class ResearchStatus(Enum):
    NOT_STARTED = "未开始"
    TOPIC_CONFIRMED = "主题确认"
    LITERATURE_REVIEW = "文献调研"
    # ... 其他状态

class TaskStatus(Enum):
    PENDING = "待执行"
    IN_PROGRESS = "执行中"
    # ... 其他状态

from dataclasses import dataclass

@dataclass
class Belief:
    content: str
    confidence: float
    source: str
    timestamp: float

@dataclass
class TaskResult:
    task_name: str
    status: TaskStatus
    output: Any
    execution_time: float
    lessons_learned: List[str]

def initialize_belief_system(self):
    self.beliefs = {
        'current_topic': None,
        'research_status': ResearchStatus.NOT_STARTED,
        'knowledge_base': {
            'ai_techniques': [
                Belief("机器学习在医疗诊断中应用广泛", 0.9, "领域知识", time.time()),
                # ...
            ]
        }
    }

def initialize_desire_system(self):
    self.desires = {
        'primary_goals': [
            "完成高质量研究报告",
            # ...
        ],
        'quality_standards': {
            'completeness': 0.9,
            # ...
        }
    }

def perceive_environment(self, research_topic: str) -> Dict[str, Any]:
    print(f" {self.name} 正在感知环境...")
    
    perception = {
        'research_topic': research_topic,
        'timestamp': time.time(),
        'complexity': self.assess_topic_complexity(research_topic)
    }
    
    return perception

def assess_topic_complexity(self, topic: str) -> str:
    complexity_keywords = {
        '高': ['趋势', '发展', '现状', '综述', '分析'],
        # ...
    }

def update_beliefs(self, perception: Dict[str, Any]):
    # 更新研究主题
    self.beliefs['current_topic'] = perception['research_topic']
    
    # 更新研究状态
    self.beliefs['research_status'] = ResearchStatus.TOPIC_CONFIRMED
    
    # 添加新的信念
    new_belief = Belief(
        content=f"研究主题'{perception['research_topic']}'复杂度为{perception['complexity']}",
        confidence=0.8,
        source='复杂度评估',
        timestamp=time.time()
    )

def evaluate_goals(self) -> List[str]:
    # 基于当前信念激活相关愿望
    activated_goals = []
    
    if self.beliefs['research_status'] == ResearchStatus.TOPIC_CONFIRMED:
        activated_goals = [
            "收集相关文献资料",
            "分析技术应用现状", 
            # ...
        ]
    
    # 设置目标优先级
    prioritized_goals = self.prioritize_goals(activated_goals)
    return prioritized_goals

def prioritize_goals(self, goals: List[str]) -> List[str]:
    priority_order = {
        "收集相关文献资料": 1,
        "分析技术应用现状": 2,
        # ...
    }

def generate_research_plan(self) -> Dict[str, Any]:
    research_plan = {
        'research_topic': self.beliefs['current_topic'],
        'phases': {
            'phase_1': {
                'name': '文献调研阶段',
                'description': '全面收集和阅读相关文献',
                'tasks': [
                    '检索学术数据库',
                    '收集相关论文',
                    # ...
                ],
                'success_criteria': ['收集30+篇相关论文', '完成文献综述'],
                'time_estimate': 3,
                'resource_allocation': 0.3
            },
            # ... 其他阶段
        }
    }

def execute_plan(self) -> List[TaskResult]:
    all_results = []
    plan = self.intentions['current_plan']
    
    for phase_id, phase_details in plan['phases'].items():
        phase_results = self.execute_phase(phase_id, phase_details)
        all_results.extend(phase_results)
        
        # 更新阶段进度信念
        self.update_phase_progress(phase_id, phase_results)
    
    return all_results

def execute_phase(self, phase_id: str, phase_details: Dict[str, Any]) -> List[TaskResult]:
    phase_results = []
    
    for task in phase_details['tasks']:
        result = self.execute_task(task, phase_id)
        phase_results.append(result)
        
        # 更新任务历史信念
        self.beliefs['task_history'].append({
            'task': task,
            'phase': phase_id,
            'result': result,
            'timestamp': time.time()
        })
    
    return phase_results

def execute_task(self, task: str, phase: str) -> TaskResult:
    task_executors = {
        '检索学术数据库': self.search_databases,
        '收集相关论文': self.collect_papers,
        # ...
    }
    
    executor = task_executors.get(task, self.default_task_executor)
    output = executor()
    
    # 评估任务成功率（模拟）
    success_rate = 0.85 + (hash(task) % 100) * 0.001
    
    status = TaskStatus.COMPLETED if success_rate > 0.8 else TaskStatus.FAILED
    
    return TaskResult(
        task_name=task,
        status=status,
        output=output,
        execution_time=execution_time,
        lessons_learned=[f"任务{task}执行{'成功' if status == TaskStatus.COMPLETED else '失败'}"]
    )

def search_databases(self):
    return {
        'databases_searched': ['CNKI', 'Web of Science', 'IEEE Xplore'],
        'papers_found': 35,
        'keywords_used': ['AI医疗诊断', '深度学习医疗', '智能诊断系统']
    }

def reflect_on_results(self, execution_results: List[TaskResult]):
    # 计算整体成功率
    successful_tasks = [r for r in execution_results if r.status == TaskStatus.COMPLETED]
    overall_success_rate = len(successful_tasks) / len(execution_results)
    
    # 提取经验教训
    lessons = self.extract_lessons(execution_results)
    
    # 更新信念系统
    self.update_beliefs_from_reflection(lessons, overall_success_rate)

def extract_lessons(self, results: List[TaskResult]) -> List[str]:
    lessons = []
    
    # 分析成功任务模式
    successful_tasks = [r for r in results if r.status == TaskStatus.COMPLETED]
    if successful_tasks:
        common_success_patterns = self.identify_success_patterns(successful_tasks)
        lessons.append(f"成功模式: {common_success_patterns}")
    
    return lessons

def run_complete_cycle(self, research_topic: str):
    # 1. 感知环境
    perception = self.perceive_environment(research_topic)
    
    # 2. 更新信念
    self.update_beliefs(perception)
    
    # 3. 评估目标
    goals = self.evaluate_goals()
    
    # 4. 生成计划
    plan = self.generate_research_plan()
    
    # 5. 执行计划
    results = self.execute_plan()
    
    # 6. 反思结果
    reflection = self.reflect_on_results(results)

def generate_final_report(self, reflection: Dict[str, Any]):
    print(f"\n📄 最终研究报告摘要 - {self.beliefs['current_topic']}")
    print("-" * 50)
    
    print(f"研究状态: {self.beliefs['research_status'].value}")
    print(f"任务成功率: {reflection['success_rate']:.1%}")
    
    print("\n📚 知识积累:")
    for category, beliefs in self.beliefs['knowledge_base'].items():
        if beliefs:
            print(f"  {category}: {len(beliefs)}条知识")