151 lines
4.7 KiB
Python
Executable File
151 lines
4.7 KiB
Python
Executable File
"""
|
|
ai_chat_processor.py
|
|
|
|
Author: Descomplicar® Crescimento Digital
|
|
Link: https://descomplicar.pt
|
|
Copyright: 2025 Descomplicar®
|
|
"""
|
|
|
|
"""
|
|
Processador para chatbot AI integrado com Chatwoot.
|
|
"""
|
|
import json
|
|
import requests
|
|
from typing import Dict, List, Optional
|
|
from .db_processor import DBProcessor
|
|
from .embedding_processor import EmbeddingProcessor
|
|
|
|
class AIChatProcessor:
|
|
def __init__(self, chatwoot_api_key: str, chatwoot_account_id: str, chatwoot_base_url: str):
|
|
"""
|
|
Inicializa o processador de chat AI.
|
|
|
|
Args:
|
|
chatwoot_api_key: Chave API do Chatwoot
|
|
chatwoot_account_id: ID da conta Chatwoot
|
|
chatwoot_base_url: URL base da API Chatwoot (ex: https://app.chatwoot.com)
|
|
"""
|
|
self.chatwoot_api_key = chatwoot_api_key
|
|
self.chatwoot_account_id = chatwoot_account_id
|
|
self.chatwoot_base_url = chatwoot_base_url.rstrip('/')
|
|
|
|
# Inicializa processadores
|
|
self.db = DBProcessor()
|
|
self.embedding_processor = EmbeddingProcessor()
|
|
|
|
# Headers para API Chatwoot
|
|
self.headers = {
|
|
'api_access_token': chatwoot_api_key,
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
def _search_knowledge_base(self, query: str, limit: int = 3) -> List[Dict]:
|
|
"""
|
|
Pesquisa na base de conhecimento usando embeddings.
|
|
|
|
Args:
|
|
query: Pergunta do usuário
|
|
limit: Número máximo de resultados
|
|
|
|
Returns:
|
|
Lista de resultados mais relevantes
|
|
"""
|
|
# Gera embedding da pergunta
|
|
query_embedding = self.embedding_processor.generate_embedding(query)
|
|
|
|
# Busca resultados similares
|
|
results = self.db.search_similar_chunks(
|
|
query_embedding,
|
|
limit=limit
|
|
)
|
|
|
|
return results
|
|
|
|
def _format_response(self, query: str, results: List[Dict]) -> str:
|
|
"""
|
|
Formata resposta com base nos resultados da pesquisa.
|
|
|
|
Args:
|
|
query: Pergunta original
|
|
results: Resultados da pesquisa
|
|
|
|
Returns:
|
|
Resposta formatada
|
|
"""
|
|
if not results:
|
|
return "Desculpe, não encontrei informações relevantes sobre isso na base de conhecimento."
|
|
|
|
# Formata resposta
|
|
response = "Com base na nossa base de conhecimento:\n\n"
|
|
|
|
for i, result in enumerate(results, 1):
|
|
relevance = result.get('relevance', 0) * 100
|
|
content = result.get('content', '').strip()
|
|
source = result.get('source', 'Desconhecida')
|
|
|
|
response += f"{i}. Relevância: {relevance:.1f}%\n"
|
|
response += f"Fonte: {source}\n"
|
|
response += f"Conteúdo: {content}\n\n"
|
|
|
|
return response
|
|
|
|
def process_message(self, conversation_id: str, message: str) -> None:
|
|
"""
|
|
Processa mensagem recebida e envia resposta via Chatwoot.
|
|
|
|
Args:
|
|
conversation_id: ID da conversa no Chatwoot
|
|
message: Mensagem recebida
|
|
"""
|
|
# Pesquisa na base de conhecimento
|
|
results = self._search_knowledge_base(message)
|
|
|
|
# Formata resposta
|
|
response = self._format_response(message, results)
|
|
|
|
# Envia resposta via API Chatwoot
|
|
endpoint = f"{self.chatwoot_base_url}/api/v1/accounts/{self.chatwoot_account_id}/conversations/{conversation_id}/messages"
|
|
|
|
payload = {
|
|
'content': response,
|
|
'message_type': 'outgoing',
|
|
'private': False
|
|
}
|
|
|
|
try:
|
|
r = requests.post(
|
|
endpoint,
|
|
headers=self.headers,
|
|
data=json.dumps(payload)
|
|
)
|
|
r.raise_for_status()
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Erro ao enviar mensagem para Chatwoot: {e}")
|
|
|
|
def handle_webhook(self, data: Dict) -> Optional[str]:
|
|
"""
|
|
Processa webhook recebido do Chatwoot.
|
|
|
|
Args:
|
|
data: Dados do webhook
|
|
|
|
Returns:
|
|
ID da conversa se mensagem deve ser processada
|
|
"""
|
|
# Verifica se é mensagem de entrada
|
|
if data.get('message_type') != 'incoming':
|
|
return None
|
|
|
|
# Extrai dados relevantes
|
|
conversation_id = data.get('conversation', {}).get('id')
|
|
message = data.get('content')
|
|
|
|
if not all([conversation_id, message]):
|
|
return None
|
|
|
|
# Processa mensagem
|
|
self.process_message(conversation_id, message)
|
|
|
|
return conversation_id
|