DeepSeek V4.1 Flash深度解析：552B MoE新架构、非对称Causal-Encoder-Decoder、KV Cache暴降437倍

package main

import (
	"fmt"
	"math"
)

// 模拟CED架构与标准Decoder-Only架构的推理成本对比

type ModelConfig struct {
	Name            string
	EncoderActiveB  float64 // Encoder激活参数量(B)
	DecoderActiveB  float64 // Decoder激活参数量(B)
	TotalParamsB    float64 // 总参数量(B)
	HasEngram       bool
	EngramParamsB   float64
}

func main() {
	ced := ModelConfig{
		Name:           "DeepSeek V4.1 Flash (CED)",
		EncoderActiveB: 8,
		DecoderActiveB: 16,
		TotalParamsB:   552,
		HasEngram:      true,
		EngramParamsB:  196,
	}

	standard := ModelConfig{
		Name:           "Standard Decoder-Only (V4 Pro)",
		EncoderActiveB: 37, // V4 Pro每token激活约37B
		DecoderActiveB: 37,
		TotalParamsB:   1016,
	}

	// Agent场景：输入100K tokens, 输出200 tokens
	inputTokens := 100000
	outputTokens := 200
	agentRatio := float64(inputTokens) / float64(outputTokens)

	fmt.Println("=== Agent场景推理成本对比 ===")
	fmt.Printf("输入: %d tokens, 输出: %d tokens (比例 %.0f:1)\n",
		inputTokens, outputTokens, agentRatio)
	fmt.Println()

	// Prefill阶段计算量 (以激活参数 * 输入token数 近似)
	cedPrefillFLOPs := ced.EncoderActiveB * float64(inputTokens)
	stdPrefillFLOPs := standard.EncoderActiveB * float64(inputTokens)

	fmt.Printf("Prefill计算量 (B·token):\n")
	fmt.Printf("  V4.1 Flash (CED):    %.2f B·token (Encoder %dB × %dtok)\n",
		cedPrefillFLOPs, int(ced.EncoderActiveB), inputTokens)
	fmt.Printf("  V4 Pro (Standard):   %.2f B·token (激活 %dB × %dtok)\n",
		stdPrefillFLOPs, int(standard.EncoderActiveB), inputTokens)
	fmt.Printf("  CED节省:            %.1f%%\n",
		(1-cedPrefillFLOPs/stdPrefillFLOPs)*100)
	fmt.Println()

	// Decode阶段计算量 (每个输出token)
	cedDecodeFLOPsPerToken := ced.DecoderActiveB
	stdDecodeFLOPsPerToken := standard.DecoderActiveB
	totalCedDecode := cedDecodeFLOPsPerToken * float64(outputTokens)
	totalStdDecode := stdDecodeFLOPsPerToken * float64(outputTokens)

	fmt.Printf("Decode计算量 (每个output token):\n")
	fmt.Printf("  V4.1 Flash (CED):    %.0f B·param (Decoder %dB)\n",
		cedDecodeFLOPsPerToken, int(ced.DecoderActiveB))
	fmt.Printf("  V4 Pro (Standard):   %.0f B·param (激活 %dB)\n",
		stdDecodeFLOPsPerToken, int(standard.DecoderActiveB))
	fmt.Println()

	// 总计算量
	totalCed := cedPrefillFLOPs + totalCedDecode
	totalStd := stdPrefillFLOPs + totalStdDecode

	fmt.Printf("总计算量对比:\n")
	fmt.Printf("  V4.1 Flash: %.2f B·token\n", totalCed)
	fmt.Printf("  V4 Pro:     %.2f B·token\n", totalStd)
	fmt.Printf("  总节省:     %.1f%%\n", (1-totalCed/totalStd)*100)

	// 敏感性分析：不同输入输出比下的成本
	fmt.Println()
	fmt.Println("=== 输入输出比敏感性分析 ===")
	ratios := []float64{10, 100, 500, 1000, 5000}
	for _, ratio := range ratios {
		inTok := 100000
		outTok := int(math.Ceil(float64(inTok) / ratio))
		cedCost := ced.EncoderActiveB*float64(inTok) + ced.DecoderActiveB*float64(outTok)
		stdCost := standard.EncoderActiveB*float64(inTok) + standard.DecoderActiveB*float64(outTok)
		saving := (1 - cedCost/stdCost) * 100
		fmt.Printf("  比 %4.0f:1 (in=%d, out=%d) → CED节省 %5.1f%%\n",
			ratio, inTok, outTok, saving)
	}

	// 最大输出场景对比（384K输出）
	fmt.Println()
	fmt.Println("=== 最大输出场景 (384K tokens) ===")
	longOutput := 384000
	maxCed := ced.EncoderActiveB*100000 + ced.DecoderActiveB*float64(longOutput)
	maxStd := standard.EncoderActiveB*100000 + standard.DecoderActiveB*float64(longOutput)
	fmt.Printf("  V4.1 Flash: %.2f B·token (Encoder %dB×100K + Decoder %dB×384K)\n",
		maxCed, int(ced.EncoderActiveB), int(ced.DecoderActiveB))
	fmt.Printf("  V4 Pro:     %.2f B·token\n", maxStd)
	fmt.Printf("  CED节省:    %.1f%%\n", (1-maxCed/maxStd)*100)
}

import numpy as np
from dataclasses import dataclass
from typing import Optional, Tuple

@dataclass
class CSA2Config:
    d_model: int = 7168      # 模型维度
    n_heads: int = 64        # 注意力头数
    d_kv: int = 128          # KV投影维度
    n_shared_layers: int = 3 # 每组的共享层数
    top_k: int = 32          # 稀疏注意力K值
    block_size: int = 64     # 层次化索引块大小

class CSA2Attention:
    """Compressed Sparse Attention 2 实现"""
    
    def __init__(self, config: CSA2Config, layer_idx: int):
        self.config = config
        self.layer_idx = layer_idx
        
        # 主KV投影
        self.W_k_main = np.random.randn(config.d_model, config.d_kv * config.n_heads) * 0.02
        self.W_v_main = np.random.randn(config.d_model, config.d_kv * config.n_heads) * 0.02
        
        # Indexer KV投影（用于稀疏索引）
        self.W_k_idx = np.random.randn(config.d_model, config.d_kv) * 0.02
        self.W_q_idx = np.random.randn(config.d_model, config.d_kv) * 0.02
        
        # 共享KV缓存（跨层复用）
        self.shared_kv: Optional[Tuple[np.ndarray, np.ndarray]] = None
        self.shared_indices: Optional[np.ndarray] = None
        
    def get_mode(self) -> str:
        """根据层索引确定CSA2模式"""
        group_size = self.config.n_shared_layers
        pos_in_group = (self.layer_idx - 1) % group_size
        
        if pos_in_group == 0:
            return "full"      # Full Mode: 计算自有KV+索引
        elif pos_in_group == 1:
            return "reindex"   # Reindex Mode: 复用KV,重新选索引
        else:
            return "reuse"     # Reuse Mode: 完全继承
        
    def forward_full_mode(self, x: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        """Full Mode: 计算自有KV并生成Top-K索引"""
        batch, seq_len, _ = x.shape
        
        # 计算主KV
        kv_main = x @ self.W_k_main  # (batch, seq, d_kv*n_heads)
        K_main = kv_main.reshape(batch, seq_len, self.config.n_heads, self.config.d_kv)
        V_main = (x @ self.W_v_main).reshape(batch, seq_len, self.config.n_heads, self.config.d_kv)
        
        # 计算Indexer KV
        K_idx = x @ self.W_k_idx  # (batch, seq, d_kv)
        Q_idx = x @ self.W_q_idx  # (batch, seq, d_kv)
        
        # 层次化索引: 先选块级候选池
        n_blocks = seq_len // self.config.block_size
        block_scores = np.zeros((batch, n_blocks))
        for b in range(n_blocks):
            start = b * self.config.block_size
            end = min(start + self.config.block_size, seq_len)
            block_k = K_idx[:, start:end, :]
            block_scores[:, b] = np.max(
                Q_idx[:, :, None, :] @ block_k[:, None, :, :, None],
                axis=(1, 3)
            ).mean(axis=1)
        
        # 选择Top-K块作为候选池
        top_blocks = np.argsort(-block_scores, axis=1)[:, :self.config.top_k]
        
        # 在候选块内精细评分
        indices = []
        for b in range(batch):
            pool = []
            for tb in top_blocks[b]:
                start = tb * self.config.block_size
                end = min(start + self.config.block_size, seq_len)
                pool.extend(range(start, end))
            pool = np.array(pool)
            
            scores = Q_idx[b] @ K_idx[b, pool].T  # (seq, pool_size)
            top_idx = pool[np.argsort(-scores.max(axis=0))[:self.config.top_k]]
            indices.append(top_idx)
        
        indices = np.stack(indices)
        
        # 缓存共享
        self.shared_kv = (K_main, V_main)
        self.shared_indices = indices
        
        return K_main, V_main, indices
    
    def forward_reindex_mode(self, x: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
        """Reindex Mode: 复用前层KV, 计算新索引"""
        assert self.shared_kv is not None, "需要前层Full Mode先执行"
        
        K_main, V_main = self.shared_kv
        Q_idx = x @ self.W_q_idx
        
        # 计算新Top-K索引（从共享候选池中重新选择）
        batch, seq_len, _ = x.shape
        indices = []
        for b in range(batch):
            scores = Q_idx[b] @ K_main[b, :, 0, :].T  # 使用第0个head的K
            top_idx = np.argsort(-scores.max(axis=0))[:self.config.top_k]
            indices.append(top_idx)
        
        self.shared_indices = np.stack(indices)
        return K_main, V_main
    
    def forward_reuse_mode(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        """Reuse Mode: 完全继承前层KV和索引, 零计算"""
        assert self.shared_kv is not None
        assert self.shared_indices is not None
        return (*self.shared_kv, self.shared_indices)
    
    def forward(self, x: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        mode = self.get_mode()
        if mode == "full":
            return self.forward_full_mode(x)
        elif mode == "reindex":
            return self.forward_reindex_mode(x)
        else:
            return self.forward_reuse_mode()

# 模拟40层CED中的CSA2模式分布
if __name__ == "__main__":
    config = CSA2Config()
    print("=== CSA2 模式分布 (40层CED, 3层/组) ===")
    print()
    print("Encoder Layers (1-20):")
    for i in range(1, 21):
        attn = CSA2Attention(config, i)
        mode = attn.get_mode()
        print(f"  Layer {i:2d}: {mode.upper():8s} Mode", end="")
        if mode == "full":
            print(" ← 计算自有KV,生成Top-K索引")
        elif mode == "reindex":
            print(" ← 复用KV,重新选索引")
        else:
            print(" ← 完全继承,零计算")
    
    print()
    print("Decoder Layers (21-40):")
    print("  (从Encoder H_{L/2}投影全局KV)")
    for i in range(21, 41):
        attn = CSA2Attention(config, i)
        mode = attn.get_mode()
        print(f"  Layer {i:2d}: {mode.upper():8s} Mode + SWA")
    
    # 统计计算节省
    total_encoder = 20
    full_count = 7   # 每3层1个Full
    reindex_count = 7
    reuse_count = 6
    print()
    print(f"=== 计算节省统计 (Encoder) ===")
    print(f"  Full Mode:   {full_count} layers (100% 计算)")
    print(f"  Reindex:     {reindex_count} layers (~30% 计算)")
    print(f"  Reuse:       {reuse_count} layers (~0% 计算)")
    saving = 1 - (full_count*1.0 + reindex_count*0.3 + reuse_count*0.0) / total_encoder
    print(f"  Encoder注意力计算节省: {saving*100:.0f}%")

import struct
import numpy as np

class MXFP4Quantizer:
    """
    OCP标准MXFP4格式的KV Cache量化器
    MXFP4: 4-bit浮点格式, 每组共享指数
    """
    
    def __init__(self, group_size: int = 32):
        self.group_size = group_size  # 每组32个元素共享指数
        
    def quantize(self, kv_tensor: np.ndarray) -> tuple:
        """将FP32 KV Cache量化为MXFP4格式"""
        original_shape = kv_tensor.shape
        flat = kv_tensor.flatten()
        
        # 分组成块，每块共享指数
        n_groups = (len(flat) + self.group_size - 1) // self.group_size
        padded = np.zeros(n_groups * self.group_size)
        padded[:len(flat)] = flat
        
        groups = padded.reshape(n_groups, self.group_size)
        
        # MXFP4格式：每组一个共享的E5M2指数 + 4-bit尾数
        # 指数范围: E5M2, 即FP32的指数部分取5位
        exponents = []
        mantissas_4bit = []
        
        for group in groups:
            # 找到组内最大值
            max_abs = np.max(np.abs(group))
            if max_abs == 0:
                exponents.append(0)
                mantissas_4bit.extend([0] * self.group_size)
                continue
            
            # E5M2指数: 计算2的幂次
            exp = int(np.floor(np.log2(max_abs)))
            # 限幅到E5M2范围 (-14到15)
            exp = max(-14, min(15, exp))
            exponents.append(exp)
            
            # 4-bit尾数: 归一化后量化到[-8, 7]
            scale = 2.0 ** (-exp)
            scaled = group * scale
            # 4-bit有符号: [-8, 7]
            mantissa = np.clip(np.round(scaled), -8, 7).astype(np.int8)
            mantissas_4bit.extend(mantissa.tolist())
        
        return (
            np.array(exponents, dtype=np.int8),
            np.array(mantissas_4bit, dtype=np.int8),
            original_shape
        )
    
    def dequantize(self, quantized: tuple) -> np.ndarray:
        """将MXFP4还原为FP32"""
        exponents, mantissas_4bit, original_shape = quantized
        n_groups = len(exponents)
        
        mantissa_array = np.array(mantissas_4bit).reshape(n_groups, self.group_size)
        
        # 还原: mantissa * 2^exp
        scales = 2.0 ** exponents.astype(np.float32)
        deq = mantissa_array.astype(np.float32) * scales[:, np.newaxis]
        
        # 裁剪到原始形状
        total_elements = int(np.prod(original_shape))
        return deq.flatten()[:total_elements].reshape(original_shape)
    
    def compression_ratio(self, orig_bytes: int) -> float:
        """计算压缩比"""
        # 原始: FP32 = 4 bytes per element
        # 量化后: 每个元素4-bit = 0.5 bytes + 每个group 1 byte exponent
        orig_elements = orig_bytes // 4
        n_groups = (orig_elements + self.group_size - 1) // self.group_size
        compressed = orig_elements * 0.5 + n_groups * 1
        return orig_bytes / compressed


# 测试: 模拟KV Cache量化
if __name__ == "__main__":
    np.random.seed(42)
    
    # 模拟一个注意力头的KV Cache (batch=1, seq=4096, d_kv=128)
    batch, seq_len, d_kv = 1, 4096, 128
    kv_cache = np.random.randn(batch, seq_len, d_kv).astype(np.float32)
    
    print("=== MXFP4 KV Cache 量化测试 ===")
    print(f"原始KV Cache形状: {kv_cache.shape}")
    print(f"原始大小: {kv_cache.nbytes / 1024:.1f} KB")
    
    quantizer = MXFP4Quantizer(group_size=32)
    quantized = quantizer.quantize(kv_cache)
    deq = quantizer.dequantize(quantized)
    
    # 计算量化误差
    mse = np.mean((kv_cache - deq) ** 2)
    snr = 10 * np.log10(np.var(kv_cache) / mse)
    
    print(f"量化MSE: {mse:.6f}")
    print(f"SNR: {snr:.2f} dB")
    print(f"压缩比: {quantizer.compression_ratio(kv_cache.nbytes):.1f}x")
    print(f"压缩后大小: {kv_cache.nbytes / quantizer.compression_ratio(kv_cache.nbytes) / 1024:.1f} KB")
    
    # 跨层KV复用节省估算
    print()
    print("=== 跨层KV复用节省 (40层CED) ===")
    layers = 40
    full_mode_layers = 14   # 每3层1个Full
    reindex_layers = 13
    reuse_layers = 13
    
    # 每层KV存储
    kv_per_layer = kv_cache.nbytes
    # Full: 存储完整KV + 索引
    full_storage = kv_per_layer * 1.5  # KV + Top-K索引
    # Reindex: 只存索引 (KV复用前层)
    reindex_storage = kv_per_layer * 0.3  # 仅新索引
    # Reuse: 零存储
    reuse_storage = 0
    
    total_standard = kv_per_layer * layers  # 无复用: 40份
    total_csa2 = (full_storage * full_mode_layers + 
                  reindex_storage * reindex_layers + 
                  reuse_storage * reuse_layers)
    
    print(f"标准KV存储: {total_standard / 1024 / 1024:.1f} MB")
    print(f"CSA2复用后: {total_csa2 / 1024 / 1024:.2f} MB")
    print(f"层维度节省: {(1 - total_csa2 / total_standard) * 100:.1f}%")

import matplotlib.pyplot as plt
import numpy as np

# 基准测试对比数据
benchmarks = {
    "GPQA Diamond": {"V4 Pro": 85.0, "V4.1 Flash": 90.9, "GPT-6 Astra": 92.1},
    "Terminal-Bench 2.1": {"V4 Pro": 87.9, "V4.1 Flash": 90.6, "Claude 5": 89.2},
    "CyberGym": {"V4 Pro": 83.3, "V4.1 Flash": 88.1, "Grok 3": 85.7},
    "DeepSWE v1.1": {"V4 Pro": 62.0, "V4.1 Flash": 74.2, "Claude 5 Opus": 72.5},
}

models = ["V4 Pro", "V4.1 Flash", "GPT-6 Astra/Claude 5"]
print("=== 多模型基准对比 ===")
print(f"{'基准测试':<20} {'V4 Pro':<10} {'V4.1 Flash':<15} {'竞品':<15} {'Flash优势':<10}")
print("-" * 70)
for bench, scores in benchmarks.items():
    v4p = scores["V4 Pro"]
    flash = scores["V4.1 Flash"]
    comp = list(scores.values())[2]
    best = max(scores.values())
    label = "🥇" if flash == best else "🥈"
    print(f"{bench:<20} {v4p:<10.1f} {flash:<13.1f} {label:<3} {comp:<12.1f} {'+' if flash > v4p else ''}{flash-v4p:.1f}")

# 效率指标：性能 vs 计算成本
print()
print("=== 性能/成本效率对比 ===")
efficiency = {
    "V4.1 Flash": {"Perf": 90.6, "Cost": 2.0, "Efficiency": 45.3},
    "V4 Pro": {"Perf": 87.9, "Cost": 8.0, "Efficiency": 11.0},
    "GPT-6 Astra Mini": {"Perf": 89.5, "Cost": 6.0, "Efficiency": 14.9},
    "Claude 5 Haiku": {"Perf": 85.2, "Cost": 3.0, "Efficiency": 28.4},
}

print(f"{'模型':<20} {'Terminal-Bench':<15} {'API价格(输入/1M)':<18} {'每美元性能':<12}")
print("-" * 65)
for name, data in efficiency.items():
    print(f"{name:<20} {data['Perf']:<15.1f} ${data['Cost']:<6.1f}         {data['Efficiency']:<12.1f}")

package main

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

// DeepSeek V4.1 Flash API 调用示例
// 展示可控制推理强度功能

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

type ChatRequest struct {
	Model       string    `json:"model"`
	Messages    []Message `json:"messages"`
	MaxTokens   int       `json:"max_tokens,omitempty"`
	Temperature float64   `json:"temperature,omitempty"`
	Effort      int       `json:"effort,omitempty"` // 推理强度 1-100
}

type ChatResponse struct {
	ID      string   `json:"id"`
	Choices []Choice `json:"choices"`
	Usage   Usage    `json:"usage"`
}

type Choice struct {
	Message Message `json:"message"`
}

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

func chatWithEffort(apiKey string, prompt string, effort int) (*ChatResponse, error) {
	url := "https://api.deepseek.com/chat/completions"
	
	req := ChatRequest{
		Model: "deepseek-flash",
		Messages: []Message{
			{Role: "system", Content: fmt.Sprintf(
				"You are a helpful AI assistant. Reasoning effort level: %d.", effort)},
			{Role: "user", Content: prompt},
		},
		MaxTokens:   8192,
		Temperature: 0.0,
		Effort:      effort,
	}
	
	body, _ := json.Marshal(req)
	
	httpReq, _ := http.NewRequest("POST", url, bytes.NewBuffer(body))
	httpReq.Header.Set("Content-Type", "application/json")
	httpReq.Header.Set("Authorization", "Bearer "+apiKey)
	
	client := &http.Client{Timeout: 60 * time.Second}
	resp, err := client.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("request failed: %w", err)
	}
	defer resp.Body.Close()
	
	respBody, _ := io.ReadAll(resp.Body)
	var chatResp ChatResponse
	json.Unmarshal(respBody, &chatResp)
	
	return &chatResp, nil
}

func main() {
	apiKey := "YOUR_DEEPSEEK_API_KEY"
	
	testCases := []struct {
		name   string
		effort int
		prompt string
	}{
		{
			name:   "低强度推理 - 快速问答",
			effort: 10,
			prompt: "什么是KV Cache?",
		},
		{
			name:   "中强度推理 - 代码生成",
			effort: 50,
			prompt: "用Go实现一个并发的KV Cache系统",
		},
		{
			name:   "高强度推理 - 复杂数学",
			effort: 100,
			prompt: "证明: 对于任意正整数n, n^3 + 2n 能被3整除",
		},
	}
	
	fmt.Println("=== DeepSeek V4.1 Flash 推理强度对比测试 ===")
	fmt.Println()
	
	for _, tc := range testCases {
		fmt.Printf("▶ %s (Effort=%d)\n", tc.name, tc.effort)
		fmt.Printf("  Prompt: %s\n", tc.prompt)
		
		resp, err := chatWithEffort(apiKey, tc.prompt, tc.effort)
		if err != nil {
			fmt.Printf("  Error: %v\n", err)
			continue
		}
		
		if resp != nil && len(resp.Choices) > 0 {
			content := resp.Choices[0].Message.Content
			if len(content) > 100 {
				content = content[:100] + "..."
			}
			fmt.Printf("  输出(前100字): %s\n", content)
			fmt.Printf("  输入Tokens: %d | 输出Tokens: %d | 总Tokens: %d\n",
				resp.Usage.PromptTokens, resp.Usage.CompletionTokens, resp.Usage.TotalTokens)
			
			// 高强度推理约使用2.5倍于低强度的输出token
			ratio := float64(resp.Usage.CompletionTokens)
			fmt.Printf("  输出Token比(相对基准): %.1fx\n", ratio/100)
		}
		fmt.Println()
	}
	
	// 不同effort下的成本-精度曲线模拟
	fmt.Println("=== Effort vs 输出Token数 (成本指标) ===")
	efforts := []int{1, 10, 25, 50, 75, 100}
	for _, e := range efforts {
		outputTokens := int(50 + float64(e)/100*200) // 模拟: 从~50到~250 tokens
		fmt.Printf("  Effort %3d: ~%4d 输出tokens (相对基准 %.1fx)\n",
			e, outputTokens, float64(outputTokens)/50)
	}
}

# API迁移成本计算器
def calculate_migration_savings(
    cache_hit_input: int,   # 月缓存命中输入 (M tokens)
    cache_miss_input: int,  # 月缓存未命中输入 (M tokens)
    output_tokens: int,     # 月输出 (M tokens)
    is_off_peak: bool = False,
):
    """计算从V4 Pro迁移到V4.1 Flash的成本节省"""
    
    # V4 Flash 旧价格 (高峰)
    old_prices = {
        "cache_hit_input": 0.10,   # 元/M tokens
        "cache_miss_input": 3.0,
        "output": 9.0,
    }
    
    # V4.1 Flash 新价格
    new_prices = {
        "cache_hit_input": 0.04 if not is_off_peak else 0.02,
        "cache_miss_input": 2.0 if not is_off_peak else 1.0,
        "output": 8.0 if not is_off_peak else 4.0,
    }
    
    old_cost = (
        cache_hit_input * old_prices["cache_hit_input"] +
        cache_miss_input * old_prices["cache_miss_input"] +
        output_tokens * old_prices["output"]
    )
    
    new_cost = (
        cache_hit_input * new_prices["cache_hit_input"] +
        cache_miss_input * new_prices["cache_miss_input"] +
        output_tokens * new_prices["output"]
    )
    
    savings = old_cost - new_cost
    saving_pct = (1 - new_cost / old_cost) * 100
    
    return {
        "old_cost": old_cost,
        "new_cost": new_cost,
        "savings": savings,
        "saving_pct": saving_pct,
        "is_off_peak": is_off_peak,
    }

# 典型Agent场景计算
print("=== API迁移成本节省 (典型Agent场景) ===")
# 假设: 月100万缓存命中输入 + 10万缓存未命中输入 + 1万输出
result = calculate_migration_savings(100, 10, 1, is_off_peak=False)
print(f"高峰时段:")
print(f"  旧成本(V4 Flash): ¥{result['old_cost']:.3f}")
print(f"  新成本(V4.1 Flash): ¥{result['new_cost']:.3f}")
print(f"  节省: ¥{result['savings']:.3f} ({result['saving_pct']:.1f}%)")

result_off = calculate_migration_savings(100, 10, 1, is_off_peak=True)
print(f"\n闲时:")
print(f"  旧成本(V4 Flash): ¥{result_off['old_cost']:.3f}")
print(f"  新成本(V4.1 Flash): ¥{result_off['new_cost']:.3f}")
print(f"  节省: ¥{result_off['savings']:.3f} ({result_off['saving_pct']:.1f}%)")

# 批处理任务调优: 闲时执行
print()
print("=== 批处理任务成本优化建议 ===")
peak_hours_per_day = 7  # 9-12 + 14-18
off_peak_hours = 17
daily_saving = (peak_hours_per_day * (1 - 0.34) + off_peak_hours * (1 - 0.50)) / 24
print(f"  如果100%任务在闲时执行, 可额外节省: {(1 - 0.50) * 100 - (1 - 0.34) * 100:.1f}%")
print(f"  高峰运行成本: 是闲时的 2x")

import base64
import requests

def analyze_image_deepseek(
    api_key: str,
    image_path: str,
    prompt: str = "请详细描述这张图片的内容"
) -> str:
    """
    使用DeepSeek V4.1 Flash分析图片
    
    Args:
        api_key: DeepSeek API密钥
        image_path: 本地图片路径
        prompt: 分析提示词
    
    Returns:
        模型的分析结果
    """
    # 读取并编码图片
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
    
    # 推断MIME类型
    ext = image_path.split(".")[-1].lower()
    mime_types = {
        "jpg": "image/jpeg",
        "jpeg": "image/jpeg",
        "png": "image/png",
        "webp": "image/webp",
        "gif": "image/gif",
    }
    mime = mime_types.get(ext, "image/png")
    
    payload = {
        "model": "deepseek-flash",
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{mime};base64,{image_data}"
                        }
                    }
                ]
            }
        ],
        "max_tokens": 4096,
        "effort": 75,  # 中等偏上推理强度
    }
    
    resp = requests.post(
        "https://api.deepseek.com/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=120
    )
    
    result = resp.json()
    return result["choices"][0]["message"]["content"]


# 多模态Agent: 从截图理解代码
def agent_from_screenshot(api_key: str, screenshot_path: str) -> dict:
    """
    从IDE截图理解代码逻辑并返回修改建议
    """
    analysis = analyze_image_deepseek(
        api_key,
        screenshot_path,
        prompt=(
            "This is a screenshot of an IDE showing Go/Python code. "
            "1. Identify the programming language and framework\n"
            "2. Extract the key logic and data flow\n"
            "3. Identify potential performance issues, bugs, or security risks\n"
            "4. Provide concrete refactoring suggestions with code snippets\n"
            "Return a structured analysis."
        )
    )
    
    # Agent下一步：自动生成修复代码
    code_gen_prompt = f"""
    Based on the code analysis below, generate the refactored version:
    
    {analysis}
    
    Output only the modified code with clear comments explaining each change.
    """
    
    # 调用文本生成生成修复代码
    payload = {
        "model": "deepseek-flash",
        "messages": [
            {"role": "system", "content": "You are a senior software engineer."},
            {"role": "user", "content": code_gen_prompt}
        ],
        "max_tokens": 8192,
        "effort": 80,
    }
    
    resp = requests.post(
        "https://api.deepseek.com/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=120
    )
    
    result = resp.json()
    return {
        "analysis": analysis,
        "refactored_code": result["choices"][0]["message"]["content"],
        "usage": result["usage"]
    }

"""
DeepSeek Harness v0.1.5 关键特性示例
"""

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class HarnessConfig:
    """Harness v0.1.5 配置"""
    model: str = "deepseek-flash"
    
    # KV Cache持久化: 支持在保留KV Cache的情况下更新系统提示词
    kv_cache_persistence: bool = True
    allow_system_prompt_update: bool = True  
    
    # Agent Teams: 主Agent创建多个成员Agent
    agent_teams_enabled: bool = True
    max_team_members: int = 5
    
    # 程序化工具调用 (PTC) 模式
    ptc_mode: bool = True
    
    # 思考模式配置
    thinking_mode: str = "auto"  # auto | low | high | max
    effort_level: int = 75


# 演示KV Cache持久化下的系统提示词更新
def demonstrate_kv_cache_update():
    """
    传统做法: 每次系统提示词变化 -> 重新Prefill -> 浪费KV Cache
    Harness v0.1.5: 保留KV Cache, 仅增量更新系统提示词部分
    """
    print("=== KV Cache持久化: 系统提示词更新 ===")
    print()
    
    # 传统方法: 全量重算
    traditional = {
        "方法": "全量Prefill",
        "输入长度": "100K tokens",
        "KV Cache利用": "❌ 丢弃旧Cache",
        "每次更新延迟": "~2-5秒 (100K上下文)",
        "N次更新总成本": "N × 全量Prefill"
    }
    
    # Harness v0.1.5方法
    harness_new = {
        "方法": "增量更新 (Harness v0.1.5)",
        "输入长度": "100K tokens",
        "KV Cache利用": "✅ 保留全局KV Cache",
        "每次更新延迟": "~0.1秒 (仅重算系统提示部分)",
        "N次更新总成本": "1次全量 + (N-1)次增量"
    }
    
    for method, details in [("传统方法", traditional), ("Harness v0.1.5", harness_new)]:
        print(f"  {method}:")
        for k, v in details.items():
            print(f"    {k}: {v}")
        print()


# Agent Teams 演示
def demonstrate_agent_teams():
    """
    Agent Teams: 主Agent创建多个成员Agent
    通过共享任务列表分配、跟踪和同步工作
    """
    print("=== Agent Teams 协作模式 ===")
    print()
    
    team_structure = {
        "主Agent (Coordinator)": {
            "职责": "任务分解、分配、进度跟踪、结果整合",
            "工具": ["任务队列", "状态监控", "冲突检测"],
        },
        "成员Agent 1 (Code Analyzer)": {
            "职责": "代码分析、缺陷检测、性能分析",
            "工具": ["静态分析器", "AST解析器"],
        },
        "成员Agent 2 (Test Generator)": {
            "职责": "生成测试用例、执行测试、报告覆盖率",
            "工具": ["测试框架", "mock生成器"],
        },
        "成员Agent 3 (Documenter)": {
            "职责": "生成文档、更新API说明、创建变更日志",
            "工具": ["文档模板", "diff工具"],
        },
    }
    
    for agent, info in team_structure.items():
        print(f"  {agent}:")
        print(f"    职责: {info['职责']}")
        print(f"    工具: {', '.join(info['工具'])}")
        print()
    
    # 协作工作流
    workflow = [
        "1. Coordinator接收任务: '审查并优化payment-service模块'",
        "2. Coordinator分解为3个子任务",
        "3. Agent 1: 扫描代码库, 发现3个性能瓶颈",
        "4. Agent 2: 为瓶颈函数生成压力测试",
        "5. Agent 3: 基于测试结果生成优化文档",
        "6. Coordinator: 整合结果, 输出最终报告"
    ]
    print("  Agent Teams工作流示例:")
    for step in workflow:
        print(f"    {step}")

# 合作伙伴集成示例: 通过WorkBuddy调用V4.1 Flash

def codebuddy_review_with_deepseek(
    code_content: str,
    language: str = "python",
    review_focus: list = None
) -> dict:
    """
    通过CodeBuddy + DeepSeek V4.1 Flash进行代码审查
    
    Args:
        code_content: 代码内容
        language: 编程语言
        review_focus: 审查重点 (performance, security, style, etc.)
    """
    if review_focus is None:
        review_focus = ["performance", "security", "correctness"]
    
    system_prompt = f"""You are CodeBuddy, powered by DeepSeek V4.1 Flash.
Review the following {language} code with focus on: {', '.join(review_focus)}.

For each issue found, provide:
1. Severity (critical/major/minor)
2. Line number
3. Description of the problem
4. Suggested fix with code example
5. Expected improvement (e.g., "reduces latency by 30%")"""
    
    # CodeBuddy内部调用DeepSeek API
    payload = {
        "model": "deepseek-flash",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"```{language}\n{code_content}\n```"}
        ],
        "max_tokens": 8192,
        "effort": 85,  # 代码审查使用高强度推理
        "temperature": 0.1,
    }
    
    # 此处为CodeBuddy内部实现
    # 实际调用通过WorkBuddy平台完成
    return payload  # 返回Payload作为示例