Spaces:
Running
Running
File size: 7,431 Bytes
35dac1d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 |
# For reading credentials from the .env file
import os
from dotenv import load_dotenv
from sentence_transformers import SentenceTransformer
from chromadb.api.types import EmbeddingFunction
# WML python SDK
from ibm_watson_machine_learning.foundation_models import Model
from ibm_watson_machine_learning.metanames import GenTextParamsMetaNames as GenParams
from ibm_watson_machine_learning.foundation_models.utils.enums import ModelTypes, DecodingMethods
import requests
from bs4 import BeautifulSoup
import spacy
import chromadb
import en_core_web_md
# Important: hardcoding the API key in Python code is not a best practice. We are using
# this approach for the ease of demo setup. In a production application these variables
# can be stored in an .env or a properties file
# URL of the hosted LLMs is hardcoded because at this time all LLMs share the same endpoint
url = "https://us-south.ml.cloud.ibm.com"
# These global variables will be updated in get_credentials() function
watsonx_project_id = ""
# Replace with your IBM Cloud key
api_key = ""
def get_credentials():
load_dotenv()
# Update the global variables that will be used for authentication in another function
globals()["api_key"] = os.getenv("api_key", None)
globals()["watsonx_project_id"] = os.getenv("project_id", None)
# The get_model function creates an LLM model object with the specified parameters
def get_model(model_type, max_tokens, min_tokens, decoding, temperature, top_k, top_p):
generate_params = {
GenParams.MAX_NEW_TOKENS: max_tokens,
GenParams.MIN_NEW_TOKENS: min_tokens,
GenParams.DECODING_METHOD: decoding,
GenParams.TEMPERATURE: temperature,
GenParams.TOP_K: top_k,
GenParams.TOP_P: top_p,
}
model = Model(
model_id=model_type,
params=generate_params,
credentials={
"apikey": api_key,
"url": url
},
project_id=watsonx_project_id
)
return model
def get_model_test(model_type, max_tokens, min_tokens, decoding, temperature):
generate_params = {
GenParams.MAX_NEW_TOKENS: max_tokens,
GenParams.MIN_NEW_TOKENS: min_tokens,
GenParams.DECODING_METHOD: decoding,
GenParams.TEMPERATURE: temperature
}
model = Model(
model_id=model_type,
params=generate_params,
credentials={
"apikey": api_key,
"url": url
},
project_id=watsonx_project_id
)
return model
# Embedding function
class MiniLML6V2EmbeddingFunction(EmbeddingFunction):
MODEL = SentenceTransformer('all-MiniLM-L6-v2')
def __call__(self, texts):
return MiniLML6V2EmbeddingFunction.MODEL.encode(texts).tolist()
def extract_text(url):
try:
# Send an HTTP GET request to the URL
response = requests.get(url)
# Check if the request was successful
if response.status_code == 200:
# Parse the HTML content of the page using BeautifulSoup
soup = BeautifulSoup(response.text, 'html.parser')
# Extract contents of <p> elements
p_contents = [p.get_text() for p in soup.find_all('p')]
# Print the contents of <p> elements
print("\nContents of <p> elements: \n")
for content in p_contents:
print(content)
raw_web_text = " ".join(p_contents)
# remove \xa0 which is used in html to avoid words break acorss lines.
cleaned_text = raw_web_text.replace("\xa0", " ")
return cleaned_text
else:
print(f"Failed to retrieve the page. Status code: {response.status_code}")
except Exception as e:
print(f"An error occurred: {str(e)}")
def split_text_into_sentences(text):
nlp = spacy.load("en_core_web_md")
doc = nlp(text)
sentences = [sent.text for sent in doc.sents]
cleaned_sentences = [s.strip() for s in sentences]
return cleaned_sentences
def create_embedding(url, collection_name):
cleaned_text = extract_text(url)
cleaned_sentences = split_text_into_sentences(cleaned_text)
client = chromadb.Client()
collection = client.get_or_create_collection(collection_name)
# Upload text to chroma
collection.upsert(
documents=cleaned_sentences,
metadatas=[{"source": str(i)} for i in range(len(cleaned_sentences))],
ids=[str(i) for i in range(len(cleaned_sentences))],
)
return collection
def create_prompt(url, question, collection_name):
# Create embeddings for the text file
collection = create_embedding(url, collection_name)
# query relevant information
relevant_chunks = collection.query(
query_texts=[question],
n_results=5,
)
context = "\n\n\n".join(relevant_chunks["documents"][0])
# Please note that this is a generic format. You can change this format to be specific to llama
prompt = (f"{context}\n\nPlease answer the following question in one sentence using this "
+ f"text. "
+ f"If the question is unanswerable, say \"unanswerable\". Do not include information that's not relevant to the question."
+ f"Question: {question}")
return prompt
def main():
# Get the API key and project id and update global variables
get_credentials()
# Try diffrent URLs and questions
url = "https://www.usbank.com/financialiq/manage-your-household/buy-a-car/own-electric-vehicles-learned-buying-driving-EVs.html"
question = "What are the incentives for purchasing EVs?"
# question = "What is the percentage of driving powered by hybrid cars?"
# question = "Can an EV be plugged in to a household outlet?"
collection_name = "test_web_RAG"
answer_questions_from_web(api_key, watsonx_project_id, url, question, collection_name)
def answer_questions_from_web(request_api_key, request_project_id, url, question, collection_name):
# Update the global variable
globals()["api_key"] = request_api_key
globals()["watsonx_project_id"] = request_project_id
# Specify model parameters
model_type = "meta-llama/llama-2-70b-chat"
max_tokens = 100
min_tokens = 50
top_k = 50
top_p = 1
decoding = DecodingMethods.GREEDY
temperature = 0.7
# Get the watsonx model = try both options
model = get_model(model_type, max_tokens, min_tokens, decoding, temperature, top_k, top_p)
# Get the prompt
complete_prompt = create_prompt(url, question, collection_name)
# Let's review the prompt
print("----------------------------------------------------------------------------------------------------")
print("*** Prompt:" + complete_prompt + "***")
print("----------------------------------------------------------------------------------------------------")
generated_response = model.generate(prompt=complete_prompt)
response_text = generated_response['results'][0]['generated_text']
# Remove trailing white spaces
response_text = response_text.strip()
# print model response
print("--------------------------------- Generated response -----------------------------------")
print(response_text)
print("*********************************************************************************************")
return response_text
# Invoke the main function
if __name__ == "__main__":
main()
|