OpenAI Codex限额优化实战:从提示词设计到成本控制

import openai
from datetime import datetime
# 设置你的 API 密钥
openai.api_key = "你的API密钥"
def check_usage():
    try:
        # 获取当前使用量摘要（通常为当月数据）
        usage = openai.Usage.retrieve()
        print(f"截至 {datetime.now().strftime('%Y-%m-%d %H:%M')} 的使用情况：")
        print(f"总 Token 使用量: {usage.total_tokens}")
        print(f"提示词 Token: {usage.prompt_tokens}")
        print(f"补全 Token: {usage.completion_tokens}")
        # 注意：具体字段名称可能随 API 版本更新，请以官方文档为准
    except Exception as e:
        print(f"查询使用量时出错: {e}")
if __name__ == "__main__":
    check_usage()

import openai
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class CodexMonitor:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.total_tokens = 0
        self.request_count = 0
    
    def generate_code(self, prompt, max_tokens=150, temperature=0.7):
        try:
            start_time = time.time()
            response = openai.Completion.create(
                engine="code-davinci-002",  # 根据实际使用的模型调整
                prompt=prompt,
                max_tokens=max_tokens,
                temperature=temperature,
                n=1,
                stop=None
            )
            end_time = time.time()
            
            # 记录使用量
            usage = response.usage
            self.total_tokens += usage.total_tokens
            self.request_count += 1
            
            logging.info(f"请求 #{self.request_count}: {usage.total_tokens} tokens, 耗时: {end_time - start_time:.2f}s")
            logging.info(f"累计 Token: {self.total_tokens}, 累计请求: {self.request_count}")
            
            return response.choices[0].text.strip()
        except openai.error.RateLimitError as e:
            logging.error(f"速率限制触发: {e}")
            # 实现指数退避重试逻辑
            time.sleep(60)  # 等待1分钟后重试
            return self.generate_code(prompt, max_tokens, temperature)
        except Exception as e:
            logging.error(f"请求异常: {e}")
            return None

# 使用示例
monitor = CodexMonitor("你的API密钥")
result = monitor.generate_code("# Python函数，计算斐波那契数列\ndef fibonacci")

"""
请写一个Python函数，它要能够接受一个整数列表作为输入参数，然后计算这个列表中所有偶数的平方和，最后返回这个和值。
这个函数应该处理空列表的情况，并且只考虑正偶数，忽略负数和奇数。
函数名最好叫做calculate_even_squares_sum，这样见名知义。
"""

# 计算列表中偶数的平方和
def calculate_even_squares_sum(numbers):
    # 处理空列表
    if not numbers:
        return 0
    total = 0
    for num in numbers:
        # 只处理正偶数
        if num > 0 and num % 2 == 0:
            total += num * num
    return total

# 优化后的参数设置
response = openai.Completion.create(
    engine="code-davinci-002",
    prompt="# 反转字符串的函数\ndef reverse_string(s):",
    max_tokens=100,  # 足够生成一个简单函数
    temperature=0.3,  # 低随机性，保证代码正确性
    stop=["\n\n", "#"],  # 遇到空行或注释开始处停止
    n=1
)

import hashlib
import json
import os

class CodexCache:
    def __init__(self, cache_file="codex_cache.json"):
        self.cache_file = cache_file
        self.cache = self._load_cache()
    
    def _load_cache(self):
        if os.path.exists(self.cache_file):
            with open(self.cache_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {}
    
    def _save_cache(self):
        with open(self.cache_file, 'w', encoding='utf-8') as f:
            json.dump(self.cache, f, ensure_ascii=False, indent=2)
    
    def get_cache_key(self, prompt, parameters):
        """基于提示词和参数生成唯一缓存键"""
        content = prompt + json.dumps(parameters, sort_keys=True)
        return hashlib.md5(content.encode('utf-8')).hexdigest()
    
    def get_cached_result(self, prompt, parameters):
        key = self.get_cache_key(prompt, parameters)
        return self.cache.get(key)
    
    def set_cached_result(self, prompt, parameters, result):
        key = self.get_cache_key(prompt, parameters)
        self.cache[key] = {
            'result': result,
            'timestamp': time.time()
        }
        self._save_cache()

# 集成缓存的使用示例
cache = CodexCache()
prompt = "# 计算阶乘的函数\ndef factorial(n):"

# 检查缓存
cached = cache.get_cached_result(prompt, {"max_tokens": 100, "temperature": 0.3})
if cached:
    print("使用缓存结果:", cached['result'])
else:
    # 调用 API
    result = monitor.generate_code(prompt, max_tokens=100, temperature=0.3)
    if result:
        cache.set_cached_result(prompt, {"max_tokens": 100, "temperature": 0.3}, result)

import openai
import time
from openai.error import RateLimitError, APIError, ServiceUnavailableError

class RobustCodexClient:
    def __init__(self, api_key, max_retries=5):
        openai.api_key = api_key
        self.max_retries = max_retries
    
    def generate_with_retry(self, prompt, **kwargs):
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                response = openai.Completion.create(
                    engine="code-davinci-002",
                    prompt=prompt,
                    **kwargs
                )
                return response.choices[0].text.strip()
                
            except RateLimitError as e:
                last_exception = e
                wait_time = (2 ** attempt) + 1  # 指数退避：2, 5, 11, 23, 47秒
                print(f"速率限制，第 {attempt+1} 次重试，等待 {wait_time} 秒")
                time.sleep(wait_time)
                
            except (APIError, ServiceUnavailableError) as e:
                last_exception = e
                wait_time = (attempt + 1) * 5  # 线性退避：5, 10, 15, 20, 25秒
                print(f"服务异常，第 {attempt+1} 次重试，等待 {wait_time} 秒")
                time.sleep(wait_time)
                
            except Exception as e:
                # 其他错误通常不需要重试
                print(f"不可重试错误: {e}")
                return None
        
        print(f"经过 {self.max_retries} 次重试后仍失败: {last_exception}")
        return None

    def generate_with_fallback(self, prompt, **kwargs):
        """主方案失败时使用备选方案"""
        result = self.generate_with_retry(prompt, **kwargs)
        
        if result is None:
            print("Codex API 调用失败，使用本地备选方案")
            # 这里可以实现简单的本地代码生成或返回预设代码片段
            result = self.local_fallback(prompt)
        
        return result
    
    def local_fallback(self, prompt):
        """简单的本地备选方案"""
        # 基于提示词关键词返回预设代码模板
        if "fibonacci" in prompt.lower():
            return """def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)"""
        elif "factorial" in prompt.lower():
            return """def factorial(n):
    if n == 0:
        return 1
    return n * factorial(n-1)"""
        else:
            return "# 无法生成代码，请检查API状态或提示词"

# 使用示例
client = RobustCodexClient("你的API密钥")
code = client.generate_with_fallback("# 生成斐波那契函数\ndef fibonacci")

import smtplib
from email.mime.text import MimeText
from datetime import datetime

class UsageMonitor:
    def __init__(self, api_key, warning_threshold=0.8, critical_threshold=0.95):
        self.api_key = api_key
        self.warning_threshold = warning_threshold
        self.critical_threshold = critical_threshold
    
    def check_usage_and_alert(self, monthly_limit=100000):
        """检查使用量并在超过阈值时发送警报"""
        try:
            usage = openai.Usage.retrieve()
            current_usage = usage.total_tokens
            usage_ratio = current_usage / monthly_limit
            
            if usage_ratio >= self.critical_threshold:
                self.send_alert("CRITICAL", current_usage, monthly_limit, usage_ratio)
            elif usage_ratio >= self.warning_threshold:
                self.send_alert("WARNING", current_usage, monthly_limit, usage_ratio)
            else:
                print(f"使用量正常: {current_usage}/{monthly_limit} ({usage_ratio:.1%})")
                
        except Exception as e:
            print(f"检查使用量失败: {e}")
    
    def send_alert(self, level, current, limit, ratio):
        subject = f"{level}: OpenAI API 使用量警报"
        message = f"""
        API 使用量已达到警戒水平:
        
        当前使用: {current} tokens
        月度限额: {limit} tokens
        使用比例: {ratio:.1%}
        检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        
        建议立即检查应用日志，优化提示词或调整调用频率。
        """
        
        # 这里实现邮件发送逻辑（需要配置SMTP）
        print(f"发送警报: {subject}")
        print(message)

# 定时执行检查
monitor = UsageMonitor("你的API密钥")
monitor.check_usage_and_alert(monthly_limit=100000)  # 根据实际限额调整