大模型应用开发极简入门：基于GPT-4和ChatGPT的实战指南

2次阅读

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

API 调用基础
大模型通过 RESTful API 提供服务，核心是发送包含 prompt 的 HTTP 请求。OpenAI 官方 SDK 封装了底层细节，开发者只需关注输入输出。关键参数包括：
model：指定模型版本（如 gpt-4 或gpt-3.5-turbo）
messages：对话历史列表
temperature：控制生成随机性（0-2）
Prompt 工程三原则
明确性：避免歧义（如用 ” 用 Python 写快速排序 ” 替代 ” 写个排序算法 ”）
结构化：使用 Markdown 分隔指令与示例
渐进式：复杂任务拆解为多轮对话
微调（Fine-tuning）适用场景
当需要：
定制输出格式（如固定 JSON 结构）
学习私有术语（如医疗专业词汇）
优化特定领域表现时考虑微调

API 限速问题
GPT- 4 默认每分钟 3,000 tokens，应对方案：
监控 x-ratelimit-remaining 响应头
实现指数退避重试机制
关键业务使用 gpt-3.5-turbo 降级
成本控制技巧
设置 max_tokens 限制单次响应长度
通过 stream=True 实时获取部分结果
日志记录每个请求的usage.total_tokens
响应延迟优化
实测 GPT- 4 平均延迟 1.2s，可采取：
预加载常用提示模板
客户端实现请求队列
超时设置建议 10-15s

import openai

# 初始化客户端（推荐环境变量管理 API 密钥）openai.api_key = os.getenv('OPENAI_KEY')

def chat_completion(prompt: str) -> str:
    """
    基础对话生成
    :param prompt: 用户输入的提示文本
    :return: 模型生成的响应内容
    """
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=500
    )
    return response.choices[0].message.content

流式响应处理

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "system", "content": "你是一个专业技术文档助手"}],
    stream=True
)

for chunk in response:
    print(chunk.choices[0].delta.get("content", ""), end="")

函数调用（Function Calling）
实现结构化数据获取：

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的天气",
            "parameters": {...}
        }
    }
]
response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo-0613",
    messages=[{"role": "user", "content": "北京现在天气怎样？"}],
    tools=tools
)

缓存层设计
对相同 prompt 的响应使用 Redis 缓存，设置 TTL 为 1 小时：

import redis
r = redis.Redis()

def cached_completion(prompt):
    cache_key = f"gpt_cache:{hash(prompt)}"
    if cached := r.get(cache_key):
        return cached
    response = chat_completion(prompt)
    r.setex(cache_key, 3600, response)
    return response

异步处理模式
使用 aiohttp 实现并发请求：

import aiohttp

async def async_completion(prompts: list):
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.post(
                "https://api.openai.com/v1/chat/completions",
                json=build_request(prompt),
                headers={"Authorization": f"Bearer {API_KEY}"}
            ) for prompt in prompts
        ]
        return await asyncio.gather(*tasks)

API 密钥管理
永远不要硬编码密钥
使用 AWS Secrets Manager 或 HashiCorp Vault
实施最小权限原则

数据脱敏处理
在发送用户输入前执行：

from presidio_analyzer import AnalyzerEngine

analyzer = AnalyzerEngine()
def anonymize_text(text):
    results = analyzer.analyze(text=text, language="zh")
    for result in results:
        text = text[:result.start] + "[REDACTED]" + text[result.end:]
    return text

错误代码 429
检查请求头中的x-ratelimit-limit-requests，建议：
降低请求频率
升级到付费计划
输出内容不符合预期
调试步骤：
检查 system 角色消息是否正确定义
调整 temperature 到 0.3-0.7 范围
使用更具体的指令（如 ” 生成包含 3 个要点的回答 ”）

推荐首个练手项目：命令行智能助手
核心功能：
1. 支持自然语言查询（如 ” 列出今天的工作项 ”）
2. 集成基础工具（计算器、单位转换等）
3. 对话历史持久化

关键实现提示：

# 对话上下文保持
history = []

def chat_loop():
    while True:
        user_input = input("You:")
        history.append({"role": "user", "content": user_input})

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=history[-10:],  # 保持最近 10 轮对话
            temperature=0.5
        )

        assistant_reply = response.choices[0].message.content
        history.append({"role": "assistant", "content": assistant_reply})
        print(f"Assistant: {assistant_reply}")

通过这个基础框架，开发者可以逐步扩展功能模块，如添加函数调用、实现多模态交互等。建议从简单场景入手，持续迭代优化。

正文完