Código utilizando una inferencia con Ollama en python creando y accediendo a una base de datos a través de un archivo .txt.
from langchain_community.document_loaders import TextLoader
from langchain_community.embeddings import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatOllama
from langchain_core.runnables import RunnablePassthrough
from langchain.retrievers.multi_query import MultiQueryRetriever
local_path = "datoscasa2.txt"
# Local .txt file uploads
if local_path:
loader = TextLoader(file_path=local_path) # Use TextFileLoader for .txt
data = loader.load() # Load with explicit encoding
else:
print("Upload a .txt file")
# Split and chunk
text_splitter = RecursiveCharacterTextSplitter(chunk_size=7500, chunk_overlap=100)
chunks = text_splitter.split_documents(data)
# Define a persistent collection name (avoid deletion after each interaction)
collection_name = "local-rag"
custom_db_directory = "/Users/cesarhernandez/Documents/PlatformIO/Projects/RAG-1/prueba2"
vector_db = Chroma.from_documents(
documents=chunks,
embedding=OllamaEmbeddings(model="phi3", show_progress=False),
collection_name=collection_name,
persist_directory=custom_db_directory
)
# LLM from Ollama
local_model = "phi3"
llm = ChatOllama(model=local_model)
QUERY_PROMPT = PromptTemplate(
input_variables=["question"],
template="""Eres una asistente y te llamas Lara.
Pregunta original: {question}""",
)
retriever = MultiQueryRetriever.from_llm(
vector_db.as_retriever(),
llm,
prompt=QUERY_PROMPT
)
# RAG prompt
template = """Debes de responder a cualquier pregunta:
{context}
Pregunta: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
def main():
while True:
texto_introducido=input("Cesar: ")
if texto_introducido.lower() == 'adios':
print('¡Hasta luego!')
break
resultado = chain.invoke(texto_introducido)
print(resultado)
if __name__ == "__main__":
main()
Código haciendo accediendo a un LLM en local y base de datos en csv excel:
import pandas as pd
from langchain_community.embeddings import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatOllama
from langchain_core.runnables import RunnablePassthrough
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.schema import Document
local_path = "datoscasa2.csv"
# Carga de datos desde un archivo CSV local
if local_path:
try:
# Lee el archivo CSV
df = pd.read_csv(local_path)
# Convierte el DataFrame a una cadena de texto
data = df.to_string()
# Crea un Documento a partir de la cadena de texto
document = Document(page_content=data)
documents = [document] # Lista que contiene el Documento
except Exception as e:
print(f"Se produjo un error al leer el archivo CSV: {e}")
documents = []
else:
print("Sube un archivo .csv")
# Verifica si se cargaron documentos correctamente
if not documents:
print("No se cargaron documentos del archivo.")
else:
# Split and chunk
text_splitter = RecursiveCharacterTextSplitter(chunk_size=7500, chunk_overlap=100)
chunks = text_splitter.split_documents(documents)
# Define a persistent collection name (avoid deletion after each interaction)
collection_name = "local-rag"
custom_db_directory = "/Users/cesarhernandez/Documents/PlatformIO/Projects/RAG-1/prueba2"
vector_db = Chroma.from_documents(
documents=chunks,
embedding=OllamaEmbeddings(model="phi3", show_progress=False),
collection_name=collection_name,
persist_directory=custom_db_directory
)
# LLM from Ollama
local_model = "phi3"
llm = ChatOllama(model=local_model)
QUERY_PROMPT = PromptTemplate(
input_variables=["question"],
template="""Eres una asistente y te llamas Lara.
Pregunta original: {question}""",
)
retriever = MultiQueryRetriever.from_llm(
vector_db.as_retriever(),
llm,
prompt=QUERY_PROMPT
)
# RAG prompt
template = """Debes de responder a cualquier pregunta:
{context}
Pregunta: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
def main():
while True:
texto_introducido = input("Cesar: ")
if texto_introducido.lower() == 'adios':
print('¡Hasta luego!')
break
resultado = chain.invoke({"question": texto_introducido})
print(resultado)
if __name__ == "__main__":
main()Codigo para interactuar en remoto y en caso de fallo en local:
from huggingface_hub import InferenceClient
from langchain.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import contextlib
import io
# Configurar el cliente de Hugging Face
client = InferenceClient(
"microsoft/Phi-3-mini-4k-instruct",
token="hf_IKVbAPjJukFCLloEKeYdupabzOvIfXqni"
)
# Configurar el modelo local (Phi-3)
local_model = "phi3"
llm = ChatOllama(model=local_model)
# Prompt para la generación de respuesta
template = """Debes de responder a cualquier pregunta:
Pregunta: {question}
"""
prompt_template = ChatPromptTemplate.from_template(template)
# Configurar la cadena para el procesamiento local
chain = (
{"question": RunnablePassthrough()}
| prompt_template
| llm
| StrOutputParser()
)
# Context manager para suprimir la salida estándar y la salida de error
@contextlib.contextmanager
def suppress_stdout_stderr():
with open(io.StringIO(), 'w') as f, contextlib.redirect_stdout(f), contextlib.redirect_stderr(f):
yield
def main():
while True:
texto_introducido = input("Cesar: ")
if texto_introducido.lower() == 'adios':
print('¡Hasta luego!')
break
# Crear el prompt usando el template
prompt = prompt_template.format(question=texto_introducido)
try:
# Suprimir la salida durante la inferencia
with suppress_stdout_stderr():
# Realizar la inferencia usando el cliente de Hugging Face
messages = [{"role": "user", "content": prompt}]
response = client.chat_completion(messages=messages, max_tokens=500)
# Imprimir el resultado
if hasattr(response, 'choices') and response.choices:
generated_text = response.choices[0].message.content
else:
generated_text = str(response) # Si la respuesta es un string
print(generated_text)
except Exception as e:
# Si hay un error, suprimir el mensaje y usar el modelo local
error_message = str(e)
if "Bad request" in error_message or "does not seem to support chat completion" in error_message:
pass # Suprimir mensaje específico
try:
resultado = chain.invoke({"question": texto_introducido})
print(resultado)
except Exception as local_e:
print(f"Error al usar el modelo local: {local_e}")
if __name__ == "__main__":
main()Ollama en local:
from huggingface_hub import InferenceClient
from langchain.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import contextlib
import io
# Configurar el cliente de Hugging Face
client = InferenceClient(
"microsoft/Phi-3-mini-4k-instruct",
token="hf_IKVbAPjJukFCLloEKeYdupabzOvIfXqni"
)
# Configurar el modelo local (Phi-3)
local_model = "phi3"
llm = ChatOllama(model=local_model)
# Prompt para la generación de respuesta
template = """Debes de responder a cualquier pregunta:
Pregunta: {question}
"""
prompt_template = ChatPromptTemplate.from_template(template)
# Configurar la cadena para el procesamiento local
chain = (
{"question": RunnablePassthrough()}
| prompt_template
| llm
| StrOutputParser()
)
# Context manager para suprimir la salida estándar y la salida de error
@contextlib.contextmanager
def suppress_stdout_stderr():
with open(io.StringIO(), 'w') as f, contextlib.redirect_stdout(f), contextlib.redirect_stderr(f):
yield
def main():
while True:
texto_introducido = input("Cesar: ")
if texto_introducido.lower() == 'adios':
print('¡Hasta luego!')
break
# Crear el prompt usando el template
prompt = prompt_template.format(question=texto_introducido)
try:
# Suprimir la salida durante la inferencia
with suppress_stdout_stderr():
# Realizar la inferencia usando el cliente de Hugging Face
messages = [{"role": "user", "content": prompt}]
response = client.chat_completion(messages=messages, max_tokens=500)
# Imprimir el resultado
if hasattr(response, 'choices') and response.choices:
generated_text = response.choices[0].message.content
else:
generated_text = str(response) # Si la respuesta es un string
print(generated_text)
except Exception as e:
# Si hay un error, suprimir el mensaje y usar el modelo local
error_message = str(e)
if "Bad request" in error_message or "does not seem to support chat completion" in error_message:
pass # Suprimir mensaje específico
try:
resultado = chain.invoke({"question": texto_introducido})
print(resultado)
except Exception as local_e:
print(f"Error al usar el modelo local: {local_e}")
if __name__ == "__main__":
main()Accediendo a Huggingface:
from huggingface_hub import InferenceClient
from langchain.prompts import ChatPromptTemplate
# Configurar el cliente de Hugging Face
client = InferenceClient(
"microsoft/Phi-3-mini-4k-instruct",
token="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"
)
# Prompt para la generación de respuesta
template = """Debes de responder a cualquier pregunta:
Pregunta: {question}
"""
prompt_template = ChatPromptTemplate.from_template(template)
def main():
while True:
texto_introducido = input("Cesar: ")
if texto_introducido.lower() == 'adios':
print('¡Hasta luego!')
break
# Crear el prompt usando el template
prompt = prompt_template.format(question=texto_introducido)
# Realizar la inferencia usando el cliente de Hugging Face
messages = [{"role": "user", "content": prompt}]
response = client.chat_completion(messages=messages, max_tokens=500)
# Imprimir el resultado
if hasattr(response, 'choices') and response.choices:
generated_text = response.choices[0].message.content
else:
generated_text = str(response) # Si la respuesta es un string
print(generated_text)
if __name__ == "__main__":
main()Video de explicacion:
Deja una respuesta Cancelar la respuesta