Pular para o conteúdo
🧠Inteligência em Código
🐍 Exemplos de Código•5 min de leitura

Chatbot com Streaming usando OpenAI API

Crie um chatbot interativo no terminal com respostas em tempo real usando a API da OpenAI com streaming.

#openai#chatbot#streaming#python

O que vamos construir

Um chatbot no terminal que:

  • Mantém histórico da conversa
  • Mostra respostas em tempo real (streaming)
  • Tem personalidade configurável via system prompt

Código Completo

"""
Chatbot com Streaming - OpenAI API
Respostas aparecem em tempo real no terminal.
"""

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

SYSTEM_PROMPT = """Você é um assistente especializado em Inteligência Artificial.
Responda sempre em português brasileiro, de forma clara e técnica.
Quando relevante, inclua exemplos de código Python."""

def chat(messages: list[dict]) -> str:
    """Envia mensagem com streaming e retorna resposta completa."""
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        stream=True,
        temperature=0.7,
        max_tokens=2000,
    )
    
    full_response = ""
    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            print(content, end="", flush=True)
            full_response += content
    
    print()  # nova linha após streaming
    return full_response

def main():
    print("🧠 Chatbot IA — digite 'sair' para encerrar")
    print("=" * 50)
    
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    
    while True:
        try:
            user_input = input("\n👤 Você: ").strip()
        except (KeyboardInterrupt, EOFError):
            break
            
        if not user_input:
            continue
        if user_input.lower() in ("sair", "exit", "quit"):
            print("👋 Até mais!")
            break
        
        messages.append({"role": "user", "content": user_input})
        
        print("\n🤖 IA: ", end="")
        response = chat(messages)
        
        messages.append({"role": "assistant", "content": response})

if __name__ == "__main__":
    main()

Como rodar

# Instalar
pip install openai

# Configurar API key
export OPENAI_API_KEY="sk-..."

# Executar
python chatbot.py

Variação: Com Histórico Salvo

import json
from pathlib import Path

HISTORY_FILE = Path("chat_history.json")

def save_history(messages):
    HISTORY_FILE.write_text(json.dumps(messages, ensure_ascii=False, indent=2))

def load_history():
    if HISTORY_FILE.exists():
        return json.loads(HISTORY_FILE.read_text())
    return [{"role": "system", "content": SYSTEM_PROMPT}]

Adicione save_history(messages) após cada resposta para persistir o contexto entre sessões!

Variação: Com Contagem de Tokens

Controlar tokens é essencial para não estourar o limite do modelo e manter custos previsíveis:

"""
Contador de tokens para controle de custo e limite de contexto.
Usa tiktoken para contagem precisa compatível com os modelos da OpenAI.
"""

import tiktoken

def contar_tokens(messages: list[dict], model: str = "gpt-4") -> int:
    """Conta tokens de uma lista de mensagens no formato da OpenAI."""
    encoding = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += 4  # overhead por mensagem (role, content, etc.)
        total += len(encoding.encode(msg["content"]))
    total += 2  # overhead de início/fim
    return total

def trimmer_por_tokens(messages: list[dict], max_tokens: int = 6000) -> list[dict]:
    """
    Remove mensagens antigas se o total de tokens ultrapassar o limite.
    Sempre mantém o system prompt (primeira mensagem).
    """
    if len(messages) <= 1:
        return messages

    system = messages[0]
    history = messages[1:]

    while contar_tokens([system] + history) > max_tokens and len(history) > 2:
        history.pop(0)  # remove a mensagem mais antiga

    return [system] + history

Integre no loop principal assim:

# Antes de chamar a API, controlar tokens:
messages = trimmer_por_tokens(messages, max_tokens=6000)
token_count = contar_tokens(messages)
print(f"[{token_count} tokens no contexto]")
response = chat(messages)

Variação: Múltiplas Personalidades

"""
Sistema de personalidades intercambiáveis para o chatbot.
O usuário troca com o comando /persona <nome>.
"""

PERSONAS = {
    "tecnico": (
        "Você é um engenheiro de software sênior. Responda de forma técnica, "
        "precisa e com exemplos de código quando relevante. Use terminologia correta."
    ),
    "professor": (
        "Você é um professor universitário paciente. Explique conceitos de forma "
        "didática, use analogias e pergunte se o aluno entendeu antes de avançar."
    ),
    "criativo": (
        "Você é um diretor criativo de uma agência. Responda com ideias originais, "
        "use metáforas visuais e sempre sugira abordagens fora da caixa."
    ),
    "coach": (
        "Você é um coach de carreira em tecnologia. Ajude com decisões profissionais, "
        "dê feedback construtivo e sugira planos de ação concretos."
    ),
}

def trocar_persona(messages: list[dict], persona: str) -> list[dict]:
    """Troca o system prompt mantendo o histórico."""
    if persona not in PERSONAS:
        print(f"Persona '{persona}' não encontrada. Opções: {list(PERSONAS.keys())}")
        return messages
    messages[0] = {"role": "system", "content": PERSONAS[persona]}
    print(f"🎭 Persona trocada para: {persona}")
    return messages

Tratamento de Erros e Retry

"""
Wrapper com retry automático para chamadas à API da OpenAI.
Lida com rate limits, timeouts e erros transitórios.
"""

import time
from openai import RateLimitError, APITimeoutError, APIConnectionError

def chat_com_retry(messages: list[dict], max_retries: int = 3) -> str:
    """Envia mensagem com retry automático em caso de erro."""
    for tentativa in range(max_retries):
        try:
            return chat(messages)
        except RateLimitError:
            wait = 2 ** tentativa  # backoff exponencial: 1s, 2s, 4s
            print(f"\n⏳ Rate limit atingido. Aguardando {wait}s...")
            time.sleep(wait)
        except APITimeoutError:
            print(f"\n⏳ Timeout. Tentativa {tentativa + 1}/{max_retries}...")
            time.sleep(1)
        except APIConnectionError:
            print(f"\n❌ Erro de conexão. Tentativa {tentativa + 1}/{max_retries}...")
            time.sleep(2)

    return "❌ Não foi possível obter resposta após várias tentativas."

Conclusão

O chatbot com streaming que construímos é uma base sólida para qualquer aplicação conversacional com IA. O streaming torna a experiência muito mais fluida — o usuário não fica olhando para uma tela em branco esperando uma resposta longa terminar de ser gerada.

Com as variações que adicionamos (persistência de histórico, contagem de tokens, múltiplas personalidades e retry robusto), você tem todos os blocos necessários para construir desde um assistente pessoal até um chatbot de atendimento ao cliente.

A chave para um bom chatbot é o system prompt: quanto mais específico e bem definido, melhores serão as respostas. Invista tempo refinando o prompt para o seu caso de uso.

Próximos Passos

  • Interface web: migrar do terminal para uma UI com Streamlit ou Gradio com streaming via WebSocket
  • Suporte a function calling: permitir que o chatbot execute ações (consultar APIs, acessar bancos de dados) usando a feature de tool use da OpenAI
  • Múltiplos provedores: abstrair o client para suportar Claude (Anthropic), Gemini (Google) e modelos locais via Ollama, permitindo trocar de modelo sem mudar a lógica
  • Moderação: usar a API de moderação da OpenAI para filtrar conteúdo impróprio antes de enviar ao modelo
  • Analytics: registrar métricas como tempo de resposta, tokens consumidos e satisfação do usuário para otimizar custos e qualidade
  • RAG: integrar Retrieval-Augmented Generation para que o chatbot responda baseado em documentos específicos da sua empresa