ChatGPT中文版API示例 | AI接口调用实践指南

随着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应用。