ChatGPT CLI 开发实战：从零构建高效命令行交互工具

1次阅读

没有评论

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

作为开发者，我经常需要在编写代码时快速查询技术问题或调试代码片段。网页版 ChatGPT 虽然强大，但在开发工作流中存在几个明显痛点：

上下文丢失问题 ：每次打开新会话都会丢失之前的对话历史，调试复杂问题时需要反复复制粘贴上下文
工作流中断 ：需要在浏览器和 IDE/ 终端之间频繁切换，影响思维连贯性
功能局限 ：无法直接执行代码片段或集成到现有开发工具链中
响应延迟 ：网页界面加载和网络请求带来的额外延迟

实现 CLI 工具主要有两种技术路线：

LangChain 方案
优点：提供高层抽象，内置对话记忆等组件
缺点：依赖较重，定制灵活性较低
直接调用 OpenAI API
优点：轻量级，完全控制请求 / 响应流程
缺点：需要自行实现上下文管理等基础功能

考虑到我们需要高度定制化和性能优化，选择直接调用 API 方案。以下是核心模块的技术栈：

CLI 框架：Python 标准库 argparse
网络请求：aiohttp（异步 HTTP 客户端）
终端 UI：rich（富文本渲染）
配置管理：configparser

使用 argparse 构建基础命令行界面：

import argparse

def create_parser():
    parser = argparse.ArgumentParser(description='ChatGPT CLI Tool')
    parser.add_argument('query', nargs='?', help='直接查询内容')
    parser.add_argument('--session', help='指定会话名称')
    parser.add_argument('--model', default='gpt-3.5-turbo', 
                       help='指定模型版本')
    return parser

使用 aiohttp 实现高性能异步请求：

import aiohttp

async def chat_completion(messages, api_key, model='gpt-3.5-turbo'):
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }

    payload = {
        'model': model,
        'messages': messages,
        'temperature': 0.7
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            json=payload,
            headers=headers
        ) as response:
            return await response.json()

实现带持久化的对话历史记录：

import json
import os

class ChatContext:
    def __init__(self, session_name='default'):
        self.session_name = session_name
        self.history = []
        self.load_history()

    def add_message(self, role, content):
        self.history.append({'role': role, 'content': content})
        self.persist_history()

    def persist_history(self):
        os.makedirs('sessions', exist_ok=True)
        with open(f'sessions/{self.session_name}.json', 'w') as f:
            json.dump(self.history, f)

    def load_history(self):
        try:
            with open(f'sessions/{self.session_name}.json') as f:
                self.history = json.load(f)
        except FileNotFoundError:
            self.history = []

利用 rich 库实现美观的输出渲染：

from rich.console import Console
from rich.markdown import Markdown

console = Console()

def print_markdown(content):
    md = Markdown(content)
    console.print(md)

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 robust_request(messages, api_key, model):
    try:
        return await chat_completion(messages, api_key, model)
    except asyncio.TimeoutError:
        print('请求超时，正在重试...')
        raise

import time

class RateLimiter:
    def __init__(self, calls_per_minute):
        self.calls_per_minute = calls_per_minute
        self.timestamps = []

    async def wait_if_needed(self):
        now = time.time()
        # 移除 1 分钟外的记录
        self.timestamps = [t for t in self.timestamps if now - t < 60]

        if len(self.timestamps) >= self.calls_per_minute:
            sleep_time = 60 - (now - self.timestamps[0])
            await asyncio.sleep(sleep_time)

        self.timestamps.append(time.time())

使用 keyring 库安全存储 API 密钥：

import keyring

def store_api_key(key):
    keyring.set_password('chatgpt_cli', 'api_key', key)

def get_api_key():
    return keyring.get_password('chatgpt_cli', 'api_key')

Token 计算误差 ：
实际使用的 token 数可能比预估多 10-15%
解决方案：使用 tiktoken 库精确计算

import tiktoken

def count_tokens(text, model='gpt-3.5-turbo'):
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

流式输出刷新问题 ：
直接打印会导致终端闪烁
解决方案：使用 rich 的 Live 显示

from rich.live import Live

async def stream_response(response):
    with Live(console=console) as live:
        full_content = ''
        async for chunk in response.content:
            full_content += chunk.decode()
            live.update(Markdown(full_content))

代理配置技巧 ：
国内用户需要配置代理
最佳实践：支持环境变量和配置文件两种方式

proxy = os.getenv('OPENAI_PROXY') or config.get('DEFAULT', 'proxy', fallback=None)
if proxy:
    connector = aiohttp.TCPConnector(proxy=proxy)
    session = aiohttp.ClientSession(connector=connector)

所有代码遵循 PEP8 规范，关键函数包含类型标注和文档字符串：

def process_response(response: dict) -> str:
    """
    处理 OpenAI API 响应，提取回复内容

    Args:
        response: OpenAI API 返回的 JSON 响应

    Returns:
        模型生成的文本内容
    """choices = response.get('choices', [])
    if not choices:
        raise ValueError('无效的 API 响应')
    return choices[0]['message']['content']

这个基础 CLI 工具可以进一步扩展：