DeepSeek V4正式版全量上线深度解析：1.6万亿MoE + 峰谷分时定价 + DSpark推理加速，国产大模型把API当成电费卖

"""
DeepSeek V4 1M上下文注意力机制实现（简化版）
核心：ALiBi位置编码 + 滑动窗口 + 稀疏注意力
"""

import math
import torch
import torch.nn as nn
import torch.nn.functional as F

class DeepSeekLongContextAttention(nn.Module):
    """支持1M上下文的层级注意力机制"""
    
    def __init__(self, d_model=7168, n_heads=64, window_size=8192):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.head_dim = d_model // n_heads
        self.window_size = window_size
        
        # 投影矩阵
        self.wq = nn.Linear(d_model, d_model, bias=False)
        self.wk = nn.Linear(d_model, d_model, bias=False)
        self.wv = nn.Linear(d_model, d_model, bias=False)
        self.wo = nn.Linear(d_model, d_model, bias=False)
        
        # ALiBi斜率（每组头不同）
        self.register_buffer(
            "alibi_slopes",
            self._compute_alibi_slopes(n_heads)
        )
    
    def _compute_alibi_slopes(self, n_heads):
        """计算ALiBi斜率，支持1M上下文"""
        def get_slopes(n):
            # 2^(-8/n * i) 的几何序列
            def _get_slopes_power_of_2(n):
                start = 2 ** (-8 / n)
                return [start ** i for i in range(n)]
            
            if math.log2(n).is_integer():
                return _get_slopes_power_of_2(n)
            else:
                closest_power = 2 ** math.floor(math.log2(n))
                slopes = _get_slopes_power_of_2(closest_power)
                extra = n - closest_power
                extra_slopes = _get_slopes_power_of_2(2 * closest_power)
                return slopes + extra_slopes[0:extra:2]
        
        return torch.tensor(get_slopes(n_heads))
    
    def forward(self, x, layer_idx=0):
        batch, seq_len, _ = x.shape
        
        Q = self.wq(x).view(batch, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
        K = self.wk(x).view(batch, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
        V = self.wv(x).view(batch, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
        
        # 滑动窗口注意力（近层使用全窗口，深层逐步缩小）
        effective_window = min(self.window_size, max(4096, self.window_size - layer_idx * 512))
        
        # 分块处理长序列
        chunk_size = 16384  # 16K分块
        outputs = []
        
        for i in range(0, seq_len, chunk_size):
            chunk_start = max(0, i - effective_window)
            chunk_end = min(seq_len, i + chunk_size)
            
            Q_chunk = Q[:, :, i:min(i+chunk_size, seq_len), :]
            K_chunk = K[:, :, chunk_start:chunk_end, :]
            V_chunk = V[:, :, chunk_start:chunk_end, :]
            
            # 计算注意力分数
            scores = torch.matmul(Q_chunk, K_chunk.transpose(-2, -1)) / math.sqrt(self.head_dim)
            
            # 应用ALiBi偏置
            pos_diff = torch.arange(
                chunk_start - i, chunk_end - i, 
                device=x.device
            ).unsqueeze(0).unsqueeze(0)
            alibi_bias = -self.alibi_slopes.view(-1, 1, 1) * pos_diff.abs()
            scores = scores + alibi_bias[:, :, :, :scores.size(-1)]
            
            # 滑动窗口掩码
            if seq_len > effective_window:
                causal_mask = torch.triu(
                    torch.ones(scores.size(-2), scores.size(-1), device=x.device),
                    diagonal=1
                ) * -1e10
                scores = scores + causal_mask
            
            attn_weights = F.softmax(scores, dim=-1)
            chunk_output = torch.matmul(attn_weights, V_chunk)
            outputs.append(chunk_output)
        
        # 合并分块结果
        output = torch.cat(outputs, dim=2)
        output = output.transpose(1, 2).contiguous().view(batch, seq_len, self.d_model)
        return self.wo(output)


# 1M上下文内存占用估算
def estimate_memory_for_1m_context():
    """计算1M上下文在不同优化策略下的显存占用"""
    d_model = 7168
    n_heads = 64
    head_dim = d_model // n_heads
    seq_len = 1_000_000
    
    # 标准Full Attention（不可行）
    full_attn_memory = (seq_len ** 2 * n_heads * 2) / (1024 ** 3)  # GB
    print(f"标准Full Attention分数矩阵: {full_attn_memory:.1f} GB")
    
    # 滑动窗口（窗口8192）
    window_size = 8192
    window_attn_memory = (seq_len * window_size * n_heads * 2) / (1024 ** 3)
    print(f"滑动窗口注意力分数矩阵: {window_attn_memory:.1f} GB")
    
    # KV缓存（16bit）
    kv_cache = (seq_len * d_model * 2 * 2) / (1024 ** 3)  # K+V各一个
    print(f"KV缓存（16bit）: {kv_cache:.1f} GB")
    
    # 实际运行中通过分块+稀疏化进一步降低
    print(f"DeepSeek V4实际推理峰值: ≈80 GB (H100-80GB)")

estimate_memory_for_1m_context()

// SWE-bench 评估结果分析工具
package main

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"
    "sort"
)

type SWEBenchResult struct {
    ModelName     string             `json:"model_name"`
    TotalIssues   int                `json:"total_issues"`
    Resolved      int                `json:"resolved"`
    PassRate      float64            `json:"pass_rate"`
    CategoryStats map[string]float64 `json:"category_stats"`
    AvgTimeCost   float64            `json:"avg_time_cost_seconds"`
}

type CategoryBreakdown struct {
    Category    string  `json:"category"`
    Resolved    int     `json:"resolved"`
    Total       int     `json:"total"`
    PassRate    float64 `json:"pass_rate"`
    AvgTokens   int     `json:"avg_tokens_used"`
}

func analyzeSWEBench(results []SWEBenchResult) {
    // 排序
    sort.Slice(results, func(i, j int) bool {
        return results[i].PassRate > results[j].PassRate
    })

    fmt.Println("=== SWE-bench Verified 排名 ===")
    for i, r := range results {
        fmt.Printf("第%d名: %s | 解决率: %.1f%% | 平均耗时: %.0f秒\n",
            i+1, r.ModelName, r.PassRate*100, r.AvgTimeCost)
    }

    // 成本效率分析
    fmt.Println("\n=== 成本效率分析 ===")
    type CostEfficiency struct {
        Model    string
        PerfCost float64 // 每1%解决率的价格（美元）
    }
    
    // 模拟数据
    costData := []CostEfficiency{
        {"DeepSeek V4 Pro", 0.15},   // 假设每百万token $0.82
        {"GPT-5.6 Terra", 2.08},      // 每百万token $15输出
        {"Claude Fable 5", 6.21},     // 每百万token $50输出
    }
    
    for _, c := range costData {
        fmt.Printf("%s: 每1%解决率成本 $%.2f\n", c.Model, c.PerfCost)
    }
}

func main() {
    // 模拟API调用示例
    results := []SWEBenchResult{
        {
            ModelName:   "DeepSeek V4 Pro",
            TotalIssues: 500,
            Resolved:    403,
            PassRate:    0.806,
            AvgTimeCost: 89.5,
            CategoryStats: map[string]float64{
                "bug_fix":   0.82,
                "feature":   0.78,
                "refactor":  0.85,
                "test":      0.76,
            },
        },
        {
            ModelName:   "GPT-5.6 Terra",
            TotalIssues: 500,
            Resolved:    390,
            PassRate:    0.781,
            AvgTimeCost: 67.2,
        },
        {
            ModelName:   "Claude Fable 5",
            TotalIssues: 500,
            Resolved:    401,
            PassRate:    0.803,
            AvgTimeCost: 145.8,
        },
    }

    analyzeSWEBench(results)
}

"""
DeepSeek V4 错峰调度客户端
实现智能时段感知的API调用，自动选择最优时段和模型
"""

import time
import datetime
from dataclasses import dataclass
from typing import Optional, Callable
import requests
import hashlib
import hmac

@dataclass
class DeepSeekPricing:
    """DeepSeek V4定价模型"""
    model: str  # 'pro' 或 'flash'
    peak_input: float
    peak_output: float
    offpeak_input: float
    offpeak_output: float
    night_input: float
    night_output: float

class DeepSeekScheduler:
    """智能调度客户端"""
    
    PRICING = {
        'pro': DeepSeekPricing(
            model='pro',
            peak_input=4.0, peak_output=12.0,
            offpeak_input=2.0, offpeak_output=6.0,
            night_input=1.6, night_output=4.8
        ),
        'flash': DeepSeekPricing(
            model='flash',
            peak_input=1.0, peak_output=4.0,
            offpeak_input=0.5, offpeak_output=2.0,
            night_input=0.4, night_output=1.6
        )
    }
    
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v4"):
        self.api_key = api_key
        self.base_url = base_url
        self.task_queue = []
    
    def get_current_price_period(self) -> str:
        """判断当前时段"""
        now = datetime.datetime.now()
        hour = now.hour
        weekday = now.weekday()
        
        # 周末全天低谷
        if weekday >= 5:
            return 'offpeak'
        
        # 夜间特惠
        if 23 <= hour or hour < 7:
            return 'night'
        
        # 低谷时段
        if (12 <= hour < 14) or (18 <= hour < 23):
            return 'offpeak'
        
        # 高峰时段
        if (9 <= hour < 12) or (14 <= hour < 18):
            return 'peak'
        
        return 'offpeak'
    
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> dict:
        """估算当前API调用成本"""
        pricing = self.PRICING[model]
        period = self.get_current_price_period()
        
        input_price = getattr(pricing, f'{period}_input')
        output_price = getattr(pricing, f'{period}_output')
        
        input_cost = (input_tokens / 1_000_000) * input_price
        output_cost = (output_tokens / 1_000_000) * output_price
        
        return {
            'period': period,
            'input_cost_cny': round(input_cost, 4),
            'output_cost_cny': round(output_cost, 4),
            'total_cost_cny': round(input_cost + output_cost, 4),
            'input_price_per_m': input_price,
            'output_price_per_m': output_price
        }
    
    def auto_select_model(self, task_difficulty: str, 
                          input_tokens: int, 
                          output_tokens: int,
                          deadline: Optional[datetime.datetime] = None) -> str:
        """根据任务难度和时限自动选择模型"""
        period = self.get_current_price_period()
        
        # 简单任务一律用Flash
        if task_difficulty == 'easy':
            return 'flash'
        
        # 复杂任务：高峰用Pro，低谷用Flash Pro
        if task_difficulty == 'hard':
            if period in ('peak',):
                return 'pro'  # 复杂任务不因价格降级
            return 'pro'
        
        # 中等任务：价格敏感
        if task_difficulty == 'medium':
            if period == 'peak':
                # 高峰期中等任务可以考虑Flash
                if output_tokens > 4000:
                    return 'pro'  # 长输出用Pro保证质量
                return 'flash'
            return 'pro'
        
        return 'flash'
    
    def schedule_task(self, prompt: str, task_type: str = 'medium',
                      max_tokens: int = 4096,
                      deadline: Optional[datetime.datetime] = None,
                      callback: Optional[Callable] = None):
        """调度任务：支持延迟执行到低谷时段"""
        input_tokens = len(prompt) // 2  # 粗略估算
        
        # 检查是否可延迟
        if deadline and datetime.datetime.now() + datetime.timedelta(hours=2) < deadline:
            period = self.get_current_price_period()
            if period == 'peak':
                print(f"[调度器] 当前高峰时段，任务 deadline={deadline}")
                print(f"[调度器] 建议延迟到低谷时段执行，预计节省50%成本")
                self.task_queue.append({
                    'prompt': prompt,
                    'max_tokens': max_tokens,
                    'callback': callback,
                    'scheduled_time': datetime.datetime.now() + datetime.timedelta(hours=3)
                })
                return {'status': 'queued', 'message': '任务已排队，将在低谷时段执行'}
        
        # 立即执行
        model = self.auto_select_model(task_type, input_tokens, max_tokens)
        cost = self.estimate_cost(model, input_tokens, max_tokens)
        
        print(f"[执行] 选择模型: {model}")
        print(f"[成本] 时段: {cost['period']}, 总计: ¥{cost['total_cost_cny']}")
        
        # 实际API调用
        result = self._call_api(prompt, model, max_tokens)
        
        if callback:
            callback(result)
        
        return result
    
    def _call_api(self, prompt: str, model: str, max_tokens: int) -> dict:
        """实际的API调用（简化示例）"""
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'model': f'deepseek-v4-{model}',
            'messages': [{'role': 'user', 'content': prompt}],
            'max_tokens': max_tokens,
            'temperature': 0.7
        }
        
        # 实际调用省略，模拟返回
        return {
            'model': model,
            'usage': {
                'prompt_tokens': len(prompt) // 2,
                'completion_tokens': max_tokens
            },
            'choices': [{'message': {'content': '模拟响应内容...'}}]
        }
    
    def process_queue(self):
        """处理排队任务"""
        now = datetime.datetime.now()
        completed = []
        
        for task in self.task_queue[:]:
            if task['scheduled_time'] <= now:
                print(f"[调度器] 执行排队任务，原计划时间: {task['scheduled_time']}")
                model = self.auto_select_model('medium', 
                    len(task['prompt']) // 2, task['max_tokens'])
                cost = self.estimate_cost(model, len(task['prompt']) // 2, task['max_tokens'])
                print(f"[成本] 当前时段: {cost['period']}, 总计: ¥{cost['total_cost_cny']}")
                
                result = self._call_api(task['prompt'], model, task['max_tokens'])
                if task['callback']:
                    task['callback'](result)
                completed.append(task)
                self.task_queue.remove(task)
        
        return completed


# 使用示例
def main():
    scheduler = DeepSeekScheduler(api_key="sk-ds-v4-example")
    
    # 场景1：紧急任务，立即执行
    now = datetime.datetime.now()
    print(f"当前时间: {now}")
    print(f"当前时段: {scheduler.get_current_price_period()}")
    
    result = scheduler.schedule_task(
        prompt="实现一个高性能的分布式缓存系统，支持LRU和TTL过期策略",
        task_type='hard',
        max_tokens=8192
    )
    print(f"执行结果: {result}\n")
    
    # 场景2：成本分析对比
    print("=== 成本对比分析 ===")
    for period in ['peak', 'offpeak', 'night']:
        # 模拟时间
        print(f"\n[时段: {period}]")
        for model in ['pro', 'flash']:
            cost = scheduler.estimate_cost(model, 10000, 4000)
            print(f"  {model}: 输入¥{cost['input_price_per_m']}/M, "
                  f"输出¥{cost['output_price_per_m']}/M, "
                  f"总计¥{cost['total_cost_cny']}")
    
    # 场景3：错峰调度
    far_deadline = datetime.datetime.now() + datetime.timedelta(hours=4)
    queued = scheduler.schedule_task(
        prompt="分析1000份财报PDF并提取关键指标",
        task_type='medium',
        max_tokens=16384,
        deadline=far_deadline
    )
    print(f"\n排队结果: {queued}")

if __name__ == "__main__":
    main()

// DSpark推理加速框架核心组件
package main

import (
    "fmt"
    "math"
    "sync"
    "time"
)

// DSparkConfig 推理加速配置
type DSparkConfig struct {
    EnableDynamicSparsity bool    // 动态稀疏注意力
    SparsityThreshold     float64 // 稀疏阈值
    SpeculativeDecoding   bool    // 推测解码
    DraftModelSize        string  // draft模型规模: tiny/small/medium
    KVQuantization        string  // KV缓存量化: fp8/int8
    BatchSize             int     // 批处理大小
    MaxTokens             int     // 最大生成token数
}

// InferenceStats 推理统计
type InferenceStats struct {
    TotalTokens       int
    TotalTime         time.Duration
    TokensPerSecond   float64
    CacheHitRate      float64
    SpecAcceptRate    float64
    EffectiveSparsity float64
}

// DSparkEngine DSpark推理加速引擎
type DSparkEngine struct {
    config DSparkConfig
    stats  InferenceStats
    mu     sync.Mutex
}

func NewDSparkEngine(config DSparkConfig) *DSparkEngine {
    return &DSparkEngine{
        config: config,
        stats:  InferenceStats{},
    }
}

// 推测解码的draft模型
type DraftModel struct {
    Size      string
    Params    int
    Speedup   float64
    AcceptRate float64
}

func getDraftModel(size string) DraftModel {
    models := map[string]DraftModel{
        "tiny":   {Size: "tiny", Params: 120_000_000, Speedup: 3.2, AcceptRate: 0.75},
        "small":  {Size: "small", Params: 350_000_000, Speedup: 2.5, AcceptRate: 0.85},
        "medium": {Size: "medium", Params: 1_200_000_000, Speedup: 1.8, AcceptRate: 0.92},
    }
    if m, ok := models[size]; ok {
        return m
    }
    return models["small"]
}

// SimulateInference 模拟推理加速
func (e *DSparkEngine) SimulateInference(promptTokens, outputTokens int) InferenceStats {
    start := time.Now()
    
    // 基准推理时间（H100，无加速）
    baseTimePerToken := 15.0 * time.Millisecond // 约66 tokens/s
    baseTotalTime := time.Duration(float64(outputTokens) * float64(baseTimePerToken))
    
    // 应用各层加速
    speedup := 1.0
    
    // 1. 动态稀疏注意力
    if e.config.EnableDynamicSparsity {
        // 长序列时稀疏注意力效果更显著
        seqLen := promptTokens + outputTokens
        var sparsityGain float64
        if seqLen > 100000 {
            sparsityGain = 2.8 // 1M上下文时2.8x加速
        } else if seqLen > 32000 {
            sparsityGain = 1.8
        } else {
            sparsityGain = 1.2
        }
        speedup *= sparsityGain
        e.mu.Lock()
        e.stats.EffectiveSparsity = 1.0 - 1.0/sparsityGain
        e.mu.Unlock()
    }
    
    // 2. 推测解码
    if e.config.SpeculativeDecoding {
        draft := getDraftModel(e.config.DraftModelSize)
        // 推测解码加速比 = 1 / (1/draft_speedup + (1-accept_rate)/draft_speedup)
        specSpeedup := 1.0 / (1.0/draft.Speedup + (1.0-draft.AcceptRate)/draft.Speedup)
        speedup *= specSpeedup
        e.mu.Lock()
        e.stats.SpecAcceptRate = draft.AcceptRate
        e.mu.Unlock()
    }
    
    // 3. KV缓存量化
    if e.config.KVQuantization != "" {
        var quantGain float64
        switch e.config.KVQuantization {
        case "fp8":
            quantGain = 1.4 // FP8: 显存带宽减半，约1.4x
        case "int8":
            quantGain = 1.6 // INT8: 更激进，约1.6x
        }
        speedup *= quantGain
    }
    
    // 4. 批处理
    if e.config.BatchSize > 1 {
        batchGain := 1.0 + float64(e.config.BatchSize-1)*0.15
        speedup *= batchGain
    }
    
    // 5. 算子融合
    fuseGain := 1.25
    speedup *= fuseGain
    
    // 计算最终性能
    acceleratedTime := time.Duration(float64(baseTotalTime) / speedup)
    actualTokensPerSec := float64(outputTokens) / acceleratedTime.Seconds()
    
    baseTokensPerSec := float64(outputTokens) / baseTotalTime.Seconds()
    
    stats := InferenceStats{
        TotalTokens:     outputTokens,
        TotalTime:       acceleratedTime,
        TokensPerSecond: actualTokensPerSec,
        CacheHitRate:    0.35, // 缓存命中率≈35%
    }
    
    e.mu.Lock()
    e.stats = stats
    e.mu.Unlock()
    
    fmt.Printf("=== DSpark推理加速分析 ===\n")
    fmt.Printf("基准速度: %.0f tokens/s\n", baseTokensPerSec)
    fmt.Printf("加速后速度: %.0f tokens/s\n", actualTokensPerSec)
    fmt.Printf("总加速比: %.1fx\n", speedup)
    fmt.Printf("吞吐提升: %.1f%%\n", (speedup-1.0)*100)
    fmt.Printf("推测解码接受率: %.1f%%\n", stats.SpecAcceptRate*100)
    fmt.Printf("有效稀疏率: %.1f%%\n", stats.EffectiveSparsity*100)
    
    return stats
}

func main() {
    // 多场景测试
    scenarios := []struct {
        name   string
        config DSparkConfig
    }{
        {
            name: "长上下文场景（100K+）",
            config: DSparkConfig{
                EnableDynamicSparsity: true,
                SparsityThreshold:     0.3,
                SpeculativeDecoding:   true,
                DraftModelSize:        "small",
                KVQuantization:        "fp8",
                BatchSize:             1,
                MaxTokens:             4096,
            },
        },
        {
            name: "高吞吐批处理场景",
            config: DSparkConfig{
                EnableDynamicSparsity: true,
                SparsityThreshold:     0.5,
                SpeculativeDecoding:   true,
                DraftModelSize:        "medium",
                KVQuantization:        "int8",
                BatchSize:             8,
                MaxTokens:             1024,
            },
        },
        {
            name: "低延迟场景",
            config: DSparkConfig{
                EnableDynamicSparsity: false,
                SpeculativeDecoding:   true,
                DraftModelSize:        "tiny",
                KVQuantization:        "fp8",
                BatchSize:             1,
                MaxTokens:             512,
            },
        },
    }
    
    for _, s := range scenarios {
        fmt.Printf("\n--- %s ---\n", s.name)
        engine := NewDSparkEngine(s.config)
        engine.SimulateInference(100000, s.config.MaxTokens)
        fmt.Println()
    }
    
    // 官方声称的60%-85%吞吐提升验证
    fmt.Println("=== 官方声称验证 ===")
    fmt.Printf("DSpark官方声称吞吐提升: 60%%-85%%\n")
    
    // 在典型场景下验证
    typicalConfig := DSparkConfig{
        EnableDynamicSparsity: true,
        SparsityThreshold:     0.4,
        SpeculativeDecoding:   true,
        DraftModelSize:        "small",
        KVQuantization:        "fp8",
        BatchSize:             4,
        MaxTokens:             2048,
    }
    engine := NewDSparkEngine(typicalConfig)
    stats := engine.SimulateInference(32000, 2048)
    
    speedup := float64(stats.TokensPerSecond) / 66.0 // 66 tokens/s基准
    throughputGain := (speedup - 1.0) * 100
    fmt.Printf("\n典型场景实测吞吐提升: %.1f%%\n", throughputGain)
    if throughputGain >= 60 && throughputGain <= 85 {
        fmt.Println("✅ 验证通过：在官方声称范围内")
    } else if throughputGain > 85 {
        fmt.Println("⚠️ 超出上限：可能受益于批处理等其他因素")
    } else {
        fmt.Println("⚠️ 低于下限：可能需要更激进的配置")
    }
}

"""
DeepSeek V4 Python SDK 完整接入示例
"""

import json
import time
from typing import AsyncGenerator, Optional
import aiohttp
import asyncio

class DeepSeekV4Client:
    """DeepSeek V4完整客户端"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v4"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def chat_completion(self, messages: list, model: str = "deepseek-v4-pro",
                        max_tokens: int = 4096, temperature: float = 0.7,
                        stream: bool = False) -> dict:
        """标准聊天补全"""
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": stream
        }
        
        resp = self.session.post(f"{self.base_url}/chat/completions", json=payload)
        resp.raise_for_status()
        return resp.json()
    
    def agentic_coding(self, task: str, repo_context: Optional[str] = None) -> dict:
        """Agentic Coding专用接口"""
        system_prompt = """你是一个专业的AI编程助手。对于给定的编程任务，请：
1. 分析任务需求
2. 设计解决方案
3. 编写完整可运行的代码
4. 添加详细的注释和错误处理
5. 提供测试用例"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"任务描述：{task}\n\n仓库上下文：{repo_context or '无'}"}
        ]
        
        return self.chat_completion(messages, model="deepseek-v4-pro", max_tokens=16384)
    
    def long_context_analysis(self, document: str, query: str) -> dict:
        """1M超长上下文分析"""
        messages = [
            {"role": "system", "content": "你是一个长文档分析专家。请基于提供的文档内容回答问题。"},
            {"role": "user", "content": f"文档内容：\n{document}\n\n问题：{query}"}
        ]
        
        return self.chat_completion(messages, model="deepseek-v4-pro", max_tokens=8192)


# 使用示例
if __name__ == "__main__":
    client = DeepSeekV4Client(api_key="sk-ds-v4-example")
    
    # 1. 代码生成
    response = client.agentic_coding(
        "实现一个Go语言的分布式任务调度器，支持cron表达式、任务依赖和失败重试"
    )
    print(json.dumps(response, indent=2, ensure_ascii=False))

// DeepSeek V4 Go SDK
package deepseek

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

// V4Client DeepSeek V4客户端
type V4Client struct {
    apiKey     string
    baseURL    string
    httpClient *http.Client
}

// ChatMessage 消息结构
type ChatMessage struct {
    Role    string `json:"role"`
    Content string `json:"content"`
}

// ChatRequest 请求结构
type ChatRequest struct {
    Model       string        `json:"model"`
    Messages    []ChatMessage `json:"messages"`
    MaxTokens   int           `json:"max_tokens"`
    Temperature float64       `json:"temperature"`
    Stream      bool          `json:"stream"`
}

// Usage 用量统计
type Usage struct {
    PromptTokens     int `json:"prompt_tokens"`
    CompletionTokens int `json:"completion_tokens"`
    TotalTokens      int `json:"total_tokens"`
}

// ChatResponse 响应结构
type ChatResponse struct {
    ID      string `json:"id"`
    Object  string `json:"object"`
    Created int64  `json:"created"`
    Model   string `json:"model"`
    Choices []struct {
        Index   int         `json:"index"`
        Message ChatMessage `json:"message"`
    } `json:"choices"`
    Usage Usage `json:"usage"`
}

// NewV4Client 创建客户端
func NewV4Client(apiKey string) *V4Client {
    return &V4Client{
        apiKey:  apiKey,
        baseURL: "https://api.deepseek.com/v4",
        httpClient: &http.Client{
            Timeout: 180 * time.Second,
        },
    }
}

// Chat 发送聊天请求
func (c *V4Client) Chat(req ChatRequest) (*ChatResponse, error) {
    body, err := json.Marshal(req)
    if err != nil {
        return nil, fmt.Errorf("marshal request: %w", err)
    }

    httpReq, err := http.NewRequest("POST",
        c.baseURL+"/chat/completions",
        bytes.NewReader(body))
    if err != nil {
        return nil, fmt.Errorf("create request: %w", err)
    }

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

    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return nil, fmt.Errorf("do request: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        respBody, _ := io.ReadAll(resp.Body)
        return nil, fmt.Errorf("API error: status=%d, body=%s",
            resp.StatusCode, string(respBody))
    }

    var result ChatResponse
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, fmt.Errorf("decode response: %w", err)
    }

    return &result, nil
}

// AgenticCodeReview 代码审查Agent
func (c *V4Client) AgenticCodeReview(code string, language string) (*ChatResponse, error) {
    req := ChatRequest{
        Model: "deepseek-v4-pro",
        Messages: []ChatMessage{
            {
                Role:    "system",
                Content: "你是一个专业的代码审查专家。请对以下代码进行审查，包括：\n" +
                    "1. 安全性分析\n2. 性能问题\n3. 代码风格\n4. 潜在Bug\n5. 改进建议",
            },
            {
                Role:    "user",
                Content: fmt.Sprintf("请审查以下%s代码：\n\n```%s\n%s\n```",
                    language, language, code),
            },
        },
        MaxTokens:   8192,
        Temperature: 0.3,
    }

    return c.Chat(req)
}

// LongContextRAG 长上下文RAG查询
func (c *V4Client) LongContextRAG(documents []string, query string) (*ChatResponse, error) {
    context := ""
    for i, doc := range documents {
        context += fmt.Sprintf("\n--- 文档%d ---\n%s\n", i+1, doc)
    }

    req := ChatRequest{
        Model: "deepseek-v4-pro",
        Messages: []ChatMessage{
            {
                Role: "system",
                Content: "你是基于提供文档的问答助手。请严格基于文档内容回答问题，" +
                    "如果文档中没有相关信息，请明确说明。",
            },
            {
                Role:    "user",
                Content: fmt.Sprintf("文档内容：\n%s\n\n问题：%s", context, query),
            },
        },
        MaxTokens:   4096,
        Temperature: 0.2,
    }

    return c.Chat(req)
}

// CostOptimizer 成本优化器
type CostOptimizer struct {
    client *V4Client
}

func NewCostOptimizer(client *V4Client) *CostOptimizer {
    return &CostOptimizer{client: client}
}

// OptimizeAndCall 优化调用
func (co *CostOptimizer) OptimizeAndCall(req ChatRequest) (*ChatResponse, error) {
    currentHour := time.Now().Hour()
    currentWeekday := time.Now().Weekday()

    // 周末自动选择Flash
    if currentWeekday == time.Saturday || currentWeekday == time.Sunday {
        req.Model = "deepseek-v4-flash"
    }

    // 夜间（23:00-7:00）自动选择Flash
    if currentHour >= 23 || currentHour < 7 {
        req.Model = "deepseek-v4-flash"
    }

    // 午休低谷（12:00-14:00）自动选择Flash
    if currentHour >= 12 && currentHour < 14 {
        req.Model = "deepseek-v4-flash"
    }

    // 长输出任务留在Pro
    if req.MaxTokens > 8192 {
        req.Model = "deepseek-v4-pro"
    }

    return co.client.Chat(req)
}