DeepSeek V4 Pro正式版深度解析——Agent能力增强、Responses API与对标Fable 5的技术路径

"""
模拟DeepSeek V4 Pro CSA+HCA混合注意力计算流程
展示从标准全量注意力到混合注意力的计算量缩减
"""
import math
import numpy as np
from typing import Dict, Tuple

class HybridAttentionSimulator:
    """
    混合注意力机制模拟器
    用于对比标准注意力与CSA+HCA的计算开销
    """

    def __init__(
        self,
        seq_len: int = 1_000_000,
        d_model: int = 7168,
        n_heads: int = 64,
        head_dim: int = 128,
    ):
        self.seq_len = seq_len
        self.d_model = d_model
        self.n_heads = n_heads
        self.head_dim = head_dim

    def compute_standard_attention_flops(self) -> float:
        """
        计算标准全量注意力的FLOPs
        公式: 2 * seq_len * d_model * n_heads + 2 * seq_len^2 * n_heads
        """
        # QKV投影
        qkv_proj = 3 * self.seq_len * self.d_model * (self.d_model)
        # 注意力分数
        attn_scores = 2 * self.seq_len * self.seq_len * self.n_heads * self.head_dim
        # 输出投影
        output_proj = self.seq_len * self.d_model * self.d_model

        total_flops = qkv_proj + attn_scores + output_proj
        return total_flops / 1e12  # 转换为TFLOPs

    def compute_csa_flops(self) -> float:
        """
        计算CSA（压缩稀疏注意力）的FLOPs
        CSA流程: KV压缩 → Lightning Indexer → Top-K注意力 → 分组投影
        """
        compression_rate = 4  # 每4个token压缩为1个
        n_compressed = self.seq_len // compression_rate
        top_k = 1024  # 稀疏选择Top-K块

        # Step 1: KV压缩（轻量级attention-like压缩）
        compression_flops = self.seq_len * self.d_model * 64

        # Step 2: Lightning Indexer打分
        indexer_flops = n_compressed * self.d_model * 64

        # Step 3: Core Attention - Multi-Query Attention
        # 只对Top-K块做注意力计算
        sparse_attn_flops = 2 * top_k * self.d_model * self.n_heads

        # Step 4: Grouped Output Projection
        output_proj = self.seq_len * self.d_model * self.d_model // 4

        total = compression_flops + indexer_flops + sparse_attn_flops + output_proj
        return total / 1e12

    def compute_hca_flops(self) -> float:
        """
        计算HCA（重度压缩注意力）的FLOPs
        HCA: 128:1压缩 + 对所有压缩块做dense attention
        """
        hca_compression_rate = 128
        n_hca_blocks = self.seq_len // hca_compression_rate

        # 压缩
        compression_flops = self.seq_len * self.d_model * 64
        # 稠密注意力
        dense_attn_flops = 2 * n_hca_blocks * self.d_model * self.n_heads
        # 输出投影
        output_proj = self.seq_len * self.d_model * self.d_model // 4

        total = compression_flops + dense_attn_flops + output_proj
        return total / 1e12

    def compute_kv_cache(self) -> Dict[str, float]:
        """
        计算KV Cache占用对比
        假设FP16存储（2 bytes per element）
        """
        # 标准KV Cache: 2 * seq_len * d_model * 2 bytes * 2 (K和V)
        standard_kv = 2 * self.seq_len * self.d_model * 2 * 2 / (1024**3)  # GiB

        # 混合KV Cache: 压缩后的CSA + HCA
        csa_kv = 2 * (self.seq_len // 4) * 64 * 2 * 2 / (1024**3)  # CSA压缩KV
        hca_kv = 2 * (self.seq_len // 128) * 64 * 2 * 2 / (1024**3)  # HCA压缩KV
        hybrid_kv = (csa_kv + hca_kv) / 2

        return {
            "standard_kv_gib": round(standard_kv, 1),
            "hybrid_kv_gib": round(hybrid_kv, 1),
            "reduction_ratio": round(hybrid_kv / standard_kv * 100, 1),
        }

    def full_report(self) -> Dict:
        """生成完整对比报告"""
        standard = self.compute_standard_attention_flops()
        csa = self.compute_csa_flops()
        hca = self.compute_hca_flops()
        hybrid = (csa + hca) / 2
        kv = self.compute_kv_cache()

        print("=" * 60)
        print("DeepSeek V4 Pro 混合注意力 FLOPs 对比")
        print("=" * 60)
        print(f"  序列长度: {self.seq_len:,} tokens")
        print(f"  模型维度: {self.d_model}")
        print(f"  注意力头数: {self.n_heads}")
        print("-" * 60)
        print(f"  标准全量注意力: {standard:.2f} TFLOPs")
        print(f"  CSA稀疏注意力:  {csa:.2f} TFLOPs")
        print(f"  HCA稠密注意力:  {hca:.2f} TFLOPs")
        print(f"  CSA+HCA混合:    {hybrid:.2f} TFLOPs")
        print(f"  计算量缩减至:   {hybrid/standard*100:.1f}%")
        print("-" * 60)
        print(f"  KV Cache对比:")
        print(f"    标准: {kv['standard_kv_gib']} GiB")
        print(f"    混合: {kv['hybrid_kv_gib']} GiB")
        print(f"    缩减至: {kv['reduction_ratio']}%")
        print("=" * 60)

        return {
            "standard_tflops": standard,
            "csa_tflops": csa,
            "hca_tflops": hca,
            "hybrid_tflops": hybrid,
            "compression_pct": round(hybrid / standard * 100, 1),
            **kv,
        }

    def simulate_layer_alternation(self, n_layers: int = 80) -> None:
        """
        模拟CSA和HCA层的交替排列
        V4 Pro中两种注意力层交替叠加
        """
        print(f"\n模拟 {n_layers} 层 Transformer 中 CSA/HCA 交替分布:")
        print("-" * 40)

        pattern = []
        for i in range(n_layers):
            if i % 3 == 0:
                layer_type = "CSA"
            elif i % 3 == 1:
                layer_type = "HCA"
            else:
                layer_type = "CSA"

            pattern.append(layer_type)

        # 统计
        csa_count = pattern.count("CSA")
        hca_count = pattern.count("HCA")
        print(f"  CSA层: {csa_count} ({csa_count/n_layers*100:.0f}%)")
        print(f"  HCA层: {hca_count} ({hca_count/n_layers*100:.0f}%)")
        print(f"  层排列: {' → '.join(pattern[:12])}...")

        # 计算每层FLOPs
        total_flops = 0
        for lt in pattern:
            if lt == "CSA":
                total_flops += self.compute_csa_flops()
            else:
                total_flops += self.compute_hca_flops()

        std_total = self.compute_standard_attention_flops() * n_layers
        print(f"  混合注意力总FLOPs: {total_flops:.2f} TFLOPs")
        print(f"  标准注意力总FLOPs: {std_total:.2f} TFLOPs")
        print(f"  总缩减比例: {total_flops/std_total*100:.1f}%")


# 执行模拟
if __name__ == "__main__":
    sim = HybridAttentionSimulator(
        seq_len=1_000_000,
        d_model=7168,
        n_heads=64,
    )
    report = sim.full_report()
    sim.simulate_layer_alternation(n_layers=80)

"""
mHC (Manifold-Constrained Hyper-Connections) 完整实现
包含Sinkhorn-Knopp迭代、多通道映射、梯度稳定性分析
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class ManifoldConstrainedHyperConnection(nn.Module):
    """
    流形约束超连接
    将层间信息流约束在学习的流形上，避免梯度爆炸/消失

    Args:
        d_model: 模型维度
        n_channels: 并行通道数
        sinkhorn_iters: Sinkhorn-Knopp迭代次数
        clamp_value: Sigmoid限幅阈值
    """

    def __init__(
        self,
        d_model: int,
        n_channels: int = 4,
        sinkhorn_iters: int = 20,
        clamp_value: float = 5.0,
    ):
        super().__init__()
        self.d_model = d_model
        self.n_channels = n_channels
        self.sinkhorn_iters = sinkhorn_iters
        self.clamp_value = clamp_value

        # 映射矩阵A: 将输入映射到多通道空间
        self.map_A = nn.Linear(d_model, d_model * n_channels, bias=False)

        # 双随机矩阵B: 通道间混合
        self.B = nn.Parameter(torch.randn(n_channels, n_channels) * 0.1)

        # 映射矩阵C: 将混合后的通道投影回输出空间
        self.map_C = nn.Linear(d_model * n_channels, d_model, bias=False)

        # 可选的层归一化
        self.norm = nn.LayerNorm(d_model)

        # 初始化参数
        self._init_weights()

    def _init_weights(self):
        """初始化权重，确保训练初期稳定"""
        nn.init.orthogonal_(self.map_A.weight, gain=0.5)
        nn.init.orthogonal_(self.map_C.weight, gain=0.5)
        nn.init.eye_(self.B)  # 初始化为单位矩阵

    def _sinkhorn_knopp(self, X: torch.Tensor) -> torch.Tensor:
        """
        Sinkhorn-Knopp迭代生成双随机矩阵
        交替进行行归一化和列归一化

        Args:
            X: [n_channels, n_channels] 输入矩阵

        Returns:
            [n_channels, n_channels] 双随机矩阵
        """
        # 使用softplus确保正值
        X = F.softplus(X)

        for i in range(self.sinkhorn_iters):
            # 行归一化
            X = X / (X.sum(dim=-1, keepdim=True) + 1e-8)
            # 列归一化
            X = X / (X.sum(dim=-2, keepdim=True) + 1e-8)

            # 渐进式clamp防止数值溢出
            if i % 5 == 0:
                X = torch.clamp(X, max=self.clamp_value)

        return X

    def forward(
        self,
        x: torch.Tensor,
        residual: Optional[torch.Tensor] = None,
    ) -> torch.Tensor:
        """
        前向传播

        Args:
            x: [batch, seq_len, d_model] 输入张量
            residual: 可选的残差连接

        Returns:
            [batch, seq_len, d_model] 输出张量
        """
        B, S, D = x.shape

        # 保存输入用于残差
        if residual is None:
            residual = x

        # Step 1: 映射到多通道空间
        # [B, S, D*n_channels]
        channels = self.map_A(x)
        channels = channels.view(B, S, self.n_channels, D)

        # Sigmoid限幅，将值约束在(0,1)区间
        channels = torch.sigmoid(channels)

        # Step 2: 生成双随机矩阵并混合通道
        B_matrix = self._sinkhorn_knopp(self.B)

        # 通道混合: [B,S,n_channels,D] @ [n_channels,n_channels]
        # -> [B,S,n_channels,D]
        mixed = torch.einsum('bsnd,nm->bsmd', channels, B_matrix)

        # 再次Sigmoid限幅
        mixed = torch.sigmoid(mixed)

        # Step 3: 合并通道
        mixed = mixed.reshape(B, S, self.n_channels * D)

        # Step 4: 投影回输出空间
        output = self.map_C(mixed)

        # 残差连接 + LayerNorm
        output = self.norm(residual + output)

        return output

    def compute_gradient_flow(self, x: torch.Tensor) -> Dict[str, float]:
        """
        分析梯度流特性
        用于诊断训练稳定性
        """
        B, S, D = x.shape

        # 前向传播
        channels = self.map_A(x)
        channels = channels.view(B, S, self.n_channels, D)
        channels = torch.sigmoid(channels)

        B_matrix = self._sinkhorn_knopp(self.B)
        mixed = torch.einsum('bsnd,nm->bsmd', channels, B_matrix)
        mixed = torch.sigmoid(mixed)
        mixed = mixed.reshape(B, S, self.n_channels * D)
        output = self.map_C(mixed)

        # 计算梯度范数
        loss = output.norm()
        loss.backward()

        grad_norm = 0.0
        total_params = 0
        for p in self.parameters():
            if p.grad is not None:
                grad_norm += p.grad.norm().item() ** 2
                total_params += p.numel()

        grad_norm = math.sqrt(grad_norm)

        return {
            "gradient_norm": round(grad_norm, 4),
            "gradient_per_param": round(grad_norm / total_params, 6),
            "b_matrix_condition_number": round(
                torch.linalg.cond(B_matrix).item(), 2
            ),
        }


def test_mhc_stability():
    """测试mHC在不同规模下的训练稳定性"""
    print("=" * 60)
    print("mHC 训练稳定性测试")
    print("=" * 60)

    configs = [
        (1024, 2, 10),
        (2048, 4, 15),
        (4096, 4, 20),
        (7168, 4, 20),
        (7168, 8, 20),
    ]

    for d_model, n_channels, sinkhorn_iters in configs:
        mhc = ManifoldConstrainedHyperConnection(
            d_model=d_model,
            n_channels=n_channels,
            sinkhorn_iters=sinkhorn_iters,
        )

        x = torch.randn(2, 128, d_model)
        out = mhc(x)
        grad_info = mhc.compute_gradient_flow(x)

        print(f"\n  d_model={d_model}, channels={n_channels}:")
        print(f"    输出形状: {out.shape}")
        print(f"    梯度范数: {grad_info['gradient_norm']}")
        print(f"    条件数:   {grad_info['b_matrix_condition_number']}")

        # 验证残差连接生效
        assert out.shape == x.shape, "输出形状应与输入一致"
        print(f"    ✓ 残差连接验证通过")

    print("\n" + "=" * 60)
    print("mHC 稳定性测试完成")
    print("=" * 60)


if __name__ == "__main__":
    test_mhc_stability()

    # 展示mHC在V4 Pro中的位置
    print("\n\nmHC在V4 Pro架构中的位置:")
    print("""
    Input Embedding
         ↓
    ┌────────────────────────────┐
    │  Transformer Layer × 80    │
    │  ┌──────────────────────┐  │
    │  │  mHC残差连接          │  │
    │  │    ↓                  │  │
    │  │  RMSNorm              │  │
    │  │    ↓                  │  │
    │  │  CSA/HCA注意力层      │  │
    │  │    ↓                  │  │
    │  │  mHC残差连接          │  │
    │  │    ↓                  │  │
    │  │  RMSNorm              │  │
    │  │    ↓                  │  │
    │  │  MoE FFN (384专家)    │  │
    │  └──────────────────────┘  │
    └────────────────────────────┘
         ↓
    Output Projection
    """)

"""
Agent基准测试数据对比与分析工具
包含V4 Pro预览版、正式版、Fable 5的三方对比
"""
from dataclasses import dataclass
from typing import List, Dict, Optional
import json

@dataclass
class BenchmarkResult:
    """单个基准测试结果"""
    name: str
    description: str
    category: str  # "coding", "reasoning", "agent", "math"
    v4_pro_preview: Optional[float] = None
    v4_pro_ga: Optional[float] = None
    fable_5: Optional[float] = None
    gpt_56_sol: Optional[float] = None

    def v4_improvement(self) -> Optional[float]:
        """V4 Pro从预览版到正式版的提升百分比"""
        if self.v4_pro_preview and self.v4_pro_ga:
            return round(
                (self.v4_pro_ga - self.v4_pro_preview)
                / self.v4_pro_preview * 100, 1
            )
        return None

    def gap_to_fable5(self) -> Optional[float]:
        """与Fable 5的差距"""
        if self.v4_pro_ga and self.fable_5:
            return round(self.v4_pro_ga - self.fable_5, 1)
        return None


class BenchmarkAnalyzer:
    """
    基准测试数据分析器
    支持多维度对比、可视化数据生成
    """

    # 完整的基准测试数据集
    BENCHMARKS = [
        BenchmarkResult(
            "DeepSWE", "软件工程Agent（复杂多步骤代码修改）",
            "agent", 12.8, 62.7, 65.0, 63.0,
        ),
        BenchmarkResult(
            "NL2Repo", "自然语言生成完整代码仓库",
            "agent", 38.5, 61.5, 63.0, 60.0,
        ),
        BenchmarkResult(
            "DSBench-Hard", "全栈高难度编码",
            "coding", 33.0, 67.2, 68.0, 65.0,
        ),
        BenchmarkResult(
            "Terminal Bench 2.1", "终端环境自主操作",
            "agent", 72.1, 87.9, 88.0, 86.5,
        ),
        BenchmarkResult(
            "CyberGym", "网络安全攻防Agent",
            "agent", 52.7, 83.3, 83.1, 82.0,
        ),
        BenchmarkResult(
            "Toolathlon-Verified", "多工具编排与验证",
            "agent", 62.0, 74.1, 76.0, 73.0,
        ),
        BenchmarkResult(
            "DSBench-FullStack", "全栈开发",
            "coding", 55.0, 71.1, 72.0, 69.0,
        ),
        BenchmarkResult(
            "AutomationBench", "端到端工作流自动化",
            "agent", 18.0, 31.8, 30.5, 29.0,
        ),
        BenchmarkResult(
            "Agents' Last Exam", "开放性Agent推理",
            "agent", 18.0, 25.7, 28.0, 26.0,
        ),
        BenchmarkResult(
            "HLE (w/ tools)", "带工具的人类终极考试",
            "reasoning", None, 60.0, 53.3, 55.0,
        ),
        BenchmarkResult(
            "SWE-bench Verified", "软件工程验证基准（第三方）",
            "coding", None, 79.4, 85.0, 83.0,
        ),
        BenchmarkResult(
            "GPQA Diamond", "研究生级问答",
            "reasoning", None, 89.1, 91.0, 90.0,
        ),
        BenchmarkResult(
            "LiveCodeBench COT", "代码生成思维链",
            "coding", None, 89.8, 91.0, 90.0,
        ),
        BenchmarkResult(
            "BrowseComp", "网页浏览与信息提取",
            "agent", None, 80.4, 82.0, 79.0,
        ),
        BenchmarkResult(
            "MRCR 1M", "百万token上下文检索",
            "reasoning", None, 83.3, 85.0, 82.0,
        ),
        BenchmarkResult(
            "HMMT Feb 2026", "高中数学竞赛",
            "math", None, 94.0, 95.0, 93.0,
        ),
        BenchmarkResult(
            "IMOAnswerBench", "国际数学奥赛",
            "math", None, 88.0, 90.0, 87.0,
        ),
        BenchmarkResult(
            "Codeforces Rating", "编程竞赛Rating",
            "coding", None, 2919, 2950, 2900,
        ),
    ]

    def __init__(self):
        self.benchmarks = self.BENCHMARKS

    def get_by_category(self, category: str) -> List[BenchmarkResult]:
        """按类别筛选"""
        return [b for b in self.benchmarks if b.category == category]

    def compute_summary(self) -> Dict:
        """计算汇总统计"""
        agent_bm = self.get_by_category("agent")
        coding_bm = self.get_by_category("coding")

        # Agent类任务：V4 Pro vs Fable 5
        agent_gaps = []
        agent_improvements = []

        for bm in agent_bm:
            gap = bm.gap_to_fable5()
            impr = bm.v4_improvement()
            if gap is not None:
                agent_gaps.append(gap)
            if impr is not None:
                agent_improvements.append(impr)

        # 计算V4 Pro反超项
        beats_fable5 = sum(1 for g in agent_gaps if g > 0)
        total_compared = len(agent_gaps)

        return {
            "agent_avg_gap": round(
                sum(agent_gaps) / len(agent_gaps), 2
            ) if agent_gaps else None,
            "agent_avg_improvement": round(
                sum(agent_improvements) / len(agent_improvements), 1
            ) if agent_improvements else None,
            "beats_fable5_count": beats_fable5,
            "beats_fable5_pct": round(
                beats_fable5 / total_compared * 100, 1
            ) if total_compared else None,
            "total_agent_benchmarks": len(agent_bm),
            "total_coding_benchmarks": len(coding_bm),
        }

    def generate_comparison_table(self) -> str:
        """生成Markdown对比表格"""
        lines = []
        lines.append("| 测试集 | 类别 | 预览版 | 正式版 | Fable 5 | 提升 | 差距 |")
        lines.append("|--------|------|--------|--------|---------|------|------|")

        for bm in sorted(self.benchmarks, key=lambda x: x.category):
            prev = f"{bm.v4_pro_preview}" if bm.v4_pro_preview else "-"
            ga = f"{bm.v4_pro_ga}" if bm.v4_pro_ga else "-"
            f5 = f"{bm.fable_5}" if bm.fable_5 else "-"
            impr = f"+{bm.v4_improvement()}%" if bm.v4_improvement() else "-"
            gap = f"{bm.gap_to_fable5():+.1f}" if bm.gap_to_fable5() is not None else "-"

            # 反超标记
            if bm.gap_to_fable5() is not None and bm.gap_to_fable5() > 0:
                gap = f"**{gap}** ↑"

            lines.append(
                f"| {bm.name} | {bm.category} | {prev} | {ga} | {f5} | {impr} | {gap} |"
            )

        return "\n".join(lines)

    def analyze_agent_capability_breakthrough(self) -> None:
        """
        分析Agent能力突破的关键因素
        从DeepSWE的390%提升切入
        """
        deepswe = [b for b in self.benchmarks if b.name == "DeepSWE"][0]

        print("=" * 60)
        print("Agent能力突破关键分析")
        print("=" * 60)
        print(f"\n1. DeepSWE: {deepswe.v4_pro_preview} → {deepswe.v4_pro_ga}")
        print(f"   提升幅度: {deepswe.v4_improvement()}%")
        print(f"   与Fable 5差距: {deepswe.gap_to_fable5():+.1f}分")

        print(f"\n2. Agent类基准平均提升: {self.compute_summary()['agent_avg_improvement']}%")
        print(f"   反超Fable 5项数: {self.compute_summary()['beats_fable5_count']}/{self.compute_summary()['total_agent_benchmarks']}")

        print(f"\n3. 架构不变下的能力跃升")
        print(f"   说明: 所有提升来自后训练，而非架构改动")
        print(f"   推测策略: 针对性Agent轨迹数据 + 多步骤工具调用强化")

        # 计算排除HLE后的平均差距
        without_hle = [
            b for b in self.benchmarks
            if b.name != "HLE (w/ tools)" and b.gap_to_fable5() is not None
        ]
        avg_gap = sum(b.gap_to_fable5() for b in without_hle) / len(without_hle)
        print(f"\n4. 排除HLE极端值后平均差距: {avg_gap:+.1f}分")
        print(f"   这意味着V4 Pro在多数任务上与Fable 5的差距在3分以内")


# 执行分析
if __name__ == "__main__":
    analyzer = BenchmarkAnalyzer()
    print(analyzer.generate_comparison_table())
    print()
    analyzer.analyze_agent_capability_breakthrough()
    print(f"\n\n汇总统计: {json.dumps(analyzer.compute_summary(), indent=2, ensure_ascii=False)}")

"""
Agent后训练数据构建流水线（模拟DeepSeek策略）
包含多步骤工具调用轨迹生成、长上下文增强、质量过滤
"""
import json
import random
from typing import List, Dict, Any, Optional, Generator
from dataclasses import dataclass, field
from enum import Enum


class ToolType(Enum):
    """Agent可用的工具类型"""
    BASH = "bash_execute"
    FILE_READ = "file_read"
    FILE_WRITE = "file_write"
    FILE_EDIT = "file_edit"
    WEB_SEARCH = "web_search"
    CODE_REVIEW = "code_review"
    TEST_RUNNER = "test_runner"
    GIT_OPS = "git_operations"
    PACKAGE_MANAGER = "package_manager"
    DATABASE_QUERY = "database_query"


@dataclass
class ToolCall:
    """单次工具调用"""
    name: str
    arguments: Dict[str, Any]
    thought: str
    observation: str
    reflection: str
    duration_ms: int = 0
    success: bool = True


@dataclass
class AgentTrajectory:
    """完整的Agent执行轨迹"""
    task: str
    task_type: str
    steps: List[ToolCall] = field(default_factory=list)
    context_tokens: int = 0
    total_tokens: int = 0
    final_answer: Optional[str] = None

    def add_step(self, call: ToolCall):
        self.steps.append(call)
        self.total_tokens += len(json.dumps(call.__dict__))

    def is_complete(self) -> bool:
        return self.final_answer is not None and len(self.steps) > 0

    def tool_diversity(self) -> float:
        """工具使用多样性得分"""
        if not self.steps:
            return 0.0
        tools_used = set(s.name for s in self.steps)
        return len(tools_used) / len(ToolType)


class TrajectoryGenerator:
    """
    Agent轨迹生成器
    生成多样化的多步骤工具调用轨迹
    """

    TASKS = [
        ("Fix the bug in the authentication module", "bug_fix"),
        ("Implement a new REST API endpoint for user management", "feature"),
        ("Refactor the database connection pool to use connection pooling", "refactor"),
        ("Write comprehensive unit tests for the payment service", "testing"),
        ("Optimize the query performance of the search endpoint", "optimization"),
        ("Migrate the CI/CD pipeline to GitHub Actions", "devops"),
        ("Implement error handling for the file upload service", "feature"),
        ("Add distributed tracing to the microservice", "observability"),
        ("Create a Helm chart for Kubernetes deployment", "devops"),
        ("Implement a caching layer for the product catalog API", "optimization"),
    ]

    TOOLS_WITH_PARAMS = {
        ToolType.BASH: lambda: {
            "command": random.choice([
                "ls -la", "pwd", "grep -r 'error' .",
                "python3 -m pytest tests/", "npm test",
                "docker-compose up -d", "kubectl get pods",
            ]),
            "timeout": 30,
            "working_directory": "/workspace/project",
        },
        ToolType.FILE_READ: lambda: {
            "file_path": random.choice([
                "src/auth/login.py", "src/api/users.py",
                "config/database.yml", "docker-compose.yml",
                "README.md", "tests/test_auth.py",
            ]),
        },
        ToolType.FILE_WRITE: lambda: {
            "file_path": f"src/{random.choice(['api', 'auth', 'core', 'utils'])}/{random.choice(['handlers', 'models', 'services', 'middleware'])}.py",
            "content": f"# Auto-generated by agent\n# Task: {random.choice(['fix bug', 'add feature', 'refactor'])}\n",
        },
        ToolType.TEST_RUNNER: lambda: {
            "test_path": random.choice([
                "tests/", "tests/unit/", "tests/integration/",
            ]),
            "test_pattern": "test_*.py",
            "verbose": True,
        },
        ToolType.GIT_OPS: lambda: {
            "action": random.choice(["status", "diff", "log", "branch"]),
            "arguments": {},
        },
    }

    @classmethod
    def generate_trajectory(cls, min_steps: int = 3, max_steps: int = 12) -> AgentTrajectory:
        """生成一条Agent轨迹"""
        task, task_type = random.choice(cls.TASKS)
        n_steps = random.randint(min_steps, max_steps)
        trajectory = AgentTrajectory(task=task, task_type=task_type)

        # 模拟思考-行动-观察循环
        for i in range(n_steps):
            tool_type = random.choice(list(ToolType))
            tool_name = tool_type.value
            tool_params = cls.TOOLS_WITH_PARAMS.get(tool_type, lambda: {})()

            thought_templates = [
                f"我需要先了解当前状态，使用{tool_name}来检查",
                f"下一步，使用{tool_name}来{task}",
                f"分析结果后，我需要用{tool_name}进行修改",
                f"验证修改是否正确，使用{tool_name}",
                f"如果出错，用{tool_name}回滚并重试",
            ]

            call = ToolCall(
                name=tool_name,
                arguments=tool_params,
                thought=random.choice(thought_templates),
                observation=f"Tool {tool_name} executed successfully, returned: step {i} of {n_steps}",
                reflection=f"Step {i+1}/{n_steps}: {'progressing well' if random.random() > 0.2 else 'encountered issue, retrying'}",
                duration_ms=random.randint(100, 5000),
                success=random.random() > 0.15,
            )
            trajectory.add_step(call)

        trajectory.final_answer = f"Task '{task}' completed. Summary: {random.choice(['All tests pass', 'Performance improved by 30%', 'Bug fixed', 'Feature implemented'])}"
        return trajectory

    @classmethod
    def generate_dataset(
        cls, n: int = 10000, min_steps: int = 3, max_steps: int = 15
    ) -> List[AgentTrajectory]:
        """批量生成轨迹数据集"""
        return [
            cls.generate_trajectory(min_steps, max_steps)
            for _ in range(n)
        ]


class TrajectoryFilter:
    """
    轨迹质量过滤器
    保留高质量、多步骤、工具调用完整的轨迹
    """

    @staticmethod
    def quality_score(trajectory: AgentTrajectory) -> float:
        """计算轨迹质量分数 (0-100)"""
        score = 0.0

        # 步骤数
        n_steps = len(trajectory.steps)
        if n_steps >= 5:
            score += 20
        elif n_steps >= 3:
            score += 10

        # 工具多样性
        diversity = trajectory.tool_diversity()
        score += diversity * 30

        # 任务完成
        if trajectory.is_complete():
            score += 20

        # 思考链完整性
        has_reflection = all(
            s.reflection and len(s.reflection) > 10
            for s in trajectory.steps
        )
        if has_reflection:
            score += 15

        # 成功率
        success_rate = sum(1 for s in trajectory.steps if s.success) / n_steps
        score += success_rate * 15

        return round(score, 1)

    @classmethod
    def filter_dataset(
        cls,
        trajectories: List[AgentTrajectory],
        min_score: float = 50.0,
        min_steps: int = 3,
    ) -> List[AgentTrajectory]:
        """过滤数据集"""
        filtered = []
        for t in trajectories:
            if len(t.steps) < min_steps:
                continue
            if cls.quality_score(t) < min_score:
                continue
            filtered.append(t)
        return filtered


# 执行数据流水线
if __name__ == "__main__":
    print("=" * 60)
    print("Agent后训练数据流水线模拟")
    print("=" * 60)

    # 1. 生成轨迹
    print("\n1. 生成轨迹数据...")
    raw_data = TrajectoryGenerator.generate_dataset(n=5000, min_steps=3, max_steps=12)
    print(f"   生成轨迹数: {len(raw_data)}")

    # 2. 质量过滤
    print("\n2. 质量过滤...")
    filtered = TrajectoryFilter.filter_dataset(raw_data, min_score=50)
    print(f"   过滤后: {len(filtered)}")
    print(f"   保留率: {len(filtered)/len(raw_data)*100:.1f}%")

    # 3. 统计信息
    print("\n3. 数据集统计:")
    avg_steps = sum(len(t.steps) for t in filtered) / len(filtered)
    avg_diversity = sum(t.tool_diversity() for t in filtered) / len(filtered)
    avg_score = sum(TrajectoryFilter.quality_score(t) for t in filtered) / len(filtered)
    print(f"   平均步骤数: {avg_steps:.1f}")
    print(f"   平均工具多样性: {avg_diversity:.2f}")
    print(f"   平均质量分数: {avg_score:.1f}")

    # 4. 任务类型分布
    print("\n4. 任务类型分布:")
    task_types = {}
    for t in filtered:
        task_types[t.task_type] = task_types.get(t.task_type, 0) + 1
    for tt, count in sorted(task_types.items(), key=lambda x: -x[1]):
        print(f"   {tt}: {count} ({count/len(filtered)*100:.1f}%)")

"""
三档推理强度控制实战
包含动态调度策略、成本追踪、性能基准测试
"""
from openai import OpenAI
from enum import Enum
from dataclasses import dataclass
from typing import Dict, Optional
import time
import json


class ReasoningEffort(Enum):
    """推理强度枚举"""
    NONE = "none"
    HIGH = "high"
    MAX = "max"


@dataclass
class TaskResult:
    """任务执行结果"""
    task_type: str
    effort: ReasoningEffort
    content: str
    reasoning_content: Optional[str]
    latency_ms: int
    tokens_input: int
    tokens_output: int
    tokens_reasoning: int
    cost: float


class DynamicReasoningRouter:
    """
    动态推理强度路由
    基于任务复杂度自动选择最优推理强度
    """

    # 定价（元/百万token）
    PRICING = {
        "input": 3.0,
        "output": 6.0,
    }

    # 任务类型到推理强度的映射规则
    TASK_EFFORT_MAP: Dict[str, ReasoningEffort] = {
        "simple_qa": ReasoningEffort.NONE,
        "code_completion": ReasoningEffort.NONE,
        "text_formatting": ReasoningEffort.NONE,
        "data_extraction": ReasoningEffort.NONE,
        "code_review": ReasoningEffort.HIGH,
        "debugging": ReasoningEffort.HIGH,
        "multi_step_reasoning": ReasoningEffort.HIGH,
        "code_generation": ReasoningEffort.HIGH,
        "architecture_design": ReasoningEffort.MAX,
        "competitive_programming": ReasoningEffort.MAX,
        "long_horizon_agent": ReasoningEffort.MAX,
        "complex_analysis": ReasoningEffort.MAX,
    }

    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.deepseek.com/v1",
        )
        self.stats = {
            "total_calls": 0,
            "total_cost": 0.0,
            "total_latency_ms": 0,
            "effort_distribution": {},
        }

    def route(
        self,
        task_type: str,
        prompt: str,
        system_prompt: Optional[str] = None,
        max_tokens: int = 4096,
    ) -> TaskResult:
        """根据任务类型路由到合适的推理强度"""
        effort = self.TASK_EFFORT_MAP.get(task_type, ReasoningEffort.HIGH)
        return self._call(prompt, effort, system_prompt, max_tokens)

    def _call(
        self,
        prompt: str,
        effort: ReasoningEffort,
        system_prompt: Optional[str] = None,
        max_tokens: int = 4096,
    ) -> TaskResult:
        """执行API调用"""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        start = time.time()

        response = self.client.chat.completions.create(
            model="deepseek-v4-pro",
            reasoning_effort=effort.value,
            messages=messages,
            max_tokens=max_tokens,
        )

        latency_ms = int((time.time() - start) * 1000)
        choice = response.choices[0]
        usage = response.usage

        # 获取思考链内容
        reasoning_content = None
        if hasattr(choice.message, 'reasoning_content'):
            reasoning_content = choice.message.reasoning_content

        # 计算成本
        tokens_input = usage.prompt_tokens
        tokens_output = usage.completion_tokens
        tokens_reasoning = len(reasoning_content.split()) if reasoning_content else 0

        cost = (
            tokens_input * self.PRICING["input"]
            + tokens_output * self.PRICING["output"]
        ) / 1_000_000

        # 更新统计
        self.stats["total_calls"] += 1
        self.stats["total_cost"] += cost
        self.stats["total_latency_ms"] += latency_ms
        self.stats["effort_distribution"][effort.value] = (
            self.stats["effort_distribution"].get(effort.value, 0) + 1
        )

        return TaskResult(
            task_type="unknown",
            effort=effort,
            content=choice.message.content,
            reasoning_content=reasoning_content,
            latency_ms=latency_ms,
            tokens_input=tokens_input,
            tokens_output=tokens_output,
            tokens_reasoning=tokens_reasoning,
            cost=cost,
        )

    def print_stats(self):
        """打印统计信息"""
        print("\n" + "=" * 50)
        print("推理路由统计")
        print("=" * 50)
        print(f"总调用次数: {self.stats['total_calls']}")
        print(f"总成本: ¥{self.stats['total_cost']:.4f}")
        print(f"平均延迟: {self.stats['total_latency_ms']/max(self.stats['total_calls'],1):.0f}ms")
        print(f"\n强度分布:")
        for effort, count in self.stats["effort_distribution"].items():
            print(f"  {effort}: {count} ({count/self.stats['total_calls']*100:.1f}%)")


# 成本效益对比测试
def benchmark_effort_levels():
    """对比不同推理强度的成本和延迟"""
    print("=" * 60)
    print("推理强度成本效益对比")
    print("=" * 60)

    test_prompts = [
        ("简单问答", "Python中list和tuple的区别是什么？", ReasoningEffort.NONE),
        ("代码审查", "审查以下代码的性能问题：\n```python\ndef process(data):\n    result = []\n    for i in range(len(data)):\n        for j in range(len(data)):\n            result.append(data[i] + data[j])\n    return result\n```", ReasoningEffort.HIGH),
        ("架构设计", "设计一个支持每日10亿请求的实时推荐系统架构，包含数据管道、模型服务、AB测试等组件", ReasoningEffort.MAX),
    ]

    for task_name, prompt, effort in test_prompts:
        messages = [{"role": "user", "content": prompt}]
        start = time.time()

        # 模拟调用（不实际调API，展示思路）
        time.sleep(0.1)  # 模拟延迟

        # 估算token消耗
        input_tokens = len(prompt) // 2
        if effort == ReasoningEffort.NONE:
            output_tokens = 200
            reasoning_tokens = 0
        elif effort == ReasoningEffort.HIGH:
            output_tokens = 800
            reasoning_tokens = 600
        else:
            output_tokens = 2000
            reasoning_tokens = 4000

        cost = (input_tokens * 3 + output_tokens * 6) / 1_000_000
        latency = int((time.time() - start) * 1000)

        print(f"\n{task_name} ({effort.value}):")
        print(f"  输入token: {input_tokens}")
        print(f"  输出token: {output_tokens}")
        print(f"  思考链token: {reasoning_tokens}")
        print(f"  成本: ¥{cost:.4f}")
        print(f"  延迟: {latency}ms")


if __name__ == "__main__":
    # 创建路由
    router = DynamicReasoningRouter(api_key="sk-your-key")

    # 模拟不同任务
    tasks = [
        ("simple_qa", "Python的GIL是什么？"),
        ("code_review", "审查这段代码：\ndef get_user(user_id):\n    return db.query(f\"SELECT * FROM users WHERE id = {user_id}\")"),
        ("architecture_design", "设计一个微服务架构的API网关"),
    ]

    for task_type, prompt in tasks:
        result = router.route(task_type, prompt)
        effort_label = result.effort.value
        print(f"[{task_type}] → {effort_label}: {len(result.content)} chars, ¥{result.cost:.4f}")

    router.print_stats()

    # 运行基准测试
    benchmark_effort_levels()

"""
DeepSeek V4 Pro Responses API Agent框架完整实现
包含：工具定义、工具调用链、多轮对话管理、状态持久化
"""
from openai import OpenAI
import json
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum


class ToolCallStatus(Enum):
    """工具调用状态"""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"


@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    parameters: Dict[str, Any]

    def to_openai_format(self) -> Dict:
        """转换为OpenAI工具格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": self.parameters,
                    "required": list(self.parameters.keys()),
                },
            },
        }


@dataclass
class ConversationState:
    """对话状态"""
    messages: List[Dict] = field(default_factory=list)
    current_turn: int = 0
    max_turns: int = 20
    tool_results: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)


class V4ProAgent:
    """
    基于Responses API的完整Agent框架
    支持多轮工具调用、状态管理、错误恢复
    """

    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.tools: Dict[str, ToolDefinition] = {}
        self.tool_handlers: Dict[str, Callable] = {}
        self._register_default_tools()

    def _register_default_tools(self):
        """注册默认工具集"""
        default_tools = [
            ToolDefinition(
                "bash",
                "Execute a shell command in the workspace",
                {
                    "command": {
                        "type": "string",
                        "description": "Shell command to execute",
                    },
                    "timeout": {
                        "type": "integer",
                        "description": "Timeout in seconds",
                        "default": 30,
                    },
                },
            ),
            ToolDefinition(
                "read",
                "Read the contents of a file",
                {
                    "file_path": {
                        "type": "string",
                        "description": "Path to the file",
                    },
                    "offset": {
                        "type": "integer",
                        "description": "Starting line (optional)",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Number of lines (optional)",
                    },
                },
            ),
            ToolDefinition(
                "write",
                "Write content to a file",
                {
                    "file_path": {
                        "type": "string",
                        "description": "Path to the file",
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write",
                    },
                },
            ),
            ToolDefinition(
                "edit",
                "Edit a file using string replacement",
                {
                    "file_path": {
                        "type": "string",
                        "description": "Path to the file",
                    },
                    "old_string": {
                        "type": "string",
                        "description": "String to replace",
                    },
                    "new_string": {
                        "type": "string",
                        "description": "Replacement string",
                    },
                },
            ),
            ToolDefinition(
                "search",
                "Search the web for information",
                {
                    "query": {
                        "type": "string",
                        "description": "Search query",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum results",
                        "default": 5,
                    },
                },
            ),
        ]

        for tool in default_tools:
            self.register_tool(tool)

    def register_tool(self, tool: ToolDefinition, handler: Optional[Callable] = None):
        """注册工具"""
        self.tools[tool.name] = tool
        if handler:
            self.tool_handlers[tool.name] = handler

    def run(
        self,
        task: str,
        system_prompt: Optional[str] = None,
        reasoning_effort: str = "high",
        max_turns: int = 15,
    ) -> str:
        """
        运行Agent任务，自动处理多轮工具调用

        Args:
            task: 任务描述
            system_prompt: 系统提示词
            reasoning_effort: 推理强度
            max_turns: 最大轮数

        Returns:
            最终答案
        """
        state = ConversationState(max_turns=max_turns)

        if system_prompt:
            state.messages.append({
                "role": "system",
                "content": system_prompt,
            })

        state.messages.append({"role": "user", "content": task})

        for turn in range(max_turns):
            print(f"\n{'='*50}")
            print(f"Turn {turn + 1}/{max_turns}")
            print(f"{'='*50}")

            # 调用Responses API
            response = self.client.responses.create(
                model="deepseek-v4-pro",
                input=state.messages,
                tools=[t.to_openai_format() for t in self.tools.values()],
                tool_choice="auto",
                reasoning=reasoning_effort,
            )

            output = response.output

            # 检查是否有工具调用
            tool_calls = []
            if hasattr(output, 'tool_calls') and output.tool_calls:
                tool_calls = output.tool_calls
            elif isinstance(output, list):
                for item in output:
                    if hasattr(item, 'type') and item.type == 'function_call':
                        tool_calls.append(item)

            if not tool_calls:
                # 没有工具调用，返回最终答案
                if hasattr(output, 'output_text'):
                    return output.output_text
                elif isinstance(output, list):
                    texts = [o.text for o in output if hasattr(o, 'text')]
                    return "\n".join(texts)
                return str(output)

            # 处理工具调用
            for tool_call in tool_calls:
                tool_name = tool_call.function.name
                try:
                    tool_args = json.loads(tool_call.function.arguments)
                except json.JSONDecodeError:
                    tool_args = {}

                print(f"  🔧 Calling: {tool_name}({json.dumps(tool_args, ensure_ascii=False)[:100]})")

                # 执行工具
                handler = self.tool_handlers.get(tool_name)
                if handler:
                    try:
                        result = handler(**tool_args)
                        tool_result = str(result)
                    except Exception as e:
                        tool_result = f"Error: {str(e)}"
                else:
                    tool_result = f"[Simulated] Tool {tool_name} executed with args: {tool_args}"

                print(f"  ✅ Result: {tool_result[:100]}")

                # 将工具调用和结果加入对话
                state.messages.append({
                    "role": "assistant",
                    "content": None,
                    "tool_calls": [{
                        "id": tool_call.id,
                        "type": "function",
                        "function": {
                            "name": tool_name,
                            "arguments": json.dumps(tool_args),
                        },
                    }],
                })

                state.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                })

            state.current_turn = turn + 1

        return "Agent reached maximum turns without completing the task."


# 完整示例：代码审查Agent
def code_review_agent_example():
    """代码审查Agent示例"""
    agent = V4ProAgent(api_key="sk-your-key")

    task = """
    请审查以下代码，完成以下步骤：
    1. 读取 src/auth.py 文件
    2. 分析代码中的安全漏洞
    3. 编写修复后的代码
    4. 运行测试验证修复
    """

    result = agent.run(
        task=task,
        system_prompt="你是一个资深安全工程师，擅长代码审查和安全漏洞修复。",
        reasoning_effort="high",
        max_turns=10,
    )

    print(f"\n{'='*50}")
    print("最终结果:")
    print(f"{'='*50}")
    print(result)
    return result


if __name__ == "__main__":
    # 运行示例
    # code_review_agent_example()

    # 展示工具定义
    agent = V4ProAgent(api_key="sk-demo")
    print("注册的工具:")
    for name, tool in agent.tools.items():
        print(f"  - {name}: {tool.description}")

// DeepSeek V4 Pro Anthropic API兼容网关
// 将Anthropic Messages协议转换为DeepSeek内部格式
// 支持Claude Code、Cursor等工具无缝接入
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"sync"
	"time"
)

// Anthropic消息格式
type AnthropicMessage struct {
	Model     string           `json:"model"`
	MaxTokens int              `json:"max_tokens"`
	System    string           `json:"system,omitempty"`
	Messages  []AnthropicTurn  `json:"messages"`
	Tools     []AnthropicTool  `json:"tools,omitempty"`
	Stream    bool             `json:"stream,omitempty"`
}

type AnthropicTurn struct {
	Role    string             `json:"role"`
	Content []AnthropicContent `json:"content"`
}

type AnthropicContent struct {
	Type  string          `json:"type"`
	Text  string          `json:"text,omitempty"`
	ID    string          `json:"id,omitempty"`
	Name  string          `json:"name,omitempty"`
	Input json.RawMessage `json:"input,omitempty"`
}

type AnthropicTool struct {
	Name        string      `json:"name"`
	Description string      `json:"description"`
	InputSchema interface{} `json:"input_schema"`
}

// DeepSeek Chat格式
type DeepSeekRequest struct {
	Model       string           `json:"model"`
	Messages    []DeepSeekMsg    `json:"messages"`
	MaxTokens   int              `json:"max_tokens,omitempty"`
	Stream      bool             `json:"stream,omitempty"`
	Temperature float64          `json:"temperature,omitempty"`
	TopP        float64          `json:"top_p,omitempty"`
	Tools       []DeepSeekTool   `json:"tools,omitempty"`
}

type DeepSeekMsg struct {
	Role       string     `json:"role"`
	Content    string     `json:"content"`
	ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
	ToolCallID string     `json:"tool_call_id,omitempty"`
}

type DeepSeekTool struct {
	Type     string          `json:"type"`
	Function json.RawMessage `json:"function"`
}

type ToolCall struct {
	ID       string `json:"id"`
	Type     string `json:"type"`
	Function struct {
		Name      string `json:"name"`
		Arguments string `json:"arguments"`
	} `json:"function"`
}

type DeepSeekResponse struct {
	ID      string          `json:"id"`
	Object  string          `json:"object"`
	Created int64           `json:"created"`
	Choices []DeepSeekChoice `json:"choices"`
	Usage   *Usage          `json:"usage,omitempty"`
}

type DeepSeekChoice struct {
	Index   int            `json:"index"`
	Message DeepSeekMsgRes `json:"message"`
}

type DeepSeekMsgRes struct {
	Role             string     `json:"role"`
	Content          string     `json:"content"`
	ReasoningContent string     `json:"reasoning_content,omitempty"`
	ToolCalls        []ToolCall `json:"tool_calls,omitempty"`
}

type Usage struct {
	PromptTokens     int `json:"prompt_tokens"`
	CompletionTokens int `json:"completion_tokens"`
	TotalTokens      int `json:"total_tokens"`
}

// 协议转换网关
type AnthropicGateway struct {
	deepseekURL string
	apiKey      string
	client      *http.Client
	metrics     *GatewayMetrics
}

type GatewayMetrics struct {
	mu              sync.RWMutex
	requestsTotal   int64
	requestsSuccess int64
	requestsFailed  int64
	latencyTotalMs  int64
}

func NewGateway(apiKey string) *AnthropicGateway {
	return &AnthropicGateway{
		deepseekURL: "https://api.deepseek.com/v1",
		apiKey:      apiKey,
		client: &http.Client{
			Timeout: 180 * time.Second,
			Transport: &http.Transport{
				MaxIdleConns:        100,
				MaxIdleConnsPerHost: 20,
				IdleConnTimeout:     90 * time.Second,
			},
		},
		metrics: &GatewayMetrics{},
	}
}

func (g *AnthropicGateway) ConvertToDeepSeek(anthReq *AnthropicMessage) *DeepSeekRequest {
	dsReq := &DeepSeekRequest{
		Model:       "deepseek-v4-pro",
		MaxTokens:   anthReq.MaxTokens,
		Stream:      anthReq.Stream,
		Temperature: 0.7,
		TopP:        0.95,
	}

	// 转换System Prompt
	if anthReq.System != "" {
		dsReq.Messages = append(dsReq.Messages, DeepSeekMsg{
			Role:    "system",
			Content: anthReq.System,
		})
	}

	// 转换Tools
	if len(anthReq.Tools) > 0 {
		for _, t := range anthReq.Tools {
			fn, _ := json.Marshal(map[string]interface{}{
				"name":        t.Name,
				"description": t.Description,
				"parameters":  t.InputSchema,
			})
			dsReq.Tools = append(dsReq.Tools, DeepSeekTool{
				Type:     "function",
				Function: fn,
			})
		}
	}

	// 转换消息历史
	for _, turn := range anthReq.Messages {
		var content string
		var toolCalls []ToolCall
		var toolCallID string
		role := turn.Role

		for _, c := range turn.Content {
			switch c.Type {
			case "text":
				content += c.Text
			case "tool_use":
				toolCalls = append(toolCalls, ToolCall{
					ID:   c.ID,
					Type: "function",
					Function: struct {
						Name      string `json:"name"`
						Arguments string `json:"arguments"`
					}{
						Name:      c.Name,
						Arguments: string(c.Input),
					},
				})
			case "tool_result":
				role = "tool"
				toolCallID = c.ID
				content += c.Text
			}
		}

		msg := DeepSeekMsg{
			Role:    role,
			Content: content,
		}

		if len(toolCalls) > 0 {
			msg.ToolCalls = toolCalls
		}
		if toolCallID != "" {
			msg.ToolCallID = toolCallID
		}

		dsReq.Messages = append(dsReq.Messages, msg)
	}

	return dsReq
}

func (g *AnthropicGateway) ConvertToAnthropic(dsResp *DeepSeekResponse) map[string]interface{} {
	response := map[string]interface{}{
		"id":      fmt.Sprintf("msg_%d", time.Now().UnixNano()),
		"type":    "message",
		"role":    "assistant",
		"model":   "deepseek-v4-pro",
		"content": []interface{}{},
	}

	if len(dsResp.Choices) == 0 {
		return response
	}

	msg := dsResp.Choices[0].Message
	var contentList []interface{}

	// 添加思考链（如果有）
	if msg.ReasoningContent != "" {
		contentList = append(contentList, map[string]interface{}{
			"type": "thinking",
			"text": msg.ReasoningContent,
		})
	}

	// 添加文本内容
	if msg.Content != "" {
		contentList = append(contentList, map[string]interface{}{
			"type": "text",
			"text": msg.Content,
		})
	}

	// 添加工具调用
	for _, tc := range msg.ToolCalls {
		contentList = append(contentList, map[string]interface{}{
			"type":  "tool_use",
			"id":    tc.ID,
			"name":  tc.Function.Name,
			"input": json.RawMessage(tc.Function.Arguments),
		})
	}

	response["content"] = contentList

	// Token用量
	if dsResp.Usage != nil {
		response["usage"] = map[string]int{
			"input_tokens":  dsResp.Usage.PromptTokens,
			"output_tokens": dsResp.Usage.CompletionTokens,
		}
	}

	return response
}

func (g *AnthropicGateway) HandleRequest(w http.ResponseWriter, r *http.Request) {
	start := time.Now()

	// 解析请求体
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Failed to read request body", http.StatusBadRequest)
		return
	}

	var anthReq AnthropicMessage
	if err := json.Unmarshal(body, &anthReq); err != nil {
		http.Error(w, fmt.Sprintf("Invalid request format: %v", err), http.StatusBadRequest)
		return
	}

	// 协议转换
	dsReq := g.ConvertToDeepSeek(&anthReq)

	// 序列化
	dsBody, err := json.Marshal(dsReq)
	if err != nil {
		http.Error(w, "Failed to serialize request", http.StatusInternalServerError)
		return
	}

	// 调用DeepSeek API
	httpReq, err := http.NewRequest(
		"POST",
		g.deepseekURL+"/chat/completions",
		bytes.NewReader(dsBody),
	)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	httpReq.Header.Set("Content-Type", "application/json")
	httpReq.Header.Set("Authorization", "Bearer "+g.apiKey)

	resp, err := g.client.Do(httpReq)
	if err != nil {
		g.metrics.mu.Lock()
		g.metrics.requestsFailed++
		g.metrics.mu.Unlock()
		http.Error(w, fmt.Sprintf("Upstream error: %v", err), http.StatusBadGateway)
		return
	}
	defer resp.Body.Close()

	// 读取响应
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		http.Error(w, "Failed to read upstream response", http.StatusBadGateway)
		return
	}

	// 解析DeepSeek响应
	var dsResp DeepSeekResponse
	if err := json.Unmarshal(respBody, &dsResp); err != nil {
		http.Error(w, "Failed to parse upstream response", http.StatusBadGateway)
		return
	}

	// 转换为Anthropic格式
	anthResp := g.ConvertToAnthropic(&dsResp)

	// 更新指标
	latency := time.Since(start).Milliseconds()
	g.metrics.mu.Lock()
	g.metrics.requestsTotal++
	g.metrics.requestsSuccess++
	g.metrics.latencyTotalMs += latency
	g.metrics.mu.Unlock()

	// 返回响应
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(anthResp)
}

func (g *AnthropicGateway) HandleMetrics(w http.ResponseWriter, r *http.Request) {
	g.metrics.mu.RLock()
	defer g.metrics.mu.RUnlock()

	metrics := map[string]interface{}{
		"total_requests":   g.metrics.requestsTotal,
		"success_rate":     float64(g.metrics.requestsSuccess) / float64(max(g.metrics.requestsTotal, 1)) * 100,
		"avg_latency_ms":   g.metrics.latencyTotalMs / max(g.metrics.requestsTotal, 1),
		"uptime_seconds":   time.Now().Unix(),
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(metrics)
}

func max(a, b int64) int64 {
	if a > b {
		return a
	}
	return b
}

func main() {
	apiKey := os.Getenv("DEEPSEEK_API_KEY")
	if apiKey == "" {
		log.Fatal("Please set DEEPSEEK_API_KEY environment variable")
	}

	gateway := NewGateway(apiKey)

	http.HandleFunc("/v1/messages", gateway.HandleRequest)
	http.HandleFunc("/metrics", gateway.HandleMetrics)
	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
	})

	addr := ":8080"
	log.Printf("Anthropic compatibility gateway starting on %s", addr)
	log.Printf("Usage: Point Claude Code/Cursor to http://localhost:%s/v1/messages", addr)
	log.Fatal(http.ListenAndServe(addr, nil))
}

"""
DeepSeek Harness 完整使用示例
包含四种运行模式、会话管理、轨迹追踪
"""
import json
import time
import requests
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field


@dataclass
class HarnessSession:
    """Harness会话状态"""
    session_id: str
    mode: str
    model: str
    created_at: float = field(default_factory=time.time)
    message_count: int = 0
    status: str = "active"


class DeepSeekHarnessClient:
    """
    DeepSeek Harness HTTP客户端
    支持所有四种运行模式
    """

    MODES = ["minimal", "standard", "code", "creator"]

    def __init__(self, base_url: str = "http://127.0.0.1:3080"):
        self.base_url = base_url
        self.session: Optional[HarnessSession] = None
        self.trajectory: List[Dict] = []

    def start_session(
        self,
        mode: str = "standard",
        model: str = "deepseek-v4-pro",
        reasoning_effort: str = "high",
        tools: Optional[List[str]] = None,
    ) -> HarnessSession:
        """
        启动新会话

        Args:
            mode: 运行模式 (minimal/standard/code/creator)
            model: 模型名称
            reasoning_effort: 推理强度
            tools: 启用的工具列表
        """
        if mode not in self.MODES:
            raise ValueError(f"Invalid mode: {mode}. Must be one of {self.MODES}")

        payload = {
            "mode": mode,
            "model": model,
            "reasoning_effort": reasoning_effort,
        }
        if tools:
            payload["tools"] = tools

        resp = requests.post(
            f"{self.base_url}/sessions",
            json=payload,
            timeout=30,
        )
        resp.raise_for_status()
        data = resp.json()

        self.session = HarnessSession(
            session_id=data["session_id"],
            mode=mode,
            model=model,
        )
        print(f"✅ Session started: {self.session.session_id}")
        print(f"   Mode: {mode}, Model: {model}")
        return self.session

    def send_message(
        self,
        content: str,
        stream: bool = False,
    ) -> Dict[str, Any]:
        """发送消息到当前会话"""
        if not self.session:
            raise RuntimeError("No active session. Call start_session() first.")

        payload = {"content": content, "stream": stream}
        resp = requests.post(
            f"{self.base_url}/sessions/{self.session.session_id}/messages",
            json=payload,
            timeout=300,
        )
        resp.raise_for_status()
        self.session.message_count += 1
        return resp.json()

    def get_trajectory(self) -> List[Dict]:
        """获取Agent运行轨迹"""
        if not self.session:
            raise RuntimeError("No active session.")

        resp = requests.get(
            f"{self.base_url}/sessions/{self.session.session_id}/trajectory",
            timeout=30,
        )
        resp.raise_for_status()
        trajectory = resp.json()
        self.trajectory = trajectory
        return trajectory

    def fork_session(self, from_message_id: str) -> str:
        """从指定消息分叉新会话"""
        if not self.session:
            raise RuntimeError("No active session.")

        resp = requests.post(
            f"{self.base_url}/sessions/{self.session.session_id}/fork",
            json={"from_message_id": from_message_id},
            timeout=30,
        )
        resp.raise_for_status()
        new_session_id = resp.json()["session_id"]
        print(f"🔀 Forked new session: {new_session_id}")
        return new_session_id

    def analyze_trajectory(self) -> Dict[str, Any]:
        """分析Agent运行轨迹"""
        if not self.trajectory:
            self.get_trajectory()

        analysis = {
            "total_events": len(self.trajectory),
            "tool_calls": 0,
            "reasoning_steps": 0,
            "errors": 0,
            "total_tokens": 0,
            "event_types": {},
        }

        for event in self.trajectory:
            etype = event.get("type", "unknown")
            analysis["event_types"][etype] = analysis["event_types"].get(etype, 0) + 1

            if etype == "tool_call":
                analysis["tool_calls"] += 1
            elif etype == "reasoning":
                analysis["reasoning_steps"] += 1
            elif etype == "error":
                analysis["errors"] += 1

            if "tokens" in event:
                analysis["total_tokens"] += event["tokens"]

        return analysis

    def close_session(self):
        """关闭当前会话"""
        if not self.session:
            return
        try:
            requests.delete(
                f"{self.base_url}/sessions/{self.session.session_id}",
                timeout=10,
            )
            print(f"Session {self.session.session_id} closed")
        except Exception as e:
            print(f"Warning: Failed to close session: {e}")
        self.session = None


# 四种模式使用示例
def demo_all_modes():
    """演示所有四种运行模式"""
    client = DeepSeekHarnessClient()

    # 1. Minimal模式 - 模型基准测试
    print("\n" + "=" * 60)
    print("1. Minimal Mode - 模型基准测试")
    print("=" * 60)
    client.start_session(mode="minimal", model="deepseek-v4-pro")
    result = client.send_message(
        "Write a Python function to calculate Fibonacci numbers"
    )
    print(f"Response: {str(result)[:200]}...")
    client.close_session()

    # 2. Standard模式 - 日常开发
    print("\n" + "=" * 60)
    print("2. Standard Mode - 日常开发")
    print("=" * 60)
    client.start_session(mode="standard", model="deepseek-v4-pro")
    result = client.send_message(
        "Create a REST API server with Flask and add a /health endpoint"
    )
    trajectory = client.get_trajectory()
    analysis = client.analyze_trajectory()
    print(f"工具调用次数: {analysis['tool_calls']}")
    print(f"推理步骤数: {analysis['reasoning_steps']}")
    client.close_session()

    # 3. Code模式 - 程序化工具调用
    print("\n" + "=" * 60)
    print("3. Code Mode - 程序化工具调用")
    print("=" * 60)
    client.start_session(mode="code", model="deepseek-v4-pro")
    result = client.send_message(
        "Use the Code Mode SDK to batch process all files in the ./src directory"
    )
    print(f"Response: {str(result)[:200]}...")
    client.close_session()

    return "All modes demonstrated successfully"


if __name__ == "__main__":
    # 启动Harness
    print("DeepSeek Harness 使用示例")
    print("=" * 60)
    print("\n启动方式:")
    print("  npx @deepseek-ai/dsh web")
    print("  然后访问 http://127.0.0.1:3080")
    print("\n或者使用CLI:")
    print("  npx @deepseek-ai/dsh run --mode minimal")

    # 展示Harness的Cordis架构
    print("\n\nHarness Cordis插件架构:")
    print("""
    ┌─────────────────────────────────────────────┐
    │            Cordis Kernel                     │
    │   Plugin Loading / Unloading / Dependency    │
    └────────────────┬────────────────────────────┘
                     │
    ┌────────────────┼────────────────────┐
    │                │                    │
    ▼                ▼                    ▼
┌────────┐    ┌────────────┐    ┌────────────┐
│ Models │    │   Tools    │    │   Skills   │
├────────┤    ├────────────┤    ├────────────┤
│DS V4Pro│    │ file_edit  │    │ Code Review│
│ OpenAI │    │ bash_shell │    │  Architect │
│Anthropic│   │ web_search │    │  Test Gen  │
└────────┘    └────────────┘    └────────────┘

┌────────┐    ┌────────────┐    ┌────────────┐
│Session │    │  Sandbox   │    │  Storage   │
├────────┤    ├────────────┤    ├────────────┤
│ State  │    │  Docker    │    │  Local FS  │
│Compress│    │  Local     │    │  S3/OSS    │
│ History│    │  Remote    │    │  Vector DB │
└────────┘    └────────────┘    └────────────┘
    """)