Optimize question responses with person suggestions utilizing Amazon Bedrock embedding and few-shot prompting

# Load the dataset and specify the subset
dataset = load_dataset("llm-blender/Unified-Suggestions", "synthetic-instruct-gptj-pairwise")

# Entry the 'practice' cut up
train_dataset = dataset["train"]

# Convert the dataset to Pandas DataFrame
df = train_dataset.to_pandas()

# Flatten the nested dialog buildings for conv_A and conv_B safely
df['conv_A_user'] = df['conv_A'].apply(lambda x: x[0]['content'] if len(x) > 0 else None)
df['conv_A_assistant'] = df['conv_A'].apply(lambda x: x[1]['content'] if len(x) > 1 else None)

# Drop the unique nested columns if they're not wanted
df = df.drop(columns=['conv_A', 'conv_B'])

import random import bedrock # Take a pattern of 6000 queries 
df = df.shuffle(seed=42).choose(vary(6000)) 
# AWS credentials
session = boto3.Session()
area = 'us-east-1'
# Initialize the S3 shopper
s3_client = boto3.shopper('s3')

boto3_bedrock = boto3.shopper('bedrock-runtime', area)
titan_embed_v2 = BedrockEmbeddings(
    shopper=boto3_bedrock, model_id="amazon.titan-embed-text-v2:0")
    
# Operate to transform textual content to embeddings
def get_embeddings(textual content):
    response = titan_embed_v2.embed_query(textual content)
    return response  # This could return the embedding vector

# Apply the perform to the 'immediate' column and retailer in a brand new column
df_test['conv_A_user_vec'] = df_test['conv_A_user'].apply(get_embeddings)

# Step 2: Outline cosine similarity perform
def compute_cosine_similarity(embedding1, embedding2):
embedding1 = np.array(embedding1).reshape(1, -1) # Reshape to 2D array
embedding2 = np.array(embedding2).reshape(1, -1) # Reshape to 2D array
return cosine_similarity(embedding1, embedding2)[0][0]

# Pattern question embedding
def get_matched_convo(question, df):
    query_embedding = get_embeddings(question)
    
    # Step 3: Compute similarity with every row within the DataFrame
    df['similarity'] = df['conv_A_user_vec'].apply(lambda x: compute_cosine_similarity(query_embedding, x))
    
    # Step 4: Kind rows based mostly on similarity rating (descending order)
    df_sorted = df.sort_values(by='similarity', ascending=False)
    
    # Step 5: Filter or get high matching rows (e.g., high 10 matches)
    top_matches = df_sorted.head(10) 
    
    # Print high matches
    return top_matches[['conv_A_user', 'conv_A_assistant','conv_A_rating','similarity']]

import boto3
from langchain_aws import ChatBedrock
from pydantic import BaseModel

# Initialize Amazon Bedrock shopper
bedrock_runtime = boto3.shopper(service_name="bedrock-runtime", region_name="us-east-1")

# Configure the mannequin to make use of
model_id = "us.anthropic.claude-3-5-haiku-20241022-v1:0"
model_kwargs = {
"max_tokens": 2048,
"temperature": 0.1,
"top_k": 250,
"top_p": 1,
"stop_sequences": ["nnHuman"],
}

# Create the LangChain Chat object for Bedrock
llm = ChatBedrock(
shopper=bedrock_runtime,
model_id=model_id,
model_kwargs=model_kwargs,
)

# Pydantic mannequin to validate the output immediate
class OptimizedPromptOutput(BaseModel):
optimized_prompt: str

# Operate to generate the few-shot immediate
def generate_few_shot_prompt_only(user_query, nearest_examples):
    # Make sure that df_examples is a DataFrame
    if not isinstance(nearest_examples, pd.DataFrame):
    elevate ValueError("Anticipated df_examples to be a DataFrame")
    # Assemble the few-shot immediate utilizing nearest matching examples
    few_shot_prompt = "Listed below are examples of person queries, LLM responses, and suggestions:nn"
    for i in vary(len(nearest_examples)):
    few_shot_prompt += f"Consumer Question: {nearest_examples.loc[i,'conv_A_user']}n"
    few_shot_prompt += f"LLM Response: {nearest_examples.loc[i,'conv_A_assistant']}n"
    few_shot_prompt += f"Consumer Suggestions: {'👍' if nearest_examples.loc[i,'conv_A_rating'] == 1.0 else '👎'}nn"
    
    # Add the person question for which the optimized immediate is required
    few_shot_prompt += f"Based mostly on these examples, generate a common optimized immediate for the next person question:nn"
    few_shot_prompt += f"Consumer Question: {user_query}n"
    few_shot_prompt += "Optimized Immediate: Present a transparent, well-researched response based mostly on correct information and credible sources. Keep away from pointless data or hypothesis."
    
    return few_shot_prompt

# Operate to generate an optimized immediate utilizing Bedrock and return solely the immediate utilizing Pydantic
def get_optimized_prompt(user_query, nearest_examples):
    # Generate the few-shot immediate
    few_shot_prompt = generate_few_shot_prompt_only(user_query, nearest_examples)
    
    # Name the LLM to generate the optimized immediate
    response = llm.invoke(few_shot_prompt)
    
    # Extract and validate solely the optimized immediate utilizing Pydantic
    optimized_prompt = response.content material # Mounted to entry the 'content material' attribute of the AIMessage object
    optimized_prompt_output = OptimizedPromptOutput(optimized_prompt=optimized_prompt)
    
    return optimized_prompt_output.optimized_prompt

# Instance utilization
question = "Is the US greenback weakening over time?"
nearest_examples = get_matched_convo(question, df_test)
nearest_examples.reset_index(drop=True, inplace=True)

# Generate optimized immediate
optimized_prompt = get_optimized_prompt(question, nearest_examples)
print("Optimized Immediate:", optimized_prompt)

# Operate to make the LLM name utilizing the optimized immediate and person question
def make_llm_call_with_optimized_prompt(optimized_prompt, user_query):
    start_time = time.time()
    # Mix the optimized immediate and person question to type the enter for the LLM
    final_prompt = f"{optimized_prompt}nnUser Question: {user_query}nResponse:"

    # Make the decision to the LLM utilizing the mixed immediate
    response = llm.invoke(final_prompt)
    
    # Extract solely the content material from the LLM response
    final_response = response.content material  # Extract the response content material with out including any labels
    time_taken = time.time() - start_time
    return final_response,time_taken

# Instance utilization
user_query = "The right way to develop avocado indoor?"
# Assume 'optimized_prompt' has already been generated from the earlier step
final_response,time_taken = make_llm_call_with_optimized_prompt(optimized_prompt, user_query)
print("LLM Response:", final_response)

def get_unoptimized_prompt_response(df_eval):
    # Iterate over the dataframe and make LLM calls
    for index, row in tqdm(df_eval.iterrows()):
        # Get the person question from 'conv_A_user'
        user_query = row['conv_A_user']
        
        # Make the Bedrock LLM name
        response = llm.invoke(user_query)
        
        # Retailer the response content material in a brand new column 'unoptimized_prompt_response'
        df_eval.at[index, 'unoptimized_prompt_response'] = response.content material  # Extract 'content material' from the response object
    
    return df_eval

def get_optimized_prompt_response(df_eval):
    # Iterate over the dataframe and make LLM calls
    for index, row in tqdm(df_eval.iterrows()):
        # Get the person question from 'conv_A_user'
        user_query = row['conv_A_user']
        nearest_examples = get_matched_convo(user_query, df_test)
        nearest_examples.reset_index(drop=True, inplace=True)
        optimized_prompt = get_optimized_prompt(user_query, nearest_examples)
        # Make the Bedrock LLM name
        final_response,time_taken = make_llm_call_with_optimized_prompt(optimized_prompt, user_query)
        
        # Retailer the response content material in a brand new column 'unoptimized_prompt_response'
        df_eval.at[index, 'optimized_prompt_response'] = final_response  # Extract 'content material' from the response object
    
    return df_eval

# Outline Pydantic mannequin to implement predicted suggestions as 0 or 1
class FeedbackPrediction(BaseModel):
    predicted_feedback: conint(ge=0, le=1)  # Solely enable values 0 or 1

# Operate to generate few-shot immediate
def generate_few_shot_prompt(df_examples, unoptimized_response):
    few_shot_prompt = (
        "You're an neutral decide evaluating the standard of LLM responses. "
        "Based mostly on the person queries and the LLM responses offered under, your activity is to find out whether or not the response is sweet or unhealthy, "
        "utilizing the examples offered. Return 1 if the response is sweet (thumbs up) or 0 if the response is unhealthy (thumbs down).nn"
    )
    few_shot_prompt += "Beneath are examples of person queries, LLM responses, and person suggestions:nn"
    
    # Iterate over few-shot examples
    for i, row in df_examples.iterrows():
        few_shot_prompt += f"Consumer Question: {row['conv_A_user']}n"
        few_shot_prompt += f"LLM Response: {row['conv_A_assistant']}n"
        few_shot_prompt += f"Consumer Suggestions: {'👍' if row['conv_A_rating'] == 1 else '👎'}nn"
    
    # Present the unoptimized response for suggestions prediction
    few_shot_prompt += (
        "Now, consider the next LLM response based mostly on the examples above. Return 0 for unhealthy response or 1 for good response.nn"
        f"Consumer Question: {unoptimized_response}n"
        f"Predicted Suggestions (0 for 👎, 1 for 👍):"
    )
    return few_shot_prompt

# Operate to foretell suggestions utilizing few-shot examples
def predict_feedback(df_examples, df_to_rate, response_column, target_col):
    # Create a brand new column to retailer predicted suggestions
    df_to_rate[target_col] = None
    
    # Iterate over every row within the dataframe to charge
    for index, row in tqdm(df_to_rate.iterrows(), whole=len(df_to_rate)):
        # Get the unoptimized immediate response
        strive:
            time.sleep(2)
            unoptimized_response = row[response_column]

            # Generate few-shot immediate
            few_shot_prompt = generate_few_shot_prompt(df_examples, unoptimized_response)

            # Name the LLM to foretell the suggestions
            response = llm.invoke(few_shot_prompt)

            # Extract the expected suggestions (assuming the mannequin returns '0' or '1' as suggestions)
            predicted_feedback_str = response.content material.strip()  # Clear and extract the expected suggestions

            # Validate the suggestions utilizing Pydantic
            strive:
                feedback_prediction = FeedbackPrediction(predicted_feedback=int(predicted_feedback_str))
                # Retailer the expected suggestions within the dataframe
                df_to_rate.at[index, target_col] = feedback_prediction.predicted_feedback
            besides (ValueError, ValidationError):
                # In case of invalid information, assign default worth (e.g., 0)
                df_to_rate.at[index, target_col] = 0
        besides:
            cross

    return df_to_rate

df_eval = df.drop(df_test.index).pattern(100)
df_eval['unoptimized_prompt_response'] = "" # Create an empty column to retailer responses
df_eval = get_unoptimized_prompt_response(df_eval)
df_eval['optimized_prompt_response'] = "" # Create an empty column to retailer responses
df_eval = get_optimized_prompt_response(df_eval)
Name the perform to foretell suggestions
df_with_predictions = predict_feedback(df_eval, df_eval, 'unoptimized_prompt_response', 'predicted_unoptimized_feedback')
df_with_predictions = predict_feedback(df_with_predictions, df_with_predictions, 'optimized_prompt_response', 'predicted_optimized_feedback')

# Calculate accuracy for unoptimized and optimized responses
original_success = df_with_predictions.conv_A_rating.sum()*100.0/len(df_with_predictions)
unoptimized_success  = df_with_predictions.predicted_unoptimized_feedback.sum()*100.0/len(df_with_predictions) 
optimized_success = df_with_predictions.predicted_optimized_feedback.sum()*100.0/len(df_with_predictions) 

# Show outcomes
print(f"Authentic success: {original_success:.2f}%")
print(f"Unoptimized Immediate success: {unoptimized_success:.2f}%")
print(f"Optimized Immediate success: {optimized_success:.2f}%")

from scipy import stats
# Pattern person satisfaction scores from the pocket book
unopt = [] #20 samples of scores for the unoptimized promt
decide = [] # 20 samples of scores for the optimized promt]
# Paired pattern t-test
t_stat, p_val = stats.ttest_rel(unopt, decide)
print(f"t-statistic: {t_stat}, p-value: {p_val}")