AI Agent 垂直化工作空间：从通用编程助手到专业领域自主智能体

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SEOContentPipeline:
    """seomachine 的 SEO 内容流水线"""
    target_keyword: str
    secondary_keywords: list[str]
    target_audience: str
    content_type: str  # "blog", "guide", "review", "comparison"
    
    # 调研阶段数据
    search_volume: int = 0
    keyword_difficulty: int = 0
    top_competitors: list[dict] = field(default_factory=list)
    search_intent: str = ""  # "informational", "commercial", "transactional"
    
    # 内容阶段数据
    outline: Optional[dict] = None
    draft: Optional[str] = None
    seo_score: float = 0.0  # 0-100
    readability_score: float = 0.0  # Flesch score
    
    # 发布阶段数据
    published_url: Optional[str] = None
    ga4_impressions: int = 0
    ga4_clicks: int = 0
    ga4_ctr: float = 0.0
    
    def calculate_competitive_gap(self) -> dict:
        """计算竞争差距，找出内容机会"""
        if not self.top_competitors:
            return {"opportunity": "no_data"}
        
        # 分析竞品覆盖的主题
        competitor_topics = set()
        for comp in self.top_competitors:
            competitor_topics.update(comp.get("covered_topics", []))
        
        # 找出未被覆盖但有搜索量的子话题
        all_subtopics = self._get_related_subtopics()
        gaps = [t for t in all_subtopics if t not in competitor_topics]
        
        return {
            "content_gaps": gaps[:5],  # Top 5 机会
            "avg_competitor_score": sum(
                c.get("seo_score", 0) for c in self.top_competitors
            ) / len(self.top_competitors),
            "target_score": 85,  # 超越竞品所需分数
        }
    
    def _get_related_subtopics(self) -> list[str]:
        """获取相关子话题（模拟）"""
        return [
            f"{self.target_keyword} 入门指南",
            f"{self.target_keyword} 最佳实践",
            f"{self.target_keyword} vs 竞品对比",
            f"{self.target_keyword} 常见错误",
            f"{self.target_keyword} 工具推荐",
        ]

# 示例：创建一个 SEO 内容项目
pipeline = SEOContentPipeline(
    target_keyword="AI Agent 垂直化",
    secondary_keywords=["专业 AI Agent", "垂直领域 AI", "Agent 工作空间"],
    target_audience="AI 从业者、产品经理",
    content_type="guide",
    search_volume=1200,
    keyword_difficulty=45,
    top_competitors=[
        {"url": "competitor-a.com", "seo_score": 72, "covered_topics": ["入门", "工具"]},
        {"url": "competitor-b.com", "seo_score": 68, "covered_topics": ["入门", "案例"]},
    ],
)

gaps = pipeline.calculate_competitive_gap()
print(f"内容机会: {gaps['content_gaps']}")
print(f"目标分数: {gaps['target_score']}（竞品平均: {gaps['avg_competitor_score']:.0f}）")

import json
from enum import Enum

class EEATCategory(Enum):
    EXPERIENCE = "experience"      # 经验
    EXPERTISE = "expertise"        # 专业性
    AUTHORITATIVENESS = "authoritativeness"  # 权威性
    TRUSTWORTHINESS = "trustworthiness"  # 可信度

class EEATEvaluator:
    """E-E-A-T 内容质量评估器"""
    
    def __init__(self, content: str, metadata: dict):
        self.content = content
        self.metadata = metadata
        self.scores = {}
    
    def evaluate(self) -> dict:
        """执行完整的 E-E-A-T 评估"""
        self.scores["experience"] = self._check_experience()
        self.scores["expertise"] = self._check_expertise()
        self.scores["authoritativeness"] = self._check_authoritativeness()
        self.scores["trustworthiness"] = self._check_trustworthiness()
        
        overall = sum(self.scores.values()) / len(self.scores)
        
        return {
            "overall_score": round(overall, 1),
            "breakdown": {k.value: v for k, v in self.scores.items()},
            "passed": overall >= 70,
            "recommendations": self._get_recommendations(),
        }
    
    def _check_experience(self) -> float:
        """检查经验信号：第一人称叙述、实际案例、个人经历"""
        score = 50.0
        first_person_signals = ["我", "我们", "我们的实践", "我们测试过"]
        case_study_signals = ["案例", "实际", "实测", "我们的客户"]
        
        for signal in first_person_signals:
            if signal in self.content:
                score += 5
        
        for signal in case_study_signals:
            if signal in self.content:
                score += 8
        
        return min(score, 100)
    
    def _check_expertise(self) -> float:
        """检查专业性信号：数据引用、技术深度、专业术语"""
        score = 50.0
        data_signals = ["数据", "统计", "研究表明", "%", "增长率"]
        technical_signals = ["架构", "API", "模型", "算法", "框架"]
        
        for signal in data_signals:
            if signal in self.content:
                score += 4
        
        for signal in technical_signals:
            if signal in self.content:
                score += 3
        
        return min(score, 100)
    
    def _check_authoritativeness(self) -> float:
        """检查权威性信号：外部引用、作者资质、来源可信度"""
        score = 50.0
        if "引用" in self.content or "来源" in self.content:
            score += 15
        if "作者" in self.metadata:
            score += 10
        if "参考文献" in self.content:
            score += 15
        
        return min(score, 100)
    
    def _check_trustworthiness(self) -> float:
        """检查可信度信号：透明度、免责声明、数据时效性"""
        score = 50.0
        if "免责声明" in self.content or "风险提示" in self.content:
            score += 15
        if "更新于" in self.content or "发布日期" in self.content:
            score += 10
        if "数据来源" in self.content:
            score += 15
        
        return min(score, 100)
    
    def _get_recommendations(self) -> list[str]:
        """根据评估结果给出改进建议"""
        recs = []
        if self.scores.get("experience", 0) < 70:
            recs.append("增加第一人称实践经验描述")
        if self.scores.get("expertise", 0) < 70:
            recs.append("增加数据支撑和技术深度分析")
        if self.scores.get("authoritativeness", 0) < 70:
            recs.append("增加外部权威来源引用")
        if self.scores.get("trustworthiness", 0) < 70:
            recs.append("添加数据来源说明和免责声明")
        return recs

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class SignalType(Enum):
    BULLISH = "bullish"
    BEARISH = "bearish"
    NEUTRAL = "neutral"

class StrategyType(Enum):
    MOMENTUM = "momentum"
    MEAN_REVERSION = "mean_reversion"
    ARBITRAGE = "arbitrage"
    SENTIMENT = "sentiment"

@dataclass
class TradingSignal:
    """交易信号"""
    symbol: str
    signal_type: SignalType
    confidence: float  # 0.0 - 1.0
    reasoning: str  # Agent 的推理过程
    target_price: Optional[float] = None
    stop_loss: Optional[float] = None

@dataclass
class TradePosition:
    """交易仓位"""
    symbol: str
    direction: str  # "long" / "short"
    entry_price: float
    quantity: float
    strategy: StrategyType
    stop_loss: float
    take_profit: float
    unrealized_pnl: float = 0.0
    
    def check_risk_limits(self, max_position_pct: float = 0.05) -> bool:
        """检查仓位是否符合风控限制"""
        position_value = self.quantity * self.entry_price
        return position_value <= max_position_pct

class AgentTrader:
    """Agent 自主交易器"""
    
    def __init__(self, initial_capital: float = 1_000_000):
        self.capital = initial_capital
        self.positions: dict[str, TradePosition] = {}
        self.trade_history: list[dict] = []
        self.max_drawdown = 0.0
        self.current_drawdown = 0.0
        self.peak_capital = initial_capital
    
    def process_signal(self, signal: TradingSignal) -> Optional[TradePosition]:
        """处理交易信号，生成交易决策"""
        if signal.confidence < 0.7:
            return None  # 置信度不足，不交易
        
        # Agent 自主决定策略
        strategy = self._select_strategy(signal)
        
        # Agent 自主计算仓位
        position_size = self._calculate_position_size(signal, strategy)
        
        # Agent 自主设置止损止盈
        stop_loss, take_profit = self._set_risk_levels(signal, strategy)
        
        position = TradePosition(
            symbol=signal.symbol,
            direction="long" if signal.signal_type == SignalType.BULLISH else "short",
            entry_price=signal.target_price or 0,
            quantity=position_size,
            strategy=strategy,
            stop_loss=stop_loss,
            take_profit=take_profit,
        )
        
        if position.check_risk_limits():
            self.positions[signal.symbol] = position
            return position
        
        return None
    
    def _select_strategy(self, signal: TradingSignal) -> StrategyType:
        """Agent 自主选择合适的交易策略"""
        if "动量" in signal.reasoning:
            return StrategyType.MOMENTUM
        elif "均值" in signal.reasoning or "回归" in signal.reasoning:
            return StrategyType.MEAN_REVERSION
        elif "套利" in signal.reasoning:
            return StrategyType.ARBITRAGE
        elif "情绪" in signal.reasoning or "舆情" in signal.reasoning:
            return StrategyType.SENTIMENT
        return StrategyType.MOMENTUM  # 默认
    
    def _calculate_position_size(self, signal: TradingSignal, 
                                  strategy: StrategyType) -> float:
        """Agent 自主计算仓位大小（Kelly 公式变体）"""
        win_rate = signal.confidence
        avg_win = 0.02  # 假设平均盈利 2%
        avg_loss = 0.01  # 假设平均亏损 1%
        
        kelly_pct = win_rate - (1 - win_rate) * (avg_loss / avg_win)
        kelly_pct = max(0, min(kelly_pct, 0.25))  # 限制在 0-25%
        
        return self.capital * kelly_pct
    
    def _set_risk_levels(self, signal: TradingSignal,
                         strategy: StrategyType) -> tuple[float, float]:
        """Agent 自主设置止损和止盈"""
        price = signal.target_price or 100
        
        if strategy == StrategyType.MOMENTUM:
            stop_loss = price * 0.97  # 3% 止损
            take_profit = price * 1.06  # 6% 止盈
        elif strategy == StrategyType.MEAN_REVERSION:
            stop_loss = price * 0.95
            take_profit = price * 1.04
        else:
            stop_loss = price * 0.98
            take_profit = price * 1.05
        
        return stop_loss, take_profit