前言

大家好，我是船长。

今天DeepSeek V4开放了百万级上下文能力，这对于AI应用开发者来说是一个重大利好。作为数据分析师和Python开发者，如何快速上手调用这些大模型API，是一个很实用的技能。

今天分享Python调用大模型API的完整实战代码，包括DeepSeek、GPT、Claude等主流模型的调用方式，以及token计算、流式输出、错误处理等实战技巧。

一、环境准备

首先安装必要的依赖：

pip install openai anthropic requests tiktoken

二、调用DeepSeek API

DeepSeek是国内领先的大模型提供商，API调用方式与OpenAI兼容：

from openai import OpenAI

# 初始化客户端
client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)

# 同步调用
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个数据分析助手"},
        {"role": "user", "content": "请解释什么是数据清洗"}
    ],
    max_tokens=1000,
    temperature=0.7
)

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

三、流式输出实现

流式输出可以提升用户体验，特别适合长文本生成场景：

# 流式调用示例
stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "写一个Python快速排序"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

四、Token计算与成本控制

使用tiktoken库计算token数量，控制API调用成本：

import tiktoken

def count_tokens(text, model="gpt-3.5-turbo"):
    """计算文本token数量"""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def estimate_cost(text, model="gpt-3.5-turbo"):
    """估算API调用成本"""
    tokens = count_tokens(text, model)
    # gpt-3.5-turbo: $0.002/1K tokens (2024价格)
    cost_per_1k = 0.002
    return tokens / 1000 * cost_per_1k

# 使用示例
text = "这是一段测试文本，用于计算token数量"
print(f"Token数: {count_tokens(text)}")
print(f"预估成本: ${estimate_cost(text):.6f}")

五、错误处理与重试机制

生产环境必须做好错误处理和重试机制：

import time
from openai import RateLimitError, APIError

def call_with_retry(client, messages, max_retries=3, delay=1):
    """带重试机制的API调用"""
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="deepseek-chat",
                messages=messages
            )
            return response
        except RateLimitError:
            print(f"限流，第{attempt+1}次重试...")
            time.sleep(delay * (attempt + 1))
        except APIError as e:
            print(f"API错误: {e}")
            if attempt == max_retries - 1:
                raise
    return None

# 使用示例
messages = [{"role": "user", "content": "你好"}]
response = call_with_retry(client, messages)

六、实用封装类

封装一个通用的LLM调用类，方便项目使用：

class LLMClient:
    def __init__(self, provider="deepseek", api_key=None):
        self.provider = provider
        if provider == "deepseek":
            self.client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")
            self.model = "deepseek-chat"
        elif provider == "openai":
            self.client = OpenAI(api_key=api_key)
            self.model = "gpt-3.5-turbo"
    
    def chat(self, prompt, system=None, **kwargs):
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        return response.choices[0].message.content
    
    def batch_chat(self, prompts):
        """批量处理多个prompt"""
        return [self.chat(p) for p in prompts]

# 使用示例
llm = LLMClient(provider="deepseek", api_key="your-key")
result = llm.chat("解释什么是数据分析")

七、总结

今天分享了Python调用大模型API的完整实战技能：

• DeepSeek API调用（与OpenAI兼容）
• 流式输出实现
• Token计算与成本控制
• 错误处理与重试机制
• 通用LLM封装类

大模型API调用是AI应用开发的基础技能。掌握了这些技能，你就可以快速构建各种AI辅助工具，提升工作效率。

觉得有用，转给需要的朋友。