命令行部署ChatGPT全指南：从零搭建到生产环境优化

2次阅读

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

在直接使用 OpenAI 官方 API 时，开发者常遇到几个典型问题：

网络延迟不稳定 ：国内直连 API 常因网络问题导致响应超时
Token 管理复杂 ：需要手动处理对话历史拼接，容易超出模型 token 限制
成本不可控 ：频繁调用 API 会产生意外费用，特别是流式响应场景

纯 API 调用方案
优点：零部署成本，实时获取最新模型
缺点：完全依赖网络，无法离线使用
本地部署开源模型
优点：数据完全可控，响应速度快
缺点：需要 GPU 资源，模型效果差距明显
混合方案（本文重点）
用容器封装 API 调用层
添加本地缓存和批处理优化

import click
from openai import OpenAI

@click.group()
def cli():
    """ChatGPT 命令行工具主入口"""
    pass

@cli.command()
@click.option('--prompt', required=True, help='输入对话内容')
def chat(prompt):
    """执行单轮对话"""
    client = OpenAI(api_key=load_api_key())  # 安全加载密钥
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    click.echo(response.choices[0].message.content)

if __name__ == '__main__':
    cli()

# Dockerfile
FROM python:3.8-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 安全建议：不要在镜像中固化 API 密钥
ENV OPENAI_KEY=""

COPY . .
CMD ["python", "main.py"]

class ConversationManager:
    """管理多轮对话上下文"""

    def __init__(self, max_tokens=4096):
        self.history = []
        self.max_tokens = max_tokens

    def add_message(self, role, content):
        """添加对话记录并自动修剪超出 token 限制的历史"""
        self.history.append({"role": role, "content": content})
        self._trim_history()

    def _trim_history(self):
        """优先保留最近对话"""
        while self._count_tokens() > self.max_tokens * 0.9:  # 保留 10% 余量
            self.history.pop(0)

def batch_process(queries):
    """批量处理多个查询"""
    messages = [{"role": "user", "content": q} 
        for q in queries
    ]
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        n=len(queries)  # 并行生成
    )
    return [choice.message.content for choice in response.choices]

from diskcache import Cache

cache = Cache("./chat_cache")

def cached_chat(prompt):
    """带缓存的对话处理"""
    if prompt in cache:
        return cache[prompt]

    response = generate_response(prompt)
    cache.set(prompt, response, expire=3600)  # 缓存 1 小时
    return response

import keyring

def save_api_key(key):
    """使用系统密钥环存储"""
    keyring.set_password("chatgpt_cli", "api_key", key)

def load_api_key():
    """安全加载 API 密钥"""
    return keyring.get_password("chatgpt_cli", "api_key")

from ratelimit import limits, sleep_and_retry

# 限制每分钟 30 次调用
@sleep_and_retry
@limits(calls=30, period=60)
def safe_api_call(prompt):
    return client.chat.completions.create(...)

429 Too Many Requests
实现自动退避重试机制
建议：tenacity 库实现指数退避
503 Service Unavailable
检查 OpenAI 状态页（status.openai.com）
降级到本地缓存响应
模型版本过时
固定模型版本号：如 gpt-3.5-turbo-0613
实现版本检测告警

MODELS = {
    "default": "gpt-3.5-turbo",
    "creative": "gpt-4",
    "fast": "gpt-3.5-turbo-16k"
}

def select_model(model_alias):
    """通过别名选择模型"""
    return MODELS.get(model_alias, MODELS["default"])

# 使用 click 的自动补全功能
@cli.command()
@click.argument("query", autocompletion=lambda: ["help", "version", "exit"])
def complete(query):
    """支持 tab 补全的命令"""
    pass