botpoet / app.py
segoedu's picture
Update app.py
811cd8b verified
raw
history blame
26.2 kB
from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI
from langchain_community.embeddings import HuggingFaceEmbeddings
#from langchain.chains import RetrievalQA
from langchain_pinecone import PineconeVectorStore
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from streamlit_option_menu import option_menu
#from elevenlabs import save, play, stream
from elevenlabs.client import ElevenLabs
from supabase import create_client, Client
from openai import OpenAI
from dotenv import load_dotenv
import os
import time
import streamlit as st
#####################
# CONFIGURACIÓN #
#####################
# Función de configuración de embedding
@st.cache_resource()
def load_embedding_model(model_name):
return HuggingFaceEmbeddings(model_name=model_name)
# Función de configuración
def configuracion():
# Cabecera
st.header(":gear: Configuración", divider='gray')
# API Keys #
hf_token = os.getenv('HF_TOKEN')
groq_api_key = os.getenv('GROQ_API_KEY')
openai_api_key = os.getenv('OPENAI_API_KEY')
pinecone_api_key = os.getenv('PINECONE_API_KEY')
supabase_key = os.getenv('SUPABASE_KEY')
supabase_url = os.getenv('SUPABASE_URL')
eleven_api_key = os.getenv('ELEVEN_API_KEY')
langsmith_api_key = os.getenv('LANGSMITH_API_KEY')
langchain_project = os.getenv('LANGCHAIN_PROJECT')
langchain_tracing_v2 = os.getenv('LANGCHAIN_TRACING_V2')
# Modelo de lenguaje #
modelos_llm = [
'llama3-8b-8192',
'llama3-70b-8192',
'mixtral-8x7b-32768',
'gemma-7b-it',
'gpt-4o',
]
modelo_llm = st.selectbox('Modelo de lenguaje', list(modelos_llm), help="Especifica el modelo de lenguaje utilizado")
# Temperatura
temperature = st.slider("Temperatura", 0.0, 2.0, 1.0, disabled=True, step=0.1)
if modelo_llm == "gpt-4o":
llm = ChatOpenAI(model=modelo_llm)
llm_creative = ChatOpenAI(model=modelo_llm, temperature=temperature)
else:
llm = ChatGroq(model=modelo_llm)
llm_creative = ChatGroq(model=modelo_llm, temperature=temperature)
# Embeddings #
model_name = 'intfloat/multilingual-e5-small'
model_name = st.text_input("Modelo de embedding:", value=model_name, disabled=True, help="Indica el modelo de _embedding_ utilizado en la vectorstore")
#embedding = HuggingFaceEmbeddings(model_name=model_name)
embedding = load_embedding_model(model_name)
# Pinecone #
index_name = "poemas-intfloat-multilingual-e5-small"
namespace = "poesiacastellana"
vectorstore = PineconeVectorStore(index_name=index_name, namespace=namespace, embedding=embedding)
col1, col2 = st.columns(2)
with col1:
input1 = st.text_input("Index", value=index_name, disabled=True, help="Nombre del _index_ del vectorstore en Pinecone para RAG")
with col2:
input2 = st.text_input("namespace", value=namespace, disabled=True, help="Nombre del _namespace_ del vectorstore en Pinecone")
# Supabase #
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_KEY")
supabase: Client = create_client(url, key)
# Variables globales
st.session_state.modelo_llm = modelo_llm
st.session_state.temperature = temperature
st.session_state.llm = llm
st.session_state.llm_creative = llm_creative
st.session_state.embedding = embedding
st.session_state.db = vectorstore
st.session_state.supabase = supabase
# API Keys #
with st.expander("API Keys"):
GROQ_API_KEY = st.text_input('Groq API Key', value=groq_api_key, type='password')
OPENAI_API_KEY = st.text_input('OpenAI API Key', value=openai_api_key, type='password')
HF_TOKEN = st.text_input('HuggingFace API Key', value=hf_token, type='password')
PINECONE_API_KEY = st.text_input('Pinecone API Key', value=pinecone_api_key, type='password')
ELEVEN_API_KEY = st.text_input('ElevenLabs API Key', value=eleven_api_key, type='password')
# ----- Función de configuración -----------------------------------------------------------
@st.cache_resource()
def __init__():
# API Keys #
load_dotenv()
# Modelo de lenguaje multimodal
llm_images = OpenAI()
# Modelo de lenguaje
modelo_llm = 'llama3-8b-8192'
temperature = 1
llm = ChatGroq(model=modelo_llm)
llm_creative = ChatGroq(model=modelo_llm, temperature=temperature)
# Embeddings
model_name = 'intfloat/multilingual-e5-small'
embedding = load_embedding_model(model_name)
# Pinecone
index_name = "poemas-intfloat-multilingual-e5-small"
namespace = "poesiacastellana"
vectorstore = PineconeVectorStore(index_name=index_name, namespace=namespace, embedding=embedding)
# Supabase
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_KEY")
supabase: Client = create_client(url, key)
# Globals
st.session_state.modelo_llm = modelo_llm
st.session_state.temperature = temperature
st.session_state.llm = llm
st.session_state.llm_creative = llm_creative
st.session_state.llm_images = llm_images
st.session_state.embedding = embedding
st.session_state.db = vectorstore
st.session_state.supabase = supabase
#####################
# GRAFO #
#####################
# ----- Función para desarrollar la metodología ------------------------------------------------
def grafo():
# Añadir una imagen al lado del título
col1, col2 = st.columns([4,20])
with col1:
st.image("img/bot.png", width=100)
with col2:
st.title("Generador de poemas")
# Botones de sugerencia para el tema/estilo
tema_suggestions = ["Amor", "Vida", "Misterio", "Aventura", "Sueños", "Amistad", "Impermanencia"]
estilo_suggestions = ["Romántico", "Gótico", "Épico", "Lírico", "Soneto", "Haiku", "Verso libre", "Gen27"]
# Campo para el tema
tema = st.text_input("Tema:", value="el despertar de la conciencia", help="Especifa el *tema* sobre el que quieres escribir el poema")
#tema_seleccionado = st.multiselect("Selecciona las opciones deseadas:", tema_suggestions)
# Campo para el estilo/estructura
estilo = st.text_input("Estilo:", value="antonio machado", help="Especifa el *estilo* con el que quieres escribir el poema")
#estilo_seleccionado = st.multiselect("Selecciona las opciones deseadas:", estilo_suggestions)
# Mostrar el conocimiento experto
with st.expander("Expertise Knowledge", icon="👩‍💻"):
for item in st.session_state.expertise:
st.markdown(f"```\n{item}\n```")
# Botón para generar el poema
if st.button("Generar Poema", type="primary", use_container_width=True):
# Variables globales
st.session_state.tema = tema
st.session_state.estilo = estilo
st.session_state.professor = ["No hay revisiones del profesor."]
# Generación del contexto
with st.spinner("Recuperando conocimiento aumentado..."):
rag_retrieval()
pass
st.success(":floppy_disk: ¡Conocimiento aumentado recuperado!")
# Generación del conocimiento
with st.spinner("Generando conocimiento experto..."):
generated_knowledge()
pass
st.success(":notebook: ¡Conocimiento experto generado!")
# Generación del borrador poético
with st.spinner("Generando borrador poético..."):
draft_poem()
pass
st.success(":memo: ¡Borrador poético generado!")
# Generando reflexion
with st.spinner("El catedrático está revisando el borrador poético..."):
reflexion_catedratico()
pass
st.success(":female-student: ¡Borrador poético revisado!")
# Generando poema
with st.spinner("Generando el poema..."):
st.session_state.poema = final_poem(st.session_state.draft_poem, st.session_state.reflexion)
pass
st.success(":scroll: ¡Poema generado!")
# Almacenar poema en Supabase
almacenar("insert")
# Mostrar poema
st.markdown(f"```\n{st.session_state.poema}\n```")
#####################
# AGENTES #
#####################
# ----- Función para mostrar la recuperación aumentada -----------------------------------
def rag_retrieval():
# Define the filter condition based on the 'author' metadata field
filter = {}
# Define el número de resultados
k=20
# Recupera los valores del grafo
tema = st.session_state.tema
# Perform the search: topic, human
documents = st.session_state.db.similarity_search(tema, k=k, filter=filter) #db.similarity_search_with_score
# Armamos la lista
rag_poems = [doc.page_content for doc in documents]
# Variable global
st.session_state.rag = rag_poems
# ----- Función para mostrar el conocimiento experto -------------------------------------
def expertise_knowledge():
# Mostramos una cabecera
st.header(":female-technologist: Expertise Knowledge (HITL)", divider='red')
# Campo de texto para ingresar poemas
poema = st.text_area("Poema de referencia", height=300, help="Añade aquí los poemas de referencia que sirvan como ejemplo para la elaboración del poema. Pulsa _Actualziar_ para limpiar el conjunto de poemas existente.")
# Botones para añadir y actualizar
col1, col2 = st.columns(2, gap="small")
with col1:
if st.button("Añadir", use_container_width=True):
st.session_state.expertise.insert(0, f"Poema:\n{poema.strip()}")
poema = "" # Limpiar el campo poema
with col2:
if st.button("Actualizar", type="primary", use_container_width=True):
st.session_state.expertise = [f"Poema:\n{poema.strip()}"]
poema = "" # Limpiar el campo poema
# Mostrar el conocimiento experto
st.divider()
#st.write("Expertise Knowledge:")
for poem in st.session_state.expertise:
st.markdown(f"```\n{poem}\n```")
# ----- Función para mostrar el conocimiento generado -------------------------------------
def generated_knowledge():
# Armamos el prompt
system = """
Eres un ratón de biblioteca y un profesor experto de literatura de una prestigiosa universidad con gran capacidad de \
sintetizar características de corrientes y estilos literarios. Tu objetivo es proporcionar las características principales \
y los detalles de las consultas académicas que te hagan sobre el estilo poético de '''{style_of}'''. Escribe las \
características principales sobre el estilo literario, movimiento o generación definido en el estilo o en el poeta nombrado, \
que sirvan de guía para la elaboración de nuevos textos que imiten el estilo indicado o al poeta solicitado.
Es importante que no escribas ningún poema, tan solo enumera un listado detallado de características, que sean útiles \
para poder reproducirse y aplicarse en la generación de nuevos textos.
"""
human = 'Devuelve SOLO un listado detallado de las características en formato JSON. \
Formato: {{ "Características": ["característica 1", "característica 2", "característica 3"]}}'
prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", human),
])
# Recupera los valores del grafo
style_of = st.session_state.estilo
# Lanzamos la consulta
chain = prompt | st.session_state.llm | JsonOutputParser()
knowledge = chain.invoke({"style_of": style_of})
# Parser
knowledge_values = list(knowledge.values())[0]
# Valor global
st.session_state.generated = knowledge_values
# ----- Función para generar el borrado poético------ -------------------------------------
def draft_poem():
# Armamos el prompt
system = """
Eres un maestro de la poesía, capaz de escribir poemas que evocan emociones poderosas y obras maestras poéticas \
que capten la esencia del tema especificado de forma profunda y evocadora. Escribes desde la belleza. Cada línea \
cuidadosamente elaborada proyecta una imagen vívida, evocando emociones fuertes y hermosas con una destreza poética \
que deja sin aliento al lector. Tu estilo es una fusión de metáforas, aliteración y cadencia rítmica.
Las imágenes que evocan tus palabras son originales, creativas y únicas, alejándose de los tópicos y dotadas de un \
significado profundo y auténtico. Tus poemas están escritos para recitarse y leerse en voz alta. Se hace hincapié en \
que la escritura sea vívida y detallada, al tiempo que se garantiza que los poemas tengan un mensaje profundo que \
resuene en el lector mucho después de que haya terminado de leerlos. El objetivo es crear una poesía impactante y \
bella que deje una impresión duradera en el público.
Haces un uso de la rima es muy sutil, SOLO la utilizas cuando es necesaria. Si las características proporcionadas no \
indican lo contrario, haz un uso responsable y limitado de ella.
Esta es una lista de poemas que debes utilizar como EJEMPLOS e inspiración para escribir el poema:
'''{expertise}'''
Aquí tienes una lista de poemas con una temática similar que puedes usar para inspirarte:
'''
{rag_poems}
'''
"""
human = (
"El poema debe escribirse atendiendo a las siguientes CARACTERÍSTICAS: "
"\n'''"
"\n * 18 versos o menos, no debes superar NUNCA esta cantidad "
"\n * {knowledge} "
"\n''' "
"\n\nEscribe un BREVE poema de menos de 18 versos sobre el tema de '''{topic}'''. "
"No hagas comentarios. " + "Esto es muy importante para mi carrera."
)
prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", human),
])
# Recupera los valores del grafo
topic = st.session_state.tema
knowledge = st.session_state.generated
expertise = st.session_state.expertise
rag_poems = st.session_state.rag
# Convertir las listas en texto
knowledge = '\n * '.join(knowledge)
expertise = ' '.join(expertise)
rag_poems = ' '.join(rag_poems)
# Lanzamos la consulta
chain = prompt | st.session_state.llm_creative | StrOutputParser()
draft_poem = chain.invoke({"topic": topic, "knowledge": knowledge, "expertise": expertise, "rag_poems": rag_poems})
# Valor global
st.session_state.draft_poem = draft_poem
# ----- Función para generar el poema final -------------------------------------------------
def final_poem(poema, review):
# Armamos el prompt
system = """
Eres un experto escritor de poesía que mejora los poemas. Tu objetivo es reescribir el poema proporcionado \
a partir de los comentarios proporcionadas por el profesor experto. \
POEMA:
'''
{poema}
'''
COMENTARIOS del profesor experto:
'''
* {review}
'''
"""
human=(
"Escribe solo el poema final mejorado, el mejor posible. "
"No hagas comentarios, ni aclaraciones sobre lo que has modificado. Esto es muy importante para mi carrera."
)
prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", human),
])
# Convertir las listas en texto
review = '\n * '.join(list(review))
# Lanzamos la consulta
chain = prompt | st.session_state.llm_creative | StrOutputParser()
final_poem = chain.invoke({"poema": poema, "review": review})
# Valor global
return final_poem
# ----- Función para mostrar el conocimiento generado -------------------------------------
def reflexion_catedratico():
# Armamos el prompt
system = """
Eres un catedrático experto en poesía de una prestigiosa universidad con un profundo conocimiento y experiencia \
en el análisis de poemas. Tienes ojo crítico y una pasión por la perfección literaria. Tu objetivo es revisar y \
sugerir mejoras para los poemas.
Evalúas y analizas la calidad del poema recibido y proporcionas comentarios constructivos que permitan mejorar el resultado final.
Este es el poema:
'''
{draft_poem}
'''
"""
#Devuelve un listado detallado de las características en formato JSON con una única clave 'comentarios'.
human = 'Devuelve SOLO un listado detallado de los comentarios en formato JSON. \
Formato: {{ "comentarios": ["comentario 1", "comentario 2", "comentario 3"]}}'
prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", human),
])
# Recupera los valores del grafo
draft_poem = st.session_state.draft_poem
# Lanzamos la consulta
chain = prompt | st.session_state.llm | JsonOutputParser()
reflexion = chain.invoke({"draft_poem": draft_poem})
# Parser
comentarios = list(reflexion.values())[0]
# Valor global
st.session_state.reflexion = comentarios
# ----- Función para mostrar el conocimiento experto -------------------------------------
def professor_reflexion():
st.header(":female-student: Professor Reflexion (HITL)", divider='red')
# Mostrar el texto combinado
st.write("Poema")
st.markdown(f"```\n{st.session_state.poema}\n```")
# Valor global
if st.session_state.poema != "No hay poema elaborado.":
col1, col2 = st.columns(2)
with col1:
# Audio
voz = st.button("Recitar", use_container_width=True, help="Recita el poema en alto")
with col2:
# Imagen
imagen = st.button("Imaginar", use_container_width=True, help="Genera una imagen a partir del poema")
if voz:
try:
# Generamos el audio
audio = recitar(st.session_state.poema)
st.audio(audio, format="audio/mp3")
except Exception as e:
# Manejo de la excepción
st.error(f"No dispones de créditos suficientes para realizar esta operación.")
if imagen:
try:
# Generamos la imagen
url_imagen, enlace = pintar(st.session_state.draft_poem)
# Mostrar la imagen
st.image(url_imagen, use_column_width="always")
st.markdown(enlace, unsafe_allow_html=True)
except Exception as e:
# Manejo de la excepción
st.error(f"No dispones de créditos suficientes para realizar esta operación.")
# Crear cuatro campos de textarea
professor = st.text_area("Comentarios, crítica y sugerencias", help="Escribe aquí la revisión experta del catedrático.")
# Reflexión
if st.button("Escribir", type="primary", use_container_width=True):
# Parser
st.session_state.professor = [professor]
st.session_state.expert_poem = st.session_state.poema
with st.spinner("Reformando el poema..."):
st.session_state.poema = final_poem(st.session_state.poema, professor)
pass
st.success("¡Poema reformado! :thumbsup:")
# Almacenamos el poema en Supabase
almacenar("update")
# Mostramos el poema elaborado
st.markdown(f"```{st.session_state.poema}```")
# ----- Función para mostrar el resumen del poema generado ----------------------------------
@st.cache_resource()
def pintar(poema):
# Lanzamos la consulta al modelo de lenguaje multimodal
prompt = "Dibuja una imagen que represente de forma abstracta el siguiente poema. No incluyas NINGUN texto. No incluyas texto. POEMA:" + poema
response = st.session_state.llm_images.images.generate(
model="dall-e-3",
prompt=prompt,
size="1024x1024",
quality="standard",
n=1,
)
# Capturar el link
url_imagen = response.data[0].url
# Crear el enlace en Markdown
link_text = "Haz clic aquí para ver la imagen en grande"
link = f'[{link_text}]({url_imagen})'
return url_imagen, link
# ----- Función para mostrar el resumen del poema generado ----------------------------------
@st.cache_resource()
def recitar(poema):
eleven_api_key = os.environ.get("ELEVEN_API_KEY")
client = ElevenLabs(
api_key=eleven_api_key,
)
audio = client.generate(
text=poema,
voice="Sara Martin 3",
model="eleven_multilingual_v2"
)
audio_bytes = b''
for chunk in audio:
audio_bytes += chunk
return audio_bytes
# ----- Función para almacenar el registro en la base de datos ----------------------------
def almacenar(accion):
# Insertar los datos en la tabla
data = {
"model": st.session_state.modelo_llm,
"temperature": st.session_state.temperature,
"topic": st.session_state.tema,
"style_of": st.session_state.estilo,
"knowledge": st.session_state.generated,
"expertise": st.session_state.expertise,
"rag_poems": st.session_state.rag,
"draft_poem": st.session_state.draft_poem,
"draft_poem_fb": st.session_state.reflexion,
"professor": st.session_state.professor,
"final_poem": st.session_state.poema,
}
# Almacenamos los poemas generados
if accion == "insert":
# Añadimos registro
data, count = st.session_state.supabase.table("Poemas").insert(data).execute()
# Recuperamos el id insertado
st.session_state.id = data[1][0]['id']
if accion == "update":
id = st.session_state.id
data, count = st.session_state.supabase.table("Poemas").update(data).eq("id", id).execute()
# ----- Función para mostrar el resumen del poema generado ----------------------------------
def resumen():
st.header(":memo: Metodología desarrollada", divider='gray')
# Muestro el tema y estilo proporcionados
col1, col2 = st.columns(2)
with col1:
input1 = st.text_input("Tema", value=st.session_state.tema, disabled=True)
with col2:
input2 = st.text_input("Estilo", value=st.session_state.estilo, disabled=True)
# Separador
st.subheader('Contexto', divider='red', help="Se muestra el *contexto* generado que se incrustará en el _prompt_")
# Mostrar el conocimiento experto
with st.expander("Expertise Knowledge (HITL)", icon="👩‍💻"):
for item in st.session_state.expertise:
st.markdown(f"```\n{item}\n```")
# Imprimir la conocimiento generado
with st.expander("Conocimiento generado", icon="💻"):
for item in st.session_state.generated:
st.markdown(f"""<span>{item}</span>""", unsafe_allow_html=True)
# Imprimir la conocimiento recuperado
with st.expander("Conocimiento recuperado", icon="💾"):
for item in st.session_state.rag:
st.markdown(f"```\n{item}\n```")
# Separador
st.subheader('Reflexión', divider='red', help="Se muestra el borrador poético y las reflexiones de los expertos")
# Imprimir el borrador poético
with st.expander("Borrador poético", icon="📃"):
st.markdown(f"```\n{st.session_state.draft_poem}\n```")
# Imprimir la reflexión del catedrático
with st.expander("Reflexión del catedrático", icon="🎓"):
for item in st.session_state.reflexion:
st.markdown(f"""<span>{item}</span>""", unsafe_allow_html=True)
# Imprimir el poema
if "expert_poem" in st.session_state:
with st.expander("Poema revisado", icon="📃"):
st.markdown(f"```{st.session_state.expert_poem}```")
# Imprimir la reflexión del catedrático
if "professor" in st.session_state:
with st.expander("Reflexión del profesor (HITL)", icon="👩‍🎓"):
for item in st.session_state.professor:
st.markdown(f"""<span>{item}</span>""", unsafe_allow_html=True)
# Separador
st.subheader('Poema :scroll:', divider='rainbow', help="Se muestra el poema elaborado")
# Imprimir el poema
st.markdown(f"```\n{st.session_state.poema}\n```")
#####################
# Función principal #
#####################
def main():
st.set_page_config(page_title="Generador de poemas")
if "draft_poem" not in st.session_state:
st.session_state.draft_poem = "No hay borrador poético."
#st.session_state.expert_poem = ""
#st.session_state.poema = "No hay poema elaborado."
st.session_state.poema = st.__version__
if "generated" not in st.session_state:
st.session_state.generated = ["No hay conocimiento generado."]
if "expertise" not in st.session_state:
st.session_state.expertise = ["No hay conocimiento experto proporcionado."]
if "rag" not in st.session_state:
st.session_state.rag = ["No hay conocimiento aumentado por recuperación."]
if "reflexion" not in st.session_state:
st.session_state.reflexion = ["No hay revisiones del catedrático."]
if "professor" not in st.session_state:
st.session_state.professor = ["No hay revisiones del profesor."]
if 'tema' not in st.session_state:
st.session_state.tema = ""
if 'estilo' not in st.session_state:
st.session_state.estilo = ""
# Ejecutamos la configuración
__init__()
# Sidebar con los enlaces a los pasos
opciones = {
"Topic&Style": grafo,
"Metodología": resumen,
"Expertise (HITL)": expertise_knowledge,
"Professor (HITL)": professor_reflexion,
"Configuración": configuracion,
}
#st.sidebar.title("Pasos")
#selected = st.sidebar.radio("Selecciona un paso", list(opciones.keys()))
# Enlace a la librería de menús: https://github.com/victoryhb/streamlit-option-menu
with st.sidebar:
seleccion = option_menu(None, list(opciones.keys()),
icons=['envelope-paper-heart','envelope-heart','person-hearts','person-square', 'gear'],
menu_icon="cast",
default_index=0)
st.image("img/bot.webp", width=250)
# Muestra el paso seleccionado
opciones[seleccion]()
if __name__ == "__main__":
main()