Multi-Agent Orchestration 崛起 2026：从 Hermes 107K 星到 Multica 托管平台，多智能体如何重塑 AI 应用架构

# multi_agent_finance.py
# Multi-Agent 金融分析系统实战

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
import json
import random
import time


class Signal(Enum):
    BUY = "BUY"
    SELL = "SELL"
    HOLD = "HOLD"


@dataclass
class MarketData:
    """模拟市场数据"""
    symbol: str
    price: float
    volume: float
    ma_20: float  # 20日均线
    ma_50: float  # 50日均线
    rsi: float    # RSI指标
    volatility: float  # 波动率


@dataclass
class AgentOutput:
    """Agent 输出"""
    agent_name: str
    signal: Signal
    confidence: float  # 0-1
    reasoning: str
    risk_level: str = "medium"  # low, medium, high


# ============ Agent 基类 ============

class BaseAgent:
    """Agent 基类"""
    
    def __init__(self, name: str):
        self.name = name
        self.history: List[AgentOutput] = []
    
    def analyze(self, data: MarketData) -> AgentOutput:
        raise NotImplementedError


# ============ 分析师 Agent ============

class AnalystAgent(BaseAgent):
    """技术分析 Agent - 负责趋势和信号识别"""
    
    def analyze(self, data: MarketData) -> AgentOutput:
        signals = []
        reasoning_parts = []
        
        # 均线交叉分析
        if data.price > data.ma_20 > data.ma_50:
            signals.append(1)  # 看涨
            reasoning_parts.append("价格在20日和50日均线之上，趋势看涨")
        elif data.price < data.ma_20 < data.ma_50:
            signals.append(-1)  # 看跌
            reasoning_parts.append("价格在20日和50日均线之下，趋势看跌")
        else:
            signals.append(0)  # 中性
            reasoning_parts.append("均线交叉，趋势不明确")
        
        # RSI 分析
        if data.rsi < 30:
            signals.append(1)
            reasoning_parts.append(f"RSI={data.rsi:.1f}，处于超卖区域")
        elif data.rsi > 70:
            signals.append(-1)
            reasoning_parts.append(f"RSI={data.rsi:.1f}，处于超买区域")
        else:
            signals.append(0)
            reasoning_parts.append(f"RSI={data.rsi:.1f}，处于正常区间")
        
        # 成交量分析
        if data.volume > data.ma_20 * 1.5:
            signals.append(1)
            reasoning_parts.append("成交量放大，确认趋势")
        
        # 综合判断
        avg_signal = sum(signals) / len(signals)
        confidence = min(abs(avg_signal) + 0.3, 0.95)
        
        if avg_signal > 0.3:
            signal = Signal.BUY
        elif avg_signal < -0.3:
            signal = Signal.SELL
        else:
            signal = Signal.HOLD
        
        output = AgentOutput(
            agent_name=self.name,
            signal=signal,
            confidence=round(confidence, 2),
            reasoning="; ".join(reasoning_parts),
            risk_level="low" if confidence > 0.7 else "medium"
        )
        self.history.append(output)
        return output


# ============ 风险经理 Agent ============

class RiskManagerAgent(BaseAgent):
    """风险管理 Agent - 负责评估交易风险"""
    
    def __init__(self, name: str, max_drawdown: float = 0.15):
        super().__init__(name)
        self.max_drawdown = max_drawdown
    
    def analyze(self, data: MarketData) -> AgentOutput:
        reasoning_parts = []
        risk_score = 0.0
        
        # 波动率风险
        if data.volatility > 0.03:
            risk_score += 0.4
            reasoning_parts.append(f"波动率 {data.volatility:.2%}，较高风险")
        elif data.volatility > 0.015:
            risk_score += 0.2
            reasoning_parts.append(f"波动率 {data.volatility:.2%}，中等风险")
        else:
            reasoning_parts.append(f"波动率 {data.volatility:.2%}，低风险")
        
        # RSI 极端值风险
        if data.rsi > 80 or data.rsi < 20:
            risk_score += 0.3
            reasoning_parts.append("RSI 处于极端区域，反转风险高")
        
        # 均线偏离度
        ma_deviation = abs(data.price - data.ma_50) / data.ma_50
        if ma_deviation > 0.1:
            risk_score += 0.3
            reasoning_parts.append(f"价格偏离50日均线 {ma_deviation:.1%}，回归风险")
        
        # 风险等级判定
        if risk_score > 0.5:
            risk_level = "high"
            signal = Signal.HOLD
            confidence = 0.9
        elif risk_score > 0.2:
            risk_level = "medium"
            signal = Signal.HOLD
            confidence = 0.6
        else:
            risk_level = "low"
            signal = Signal.BUY
            confidence = 0.7
        
        output = AgentOutput(
            agent_name=self.name,
            signal=signal,
            confidence=round(confidence, 2),
            reasoning="; ".join(reasoning_parts),
            risk_level=risk_level
        )
        self.history.append(output)
        return output


# ============ 策略师 Agent ============

class StrategyAgent(BaseAgent):
    """策略制定 Agent - 综合信号生成最终决策"""
    
    def __init__(self, name: str):
        super().__init__(name)
        self.portfolio: Dict[str, float] = {}
    
    def make_decision(
        self,
        analyst: AgentOutput,
        risk: AgentOutput,
        data: MarketData
    ) -> AgentOutput:
        """综合各 Agent 信号生成最终决策"""
        
        # 风险管理有一票否决权
        if risk.risk_level == "high":
            output = AgentOutput(
                agent_name=self.name,
                signal=Signal.HOLD,
                confidence=0.95,
                reasoning=f"风险经理否决: {risk.reasoning}",
                risk_level="high"
            )
            self.history.append(output)
            return output
        
        # 加权投票
        weights = {"analyst": 0.4, "risk": 0.6}
        vote_score = 0.0
        
        if analyst.signal == Signal.BUY:
            vote_score += weights["analyst"] * analyst.confidence
        elif analyst.signal == Signal.SELL:
            vote_score -= weights["analyst"] * analyst.confidence
            
        if risk.signal == Signal.BUY:
            vote_score += weights["risk"] * risk.confidence
        elif risk.signal == Signal.SELL:
            vote_score -= weights["risk"] * risk.confidence
        
        if vote_score > 0.2:
            signal = Signal.BUY
        elif vote_score < -0.2:
            signal = Signal.SELL
        else:
            signal = Signal.HOLD
        
        confidence = min(abs(vote_score) + 0.4, 0.95)
        
        output = AgentOutput(
            agent_name=self.name,
            signal=signal,
            confidence=round(confidence, 2),
            reasoning=f"分析师: {analyst.signal.value}({analyst.confidence}), "
                       f"风险: {risk.signal.value}({risk.confidence}), "
                       f"综合得分: {vote_score:.2f}",
            risk_level=risk.risk_level
        )
        self.history.append(output)
        return output


# ============ 编排器 ============

class MultiAgentOrchestrator:
    """Multi-Agent 编排器"""
    
    def __init__(self):
        self.analyst = AnalystAgent("技术分析师")
        self.risk_mgr = RiskManagerAgent("风险经理", max_drawdown=0.15)
        self.strategist = StrategyAgent("策略师")
    
    def run_analysis(self, market_data: MarketData) -> Dict:
        """执行完整的 Multi-Agent 分析流程"""
        
        print(f"\n{'='*60}")
        price_str = f"{market_data.price:.2f}"
        print(f"📊 分析 {market_data.symbol} | 价格: " + price_str)
        print(f"{'='*60}")
        
        # Step 1: 分析师分析
        analyst_output = self.analyst.analyze(market_data)
        print(f"\n📈 {analyst_output.agent_name}:")
        print(f"   信号: {analyst_output.signal.value}")
        print(f"   置信度: {analyst_output.confidence}")
        print(f"   理由: {analyst_output.reasoning}")
        
        # Step 2: 风险评估
        risk_output = self.risk_mgr.analyze(market_data)
        print(f"\n⚠️ {risk_output.agent_name}:")
        print(f"   风险等级: {risk_output.risk_level}")
        print(f"   信号: {risk_output.signal.value}")
        print(f"   理由: {risk_output.reasoning}")
        
        # Step 3: 策略决策
        decision = self.strategist.make_decision(
            analyst_output, risk_output, market_data
        )
        print(f"\n🎯 {decision.agent_name}:")
        print(f"   最终决策: {decision.signal.value}")
        print(f"   置信度: {decision.confidence}")
        print(f"   理由: {decision.reasoning}")
        
        return {
            "symbol": market_data.symbol,
            "analyst": vars(analyst_output),
            "risk": vars(risk_output),
            "decision": vars(decision),
        }


# ============ 运行示例 ============

if __name__ == "__main__":
    orchestrator = MultiAgentOrchestrator()
    
    # 测试场景 1: 看涨趋势
    bull_market = MarketData(
        symbol="AAPL", price=195.50, volume=85000000,
        ma_20=190.00, ma_50=185.00, rsi=55.0, volatility=0.012
    )
    orchestrator.run_analysis(bull_market)
    
    # 测试场景 2: 高风险回调
    risky_market = MarketData(
        symbol="TSLA", price=280.00, volume=120000000,
        ma_20=250.00, ma_50=230.00, rsi=82.0, volatility=0.045
    )
    orchestrator.run_analysis(risky_market)
    
    # 测试场景 3: 超卖反弹
    oversold_market = MarketData(
        symbol="NVDA", price=110.00, volume=95000000,
        ma_20=120.00, ma_50=125.00, rsi=25.0, volatility=0.028
    )
    orchestrator.run_analysis(oversold_market)