Claude命令行工具开发实战：从零构建高效AI交互终端

1次阅读

没有评论

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

在日常开发中直接调用 Claude API 时，开发者通常会遇到几个典型问题：

每次请求都需要手动构造 HTTP 请求体和处理响应，交互效率低下
调试时需要反复修改代码参数，无法快速测试不同输入
对话上下文难以维护，多轮对话时需要手动拼接历史记录
流式响应处理复杂，无法实时看到模型生成过程

这些痛点使得开发效率大打折扣，特别是当我们需要快速验证想法或进行多轮对话测试时。

实现命令行工具主要有几种技术路线：

直接使用 Python 内置的 input() + requests
采用 Click 或 Fire 等高级 CLI 框架
使用 argparse + requests 组合

经过对比评估，我们选择第三种方案，因为：

argparse 是 Python 标准库，无需额外依赖
足够灵活支持我们需要的参数解析功能
与 requests 组合能保持代码简洁
调试和部署更简单

首先我们需要封装 Claude 的基础 API 调用，核心功能包括：

class ClaudeClient:
    def __init__(self, api_key):
        self.session = requests.Session()
        self.session.headers.update({
            'x-api-key': api_key,
            'content-type': 'application/json'
        })

    def stream_completion(self, prompt, model="claude-2", max_tokens=1024):
        payload = {
            "prompt": prompt,
            "model": model,
            "max_tokens_to_sample": max_tokens,
            "stream": True
        }
        response = self.session.post(
            "https://api.anthropic.com/v1/complete",
            json=payload,
            stream=True
        )
        return response

处理流式响应需要特殊处理，这里我们实现一个生成器函数：

def handle_stream_response(response):
    buffer = ""
    for chunk in response.iter_lines():
        if chunk:
            decoded = chunk.decode('utf-8')
            if decoded.startswith('data:'):
                event_data = json.loads(decoded[6:])
                if 'completion' in event_data:
                    buffer += event_data['completion']
                    yield event_data['completion']
    return buffer

时间复杂度分析：O(n)，其中 n 为响应体大小，因为我们只需线性处理每个数据块。

为了实现多轮对话，我们需要持久化对话历史：

class Conversation:
    def __init__(self, max_history=5):
        self.history = deque(maxlen=max_history)

    def add_exchange(self, prompt, response):
        self.history.append({"prompt": prompt, "response": response})

    def get_context(self):
        return "\n".join(f"Human: {item['prompt']}\nAssistant: {item['response']}"
            for item in self.history
        )

将上述模块整合后的核心代码如下（完整版请见 GitHub 仓库）：

import argparse
import json
import os
from collections import deque

import requests

# ... 省略上述已展示的类定义 ...

def main():
    parser = argparse.ArgumentParser(description='Claude 命令行客户端')
    parser.add_argument('--api-key', help='API 密钥', required=True)
    parser.add_argument('--model', help='模型版本', default='claude-2')
    parser.add_argument('--max-tokens', help='最大 token 数', type=int, default=1024)
    args = parser.parse_args()

    client = ClaudeClient(args.api_key)
    conversation = Conversation()

    try:
        while True:
            prompt = input("You:")
            if prompt.lower() in ('exit', 'quit'):
                break

            full_prompt = f"{conversation.get_context()}\nHuman: {prompt}\nAssistant:"
            response = client.stream_completion(
                full_prompt, 
                model=args.model,
                max_tokens=args.max_tokens
            )

            print("Claude:", end="", flush=True)
            response_text = ""
            for chunk in handle_stream_response(response):
                print(chunk, end="", flush=True)
                response_text += chunk
            print()

            conversation.add_exchange(prompt, response_text)
    except KeyboardInterrupt:
        print("\n 对话结束")

if __name__ == "__main__":
    main()