09 - 常见问题与故障排除

import time
from openai import OpenAI, APIError, RateLimitError, Timeout

client = OpenAI(api_key="your-api-key")

def call_api_with_retry(
    prompt: str,
    max_retries: int = 3,
    base_delay: int = 1
):
    """带重试的 API 调用"""
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                timeout=30  # 设置超时时间
            )
            return response.choices[0].message.content
        
        except Timeout:
            print(f"⏱️  超时，重试 {attempt + 1}/{max_retries}")
            time.sleep(base_delay * (2 ** attempt))  # 指数退避
        
        except RateLimitError:
            print(f"🚦 速率限制，等待...")
            time.sleep(60)  # 等待1分钟
        
        except APIError as e:
            print(f"❌ API 错误：{e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(base_delay * (2 ** attempt))
    
    raise Exception(f"Failed after {max_retries} retries")

# 使用
result = call_api_with_retry("写一个冒泡排序")

class CodebaseManager:
    """代码库管理器"""
    
    def __init__(self, project_path: str):
        self.project_path = project_path
        self.index = self.build_index()
    
    def build_index(self) -> dict:
        """构建代码库索引"""
        
        index = {
            "files": [],
            "functions": {},
            "classes": {},
            "imports": {}
        }
        
        # 遍历项目文件
        for file_path in self.get_python_files():
            # 解析文件
            analysis = self.analyze_file(file_path)
            
            index["files"].append(file_path)
            index["functions"].update(analysis["functions"])
            index["classes"].update(analysis["classes"])
            index["imports"].update(analysis["imports"])
        
        return index
    
    def get_relevant_context(self, query: str) -> str:
        """获取相关上下文"""
        
        # 1. 向量搜索找到相关文件
        relevant_files = self.search_files(query)
        
        # 2. 提取相关代码片段
        context = []
        for file in relevant_files[:5]:  # 只取前5个最相关的
            content = self.read_file(file)
            context.append(f"# {file}\n{content}")
        
        # 3. 构建精简的上下文
        return "\n\n".join(context)
    
    def generate_with_context(self, query: str) -> str:
        """带上下文的代码生成"""
        
        context = self.get_relevant_context(query)
        
        prompt = f"""
        项目上下文：
        {context}
        
        任务：
        {query}
        
        请基于上述上下文生成代码。
        """
        
        return call_api(prompt)

# 使用
manager = CodebaseManager("/path/to/project")
result = manager.generate_with_context(
    "添加用户注销功能"
)

def security_check(code: str) -> dict:
    """安全检查"""
    
    issues = []
    
    # 1. 检查 SQL 注入
    if re.search(r'\.format\([^)]*\).*execute', code):
        issues.append({
            "type": "SQL Injection",
            "severity": "高",
            "description": "可能存在 SQL 注入风险",
            "suggestion": "使用参数化查询"
        })
    
    # 2. 检查硬编码密钥
    if re.search(r'(api_key|password|secret)\s*=\s*["\']', code):
        issues.append({
            "type": "Hardcoded Secrets",
            "severity": "高",
            "description": "检测到硬编码的密钥",
            "suggestion": "使用环境变量"
        })
    
    # 3. 检查不安全的密码存储
    if 'password' in code and not 'hash' in code:
        issues.append({
            "type": "Insecure Password Storage",
            "severity": "高",
            "description": "密码可能未加密存储",
            "suggestion": "使用 bcrypt 或 argon2"
        })
    
    return {
        "safe": len(issues) == 0,
        "issues": issues
    }

# 使用
code = """
def login(username, password):
    query = f"SELECT * FROM users WHERE username='{username}'"
    db.execute(query)
"""

result = security_check(code)
if not result["safe"]:
    print("⚠️  发现安全问题：")
    for issue in result["issues"]:
        print(f"- {issue['type']}: {issue['description']}")
        print(f"  建议：{issue['suggestion']}")

# AI 生成的代码（性能差）
def find_duplicates(arr):
    duplicates = []
    for i in range(len(arr)):
        for j in range(i+1, len(arr)):
            if arr[i] == arr[j] and arr[i] not in duplicates:
                duplicates.append(arr[i])
    return duplicates

# 时间复杂度：O(n³)

def find_duplicates_optimized(arr):
    """
    优化后的查找重复元素
    
    时间复杂度：O(n)
    空间复杂度：O(n)
    """
    seen = set()
    duplicates = set()
    
    for item in arr:
        if item in seen:
            duplicates.add(item)
        else:
            seen.add(item)
    
    return list(duplicates)

# 性能对比：
# 输入：10000 个元素
# 原始：~15秒
# 优化：~2ms
# 提升：7500倍

import os

# 方案 1：使用代理
os.environ['HTTP_PROXY'] = 'http://your-proxy:port'
os.environ['HTTPS_PROXY'] = 'http://your-proxy:port'

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    http_client=httpx.Client(proxies="http://your-proxy:port")
)

# 方案 2：使用国内镜像服务
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.mirror.com/v1"  # 镜像地址
)

# 方案 3：使用替代服务
from anthropic import Anthropic  # Claude（可直接访问）

client = Anthropic(api_key="your-api-key")
response = client.messages.create(
    model="claude-4-sonnet",
    messages=[{"role": "user", "content": "Hello"}]
)

class CostMonitor:
    """成本监控"""
    
    def __init__(self, budget_daily: float = 10.0):
        self.budget_daily = budget_daily
        self.usage_today = 0.0
        self.reset_time = datetime.now().date()
    
    def track(self, model: str, input_tokens: int, output_tokens: int):
        """追踪使用"""
        
        # 检查是否需要重置
        if datetime.now().date() > self.reset_time:
            self.usage_today = 0.0
            self.reset_time = datetime.now().date()
        
        # 计算成本
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        self.usage_today += cost
        
        # 检查预算
        if self.usage_today > self.budget_daily:
            raise BudgetExceededError(
                f"Daily budget exceeded: ${self.usage_today:.2f} / ${self.budget_daily}"
            )
        
        # 警告
        if self.usage_today > self.budget_daily * 0.8:
            print(f"⚠️  预算警告：已使用 {self.usage_today/self.budget_daily*100:.0f}%")
    
    def report(self):
        """生成报告"""
        return {
            "date": self.reset_time,
            "usage": f"${self.usage_today:.2f}",
            "budget": f"${self.budget_daily:.2f}",
            "remaining": f"${self.budget_daily - self.usage_today:.2f}"
        }

# 使用
monitor = CostMonitor(budget_daily=10.0)

def call_api_monitored(prompt: str, model: str = "gpt-4o"):
    """带成本监控的 API 调用"""
    
    # 估算 token 数量
    input_tokens = len(prompt) // 4  # 粗略估算
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # 记录实际使用量
        monitor.track(
            model,
            response.usage.prompt_tokens,
            response.usage.completion_tokens
        )
        
        return response.choices[0].message.content
    
    except BudgetExceededError as e:
        print(f"❌ {e}")
        print(f"今日使用情况：{monitor.report()}")
        return None

# 定期查看报告
print(monitor.report())

import logging

# 启用详细日志
logging.basicConfig(level=logging.DEBUG)

# 记录请求和响应
def call_api_debug(prompt: str):
    """调试模式的 API 调用"""
    
    print("\n" + "="*50)
    print("📤 请求")
    print("="*50)
    print(f"提示词：\n{prompt}")
    print(f"字符数：{len(prompt)}")
    print(f"估算 tokens：{len(prompt) // 4}")
    
    try:
        start_time = time.time()
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        
        elapsed = time.time() - start_time
        
        print("\n" + "="*50)
        print("📥 响应")
        print("="*50)
        print(f"状态：成功")
        print(f"耗时：{elapsed:.2f}秒")
        print(f"使用 tokens：{response.usage.total_tokens}")
        print(f"输入：{response.usage.prompt_tokens}")
        print(f"输出：{response.usage.completion_tokens}")
        print(f"\n生成内容：\n{response.choices[0].message.content}")
        
        return response.choices[0].message.content
    
    except Exception as e:
        print("\n" + "="*50)
        print("❌ 错误")
        print("="*50)
        print(f"类型：{type(e).__name__}")
        print(f"信息：{str(e)}")
        raise

# 使用
result = call_api_debug("写一个快速排序")