DeepSeek V4 深度解析：中国开源模型如何以 1/10 价格逼近 GPT-5.5 的性能

import requests
import json

# DeepSeek V4 API 调用示例
def deepseek_chat(messages, model="deepseek-chat", max_tokens=2048):
    """
    调用 DeepSeek V4 API 进行对话
    
    Args:
        messages: 消息列表，格式 [{"role": "user", "content": "..."}]
        model: 模型名称，"deepseek-chat" 对应 V4-Pro
        max_tokens: 最大输出 token 数
    
    Returns:
        模型回复文本
    """
    api_key = "your-api-key-here"  # 从环境变量获取
    url = "https://api.deepseek.com/v1/chat/completions"
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    payload = {
        "model": model,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": 0.7,
        "stream": False
    }
    
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    
    return response.json()["choices"][0]["message"]["content"]

# 使用示例
messages = [
    {"role": "system", "content": "你是一个 AI 技术专家"},
    {"role": "user", "content": "请解释 MoE 架构的核心原理"}
]

response = deepseek_chat(messages)
print(response)

# DeepSeek V4 vs OpenAI API 成本对比脚本
import requests
from typing import Dict, List

class ModelCostComparator:
    """对比不同 API 模型的成本和性能"""
    
    MODELS = {
        "deepseek-v4-pro": {
            "url": "https://api.deepseek.com/v1/chat/completions",
            "key_env": "DEEPSEEK_API_KEY",
            "input_price": 0.14,   # $/M tokens
            "output_price": 0.28,  # $/M tokens
        },
        "gpt-5.5": {
            "url": "https://api.openai.com/v1/chat/completions",
            "key_env": "OPENAI_API_KEY",
            "input_price": 2.50,   # $/M tokens
            "output_price": 10.00, # $/M tokens
        },
    }
    
    @staticmethod
    def estimate_cost(model_name: str, input_tokens: int, 
                       output_tokens: int) -> Dict:
        """估算一次 API 调用的成本"""
        model = ModelCostComparator.MODELS[model_name]
        input_cost = (input_tokens / 1_000_000) * model["input_price"]
        output_cost = (output_tokens / 1_000_000) * model["output_price"]
        
        return {
            "model": model_name,
            "input_cost": f"${input_cost:.6f}",
            "output_cost": f"${output_cost:.6f}",
            "total_cost": f"${input_cost + output_cost:.6f}",
        }
    
    @classmethod
    def compare(cls, input_tokens=1000, output_tokens=500):
        """对比所有模型的成本"""
        results = []
        for name in cls.MODELS:
            cost = cls.estimate_cost(name, input_tokens, output_tokens)
            results.append(cost)
        return results

# 运行对比
comparator = ModelCostComparator()
results = comparator.compare(input_tokens=2000, output_tokens=1000)

print("📊 API 调用成本对比 (2K 输入 + 1K 输出):")
print("-" * 50)
for r in results:
    print(f"{r['model']:20s} | {r['total_cost']}")

# 输出:
# deepseek-v4-pro      | $0.000560
# gpt-5.5              | $0.015000
# 
# DeepSeek V4-Pro 成本仅为 GPT-5.5 的 1/27！

# DeepSeek V4-Flash 批量处理示例
import asyncio
import aiohttp
from typing import List, Dict
import time

class DeepSeekBatchProcessor:
    """使用 V4-Flash 进行批量文本处理"""
    
    def __init__(self, api_key: str, concurrency: int = 10):
        self.api_key = api_key
        self.concurrency = concurrency
        self.base_url = "https://api.deepseek.com/v1/chat/completions"
    
    async def _process_single(self, session: aiohttp.ClientSession, 
                               text: str) -> Dict:
        """处理单条文本"""
        payload = {
            "model": "deepseek-chat",  # V4-Flash 也是此端点
            "messages": [
                {"role": "user", "content": f"请对以下文本进行分类：{text}"}
            ],
            "max_tokens": 50,
            "temperature": 0.1,  # 低温度保证一致性
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        async with session.post(self.base_url, json=payload, 
                                 headers=headers) as resp:
            data = await resp.json()
            return {
                "input": text[:50] + "...",
                "category": data["choices"][0]["message"]["content"],
                "tokens_used": data["usage"]["total_tokens"]
            }
    
    async def process_batch(self, texts: List[str]) -> List[Dict]:
        """批量处理文本列表"""
        semaphore = asyncio.Semaphore(self.concurrency)
        
        async def limited_process(text):
            async with semaphore:
                async with aiohttp.ClientSession() as session:
                    return await self._process_single(session, text)
        
        start = time.time()
        tasks = [limited_process(t) for t in texts]
        results = await asyncio.gather(*tasks)
        elapsed = time.time() - start
        
        print(f"✅ 处理 {len(texts)} 条文本，耗时 {elapsed:.2f} 秒")
        print(f"   平均每条: {elapsed/len(texts)*1000:.0f}ms")
        return results

# 使用示例
async def main():
    processor = DeepSeekBatchProcessor(api_key="your-key", concurrency=20)
    
    texts = [f"这是一条需要分类的文本 {i}" for i in range(100)]
    results = await processor.process_batch(texts)
    
    # 统计结果
    categories = {}
    total_tokens = 0
    for r in results:
        cat = r["category"]
        categories[cat] = categories.get(cat, 0) + 1
        total_tokens += r["tokens_used"]
    
    print(f"\n📊 分类结果:")
    for cat, count in sorted(categories.items()):
        print(f"  {cat}: {count}")
    print(f"\n💰 总 Token 消耗: {total_tokens}")
    print(f"   预估成本: ${total_tokens / 1_000_000 * 0.14:.6f}")

# asyncio.run(main())

# 智能代码审查助手 - 基于 DeepSeek V4
import requests
import os
from pathlib import Path
from typing import List, Dict

class CodeReviewer:
    """基于 DeepSeek V4 的智能代码审查器"""
    
    REVIEW_PROMPT = """你是一个资深代码审查专家。请对以下代码进行全面审查：

审查维度：
1. 代码质量：可读性、命名规范、代码结构
2. 安全性：常见漏洞（SQL 注入、XSS、命令注入等）
3. 性能：时间/空间复杂度、潜在瓶颈
4. 最佳实践：语言特性使用、设计模式

请以 JSON 格式返回审查结果：
{{
    "score": 0-100,
    "issues": [
        {{"severity": "critical|warning|info", "line": 行号, "message": "描述", "suggestion": "建议"}}
    ],
    "summary": "总体评价"
}}

代码：
{code}
"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
        self.base_url = "https://api.deepseek.com/v1/chat/completions"
    
    def review_file(self, filepath: str) -> Dict:
        """审查单个文件"""
        code = Path(filepath).read_text(encoding="utf-8")
        
        # 对于大文件，只审查关键部分
        if len(code) > 10000:
            code = code[:5000] + "\n...[truncated]..."
        
        return self._review_code(code, filepath)
    
    def _review_code(self, code: str, filename: str = "code") -> Dict:
        """执行代码审查"""
        prompt = self.REVIEW_PROMPT.format(code=code)
        
        response = requests.post(
            self.base_url,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "deepseek-chat",
                "messages": [
                    {"role": "system", "content": "你是一个资深代码审查专家"},
                    {"role": "user", "content": prompt}
                ],
                "max_tokens": 2048,
                "temperature": 0.2,
                "response_format": {"type": "json_object"}
            },
            timeout=60
        )
        
        import json
        result = json.loads(response.json()["choices"][0]["message"]["content"])
        result["filename"] = filename
        return result
    
    def review_project(self, directory: str) -> List[Dict]:
        """审查整个项目"""
        results = []
        
        # 只审查 Python 文件
        for filepath in Path(directory).rglob("*.py"):
            # 跳过虚拟环境和缓存目录
            if any(skip in str(filepath) for skip in ['.venv', '__pycache__', 'node_modules']):
                continue
            
            print(f"📝 审查: {filepath}")
            result = self.review_file(str(filepath))
            results.append(result)
            
            # 打印摘要
            score = result.get("score", "N/A")
            issues = len(result.get("issues", []))
            print(f"   评分: {score}/100 | 问题: {issues} 个")
        
        return results

# 使用示例
if __name__ == "__main__":
    reviewer = CodeReviewer()
    
    # 审查单个文件
    result = reviewer.review_file("my_script.py")
    print(f"\n📊 审查结果: {result['filename']}")
    print(f"   评分: {result['score']}/100")
    for issue in result.get("issues", []):
        icon = {"critical": "🔴", "warning": "🟡", "info": "ℹ️"}.get(
            issue["severity"], "❓")
        print(f"   {icon} L{issue['line']}: {issue['message']}")
        print(f"      💡 {issue['suggestion']}")
    
    # 审查整个项目
    # results = reviewer.review_project("./my_project")