大模型API开发极简实战

本文手把手教你玩转大模型API开发，覆盖OpenAI/DeepSeek/文心一言三大平台，5分钟实现首个AI对话应用！文末附【免梯子】国内镜像站地址。

Memory_mumu

856人浏览 · 2025-03-06 09:51:51

Memory_mumu · 2025-03-06 09:51:51 发布

本文手把手教你玩转大模型API开发，覆盖OpenAI/DeepSeek/文心一言三大平台，5分钟实现首个AI对话应用！文末附【免梯子】国内镜像站地址。

🌟 API服务商对比选型

服务商	代表模型	计费方式	国内访问	特点
OpenAI	GPT-4	按Token付费	需要代理	多模态能力最强
DeepSeek	DeepSeek-67B	按次/包月	直连可用	代码生成场景优化
文心一言	ERNIE-4.0	免费额度+阶梯价	毫秒级响应	中文理解能力突出

🚀 四步快速接入

步骤1：获取API密钥

OpenAI示例：

登录OpenAI平台
点击右上角「Create new secret key」
复制生成的API Key

国内镜像站（免科学）：

# config.py
API_BASE = "https://api.ai-proxy.com/v1"  # 代理中转地址
API_KEY = "sk-xxxxxxxxxxxxxxxx"  # 从镜像站获取

步骤2：安装开发套件

# 通用依赖
pip install openai python-dotenv requests aiohttp

# 异步支持
pip install nest_asyncio websockets

步骤3：基础对话实现

import openai
from openai import AsyncOpenAI

# 同步调用
def chat(prompt: str):
    client = openai.Client(api_key="YOUR_KEY")
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )
    return response.choices[0].message.content

# 异步调用（推荐）
async def async_chat(prompt: str):
    aclient = AsyncOpenAI(api_key="YOUR_KEY")
    response = await aclient.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        stream=True  # 启用流式传输
    )
    async for chunk in response:
        print(chunk.choices[0].delta.content, end="")

步骤4：处理复杂场景

上下文保持：

history = []

def chat_with_history(query: str):
    global history
    history.append({"role": "user", "content": query})
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=history[-6:],  # 保留最近3轮对话
        max_tokens=500
    )
    
    answer = response.choices[0].message.content
    history.append({"role": "assistant", "content": answer})
    return answer

🔥 开发实战

场景1：智能客服系统

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    question: str
    user_id: str

@app.post("/api/chat")
async def customer_service(req: ChatRequest):
    # 查询用户历史记录
    history = get_chat_history(req.user_id)  
    
    # 添加业务预设
    system_prompt = """你是一名手机售后客服，请用口语化中文回答，遵循以下规则：
    1. 不透露厂商敏感信息
    2. 不承诺未授权事项
    3. 遇到技术问题转接人工"""
    
    full_msg = [{"role": "system", "content": system_prompt}] 
    full_msg.extend(history[-4:])
    full_msg.append({"role": "user", "content": req.question})
    
    response = await client.chat.completions.create(
        model="ERNIE-4.0",
        messages=full_msg,
        temperature=0.3
    )
    
    return {"answer": response.choices[0].message.content}

场景2：代码生成助手

def code_generator(task: str):
    prompt = f"""你是一个资深Python工程师，请按照以下要求生成代码：
    1. 添加详细注释
    2. 使用Python 3.10+语法
    3. 包含异常处理
    4. 输出PEP8规范代码
    
    需求：{task}"""
    
    response = client.chat.completions.create(
        model="deepseek-coder",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        stop=["```"]  # 防止多余输出
    )
    
    # 提取代码块
    raw_code = response.choices[0].message.content
    return extract_code_block(raw_code)  # 自定义提取函数

⚡ 性能优化技巧

1. 流式响应加速

// 前端处理示例
const eventSource = new EventSource('/api/stream-chat');

eventSource.onmessage = (e) => {
    document.getElementById('answer').innerHTML += JSON.parse(e.data).content;
};

2. 异步批处理

import asyncio

async def batch_chat(queries: list):
    tasks = [async_chat(q) for q in queries]
    return await asyncio.gather(*tasks, return_exceptions=True)

3. 缓存优化

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def cached_chat(prompt: str):
    md5 = hashlib.md5(prompt.encode()).hexdigest()
    if redis.exists(md5):
        return redis.get(md5)
    # ...正常调用API...

⚠️ 避坑指南

高频报错处理

错误码	原因	解决方案
429	请求频率超限	添加指数退避重试机制
503	服务不可用	切换备用API端点
401	密钥无效	检查密钥是否包含特殊字符
400	输入过长	启用自动截断功能

指数退避示例：

import time
from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, max=10))
def safe_chat(prompt):
    return client.chat.completions.create(...)

🔐 安全防护策略

敏感信息过滤

blacklist = ["信用卡", "密码", "密钥"]

def sanitize_input(text: str):
    for word in blacklist:
        text = text.replace(word, "***")
    return text

用量监控

# 限流中间件
from fastapi import HTTPException

async def rate_limit_middleware(request: Request):
    user = get_current_user(request)
    if user.quota <= 0:
        raise HTTPException(402, "API配额已用尽")
    user.quota -= 1