Agent 集体决策架构：从 CHAL 到 TradingAgents 的多智能体协作范式

# 单 Agent vs 多 Agent 决策对比示例

# 单 Agent 决策：一个 Agent 完成所有分析
class SingleAgentDecision:
    def decide(self, market_data: dict) -> str:
        analysis = self.analyze_market(market_data)
        risk = self.assess_risk(market_data)
        timing = self.choose_timing(market_data)
        return self.make_final_decision(analysis, risk, timing)

# 多 Agent 集体决策：多个专业 Agent 协作
class CollectiveDecision:
    def decide(self, market_data: dict) -> str:
        # 三个专业 Agent 各自分析
        market_opinion = self.market_analyst.run(market_data)
        risk_opinion = self.risk_analyst.run(market_data)
        timing_opinion = self.timing_analyst.run(market_data)
        
        # 通过投票机制整合意见
        decision = self.voting_mechanism.vote([
            market_opinion, risk_opinion, timing_opinion
        ])
        return decision

# CHAL 三层架构的简化实现
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class AnalysisReport:
    analyst_id: str
    domain: str
    conclusion: str
    confidence: float  # 0.0 - 1.0
    evidence: List[str]

@dataclass
class ManagerRecommendation:
    manager_id: str
    recommendation: str
    confidence: float
    analyst_reports: List[AnalysisReport]
    contradictions: List[str]  # 记录发现的矛盾点

class CHALFramework:
    def __init__(self):
        self.analysts = {}      # 底层分析 Agent 注册表
        self.managers = {}      # 中层管理 Agent 注册表
        self.leader = None      # 顶层决策 Agent
        self.weights = {}       # 动态权重
    
    def run_collective_decision(self, input_data: dict) -> str:
        # 第一层：各 Analyst 独立分析
        reports = []
        for analyst_id, analyst in self.analysts.items():
            report = analyst.analyze(input_data)
            reports.append(report)
        
        # 第二层：Manager 整合分析报告
        recommendations = []
        for manager_id, manager in self.managers.items():
            assigned_reports = manager.get_assigned_reports(reports)
            rec = manager.synthesize(assigned_reports, self.weights)
            recommendations.append(rec)
        
        # 第三层：Leader 做出最终决策
        final_decision = self.leader.decide(recommendations)
        return final_decision

# 动态权重更新机制
class DynamicWeightManager:
    """根据 Agent 历史表现动态调整权重"""
    def __init__(self, decay_factor: float = 0.95):
        self.history: Dict[str, List[bool]] = {}  # agent_id -> [correct/incorrect]
        self.decay = decay_factor
    
    def record_outcome(self, agent_id: str, was_correct: bool):
        if agent_id not in self.history:
            self.history[agent_id] = []
        self.history[agent_id].append(was_correct)
    
    def get_weight(self, agent_id: str) -> float:
        """计算加权准确率（近期表现权重更高）"""
        outcomes = self.history.get(agent_id, [])
        if not outcomes:
            return 1.0  # 新 Agent 默认权重
        weighted_sum = 0.0
        weight_total = 0.0
        for i, correct in enumerate(reversed(outcomes)):
            w = self.decay ** i
            weighted_sum += w * (1 if correct else 0)
            weight_total += w
        return weighted_sum / weight_total if weight_total > 0 else 1.0

# TradingAgents 框架核心流程
class TradingAgentSystem:
    def make_trading_decision(self, portfolio: dict, market_data: dict) -> dict:
        # 阶段 1: 研究员收集信息
        research = self.researcher.collect(
            topics=["macro", "sector", "company", "sentiment"],
            market_data=market_data
        )
        
        # 阶段 2: 分析师生成交易信号
        signals = []
        for analyst in self.analysts:
            signal = analyst.analyze(research)
            signals.append(signal)
        
        # 阶段 3: 分析师辩论
        debate_result = self.debate_protocol.run(signals)
        
        # 阶段 4: 风险评估
        risk_assessment = self.risk_manager.evaluate(
            signals, debate_result, portfolio
        )
        if risk_assessment.exceeds_threshold:
            return {"action": "reject", "reason": risk_assessment.reason}
        
        # 阶段 5: 交易者决策
        decision = self.trader.decide(
            signals, debate_result, risk_assessment, portfolio
        )
        return decision

# 辩论协议简化实现
class DebateProtocol:
    def __init__(self, max_rounds: int = 3):
        self.max_rounds = max_rounds
    
    def run(self, signals: list) -> dict:
        """多轮辩论：多头 vs 空头"""
        bulls = [s for s in signals if s.direction == "buy"]
        bears = [s for s in signals if s.direction == "sell"]
        
        arguments = {"bull": [], "bear": []}
        for round_num in range(self.max_rounds):
            # 多头陈述
            for bull in bulls:
                arg = bull.generate_argument(
                    against=arguments["bear"], round=round_num
                )
                arguments["bull"].append(arg)
            
            # 空头陈述
            for bear in bears:
                arg = bear.generate_argument(
                    against=arguments["bull"], round=round_num
                )
                arguments["bear"].append(arg)
        
        # 辩论总结：提取核心论点
        return {
            "bull_points": self.summarize(arguments["bull"]),
            "bear_points": self.summarize(arguments["bear"]),
            "strength_balance": self.calculate_balance(arguments)
        }

# 三种投票机制的实现对比
from typing import List, Dict
from collections import Counter

class VotingMechanisms:
    @staticmethod
    def majority_vote(opinions: List[str]) -> str:
        """简单多数投票"""
        counts = Counter(opinions)
        return counts.most_common(1)[0][0]
    
    @staticmethod
    def weighted_vote(
        opinions: List[str], 
        weights: List[float]
    ) -> str:
        """加权投票"""
        score: Dict[str, float] = {}
        for opinion, weight in zip(opinions, weights):
            score[opinion] = score.get(opinion, 0) + weight
        return max(score, key=score.get)
    
    @staticmethod
    def borda_count(
        rankings: List[List[str]]  # 每个 Agent 的排序列表
    ) -> str:
        """Borda 计数"""
        n = len(rankings[0])
        scores: Dict[str, float] = {}
        for ranking in rankings:
            for rank, option in enumerate(ranking):
                scores[option] = scores.get(option, 0) + (n - 1 - rank)
        return max(scores, key=scores.get)

# 使用示例
opinions = ["买入", "买入", "卖出", "持有", "买入"]
weights = [0.9, 0.7, 0.8, 0.6, 0.5]
rankings = [
    ["买入", "持有", "卖出"],
    ["持有", "买入", "卖出"],
    ["买入", "卖出", "持有"],
]

print(f"多数投票: {VotingMechanisms.majority_vote(opinions)}")
print(f"加权投票: {VotingMechanisms.weighted_vote(opinions, weights)}")
print(f"Borda计数: {VotingMechanisms.borda_count(rankings)}")

# 共识度阈值机制
class ConsensusThreshold:
    def __init__(self, threshold: float = 0.75):
        self.threshold = threshold
    
    def check_consensus(
        self, 
        opinions: List[str],
        weights: List[float]
    ) -> tuple:
        """检查是否达成共识"""
        total_weight = sum(weights)
        score: Dict[str, float] = {}
        for opinion, weight in zip(opinions, weights):
            score[opinion] = score.get(opinion, 0) + weight
        
        top_opinion, top_score = max(score.items(), key=lambda x: x[1])
        consensus_ratio = top_score / total_weight
        
        if consensus_ratio >= self.threshold:
            return (True, top_opinion, consensus_ratio)
        else:
            return (False, "无共识", consensus_ratio)

# 示例
ct = ConsensusThreshold(threshold=0.75)
result = ct.check_consensus(
    ["买入", "买入", "买入", "卖出"],
    [0.8, 0.7, 0.9, 0.6]
)
# 买入权重和 = 2.4, 总权重 = 3.0, 比例 = 0.8 >= 0.75 → 达成共识
print(f"达成共识: {result[0]}, 结论: {result[1]}, 共识度: {result[2]:.2f}")

# 红蓝对抗决策框架
class RedBlueDebate:
    def __init__(self, red_agent, blue_agent, judge_agent, max_rounds=3):
        self.red = red_agent      # 红队：批判者
        self.blue = blue_agent    # 蓝队：辩护者
        self.judge = judge_agent  # 法官：中立裁决者
        self.max_rounds = max_rounds
    
    def run(self, proposal: str, context: dict) -> dict:
        # 第一轮：立场陈述
        blue_arg = self.blue.defend(proposal, context)
        red_arg = self.red.attack(proposal, context)
        
        rounds = [{"blue": blue_arg, "red": red_arg}]
        
        # 后续轮次：交叉质询
        for round_num in range(1, self.max_rounds):
            prev_blue = rounds[-1]["blue"]
            prev_red = rounds[-1]["red"]
            
            # 蓝队回应红队的质疑
            new_blue = self.blue.rebut(prev_red, prev_blue, context)
            # 红队回应蓝队的辩护
            new_red = self.red.rebut(prev_blue, prev_red, context)
            
            rounds.append({"blue": new_blue, "red": new_red})
        
        # 法官裁决
        verdict = self.judge.adjudicate(rounds, proposal, context)
        return {
            "verdict": verdict,
            "debate_log": rounds,
            "confidence": verdict.confidence
        }

# 集体决策 vs 单 Agent 的性能对比实验框架
from dataclasses import dataclass
import time

@dataclass
class BenchmarkResult:
    task_name: str
    single_agent: dict   # {accuracy, latency_ms, cost_usd, robustness}
    collective: dict     # {accuracy, latency_ms, cost_usd, robustness}
    
    def summarize(self) -> str:
        acc_gain = (
            (self.collective["accuracy"] - self.single_agent["accuracy"])
            / self.single_agent["accuracy"] * 100
        )
        cost_ratio = self.collective["cost_usd"] / self.single_agent["cost_usd"]
        latency_ratio = self.collective["latency_ms"] / self.single_agent["latency_ms"]
        return (
            f"任务: {self.task_name}\n"
            f"  准确率提升: +{acc_gain:.1f}%\n"
            f"  成本倍数: {cost_ratio:.1f}x\n"
            f"  延迟倍数: {latency_ratio:.1f}x"
        )

# 模拟实验结果（基于文献数据）
results = [
    BenchmarkResult("多跳问答",
        {"accuracy": 0.68, "latency_ms": 3000, "cost_usd": 0.05, "robustness": 0.60},
        {"accuracy": 0.82, "latency_ms": 15000, "cost_usd": 0.50, "robustness": 0.85},
    ),
    BenchmarkResult("金融交易",
        {"accuracy": 0.55, "latency_ms": 5000, "cost_usd": 0.10, "robustness": 0.50},
        {"accuracy": 0.70, "latency_ms": 25000, "cost_usd": 1.20, "robustness": 0.78},
    ),
    BenchmarkResult("代码审计",
        {"accuracy": 0.60, "latency_ms": 8000, "cost_usd": 0.15, "robustness": 0.55},
        {"accuracy": 0.78, "latency_ms": 30000, "cost_usd": 1.50, "robustness": 0.88},
    ),
]

for r in results:
    print(r.summarize())
    print()

# 使用 LangGraph 实现集体决策工作流
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class DecisionState(TypedDict):
    input_data: dict
    analyst_reports: list
    risk_score: float
    debate_result: dict
    final_decision: str
    confidence: float

# 定义工作流
workflow = StateGraph(DecisionState)

# 添加节点（每个节点对应一个 Agent）
workflow.add_node("analyst_market", market_analyst_node)
workflow.add_node("analyst_technical", technical_analyst_node)
workflow.add_node("analyst_sentiment", sentiment_analyst_node)
workflow.add_node("risk_manager", risk_manager_node)
workflow.add_node("debate", debate_node)
workflow.add_node("voting", voting_node)
workflow.add_node("final_decision", decision_node)

# 设置边：并行分析
workflow.set_entry_point("analyst_market")
workflow.add_edge("analyst_market", "analyst_technical")
workflow.add_edge("analyst_technical", "analyst_sentiment")
workflow.add_edge("analyst_sentiment", "risk_manager")
workflow.add_edge("risk_manager", "debate")
workflow.add_edge("debate", "voting")
workflow.add_edge("voting", "final_decision")
workflow.add_edge("final_decision", END)

# 编译为可执行图
app = workflow.compile()

# 运行决策流程
result = app.invoke({"input_data": market_data})
print(f"决策: {result['final_decision']}")
print(f"置信度: {result['confidence']}")

// 集体决策系统的监控与日志
interface AgentMetrics {
  agentId: string;
  totalCalls: number;
  avgLatencyMs: number;
  avgTokensUsed: number;
  successRate: number;
  lastError?: string;
}

class CollectiveDecisionMonitor {
  private metrics: Map<string, AgentMetrics> = new Map();
  
  recordCall(agentId: string, latency: number, tokens: number, success: boolean, error?: string) {
    let m = this.metrics.get(agentId) || {
      agentId, totalCalls: 0, avgLatencyMs: 0,
      avgTokensUsed: 0, successRate: 1,
    };
    m.totalCalls++;
    m.avgLatencyMs = (m.avgLatencyMs * (m.totalCalls - 1) + latency) / m.totalCalls;
    m.avgTokensUsed = (m.avgTokensUsed * (m.totalCalls - 1) + tokens) / m.totalCalls;
    m.successRate = (m.successRate * (m.totalCalls - 1) + (success ? 1 : 0)) / m.totalCalls;
    if (error) m.lastError = error;
    this.metrics.set(agentId, m);
  }
  
  // 检查是否有 Agent 表现异常
  getAnomalies(threshold: number = 0.8): AgentMetrics[] {
    return Array.from(this.metrics.values())
      .filter(m => m.successRate < threshold);
  }
}

# 未来方向：自适应角色分配的概念验证
class AdaptiveRoleSystem:
    """根据任务特性动态创建 Agent 角色"""
    def __init__(self, model_pool: dict):
        self.model_pool = model_pool  # {model_name: LLM_instance}
        self.role_templates = {}      # 预定义的角色模板
        self.active_agents = {}       # 当前活跃的 Agent
    
    def analyze_task(self, task: str) -> dict:
        """分析任务特性，确定需要的角色"""
        analysis = self._classify_task(task)
        required_roles = self._determine_roles(analysis)
        return {"task_type": analysis, "roles": required_roles}
    
    def create_agents(self, required_roles: list) -> dict:
        """为每个角色分配合适的模型"""
        agents = {}
        for role in required_roles:
            # 根据角色特性选择最合适的模型
            best_model = self._select_model_for_role(role)
            agent = self._instantiate_agent(role, best_model)
            agents[role] = agent
        return agents
    
    def _select_model_for_role(self, role: str) -> str:
        """角色到模型的映射策略"""
        model_mapping = {
            "analyst": "gpt-4",          # 需要强推理能力
            "researcher": "claude-3",     # 需要长上下文处理
            "risk_manager": "gpt-4",      # 需要高可靠性
            "creative": "claude-3",       # 需要创意生成
            "code_reviewer": "codex",     # 专用代码模型
        }
        return model_mapping.get(role, "gpt-4")