【大模型测试】Python调用大模型API接口开发指南，详细介绍，指导新人开展调试

概述

本指南将详细介绍如何使用Python调用大模型API接口，帮助新人快速掌握相关技术并完成项目开发。大模型API（如OpenAI GPT、Anthropic Claude等）提供了强大的自然语言处理能力，通过简单的API调用即可集成到各种应用中。

流程图

开发步骤详解

1. 环境准备

首先需要安装必要的Python库：

bash

pip install openai requests python-dotenv

2. 获取API密钥

通常需要在大模型服务提供商处注册账号并获取API密钥：

3. 项目结构

text

project/
├── main.py # 主程序
├── config.py # 配置文件
├── .env # 环境变量（包含API密钥）
├── requirements.txt # 依赖列表
└── utils/ # 工具函数
└── api_utils.py # API调用工具

4. 代码实现

配置文件 (config.py)

python

# 模型配置
MODEL_CONFIG = {
"openai": {
"model": "gpt-3.5-turbo",
"api_base": "https://api.openai.com/v1",
"max_tokens": 1000,
"temperature": 0.7
},
"anthropic": {
"model": "claude-2",
"api_base": "https://api.anthropic.com/v1",
"max_tokens": 1000,
"temperature": 0.7
}
}

# 其他配置
REQUEST_TIMEOUT = 30
MAX_RETRIES = 3

环境变量文件 (.env)

text

OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_api_key_here

API工具函数 (utils/api_utils.py)

python

import os
import json
import requests
from typing import Dict, Any, List
from dotenv import load_dotenv
import time
from config import MODEL_CONFIG, REQUEST_TIMEOUT, MAX_RETRIES

# 加载环境变量
load_dotenv()

class LLMAPI:
def __init__(self, provider: str = "openai"):
self.provider = provider
self.api_key = os.getenv(f"{provider.upper()}_API_KEY")
self.config = MODEL_CONFIG[provider]
self.api_base = self.config["api_base"]

if not self.api_key:
raise ValueError(f"{provider.upper()}_API_KEY not found in environment variables")

def call_api(self, messages: List[Dict[str, str]], **kwargs) -> Dict[str, Any]:
"""调用大模型API"""

# 合并配置参数
params = {
"model": self.config["model"],
"max_tokens": self.config["max_tokens"],
"temperature": self.config["temperature"],
**kwargs
}

# 根据不同提供商构建请求
if self.provider == "openai":
return self._call_openai_api(messages, params)
elif self.provider == "anthropic":
return self._call_anthropic_api(messages, params)
else:
raise ValueError(f"Unsupported provider: {self.provider}")

def _call_openai_api(self, messages: List[Dict[str, str]], params: Dict[str, Any]) -> Dict[str, Any]:
"""调用OpenAI API"""
url = f"{self.api_base}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}

data = {
"messages": messages,
**params
}

for attempt in range(MAX_RETRIES):
try:
response = requests.post(
url,
headers=headers,
json=data,
timeout=REQUEST_TIMEOUT
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == MAX_RETRIES – 1:
raise e
time.sleep(2 ** attempt) # 指数退避

return {}

def _call_anthropic_api(self, messages: List[Dict[str, str]], params: Dict[str, Any]) -> Dict[str, Any]:
"""调用Anthropic API"""
url = f"{self.api_base}/messages"
headers = {
"Content-Type": "application/json",
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01"
}

# 转换消息格式为Anthropic格式
system_message = ""
conversation_messages = []

for msg in messages:
if msg["role"] == "system":
system_message = msg["content"]
else:
conversation_messages.append({
"role": msg["role"],
"content": msg["content"]
})

data = {
"model": params["model"],
"max_tokens": params["max_tokens"],
"temperature": params["temperature"],
"messages": conversation_messages,
"system": system_message
}

for attempt in range(MAX_RETRIES):
try:
response = requests.post(
url,
headers=headers,
json=data,
timeout=REQUEST_TIMEOUT
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == MAX_RETRIES – 1:
raise e
time.sleep(2 ** attempt) # 指数退避

return {}

def extract_response_text(self, response: Dict[str, Any]) -> str:
"""从API响应中提取文本内容"""
if self.provider == "openai":
return response["choices"][0]["message"]["content"]
elif self.provider == "anthropic":
return response["content"][0]["text"]
else:
return ""

主程序 (main.py)

python

from utils.api_utils import LLMAPI

def main():
# 初始化API客户端
llm_api = LLMAPI(provider="openai") # 可以替换为"anthropic"

# 构建对话消息
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "请解释一下机器学习的基本概念。"}
]

try:
# 调用API
response = llm_api.call_api(messages)

# 提取响应文本
response_text = llm_api.extract_response_text(response)

print("API响应:")
print(response_text)

# 可选：保存响应
with open("response.txt", "w", encoding="utf-8") as f:
f.write(response_text)

except Exception as e:
print(f"调用API时出错: {e}")

if __name__ == "__main__":
main()

5. 高级功能实现

流式响应处理

python

def stream_response(messages: List[Dict[str, str]], **kwargs):
"""处理流式响应"""
llm_api = LLMAPI(provider="openai")

# 修改参数以支持流式响应
stream_params = {**kwargs, "stream": True}

try:
response = llm_api.call_api(messages, **stream_params)

# 处理流式数据
for chunk in response:
# 解析并处理每个数据块
delta = chunk.get("choices", [{}])[0].get("delta", {})
if "content" in delta:
print(delta["content"], end="", flush=True)

except Exception as e:
print(f"流式请求错误: {e}")

异步请求处理

python

import aiohttp
import asyncio

async def async_call_api(session: aiohttp.ClientSession, messages: List[Dict[str, str]], **kwargs):
"""异步调用API"""
llm_api = LLMAPI(provider="openai")

url = f"{llm_api.api_base}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {llm_api.api_key}"
}

data = {
"model": llm_api.config["model"],
"messages": messages,
"max_tokens": llm_api.config["max_tokens"],
"temperature": llm_api.config["temperature"],
**kwargs
}

async with session.post(url, json=data, headers=headers) as response:
return await response.json()

async def main_async():
"""主异步函数"""
messages = [
{"role": "user", "content": "请解释一下深度学习的基本概念。"}
]

async with aiohttp.ClientSession() as session:
response = await async_call_api(session, messages)
print(response["choices"][0]["message"]["content"])

6. 错误处理与重试机制

python

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustLLMAPI(LLMAPI):
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_api_with_retry(self, messages: List[Dict[str, str]], **kwargs):
"""带重试机制的API调用"""
try:
return self.call_api(messages, **kwargs)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# 速率限制错误，需要重试
print("达到速率限制，等待后重试…")
raise
else:
# 其他HTTP错误，不需要重试
print(f"HTTP错误: {e}")
return {}
except requests.exceptions.RequestException as e:
print(f"请求错误: {e}")
raise # 其他网络错误，需要重试

7. 实际应用示例

构建简单的聊天机器人

python

class ChatBot:
def __init__(self, provider="openai"):
self.llm_api = LLMAPI(provider)
self.conversation_history = []

def add_message(self, role: str, content: str):
"""添加消息到对话历史"""
self.conversation_history.append({"role": role, "content": content})

def get_response(self, user_input: str):
"""获取机器人响应"""
self.add_message("user", user_input)

try:
response = self.llm_api.call_api(self.conversation_history)
bot_response = self.llm_api.extract_response_text(response)

self.add_message("assistant", bot_response)
return bot_response
except Exception as e:
return f"抱歉，发生了错误: {e}"

def run(self):
"""运行聊天机器人"""
print("聊天机器人已启动! 输入'退出'结束对话。")

while True:
user_input = input("你: ")
if user_input.lower() in ["退出", "exit", "quit"]:
break

response = self.get_response(user_input)
print(f"机器人: {response}")

# 使用示例
if __name__ == "__main__":
bot = ChatBot()
bot.run()

最佳实践

总结

通过本指南，您应该已经掌握了使用Python调用大模型API的基本方法。关键步骤包括：

随着项目复杂度的增加，您可以进一步探索高级功能如流式处理、异步请求、缓存机制等，以构建更强大、高效的应用程序。