1. Python调用大模型API入门实践
大模型API已经成为开发者快速接入AI能力的重要途径。作为一名长期使用Python进行AI开发的工程师,我发现很多初学者在调用大模型API时容易陷入一些常见陷阱。本文将分享我从零开始调用各类大模型API的完整经验,包括从环境准备到实际调用的全流程,以及那些官方文档不会告诉你的实用技巧。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与基础配置
2.1 Python环境搭建
对于大模型API调用,我推荐使用Python 3.8+版本。这个版本区间既稳定又兼容大多数AI相关库。安装完成后,务必设置好环境变量:
bash复制# Linux/macOS
export PATH=$PATH:/path/to/python
# Windows
setx PATH "%PATH%;C:\path\to\python"
验证安装是否成功:
python复制import sys
print(sys.version)
注意:避免使用Python 2.x版本,许多现代AI库已不再支持。我曾在项目中因为版本问题浪费了半天时间排查兼容性问题。
2.2 必备库安装
大模型API调用通常需要以下核心库:
bash复制pip install requests openai tiktoken python-dotenv
requests:HTTP请求库(比urllib更友好)openai:官方SDK(也适配其他兼容API)tiktoken:Token计数器(避免超出上下文限制)python-dotenv:管理API密钥等敏感信息
我习惯使用虚拟环境隔离项目依赖:
bash复制python -m venv llm_env
source llm_env/bin/activate # Linux/macOS
llm_env\Scripts\activate # Windows
3. API密钥获取与管理
3.1 主流平台API申请
目前常用的大模型API平台包括:
- DeepSeek(支持deepseek-v4-pro/deepseek-v4-flash)
- 书生·浦语
- 智谱AI
- Kimi Chat
以DeepSeek为例的申请流程:
- 注册平台账号
- 进入"API管理"页面
- 创建新应用获取API Key
- 查看免费额度和使用限制
3.2 安全存储API密钥
绝对不要将API密钥硬编码在脚本中!我采用.env文件管理:
ini复制# .env文件
DEEPSEEK_API_KEY=sk-your-key-here
然后在Python中安全加载:
python复制from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("DEEPSEEK_API_KEY")
重要:将
.env加入.gitignore,避免密钥意外提交到代码仓库。我曾见过开发者因此导致数千美元的API滥用。
4. 基础API调用实践
4.1 最简单的GET请求
对于提供简单问答接口的API:
python复制import requests
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
params = {
"model": "deepseek-v4-pro",
"prompt": "Python如何连接MySQL数据库?",
"max_tokens": 500
}
response = requests.get(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
params=params
)
print(response.json())
4.2 处理流式响应
大模型API常支持流式传输,节省等待时间:
python复制def stream_response(prompt):
data = {
"model": "deepseek-v4-flash",
"messages": [{"role": "user", "content": prompt}],
"stream": True
}
with requests.post(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
json=data,
stream=True
) as response:
for chunk in response.iter_lines():
if chunk:
print(json.loads(chunk.decode('utf-8')))
5. 高级功能实现
5.1 上下文管理
维持多轮对话的关键是正确管理消息历史:
python复制conversation = [
{"role": "system", "content": "你是一位Python专家"},
{"role": "user", "content": "如何用Python发送HTTP请求?"}
]
def chat(message):
conversation.append({"role": "user", "content": message})
response = requests.post(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
json={
"model": "deepseek-v4-pro",
"messages": conversation,
"temperature": 0.7
}
)
assistant_reply = response.json()["choices"][0]["message"]
conversation.append(assistant_reply)
return assistant_reply["content"]
5.2 Token计数与限制
避免触发"maximum context length"错误:
python复制import tiktoken
def num_tokens(text, model="deepseek-v4-pro"):
enc = tiktoken.encoding_for_model(model)
return len(enc.encode(text))
message = "请解释Python的装饰器"
print(f"Token数量: {num_tokens(message)}")
当对话历史过长时,需要智能截断:
python复制def trim_conversation(conversation, max_tokens=4000):
total = sum(num_tokens(msg["content"]) for msg in conversation)
while total > max_tokens and len(conversation) > 1:
removed = conversation.pop(1) # 保留系统提示
total -= num_tokens(removed["content"])
return conversation
6. 错误处理与调试
6.1 常见API错误处理
python复制try:
response = requests.post(api_url, headers=headers, json=data)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as err:
if err.response.status_code == 400:
error_data = err.response.json()
if "maximum context length" in error_data.get("error", {}).get("message", ""):
print("错误:上下文长度超出限制")
# 自动触发对话截断逻辑
return trim_and_retry(conversation)
elif err.response.status_code == 429:
print("错误:请求过于频繁")
time.sleep(10) # 指数退避更好
return make_request() # 重试
else:
raise
6.2 请求重试机制
实现带指数退避的智能重试:
python复制def make_request_with_retry(max_retries=3):
retry_delay = 1
for attempt in range(max_retries):
try:
return make_api_call()
except (requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
if attempt == max_retries - 1:
raise
time.sleep(retry_delay)
retry_delay *= 2 # 指数退避
7. 性能优化技巧
7.1 批量处理请求
对于需要处理多个独立查询的场景:
python复制from concurrent.futures import ThreadPoolExecutor
def batch_process_queries(queries):
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [
executor.submit(process_single_query, query)
for query in queries
]
return [f.result() for f in futures]
7.2 缓存常见响应
使用磁盘缓存避免重复请求:
python复制from diskcache
