Spaces:
Build error
Build error
| import os | |
| from dotenv import load_dotenv | |
| from langchain_aws import ChatBedrock | |
| import time | |
| import requests | |
| load_dotenv() | |
| AWS_REGION = os.getenv("AWS_REGION") | |
| AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID") | |
| AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY") | |
| AWS_CHAT_MODEL_ID = os.getenv("AWS_CHAT_MODEL_ID") | |
| # Configure AWS credentials | |
| os.environ["AWS_ACCESS_KEY_ID"] = AWS_ACCESS_KEY_ID | |
| os.environ["AWS_SECRET_ACCESS_KEY"] = AWS_SECRET_ACCESS_KEY | |
| os.environ["AWS_DEFAULT_REGION"] = AWS_REGION | |
| llm_questioner = ChatBedrock( | |
| model_id=AWS_CHAT_MODEL_ID, | |
| temperature=0.5, | |
| ) | |
| llm_answerer = ChatBedrock( | |
| model_id=AWS_CHAT_MODEL_ID, | |
| temperature=0.5, | |
| ) | |
| llm_evaluator = ChatBedrock( | |
| model_id=AWS_CHAT_MODEL_ID, | |
| temperature=0.5, | |
| ) | |
| crypto_token = { | |
| "btc": "bitcoin", | |
| "eth": "ethereum", | |
| "xrp": "ripple", | |
| "virtual": "virtual-protocol", | |
| "bnb": "binancecoin", | |
| "bnb2": "bifrost-bridged-bnb-bifrost", | |
| "shiba-inu": "shiba-inu", | |
| "cardano": "cardano", | |
| } | |
| def get_crypto_data(crypto_name, data_span): | |
| crypto_name = crypto_token[crypto_name] | |
| r = requests.get( | |
| f"https://api.coingecko.com/api/v3/coins/{crypto_name}?localization=false" | |
| ) | |
| data = r.json() | |
| tw_up_votes = data["sentiment_votes_up_percentage"] | |
| tw_down_votes = data["sentiment_votes_down_percentage"] | |
| value_change = data["market_data"][f"price_change_percentage_{data_span}"] | |
| value_change_in_currency = data["market_data"][ | |
| f"price_change_percentage_{data_span}_in_currency" | |
| ]["usd"] | |
| return f""" | |
| Crypto Name: {crypto_name} | |
| Data TimeSpan: {data_span} | |
| Twitter up votes: {tw_up_votes} | |
| Twitter down votes: {tw_down_votes} | |
| Value change: {value_change} | |
| Value change in currency: {value_change_in_currency} | |
| """ | |
| def get_question(previous_qa=""): | |
| """Generate a crypto question using the questioner LLM""" | |
| prompt = f""" | |
| You are a questioner and expert in the field of Cryptocurrency and Investments. | |
| # Instructions | |
| - Your task is to ask question to the user based on Cryptocurrency and Investments. | |
| - You need to ask multiple choice questions so that the user can select the correct answer. | |
| - Make sure the question is clear and the options are distinct. | |
| - Just generate the question and options, do not include any other text. | |
| - Your question will directly be given to the user, so make sure no starting or ending texts are added. | |
| - The reason of asking the question is to evaluate the user's risk profiling. | |
| - We need to build a risk profiling model based on the answers. | |
| - You need to generate a new multiple choice question for risk profiling. | |
| # Types of Questions | |
| - Risk Tolerance | |
| - Investment Goals | |
| - Time Horizon | |
| - Risk Appetite | |
| - Investment Sentiment | |
| - Details about previous investments | |
| - Percentage of investment in Cryptocurrency | |
| # previous questions with answers | |
| {previous_qa} | |
| """ | |
| return llm_questioner.invoke(prompt) | |
| def get_answer(question, previous_qa, crypto_data): | |
| """Generate an answer explanation using the answerer LLM""" | |
| prompt = f""" | |
| You are an expert in the field of Cryptocurrency and Investments. | |
| # Instructions | |
| - Your task is to answer the question based on Cryptocurrency and Investments. | |
| - The question will be the multiple choice question. | |
| - Just select the correct answer from the options. | |
| - Do not include any other text. | |
| - Your answer should be consistent with previous answers and build upon them. | |
| - Consider the user's previous responses to maintain a coherent risk profile. | |
| # Notes | |
| - You need to give answers by looking at the following crypto data | |
| # Crypto Data | |
| {crypto_data} | |
| # Previous Q&A History | |
| {previous_qa} | |
| # Current Question with options | |
| {question} | |
| """ | |
| return llm_answerer.invoke(prompt) | |
| def evaluate_expertise(qa_history): | |
| """Evaluate the user's crypto expertise level based on Q&A history""" | |
| prompt = f""" | |
| You are an expert evaluator in the field of Cryptocurrency and Investments. | |
| # Instructions | |
| - Analyze the following Q&A history to determine the user's risk profiling. | |
| - Consider the complexity of questions asked and the quality of answers provided | |
| - Include 2-3 specific observations about their risk profiling | |
| - Just generate the evaluation, do not include any other text. | |
| - You have to decide the tranche based on user's answer. | |
| - Tranche can be Senior, Mezzanine and Junior. | |
| - You need to give a final verdict based on the user's answer. | |
| # Q&A History | |
| {qa_history} | |
| """ | |
| return llm_evaluator.invoke(prompt) | |
| def main(): | |
| print("Starting Crypto Q&A Session between AI models...\n") | |
| crypto_name = input("Enter the crypto name (Choose from: btc, eth, virtual, bnb): ") | |
| if crypto_name.lower() not in ["btc", "eth", "virtual", "bnb"]: | |
| print("Invalid crypto name. Please try again.") | |
| return | |
| data_span = input("Enter the data span (Choose from: 24h, 7d, 14d, 30d, 60d): ") | |
| if data_span.lower() not in ["24h", "7d", "14d", "30d", "60d"]: | |
| print("Invalid data span. Please try again.") | |
| return | |
| crypto_data = get_crypto_data(crypto_name, data_span) | |
| previous_qa = "" | |
| qa_history = [] | |
| for i in range(5): # Generate 5 questions | |
| print(f"\n=== Question {i+1} ===") | |
| question = get_question(previous_qa) | |
| print("\nQuestioner:") | |
| print(question.content) | |
| # Get answer from the answerer LLM | |
| explanation = get_answer(question, previous_qa, crypto_data) | |
| print("\nAnswerer:") | |
| print(explanation.content) | |
| # Store Q&A in history | |
| qa_history.append( | |
| { | |
| "question": question.content, | |
| "answer": explanation.content, | |
| } | |
| ) | |
| # Update previous Q&A for context | |
| previous_qa += f"\nQ: {question.content}\nA: {explanation.content}\n" | |
| print("\n" + "=" * 50 + "\n") | |
| time.sleep(2) | |
| # After all questions, evaluate expertise | |
| print("\n=== Expertise Evaluation ===") | |
| evaluation = evaluate_expertise(qa_history) | |
| print("\nEvaluator:") | |
| print(evaluation.content) | |
| if __name__ == "__main__": | |
| main() | |