Spaces:
Sleeping
Sleeping
File size: 5,252 Bytes
dc4de02 918fcc8 dc4de02 |
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 |
from ragas.metrics.base import MetricWithLLM, SingleTurnMetric
from ragas.prompt.pydantic_prompt import PydanticPrompt
from pydantic import BaseModel, Field
import pandas as pd
from typing import List, Tuple
from datetime import datetime
import sys
from dataclasses import dataclass, field
from ragas.metrics.base import MetricType
from ragas.messages import AIMessage, HumanMessage, ToolMessage, ToolCall
from ragas import SingleTurnSample, MultiTurnSample
import typing as t
import asyncio
import dotenv
import os
# Load environment variables from .env file
dotenv.load_dotenv()
# Access the OpenAI API key
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
class ObjectionInput(BaseModel):
user_input: str = Field(description="The objection text")
response: str = Field(default="", description="The response to the objection")
reference: str = Field(default="", description="Any reference related to the objection")
class ObjectionOutput(BaseModel):
satisfy: bool = Field(description="Boolean indicating if the objection was satisfied")
def process_salesbud_file(file_path: str) -> List[Tuple[ObjectionInput, ObjectionOutput]]:
"""
Process the salesbud CSV file and return a list of examples for ObjectionlPrompt.
Args:
file_path (str): The path to the salesbud CSV file.
Returns:
List[Tuple[ObjectionInput, ObjectionOutput]]: A list of tuples containing ObjectionInput and ObjectionOutput.
"""
# Print the timestamp and the file being processed
print(f"{datetime.now()}: Processing file: salesbud_examples.csv")
# Read the CSV file into a DataFrame
df = pd.read_csv('data/salesbud_examples.csv')
# List to hold the processed objections
examples = [] # List to hold examples
# Process each row in the DataFrame
for index, row in df.iterrows():
# Create an ObjectionInput instance for each row
objection_input = ObjectionInput(
user_input=row['objection'], # Assuming your CSV has a column named 'objection'
response=row.get('response', ""), # Use .get() to avoid KeyError if the column doesn't exist
reference=row.get('reference', "") # Use .get() to avoid KeyError if the column doesn't exist
)
# Create an ObjectionOutput instance (you can modify the logic for 'satisfy' as needed)
objection_output = ObjectionOutput(
satisfy= row['satisfy']
)
# Append the example tuple to the examples list
examples.append((objection_input, objection_output))
#print (examples[0])
return examples
class ObjectionlPrompt(PydanticPrompt[ObjectionInput, ObjectionOutput]):
instruction = "You are an expert technology sales rep that is tasked with judging if response satisfies potential customer's objection (user input). \
Given an user input and sales rep response, output True if the response satisfies the objection by the potential customer"
input_model = ObjectionInput
output_model = ObjectionOutput
examples = process_salesbud_file('salesbud_examples.csv')
@dataclass
class SatisfyRate(MetricWithLLM, SingleTurnMetric):
name: str = "satisfy_rate"
_required_columns: t.Dict[MetricType, t.Set[str]] = field(
default_factory=lambda: {MetricType.SINGLE_TURN: {"response", "reference"}}
)
objection_prompt: PydanticPrompt = ObjectionlPrompt()
async def _ascore(self, row):
pass
async def _single_turn_ascore(self, sample, callbacks):
prompt_input = ObjectionInput(
user_input=sample.user_input, response=sample.response
)
prompt_response = await self.objection_prompt.generate(
data=prompt_input, llm=self.llm
)
return int(prompt_response.satisfy)
async def generate_response_to_objection(file_path, num):
from langchain_openai import ChatOpenAI
from ragas.llms.base import LangchainLLMWrapper
import pandas as pd
user_response= pd.read_csv(file_path)
openai_model = LangchainLLMWrapper(ChatOpenAI(model_name="gpt-4o", api_key=OPENAI_API_KEY))
scorer = SatisfyRate(llm=openai_model)
sample = SingleTurnSample(user_input=user_response['objection'][num], response=user_response['response'][num])
#(user_response['objection'][num], user_response['response'][num])
satisfy_0_1 = await scorer.single_turn_ascore(sample)
print (user_response['objection'][num], user_response['response'][num], satisfy_0_1)
# Implement your logic to generate a response based on the user's input
return satisfy_0_1 #f"Response to your objection: {user_response['objection'][num]}, {user_response['response'][num]}, {satisfy_0_1}"
async def main(file_path):
# Call the async function
#examples_file = process_salesbud_file()
response = await generate_response_to_objection(file_path, 0)
if __name__ == "__main__":
# Check if the file path is provided as a command-line argument
if len(sys.argv) != 2:
print("Usage: python objection_eval.py <path_to_salesbud.csv>")
sys.exit(1)
# Get the file path from the command-line argument
file_path = sys.argv[1]
# Run the main async function
asyncio.run(main(file_path))
|