managers

管理层智能体

概述

管理层智能体是 TradingAgents 框架的协调和决策中枢，负责统筹各个专业团队的工作，确保整个投资决策流程的有序进行。管理层包括研究经理和风险经理，分别负责研究活动的协调和风险管理的统筹。

管理层架构

管理层体系设计

class ManagementLayer:
    """管理层系统 - 统筹协调各专业团队"""
    
    def __init__(self, config):
        self.config = config
        self.research_manager = ResearchManager(config)
        self.risk_manager = RiskManager(config)
        self.decision_coordinator = DecisionCoordinator(config)
        
    def orchestrate_decision_process(self, analysis_data: Dict) -> Dict:
        """协调整个决策流程"""
        
        # 1. 研究阶段管理
        research_results = self.research_manager.manage_research_process(analysis_data)
        
        # 2. 风险评估管理
        risk_assessment = self.risk_manager.manage_risk_assessment(
            research_results, analysis_data
        )
        
        # 3. 决策协调
        final_decision = self.decision_coordinator.coordinate_final_decision(
            research_results, risk_assessment
        )
        
        # 4. 质量控制
        quality_check = self._conduct_quality_control(final_decision)
        
        return {
            "research_results": research_results,
            "risk_assessment": risk_assessment,
            "final_decision": final_decision,
            "quality_check": quality_check,
            "management_approval": self._provide_management_approval(final_decision, quality_check)
        }

1. 研究经理 (Research Manager)

职责与功能

class ResearchManager:
    """研究经理 - 协调和管理研究活动"""
    
    def __init__(self, config):
        self.config = config
        self.research_standards = ResearchStandards()
        self.quality_controller = ResearchQualityController()
        self.debate_moderator = DebateModerator()
        
    核心职责:
    - 协调分析师和研究员工作
    - 主持研究员辩论
    - 确保研究质量
    - 形成研究共识
    - 管理研究流程
    
    管理范围:
    - 分析师团队协调
    - 研究员辩论管理
    - 研究质量控制
    - 共识形成机制
    - 研究效率优化

研究流程管理

def manage_research_process(self, analysis_data: Dict) -> Dict:
    """管理研究流程"""
    
    # 1. 分析师工作协调
    analyst_coordination = self._coordinate_analyst_work(analysis_data)
    
    # 2. 分析质量检查
    analysis_quality = self._check_analysis_quality(analyst_coordination)
    
    # 3. 研究员辩论管理
    debate_results = self._manage_researcher_debate(analyst_coordination)
    
    # 4. 共识形成
    research_consensus = self._facilitate_consensus_formation(debate_results)
    
    # 5. 研究报告生成
    research_report = self._generate_research_report(
        analyst_coordination, debate_results, research_consensus
    )
    
    return {
        "analyst_coordination": analyst_coordination,
        "analysis_quality": analysis_quality,
        "debate_results": debate_results,
        "research_consensus": research_consensus,
        "research_report": research_report,
        "research_confidence": self._assess_research_confidence(research_consensus)
    }

def _coordinate_analyst_work(self, analysis_data: Dict) -> Dict:
    """协调分析师工作"""
    
    analyst_reports = analysis_data.get("analyst_reports", {})
    
    coordination_results = {
        "coverage_completeness": self._assess_coverage_completeness(analyst_reports),
        "analysis_consistency": self._check_analysis_consistency(analyst_reports),
        "data_quality": self._evaluate_data_quality(analyst_reports),
        "methodology_alignment": self._check_methodology_alignment(analyst_reports)
    }
    
    # 识别需要补充的分析
    missing_analysis = self._identify_missing_analysis(analyst_reports)
    
    # 协调补充分析
    if missing_analysis:
        coordination_results["supplementary_analysis"] = self._request_supplementary_analysis(
            missing_analysis
        )
    
    # 分析师工作评估
    coordination_results["analyst_performance"] = self._evaluate_analyst_performance(
        analyst_reports
    )
    
    return coordination_results

def _manage_researcher_debate(self, analyst_coordination: Dict) -> Dict:
    """管理研究员辩论"""
    
    # 设置辩论议题
    debate_agenda = self._set_debate_agenda(analyst_coordination)
    
    # 分配辩论角色
    debate_roles = self._assign_debate_roles()
    
    # 主持辩论过程
    debate_process = self._moderate_debate_process(debate_agenda, debate_roles)
    
    # 评估辩论质量
    debate_quality = self._assess_debate_quality(debate_process)
    
    # 提取关键争议点
    key_disagreements = self._extract_key_disagreements(debate_process)
    
    return {
        "debate_agenda": debate_agenda,
        "debate_roles": debate_roles,
        "debate_process": debate_process,
        "debate_quality": debate_quality,
        "key_disagreements": key_disagreements,
        "debate_effectiveness": self._measure_debate_effectiveness(debate_process)
    }

def _facilitate_consensus_formation(self, debate_results: Dict) -> Dict:
    """促进共识形成"""
    
    # 分析辩论结果
    debate_analysis = self._analyze_debate_outcomes(debate_results)
    
    # 识别共同点
    common_ground = self._identify_common_ground(debate_analysis)
    
    # 调解分歧
    mediated_disagreements = self._mediate_disagreements(
        debate_results["key_disagreements"]
    )
    
    # 权衡不同观点
    balanced_perspective = self._create_balanced_perspective(
        common_ground, mediated_disagreements
    )
    
    # 形成最终共识
    final_consensus = self._formulate_final_consensus(balanced_perspective)
    
    return {
        "common_ground": common_ground,
        "mediated_disagreements": mediated_disagreements,
        "balanced_perspective": balanced_perspective,
        "final_consensus": final_consensus,
        "consensus_strength": self._measure_consensus_strength(final_consensus),
        "remaining_uncertainties": self._identify_remaining_uncertainties(final_consensus)
    }