CLI com IA: Construindo Ferramentas de Linha de Comando em Python e JavaScript que Usam LLMs de Verdade

# cli.py
import click
import hashlib
import json
import sqlite3
import os
 
DB_PATH = os.path.expanduser("~/.cache/aicli/cache.db")
 
def get_cache_db() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS cache "
        "(key TEXT PRIMARY KEY, response TEXT, created_at REAL DEFAULT (unixepoch()))"
    )
    return conn
 
def cache_key(model: str, prompt: str) -> str:
    # SHA256 garante chave fixa independente do tamanho do prompt
    raw = f"{model}::{prompt}"
    return hashlib.sha256(raw.encode()).hexdigest()
 
@click.command()
@click.argument("prompt")
@click.option("--model", default="gpt-4o-mini", help="Modelo a usar")
@click.option("--no-cache", is_flag=True, help="Ignora cache local")
def ask(prompt: str, model: str, no_cache: bool):
    """Envia um prompt para o LLM e exibe a resposta com streaming."""
    db = get_cache_db()
 
    if not no_cache:
        key = cache_key(model, prompt)
        row = db.execute("SELECT response FROM cache WHERE key = ?", (key,)).fetchone()
        if row:
            click.echo(row[0])
            return
 
    response_text = stream_completion(model, prompt)
 
    if response_text and not no_cache:
        db.execute(
            "INSERT OR REPLACE INTO cache (key, response) VALUES (?, ?)",
            (cache_key(model, prompt), response_text),
        )
        db.commit()

# client.py
import httpx
import sys
import os
import json
 
OPENAI_URL = "https://api.openai.com/v1/chat/completions"
ANTHROPIC_URL = "https://api.anthropic.com/v1/messages"
 
def stream_completion(model: str, prompt: str) -> str:
    """Tenta OpenAI primeiro. Se falhar com status >= 500 ou timeout, cai para Anthropic."""
    providers = [
        ("openai", _stream_openai),
        ("anthropic", _stream_anthropic),
    ]
 
    for name, fn in providers:
        try:
            return fn(model, prompt)
        except (httpx.TimeoutException, httpx.HTTPStatusError) as exc:
            # Só faz fallback em erro de servidor ou timeout, não em 401/403
            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 500:
                raise
            click.echo(f"\n[fallback] {name} falhou, tentando próximo...", err=True)
 
    raise click.ClickException("Todos os provedores falharam.")
 
def _stream_openai(model: str, prompt: str) -> str:
    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        raise click.ClickException("OPENAI_API_KEY não definida.")
 
    chunks = []
    # timeout de 30s para connect, 60s para read: LLMs demoram
    with httpx.Client(timeout=httpx.Timeout(connect=30.0, read=60.0)) as client:
        with client.stream(
            "POST",
            OPENAI_URL,
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": True,
            },
        ) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if not line.startswith("data: "):
                    continue
                payload = line.removeprefix("data: ").strip()
                if payload == "[DONE]":
                    break
                data = json.loads(payload)
                token = data["choices"][0]["delta"].get("content", "")
                sys.stdout.write(token)
                sys.stdout.flush()
                chunks.append(token)
 
    sys.stdout.write("\n")
    return "".join(chunks)

// cli.mjs
import { Command } from "commander";
import { createHash } from "node:crypto";
import Database from "better-sqlite3";
import { homedir } from "node:os";
import { mkdirSync } from "node:fs";
import { join } from "node:path";
import { streamCompletion } from "./client.mjs";
 
const DB_DIR = join(homedir(), ".cache", "aicli");
mkdirSync(DB_DIR, { recursive: true });
 
const db = new Database(join(DB_DIR, "cache.db"));
db.exec(`
  CREATE TABLE IF NOT EXISTS cache (
    key TEXT PRIMARY KEY,
    response TEXT,
    created_at INTEGER DEFAULT (unixepoch())
  )
`);
 
function cacheKey(model, prompt) {
  return createHash("sha256").update(`${model}::${prompt}`).digest("hex");
}
 
const program = new Command();
program
  .argument("<prompt>")
  .option("--model <model>", "Modelo a usar", "gpt-4o-mini")
  .option("--no-cache", "Ignora cache local")
  .action(async (prompt, opts) => {
    const key = cacheKey(opts.model, prompt);
 
    if (opts.cache !== false) {
      const row = db.prepare("SELECT response FROM cache WHERE key = ?").get(key);
      if (row) {
        process.stdout.write(row.response + "\n");
        return;
      }
    }
 
    const text = await streamCompletion(opts.model, prompt);
 
    if (text && opts.cache !== false) {
      db.prepare("INSERT OR REPLACE INTO cache (key, response) VALUES (?, ?)").run(key, text);
    }
  });
 
program.parse();

// client.mjs
const OPENAI_URL = "https://api.openai.com/v1/chat/completions";
 
export async function streamCompletion(model, prompt) {
  const apiKey = process.env.OPENAI_API_KEY;
  if (!apiKey) throw new Error("OPENAI_API_KEY não definida.");
 
  // AbortController com timeout: sem isso, a CLI trava indefinidamente
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), 60_000);
 
  const resp = await fetch(OPENAI_URL, {
    method: "POST",
    signal: controller.signal,
    headers: {
      Authorization: `Bearer ${apiKey}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      model,
      messages: [{ role: "user", content: prompt }],
      stream: true,
    }),
  });
 
  clearTimeout(timeoutId);
 
  if (!resp.ok) {
    const body = await resp.text();
    throw new Error(`OpenAI retornou ${resp.status}: ${body}`);
  }
 
  const chunks = [];
  const decoder = new TextDecoder();
 
  // resp.body é um ReadableStream no Node 18+
  for await (const raw of resp.body) {
    const text = decoder.decode(raw, { stream: true });
    // Um chunk pode conter múltiplas linhas SSE
    for (const line of text.split("\n")) {
      if (!line.startsWith("data: ")) continue;
      const payload = line.slice(6).trim();
      if (payload === "[DONE]") break;
 
      const data = JSON.parse(payload);
      const token = data.choices?.[0]?.delta?.content ?? "";
      process.stdout.write(token);
      chunks.push(token);
    }
  }
 
  process.stdout.write("\n");
  return chunks.join("");
}

# ERRADO: o usuário fica olhando para um cursor parado por 5-15 segundos
import openai
 
def ask_wrong(prompt: str) -> str:
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        # sem stream=True
    )
    return response.choices[0].message.content

// ERRADO: mesma chave para prompts iguais em modelos diferentes
function badCacheKey(prompt) {
  return createHash("sha256").update(prompt).digest("hex");
}
// "Explique closures" com gpt-4o e com claude-3-haiku geram respostas diferentes
// mas essa função retorna a mesma chave para ambos

# ERRADO: fazer fallback quando o erro é 401 (chave inválida)
# Se a chave da OpenAI está errada, a da Anthropic provavelmente também está.
# Fallback só faz sentido para 5xx e timeout.
try:
    return _stream_openai(model, prompt)
except httpx.HTTPStatusError:
    # Cai aqui mesmo com 401, 403, 429
    return _stream_anthropic(model, prompt)

-- Limpa entradas com mais de 7 dias
DELETE FROM cache WHERE created_at < unixepoch() - (7 * 86400);

@click.command()
@click.option("--days", default=7, help="Remove entradas mais antigas que N dias")
def clean(days: int):
    db = get_cache_db()
    result = db.execute(
        "DELETE FROM cache WHERE created_at < unixepoch() - (? * 86400)", (days,)
    )
    db.commit()
    click.echo(f"Removidas {result.rowcount} entradas.")

# Para usar com Ollama local, basta trocar a URL e remover a autenticação
OLLAMA_URL = "http://localhost:11434/v1/chat/completions"
 
def _stream_ollama(model: str, prompt: str) -> str:
    chunks = []
    with httpx.Client(timeout=httpx.Timeout(connect=5.0, read=120.0)) as client:
        # Timeout de read maior: modelos locais em CPU podem demorar 2min+
        with client.stream(
            "POST",
            OLLAMA_URL,
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": True,
            },
        ) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if not line.startswith("data: "):
                    continue
                payload = line.removeprefix("data: ").strip()
                if payload == "[DONE]":
                    break
                data = json.loads(payload)
                token = data["choices"][0]["delta"].get("content", "")
                sys.stdout.write(token)
                sys.stdout.flush()
                chunks.append(token)
    sys.stdout.write("\n")
    return "".join(chunks)

# Python (com pyproject.toml)
aicli/
├── pyproject.toml
├── src/
│   └── aicli/
│       ├── __init__.py
│       ├── cli.py       # entry point com @click.command
│       └── client.py    # stream_completion + fallback
└── README.md
 
# JavaScript (com package.json + bin field)
aicli/
├── package.json        # "bin": { "aicli": "./cli.mjs" }
├── cli.mjs
├── client.mjs
└── README.md

[project.scripts]
aicli = "aicli.cli:ask"

{
  "name": "aicli",
  "version": "1.0.0",
  "type": "module",
  "bin": {
    "aicli": "./cli.mjs"
  },
  "dependencies": {
    "commander": "^12.0.0",
    "better-sqlite3": "^11.0.0"
  }
}