随着AI技术的普及,ChatGPT中文版API示例的需求不断增长。本文将通过实际案例,详细介绍API的使用方法和最佳实践。
基础API调用示例
Python示例
import requests
import json
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
class ChatGPTAPI:
def __init__(self):
self.api_key = os.getenv('OPENAI_API_KEY')
self.api_base = os.getenv('OPENAI_API_BASE')
self.headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
def chat(self, message):
"""基础聊天接口"""
try:
response = requests.post(
f'{self.api_base}/chat/completions',
headers=self.headers,
json={
'model': 'gpt-3.5-turbo',
'messages': [{'role': 'user', 'content': message}],
'temperature': 0.7
}
)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
except Exception as e:
print(f"API调用错误: {str(e)}")
raise
# 使用示例
api = ChatGPTAPI()
response = api.chat("请介绍一下Python编程语言的特点。")
print(response)
Node.js示例
const axios = require('axios');
require('dotenv').config();
class ChatGPTAPI {
constructor() {
this.apiKey = process.env.OPENAI_API_KEY;
this.apiBase = process.env.OPENAI_API_BASE;
this.headers = {
'Authorization': `Bearer ${this.apiKey}`,
'Content-Type': 'application/json'
};
}
async chat(message) {
try {
const response = await axios.post(
`${this.apiBase}/chat/completions`,
{
model: 'gpt-3.5-turbo',
messages: [{ role: 'user', content: message }],
temperature: 0.7
},
{ headers: this.headers }
);
return response.data.choices[0].message.content;
} catch (error) {
console.error('API调用错误:', error);
throw error;
}
}
}
// 使用示例
const api = new ChatGPTAPI();
api.chat('请介绍一下JavaScript编程语言的特点。')
.then(response => console.log(response))
.catch(error => console.error(error));
高级API应用示例
多轮对话实现
class ChatSession:
def __init__(self, api):
self.api = api
self.messages = []
def add_message(self, role, content):
"""添加消息到对话历史"""
self.messages.append({
'role': role,
'content': content
})
async def chat(self, message):
"""进行多轮对话"""
self.add_message('user', message)
try:
response = await self.api.chat_async(self.messages)
self.add_message('assistant', response)
return response
except Exception as e:
print(f"对话错误: {str(e)}")
raise
# 使用示例
async def main():
api = ChatGPTAPI()
session = ChatSession(api)
# 第一轮对话
response1 = await session.chat("什么是机器学习?")
print("回答1:", response1)
# 第二轮对话
response2 = await session.chat("它有什么应用场景?")
print("回答2:", response2)
流式响应处理
class StreamingAPI:
def __init__(self):
self.api = ChatGPTAPI()
async def stream_chat(self, message):
"""流式响应接口"""
try:
async for chunk in self.api.chat_stream(message):
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
except Exception as e:
print(f"流式响应错误: {str(e)}")
raise
# 使用示例
async def handle_stream():
api = StreamingAPI()
async def print_chunk(chunk):
print(chunk, end='', flush=True)
async for chunk in api.stream_chat("请讲一个故事"):
await print_chunk(chunk)
实用功能示例
内容生成器
class ContentGenerator:
def __init__(self):
self.api = ChatGPTAPI()
async def generate_article(self, topic, style, length):
"""生成文章"""
prompt = f"""
请根据以下要求生成一篇文章:
主题:{topic}
风格:{style}
字数:{length}字左右
要求:
1. 结构清晰,层次分明
2. 观点新颖,论据充分
3. 语言流畅,表达准确
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"文章生成错误: {str(e)}")
raise
async def generate_code(self, language, task):
"""生成代码"""
prompt = f"""
请用{language}编写代码实现以下功能:
{task}
要求:
1. 代码规范,注释清晰
2. 考虑边界情况
3. 优化性能
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"代码生成错误: {str(e)}")
raise
智能问答系统
class QASystem:
def __init__(self):
self.api = ChatGPTAPI()
self.context = []
def add_context(self, text):
"""添加上下文信息"""
self.context.append(text)
def clear_context(self):
"""清除上下文"""
self.context = []
async def get_answer(self, question):
"""获取答案"""
prompt = f"""
基于以下背景信息回答问题:
背景信息:
{' '.join(self.context)}
问题:{question}
请提供准确、完整的答案。
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"问答错误: {str(e)}")
raise
# 使用示例
async def qa_demo():
qa = QASystem()
# 添加背景信息
qa.add_context("Python是一种解释型、面向对象的高级编程语言。")
qa.add_context("Python强调代码的可读性和简洁的语法。")
# 提问
answer = await qa.get_answer("Python有哪些主要特点?")
print(answer)
特定场景示例
文本翻译
class TranslationAPI:
def __init__(self):
self.api = ChatGPTAPI()
async def translate(self, text, source_lang, target_lang):
"""文本翻译"""
prompt = f"""
请将以下{source_lang}文本翻译为{target_lang}:
{text}
要求:
1. 准确传达原文含义
2. 符合目标语言表达习惯
3. 保持专业术语的准确性
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"翻译错误: {str(e)}")
raise
# 使用示例
async def translation_demo():
translator = TranslationAPI()
chinese_text = "人工智能正在改变我们的生活方式。"
english = await translator.translate(chinese_text, "中文", "英文")
print(f"翻译结果: {english}")
代码分析
class CodeAnalyzer:
def __init__(self):
self.api = ChatGPTAPI()
async def analyze_code(self, code, language):
"""代码分析"""
prompt = f"""
请分析以下{language}代码:
{code}
请提供:
1. 代码功能说明
2. 性能分析
3. 可能的改进建议
4. 潜在的bug
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"代码分析错误: {str(e)}")
raise
async def optimize_code(self, code, language):
"""代码优化"""
prompt = f"""
请优化以下{language}代码:
{code}
优化目标:
1. 提高性能
2. 改善可读性
3. 增强可维护性
4. 修复潜在问题
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
print(f"代码优化错误: {str(e)}")
raise
性能优化示例
并发请求处理
import asyncio
from asyncio import Semaphore
class BatchProcessor:
def __init__(self, max_concurrent=5):
self.api = ChatGPTAPI()
self.semaphore = Semaphore(max_concurrent)
async def process_single(self, message):
"""处理单个请求"""
async with self.semaphore:
return await self.api.chat_async(message)
async def process_batch(self, messages):
"""批量处理请求"""
tasks = [self.process_single(msg) for msg in messages]
return await asyncio.gather(*tasks, return_exceptions=True)
# 使用示例
async def batch_demo():
processor = BatchProcessor()
messages = [
"你好",
"介绍一下Python",
"什么是机器学习"
]
results = await processor.process_batch(messages)
for msg, result in zip(messages, results):
print(f"问题: {msg}")
print(f"回答: {result}\n")
缓存实现
import redis
from functools import lru_cache
import hashlib
import json
class CachedAPI:
def __init__(self):
self.api = ChatGPTAPI()
self.redis = redis.Redis(
host='localhost',
port=6379,
db=0
)
def generate_cache_key(self, message):
"""生成缓存键"""
message_str = json.dumps(message, sort_keys=True)
return f"chatgpt:response:{hashlib.md5(message_str.encode()).hexdigest()}"
@lru_cache(maxsize=1000)
async def get_cached_response(self, cache_key):
"""获取缓存的响应"""
return self.redis.get(cache_key)
async def chat_with_cache(self, message):
"""带缓存的聊天接口"""
cache_key = self.generate_cache_key(message)
# 尝试从缓存获取
cached_response = await self.get_cached_response(cache_key)
if cached_response:
return cached_response.decode()
# 调用API获取响应
response = await self.api.chat_async(message)
# 存入缓存
self.redis.setex(cache_key, 3600, response) # 缓存1小时
return response
错误处理示例
错误分类
class APIError(Exception):
"""API错误基类"""
def __init__(self, message, code=None, details=None):
super().__init__(message)
self.code = code
self.details = details
class AuthenticationError(APIError):
"""认证错误"""
pass
class RateLimitError(APIError):
"""频率限制错误"""
pass
class NetworkError(APIError):
"""网络错误"""
pass
class ValidationError(APIError):
"""参数验证错误"""
pass
错误处理
def handle_api_error(func):
"""错误处理装饰器"""
async def wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except AuthenticationError as e:
logger.error(f"认证错误: {str(e)}")
raise
except RateLimitError as e:
logger.error(f"频率限制错误: {str(e)}")
# 等待一分钟后重试
await asyncio.sleep(60)
return await func(*args, **kwargs)
except NetworkError as e:
logger.error(f"网络错误: {str(e)}")
# 重试3次
for i in range(3):
try:
await asyncio.sleep(2 ** i) # 指数退避
return await func(*args, **kwargs)
except NetworkError:
continue
raise
except ValidationError as e:
logger.error(f"参数验证错误: {str(e)}")
raise
except Exception as e:
logger.error(f"未知错误: {str(e)}")
raise APIError("系统错误", details=str(e))
return wrapper
# 使用示例
class SafeAPI:
def __init__(self):
self.api = ChatGPTAPI()
@handle_api_error
async def safe_chat(self, message):
"""安全的API调用"""
return await self.api.chat_async(message)
# 错误处理示例
async def error_handling_demo():
safe_api = SafeAPI()
try:
response = await safe_api.safe_chat("你好")
print(f"正常响应: {response}")
except APIError as e:
print(f"API错误: {str(e)}, 代码: {e.code}, 详情: {e.details}")
监控与日志示例
日志记录
import logging
from datetime import datetime
class APILogger:
def __init__(self):
self.logger = logging.getLogger("api")
self.logger.setLevel(logging.INFO)
# 添加文件处理器
handler = logging.FileHandler("api.log")
handler.setFormatter(
logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
)
self.logger.addHandler(handler)
def log_request(self, method, path, params):
"""记录请求"""
self.logger.info(f"API请求 - 方法: {method}, 路径: {path}, 参数: {params}")
def log_response(self, status_code, response_time, content):
"""记录响应"""
self.logger.info(
f"API响应 - 状态码: {status_code}, 响应时间: {response_time}ms, "
f"内容长度: {len(str(content))}"
)
def log_error(self, error):
"""记录错误"""
self.logger.error(f"API错误 - {str(error)}")
# 使用示例
api_logger = APILogger()
api_logger.log_request("POST", "/chat/completions", {"message": "你好"})
性能监控
import time
from collections import deque
class APIMonitor:
def __init__(self, window_size=1000):
self.response_times = deque(maxlen=window_size)
self.error_count = 0
self.request_count = 0
def record_request(self, start_time):
"""记录请求"""
self.request_count += 1
end_time = time.time()
response_time = (end_time - start_time) * 1000 # 转换为毫秒
self.response_times.append(response_time)
def record_error(self):
"""记录错误"""
self.error_count += 1
def get_statistics(self):
"""获取统计信息"""
if not self.response_times:
return {
"avg_response_time": 0,
"error_rate": 0,
"total_requests": 0
}
avg_response_time = sum(self.response_times) / len(self.response_times)
error_rate = self.error_count / self.request_count if self.request_count > 0 else 0
return {
"avg_response_time": round(avg_response_time, 2),
"error_rate": round(error_rate * 100, 2),
"total_requests": self.request_count
}
# 使用示例
monitor = APIMonitor()
async def monitored_request(api, message):
start_time = time.time()
try:
response = await api.chat(message)
monitor.record_request(start_time)
return response
except Exception as e:
monitor.record_error()
raise
实战应用示例
智能客服系统
class CustomerService:
def __init__(self):
self.api = ChatGPTAPI()
self.conversation_history = {}
def create_prompt(self, user_id, question):
"""创建提示词"""
history = self.conversation_history.get(user_id, [])
prompt = f"""
你是一个专业的客服代表。请根据以下对话历史,
用专业、友好的语气回答用户的问题。
对话历史:
{' '.join(history)}
用户问题:{question}
"""
return prompt
async def handle_question(self, user_id, question):
"""处理用户问题"""
prompt = self.create_prompt(user_id, question)
try:
response = await self.api.chat_async(prompt)
# 更新对话历史
if user_id not in self.conversation_history:
self.conversation_history[user_id] = []
self.conversation_history[user_id].append(
f"用户:{question}\n助手:{response}"
)
return response
except Exception as e:
logger.error(f"客服系统错误: {str(e)}")
return "抱歉,系统暂时无法回答您的问题,请稍后再试。"
# 使用示例
async def customer_service_demo():
cs = CustomerService()
# 模拟用户对话
user_id = "user123"
response1 = await cs.handle_question(user_id, "你们的退货政策是什么?")
print("回答1:", response1)
response2 = await cs.handle_question(user_id, "如何申请退款?")
print("回答2:", response2)
文档助手
class DocumentAssistant:
def __init__(self):
self.api = ChatGPTAPI()
async def analyze_document(self, document):
"""分析文档"""
prompt = f"""
请分析以下文档内容:
{document}
请提供:
1. 文档主要内容概述
2. 关键点提取
3. 建议或改进意见
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
logger.error(f"文档分析错误: {str(e)}")
raise
async def generate_summary(self, document, max_length=200):
"""生成文档摘要"""
prompt = f"""
请为以下文档生成一个不超过{max_length}字的摘要:
{document}
要求:
1. 保留核心内容
2. 语言简洁清晰
3. 突出重点信息
"""
try:
return await self.api.chat_async(prompt)
except Exception as e:
logger.error(f"摘要生成错误: {str(e)}")
raise
# 使用示例
async def document_assistant_demo():
assistant = DocumentAssistant()
document = """
人工智能(AI)是计算机科学的一个分支,它致力于研究和开发能够模拟、延伸和扩展人类智能的理论、方法、技术及应用系统。
AI的研究包括机器学习、计算机视觉、自然语言处理等多个领域。近年来,随着深度学习技术的突破,AI在图像识别、语音识别、
自然语言处理等任务上取得了显著进展,并在医疗、金融、教育等领域得到广泛应用。
"""
analysis = await assistant.analyze_document(document)
print("文档分析:", analysis)
summary = await assistant.generate_summary(document)
print("文档摘要:", summary)
总结
ChatGPT中文版API示例为开发者提供了丰富的实践参考。建议开发者:
- 从基础示例开始学习
- 逐步掌握高级功能
- 重视错误处理
- 注意性能优化
通过这些示例的学习和实践,开发者可以更好地理解和使用API,构建功能强大的AI应用。记住,示例代码仅供参考,实际应用中需要根据具体需求进行调整和优化。持续学习和实践,才能构建出更好的AI应用。