AI 工作流专栏 09：LLM API 接入与统一适配层

pip install openai python-dotenv

# llm_basic.py
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()  # 从 .env 读取环境变量

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    # 如用中转服务，取消下一行注释并填中转地址
    # base_url="https://your-proxy.com/v1",
)

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一位资深 Python 程序员，回答简洁、带代码示例。"},
        {"role": "user", "content": "用一句话解释什么是装饰器。"},
    ],
    temperature=0.7,
    max_tokens=500,
    top_p=1.0,
)

print(response.choices[0].message.content)

# 模拟多轮对话
history = [
    {"role": "system", "content": "你是友好助手。"},
    {"role": "user", "content": "我叫小明。"},
    {"role": "assistant", "content": "你好小明！很高兴认识你。"},
    {"role": "user", "content": "我刚才说我叫什么？"},  # 测试它"记不记得"
]

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=history,
    temperature=0,
)
print(resp.choices[0].message.content)
# 输出：你刚才说你叫小明。

pip install anthropic

# claude_basic.py
import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=500,
    system="你是一位资深 Python 程序员，回答简洁、带代码示例。",  # 注意：放在顶层
    messages=[
        {"role": "user", "content": "用一句话解释什么是装饰器。"},
    ],
)

print(response.content[0].text)

# qwen_via_openai_sdk.py，通义千问
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),                  # 通义的 Key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 通义的兼容端点
)

resp = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "你好，介绍一下自己。"}],
)
print(resp.choices[0].message.content)

# deepseek_via_openai_sdk.py，DeepSeek
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",   # DeepSeek 的兼容端点
)

resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "你好，介绍一下自己。"}],
)
print(resp.choices[0].message.content)

# llm_client.py，统一模型适配层
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
from dotenv import load_dotenv

load_dotenv()


@dataclass
class ChatResult:
    """统一的返回结构，屏蔽各家的差异"""
    content: str               # 模型输出的文本
    model: str                 # 实际用的模型名
    prompt_tokens: int = 0     # 输入 token 数（用于算钱）
    completion_tokens: int = 0 # 输出 token 数
    total_tokens: int = 0


class LLMClient(ABC):
    """抽象基类：所有模型适配器都要实现 chat()"""

    @abstractmethod
    def chat(
        self,
        messages: list[dict],
        system: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
    ) -> ChatResult:
        """
        统一的对话接口。
        :param messages: [{"role": "user"/"assistant", "content": "..."}]
        :param system: 系统提示词（单独抽出，方便适配 Claude 的顶层参数）
        """
        ...


class OpenAIClient(LLMClient):
    """OpenAI 官方 + 所有兼容 OpenAI 协议的模型（通义、DeepSeek）"""

    def __init__(self, model: str, api_key: str, base_url: Optional[str] = None):
        from openai import OpenAI
        kwargs = {"api_key": api_key}
        if base_url:
            kwargs["base_url"] = base_url
        self.client = OpenAI(**kwargs)
        self.model = model

    def chat(self, messages, system=None, temperature=0.7, max_tokens=1024):
        # OpenAI 协议：system 放进 messages 第一条
        full_messages = []
        if system:
            full_messages.append({"role": "system", "content": system})
        full_messages.extend(messages)

        resp = self.client.chat.completions.create(
            model=self.model,
            messages=full_messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        usage = resp.usage
        return ChatResult(
            content=resp.choices[0].message.content,
            model=self.model,
            prompt_tokens=usage.prompt_tokens if usage else 0,
            completion_tokens=usage.completion_tokens if usage else 0,
            total_tokens=usage.total_tokens if usage else 0,
        )


class ClaudeClient(LLMClient):
    """Claude：system 是顶层参数，max_tokens 必填"""

    def __init__(self, model: str, api_key: str):
        from anthropic import Anthropic
        self.client = Anthropic(api_key=api_key)
        self.model = model

    def chat(self, messages, system=None, temperature=0.7, max_tokens=1024):
        kwargs = {
            "model": self.model,
            "max_tokens": max_tokens,   # Claude 必填
            "messages": messages,       # system 不在这里
            "temperature": temperature,
        }
        if system:
            kwargs["system"] = system   # system 作为顶层参数

        resp = self.client.messages.create(**kwargs)
        usage = resp.usage
        return ChatResult(
            content=resp.content[0].text,
            model=self.model,
            prompt_tokens=usage.input_tokens,
            completion_tokens=usage.output_tokens,
            total_tokens=usage.input_tokens + usage.output_tokens,
        )


def create_client(provider: str = "openai") -> LLMClient:
    """工厂函数：根据 provider 名字创建对应适配器。
    切换模型 = 改这一个字符串。"""
    if provider == "openai":
        return OpenAIClient(
            model="gpt-4o-mini",
            api_key=os.getenv("OPENAI_API_KEY"),
        )
    elif provider == "qwen":
        return OpenAIClient(                          # 通义复用 OpenAIClient
            model="qwen-plus",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
    elif provider == "deepseek":
        return OpenAIClient(                          # DeepSeek 也复用
            model="deepseek-chat",
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url="https://api.deepseek.com/v1",
        )
    elif provider == "claude":
        return ClaudeClient(
            model="claude-sonnet-4-20250514",
            api_key=os.getenv("ANTHROPIC_API_KEY"),
        )
    else:
        raise ValueError(f"未知的 provider: {provider}")

# app.py，上层业务，完全不知道底层是哪家模型
from llm_client import create_client

# 切换模型只改这一行！openai / claude / qwen / deepseek
client = create_client("deepseek")

result = client.chat(
    messages=[{"role": "user", "content": "用一句话解释什么是闭包。"}],
    system="你是一位资深程序员，回答简洁。",
    temperature=0,
)

print(f"[{result.model}] {result.content}")
print(f"token 用量：输入 {result.prompt_tokens} + 输出 {result.completion_tokens} = {result.total_tokens}")

# compare.py，同一问题问 4 家
import time
from llm_client import create_client

question = [{"role": "user", "content": "用 100 字解释什么是 RAG。"}]

for provider in ["openai", "claude", "qwen", "deepseek"]:
    try:
        client = create_client(provider)
        t0 = time.time()
        result = client.chat(question, system="你是 AI 技术专家。", temperature=0)
        elapsed = time.time() - t0
        print(f"\n=== {provider}（{elapsed:.1f}s，{result.total_tokens} tokens）===")
        print(result.content[:120], "...")
    except Exception as e:
        print(f"\n=== {provider} 调用失败：{e} ===")

pip install tenacity

# llm_retry.py，给适配层加重试
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
from openai import RateLimitError, APIConnectionError, APITimeoutError
from llm_client import create_client


class RobustClient:
    """包装一层，自动对可恢复错误做指数退避重试"""

    def __init__(self, provider: str, max_attempts: int = 5):
        self.inner = create_client(provider)
        self.max_attempts = max_attempts

        # 用 tenacity 装饰：遇到这三类错误才重试
        self._chat = retry(
            stop=stop_after_attempt(max_attempts),            # 最多重试 5 次
            wait=wait_exponential(multiplier=1, min=1, max=16),  # 1,2,4,8,16 秒
            retry=retry_if_exception_type(
                (RateLimitError, APIConnectionError, APITimeoutError)
            ),
            reraise=True,  # 重试用尽后抛原异常
        )(self.inner.chat)

    def chat(self, *args, **kwargs):
        return self._chat(*args, **kwargs)


# 使用
client = RobustClient("deepseek")
result = client.chat([{"role": "user", "content": "你好"}])

def chat_with_fallback(question: str, primary="deepseek", fallback="qwen") -> str:
    """主模型挂了，自动切备用模型"""
    providers = [primary, fallback]
    for p in providers:
        try:
            client = RobustClient(p)
            return client.chat([{"role": "user", "content": question}]).content
        except Exception as e:
            print(f"[降级] {p} 失败：{e}，尝试下一个...")
    raise RuntimeError("所有模型都不可用")

# .env，这个文件绝不能提交到 Git！
OPENAI_API_KEY=sk-xxxxxxxxxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxx
DASHSCOPE_API_KEY=sk-xxxxxxxxxxxx
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxx

from dotenv import load_dotenv
load_dotenv()  # 把 .env 的内容读进环境变量

import os
key = os.getenv("DEEPSEEK_API_KEY")  # 永远从环境变量取

result = client.chat([{"role": "user", "content": "解释量子力学"}])
print(f"输入 {result.prompt_tokens} tokens，输出 {result.completion_tokens} tokens")

# cost.py，按各家价格算单次费用
# 单价：每百万 token 多少美元（输入 / 输出）
PRICING = {
    "gpt-4o-mini": (0.15, 0.60),
    "gpt-4o":      (2.5, 10.0),
    "deepseek-chat": (0.14, 0.28),    # 折算成美元，1 元约 0.14 美元
    "qwen-plus":     (0.11, 0.28),
    "claude-sonnet-4-20250514": (3.0, 15.0),
}

def calc_cost(model, prompt_tokens, completion_tokens):
    in_price, out_price = PRICING.get(model, (0, 0))
    cost = (prompt_tokens / 1_000_000) * in_price + (completion_tokens / 1_000_000) * out_price
    return cost

# 示例：一次调用
cost = calc_cost("deepseek-chat", prompt_tokens=500, completion_tokens=200)
print(f"本次调用花费：${cost:.5f}")  # 约 $0.00013，几乎免费