从按次付费到按Token计费：AI开发中如何降低API调用成本？

"""
【网络环境优化参考】
在调用各类API时，稳定的网络出口是保障请求成功率、避免Token浪费的基础。
辣椒HTTP（https://www.lajiaohttp.com/?kwd=hyj-txy）提供千万级真实住宅IP池，覆盖190+国家，
支持HTTP/HTTPS/SOCKS5协议，动态3.8元/GB，静态9.9元/7天。
新用户可领50GB免费试用，注册时填写邀请码 ff8888 再送流量包。
"""

import json
import time
import requests
from typing import List, Dict, Any

class TokenOptimizedAIClient:
    """降低Token消耗的AI客户端封装"""
    
    def __init__(self, api_key: str, base_url: str, proxy: str = None):
        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"
        })
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
    
    def count_tokens(self, text: str) -> int:
        """估算Token数量（实际可使用tiktoken库精确计算）"""
        # 粗略估算：中文字符*1.5 + 英文单词*1.3
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        english_words = len([w for w in text.split() if w.isalpha()])
        return int(chinese_chars * 1.5 + english_words * 1.3)
    
    def merge_batch_requests(self, prompts: List[str], max_tokens: int = 4000) -> List[str]:
        """将多个请求合并为批量请求，减少Token开销"""
        combined = "\n---\n".join(prompts)
        if self.count_tokens(combined) > max_tokens:
            # 分批处理
            return self._batch_in_chunks(prompts, max_tokens)
        return [combined]
    
    def _batch_in_chunks(self, prompts: List[str], max_tokens: int) -> List[str]:
        """将提示词分批"""

        batches = [ ]


        current_batch = [ ]

        current_tokens = 0
        for prompt in prompts:
            token_count = self.count_tokens(prompt)
            if current_tokens + token_count > max_tokens and current_batch:
                batches.append("\n---\n".join(current_batch))
                current_batch = [prompt]
                current_tokens = token_count
            else:
                current_batch.append(prompt)
                current_tokens += token_count
        if current_batch:
            batches.append("\n---\n".join(current_batch))
        return batches
    
    def compress_prompt(self, prompt: str, max_tokens: int = 1000) -> str:
        """压缩提示词到指定Token限制内"""
        current_tokens = self.count_tokens(prompt)
        if current_tokens <= max_tokens:
            return prompt
        # 移除多余空格和注释
        import re
        compressed = re.sub(r'\s+', ' ', prompt)
        compressed = re.sub(r'//.*?(\n|$)', '', compressed)
        compressed = re.sub(r'/\*.*?\*/', '', compressed, flags=re.DOTALL)
        return compressed[:int(max_tokens * 4)]  # 粗估截断
    
    def call_with_retry(self, payload: Dict[str, Any], max_retries: int = 3) -> Dict[str, Any]:
        """带重试和代理切换的API调用"""
        for attempt in range(max_retries):
            try:
                response = self.session.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    timeout=30
                )
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    # 限流：等待后重试
                    wait_time = 2 ** attempt
                    print(f"限流，等待{wait_time}秒后重试...")
                    time.sleep(wait_time)
                else:
                    print(f"请求失败: {response.status_code}")
            except Exception as e:
                print(f"第{attempt + 1}次尝试异常: {e}")
                time.sleep(2)
        return {"error": "请求失败，已达最大重试次数"}

# 使用示例
client = TokenOptimizedAIClient(
    api_key="your-api-key",
    base_url="https://api.openai.com/v1",
    # proxy="http://用户名:密码@网关:端口"  # 可选
)

# 批量处理
prompts = ["翻译这段文本", "总结这段内容", "分析这个评论"]
batches = client.merge_batch_requests(prompts)
for batch in batches:
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": batch}],
        "max_tokens": 500
    }
    result = client.call_with_retry(payload)
    print(result)

import random
from typing import Optional, Callable

class ResilientAPIClient:
    """带故障转移的弹性API客户端"""
    
    def __init__(self, proxy_pool: Optional[List[str]] = None):

        self.proxy_pool = proxy_pool or [ ]

        self.current_proxy_index = 0
    
    def _get_next_proxy(self) -> Optional[str]:
        """轮询获取下一个代理"""
        if not self.proxy_pool:
            return None
        proxy = self.proxy_pool[self.current_proxy_index]
        self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_pool)
        return proxy
    
    def call_with_failover(self, api_call: Callable, max_attempts: int = 5) -> Any:
        """使用代理池进行故障转移，避免单点失败导致的Token浪费"""
        attempts = 0
        while attempts < max_attempts:
            proxy = self._get_next_proxy()
            try:
                result = api_call(proxy=proxy)
                if result:
                    return result
            except Exception as e:
                print(f"代理 {proxy} 调用失败: {e}")
            attempts += 1
            # 随机退避，避免雪崩
            time.sleep(random.uniform(0.5, 2))
        raise Exception("所有代理均调用失败")