如何通过API高效调用ChatGPT实现AI代码生成工具

1次阅读

没有评论

共计 3369 个字符，预计需要花费 9 分钟才能阅读完成。

在开发 AI 代码生成工具时，我们常常会遇到以下问题：

高延迟：单个请求可能需要几秒甚至更长时间才能返回结果，影响开发效率
响应不稳定：相同提示词可能返回不同质量的代码，甚至有时会失败
成本控制难：API 调用费用可能因频繁请求而快速累积
结果质量参差不齐：需要反复调整提示词才能获得理想输出

当前主流 AI 模型 API 包括：

ChatGPT API
优势：代码理解能力强，支持长上下文，文档完善
劣势：调用成本相对较高
Claude API
优势：处理长文本能力强
劣势：代码生成能力略逊于 ChatGPT
开源模型自部署
优势：完全可控，无 API 限制
劣势：需要强大算力，维护成本高

基于代码生成质量和开发效率的综合考虑，我们选择 ChatGPT API 作为解决方案。

获取 OpenAI API 密钥
安装 openai Python 包
构建 API 请求
处理响应
错误处理和重试

import openai
from tenacity import retry, stop_after_attempt, wait_exponential

# 初始化 API 密钥
openai.api_key = "your-api-key"

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def generate_code(prompt: str, model="gpt-4", temperature=0.7) -> str:
    """
    使用 ChatGPT 生成代码
    :param prompt: 提示词
    :param model: 使用的模型
    :param temperature: 控制创造性(0-1)
    :return: 生成的代码
    """
    try:
        response = await openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

明确上下文：指定编程语言、框架版本
提供示例：展示期望的输入输出格式
分步指示：将复杂任务分解为多个步骤
限制输出：要求只返回代码，不包含解释

示例提示词：

你是一位经验丰富的 Python 开发者。请生成一个 Flask REST API 端点，它应该：1. 接收 JSON 格式的 POST 请求
2. 验证请求体包含 'username' 和 'email' 字段
3. 返回 201 状态码和创建成功的消息

只返回代码，不要包含任何解释。使用 Flask 2.0 版本。

通过将多个请求合并为一个 API 调用，可以减少网络延迟：

async def batch_generate(prompts: list[str], model="gpt-4") -> list[str]:
    messages = [{"role": "user", "content": p} for p in prompts]
    response = await openai.ChatCompletion.create(
        model=model,
        messages=messages,
    )
    return [choice.message.content for choice in response.choices]

对常见请求结果进行缓存，避免重复调用：

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_cached_response(prompt: str) -> str:
    return generate_code(prompt)

使用 asyncio 限制并发请求数，防止超过 API 限制：

import asyncio
from typing import List

async def bounded_gather(tasks: List, concurrency: int = 5):
    semaphore = asyncio.Semaphore(concurrency)

    async def sem_task(task):
        async with semaphore:
            return await task

    return await asyncio.gather(*(sem_task(task) for task in tasks))

使用环境变量存储 API 密钥
实现请求速率限制
考虑使用 API 网关管理流量

记录每个请求的耗时
监控 API 错误率
跟踪每个用户的调用量

为不同功能选择合适模型（简单任务用 gpt-3.5-turbo）
设置每月预算上限
监控 token 使用量

import openai
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
from functools import lru_cache
from typing import List

class CodeGenerator:
    def __init__(self, api_key: str):
        openai.api_key = api_key

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def _generate_code(self, prompt: str, model: str = "gpt-4", temperature: float = 0.7) -> str:
        try:
            response = await openai.ChatCompletion.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=temperature,
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"API 调用失败: {str(e)}")
            raise

    @lru_cache(maxsize=1000)
    async def generate_cached(self, prompt: str) -> str:
        return await self._generate_code(prompt)

    async def batch_generate(self, prompts: List[str], concurrency: int = 5) -> List[str]:
        semaphore = asyncio.Semaphore(concurrency)

        async def sem_task(prompt):
            async with semaphore:
                return await self.generate_cached(prompt)

        return await asyncio.gather(*(sem_task(p) for p in prompts))

# 使用示例
async def main():
    generator = CodeGenerator("your-api-key")
    prompt = "生成一个 Python 函数，计算斐波那契数列前 n 项"
    code = await generator.generate_cached(prompt)
    print(code)

    # 批量生成
    prompts = [f"生成一个计算 {n} 的平方根的 Python 函数" for n in range(1, 6)]
    results = await generator.batch_generate(prompts)
    for res in results:
        print(res)

if __name__ == "__main__":
    asyncio.run(main())