企业如何高效接入ChatGPT：从API调用到生产环境部署的完整指南

1次阅读

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

企业接入 ChatGPT 时，往往会遇到以下几个特有的挑战：

审计日志需求 ：企业通常需要记录所有 API 调用记录，包括请求内容、响应时间、用户信息等，以满足合规要求。
多租户隔离 ：在 SaaS 或多团队协作场景中，需要确保不同团队或客户的数据完全隔离，避免信息泄露。
API 调用配额管理 ：企业可能需要根据部门或项目分配不同的调用配额，避免资源滥用。
敏感信息过滤 ：企业数据往往包含敏感信息（如客户隐私、商业机密），需在 API 调用前进行严格过滤。

ChatGPT API：适合中小型企业或快速原型开发，直接通过 OpenAI 官方 API 调用，无需复杂配置。
Azure OpenAI 服务 ：适合大型企业或需要与其他 Azure 服务集成的场景，提供企业级 SLA、私有网络部署和更细粒度的权限控制。

import openai
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

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

const {Configuration, OpenAIApi} = require("openai");
const configuration = new Configuration({apiKey: process.env.OPENAI_API_KEY});
const openai = new OpenAIApi(configuration);

async function chatCompletionWithRetry(prompt, retries = 3) {
    try {
        const response = await openai.createChatCompletion({
            model: "gpt-4",
            messages: [{role: "user", content: prompt}]
        });
        return response.data.choices[0].message.content;
    } catch (error) {if (retries <= 0) throw error;
        await new Promise(res => setTimeout(res, 1000 * (4 - retries)));
        return chatCompletionWithRetry(prompt, retries - 1);
    }
}

import jwt
import time
from functools import wraps

class ChatGPTEnterpriseClient:
    def __init__(self, api_key, jwt_secret, rate_limit=60):
        self.api_key = api_key
        self.jwt_secret = jwt_secret
        self.rate_limit = rate_limit  # 每分钟最大调用次数
        self.last_called = 0
        self.call_count = 0

    def generate_token(self, user_id):
        payload = {
            "user_id": user_id,
            "exp": int(time.time()) + 3600  # 1 小时过期
        }
        return jwt.encode(payload, self.jwt_secret, algorithm="HS256")

    def rate_limiter(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_time = time.time()
            elapsed = current_time - self.last_called

            if elapsed > 60:  # 超过 1 分钟，重置计数器
                self.call_count = 0
                self.last_called = current_time

            if self.call_count >= self.rate_limit:
                time.sleep(60 - elapsed)
                self.call_count = 0
                self.last_called = time.time()

            self.call_count += 1
            return func(*args, **kwargs)
        return wrapper

    @rate_limiter
    def chat_completion(self, prompt, user_token):
        try:
            jwt.decode(user_token, self.jwt_secret, algorithms=["HS256"])
            # 实际调用 API 的代码...
        except jwt.ExpiredSignatureError:
            raise Exception("Token 已过期")
        except jwt.InvalidTokenError:
            raise Exception("无效 Token")

const sensitiveWords = ["密码", "信用卡", "机密"]; // 敏感词库
const sensitivePatterns = [/\b\d{16}\b/,  // 信用卡号
    /\b\d{3}-?\d{2}-?\d{4}\b/  // SSN
];

function filterSensitiveContent(text) {
    // 词库检查
    for (const word of sensitiveWords) {if (text.includes(word)) {throw new Error(` 内容包含敏感词: ${word}`);
        }
    }

    // 正则检查
    for (const pattern of sensitivePatterns) {if (pattern.test(text)) {throw new Error("内容包含敏感模式");
        }
    }

    return text;
}