共计 3031 个字符，预计需要花费 8 分钟才能阅读完成。
背景痛点
在日常开发中，我们经常需要频繁地与 ChatGPT 交互，而 Web 界面虽然直观，但在自动化流程和集成方面存在明显不足。命令行工具能够带来以下优势：
- 自动化集成：可以轻松嵌入脚本、CI/CD 流程或其他自动化工具中
- 批量处理：适合处理大量对话或测试不同提示词效果
- 定制化输出：可以按照需求格式化响应，方便后续处理
然而，直接使用命令行调用 ChatGPT API 也会遇到一些挑战：
- API 版本兼容性问题
- Token 计算和成本控制
- 上下文管理复杂
- 错误处理和重试机制
技术对比：官方 CLI vs 自建方案
官方 CLI 工具（chatgpt-terminal）
- 优点：
- 开箱即用，无需额外开发
- 内置基本配置和错误处理
-
官方维护，更新及时
-
缺点：
- 功能有限，扩展性差
- 无法深度定制交互逻辑
- 部分高级功能缺失
自建方案（原生 API 调用）
- 优点：
- 完全可定制
- 能实现更复杂的功能
-
可以优化性能和控制成本
-
缺点：
- 需要自行处理各种边界情况
- 开发成本较高
- 需要维护代码
核心实现
1. API 密钥配置
安全存储 API 密钥是第一步，强烈建议使用 .env 文件：
# 创建.env 文件
echo "OPENAI_API_KEY=your_api_key_here" > .env然后在 Python 中通过 python-dotenv 加载：
from dotenv import load_dotenv
load_dotenv()
import os
api_key = os.getenv("OPENAI_API_KEY")2. 带错误处理的基础实现
以下是使用 requests 和click库的基础实现：
import requests
import click
from typing import Optional, Dict, Any
@click.command()
@click.argument('prompt')
def chat(prompt: str) -> None:
"""
基础 ChatGPT 命令行交互
Args:
prompt: 用户输入的提示词
"""
try:
headers = {"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
data = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": prompt}]
}
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=data,
timeout=30
)
response.raise_for_status()
result = response.json()
click.echo(result['choices'][0]['message']['content'])
except requests.exceptions.RequestException as e:
click.echo(f"请求失败: {str(e)}", err=True)
if __name__ == "__main__":
chat()3. 实现上下文记忆
要实现多轮对话，需要管理对话历史：
from typing import List
class Conversation:
"""管理对话上下文的类"""
def __init__(self):
self.history: List[Dict[str, str]] = []
def add_message(self, role: str, content: str) -> None:
"""添加消息到历史记录"""
self.history.append({"role": role, "content": content})
def get_messages(self) -> List[Dict[str, str]]:
"""获取完整对话历史"""
return self.history.copy()
# 使用示例
conv = Conversation()
conv.add_message("user", "你好")
# 发送请求时将 conv.get_messages()作为 messages 参数生产级考量
1. 流式输出优化
处理流式响应时，需要注意终端渲染：
import sys
def handle_stream_response(response):
"""处理流式响应，优化终端显示"""
buffer = ""
for chunk in response.iter_content(chunk_size=1024):
if chunk:
buffer += chunk.decode("utf-8")
# 简单的换行处理
if "\n" in buffer:
lines = buffer.split("\n")
for line in lines[:-1]:
sys.stdout.write(line + "\n")
sys.stdout.flush()
buffer = lines[-1]2. 请求超时与速率限制
实现健壮的重试机制：
from time import sleep
import random
def make_request_with_retry(url, headers, data, max_retries=3):
"""带重试的请求函数"""
for attempt in range(max_retries):
try:
response = requests.post(url, headers=headers, json=data, timeout=30)
response.raise_for_status()
return response
except requests.exceptions.HTTPError as e:
if response.status_code == 429: # 速率限制
retry_after = int(response.headers.get("Retry-After", 5))
sleep(retry_after + random.uniform(0, 1))
continue
raise
except requests.exceptions.RequestException:
if attempt == max_retries - 1:
raise
sleep(2 ** attempt + random.uniform(0, 1))避坑指南
1. Token 成本差异
- GPT-3.5-turbo: $0.002/1k tokens
- GPT-4: $0.06/1k tokens (输入), $0.12/1k tokens (输出)
2. 合规要求
- 欧盟 GDPR 要求可能需要对数据进行清理
- 建议定期清理日志和对话历史
- 敏感信息不应存储在日志中
代码规范建议
- 所有函数都应有类型注解
- 主要函数应有完整的 docstring
- 错误处理要全面
- 关键操作应有日志记录
延伸思考
-
使用
jq工具处理 JSON 响应:curl ... | jq '.choices[0].message.content' -
探索 Assistant API 的 CLI 化可能
-
考虑添加对话历史持久化功能
-
实现多模型切换支持
通过这些实践，你可以构建出一个强大而灵活的 ChatGPT 命令行工具，满足各种开发需求。
总结
本文详细介绍了如何从零开始构建一个功能完善的 ChatGPT 命令行工具。从基础的 API 调用到生产级的错误处理，再到高级功能如上下文管理和流式输出，我们覆盖了开发者可能遇到的主要场景。希望这些内容能帮助你更高效地在命令行环境中使用 ChatGPT，提升开发效率。
正文完
