chatbot_mentor.py– Etapa #6 (Com Validações)
# ==========================================================
# IMPORTAÇÕES
# ==========================================================
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.memory import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# ==========================================================
# VERIFICAÇÃO DA CHAVE DE API
# ==========================================================
load_dotenv()
if "OPENAI_API_KEY" not in os.environ:
raise ValueError(
" ERRO: A variável OPENAI_API_KEY não foi encontrada no arquivo .env."
)
print("Chave OPENAI_API_KEY carregada com sucesso.\n")
# ==========================================================
# MODELO
# ==========================================================
llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7
)
# ==========================================================
# PROMPT TEMPLATE
# ==========================================================
prompt = ChatPromptTemplate.from_messages([
(
"system",
"Você é o 'GeoAI Mentor', um assistente especializado em ajudar "
"geocientistas a migrar para a área de Ciência de Dados. "
"Seja amigável, didático e estratégico."
),
("placeholder", "{historico}"),
("human", "{query}")
])
parser = StrOutputParser()
chain = prompt | llm | parser
# ==========================================================
# MEMÓRIA POR SESSÃO (Singleton)
# ==========================================================
memoria_sessoes = {}
def obter_historico_por_sessao(session_id: str):
if session_id not in memoria_sessoes:
memoria_sessoes[session_id] = InMemoryChatMessageHistory()
return memoria_sessoes[session_id]
cadeia_com_memoria = RunnableWithMessageHistory(
runnable=chain,
get_session_history=obter_historico_por_sessao,
input_messages_key="query",
history_messages_key="historico",
)
# ==========================================================
# TESTE DE CONSISTÊNCIA DAS CHAVES
# ==========================================================
print("Verificando consistência das chaves...")
assert "query" in prompt.input_variables, " A variável {query} não está no prompt."
assert "historico" in prompt.input_variables, " A variável {historico} não está no prompt."
print(" Variáveis do prompt estão consistentes com RunnableWithMessageHistory.\n")
# ==========================================================
# EXECUÇÃO DE TESTE
# ==========================================================
perguntas = [
"Eu sou geofísico e quero migrar para a área de dados. Qual linguagem devo aprender?",
"Que tipo de projeto posso criar usando essa linguagem?"
]
session_id = "usuario_teste"
print("Iniciando teste de memória...\n")
for pergunta in perguntas:
resposta = cadeia_com_memoria.invoke(
{"query": pergunta},
config={"configurable": {"session_id": session_id}}
)
print("Pergunta:")
print(pergunta)
print("\nResposta:")
print(resposta)
print("\n" + "-" * 60)
# ==========================================================
# VERIFICAÇÃO LÓGICA DA MEMÓRIA
# ==========================================================
print("\n Revisão Final:")
print("Se a segunda resposta mencionar a linguagem sugerida na primeira,")
print("a memória está funcionando corretamente. ")