| import os | |
| import gradio as gr | |
| from langchain.document_loaders import PyMuPDFLoader, TextLoader | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| from langchain.vectorstores import Chroma | |
| from langchain.embeddings import HuggingFaceEmbeddings | |
| from langchain.chains import RetrievalQA | |
| from langchain.prompts import PromptTemplate | |
| from langchain_groq import ChatGroq | |
| import chardet | |
| import pandas as pd | |
| import plotly.graph_objs as go | |
| try: | |
| SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| except NameError: | |
| SCRIPT_DIR = os.getcwd() | |
| CSV_PATH = os.path.join(SCRIPT_DIR, "evaluations.csv") | |
| def detect_encoding(file_path): | |
| with open(file_path, 'rb') as file: | |
| raw_data = file.read() | |
| return chardet.detect(raw_data)['encoding'] | |
| def setup(file_path): | |
| _, extension = os.path.splitext(file_path) | |
| if extension.lower() == '.pdf': | |
| loader = PyMuPDFLoader(file_path) | |
| elif extension.lower() == '.txt': | |
| encoding = detect_encoding(file_path) | |
| loader = TextLoader(file_path, encoding=encoding) | |
| else: | |
| raise ValueError("Unsupported file type. Please upload a PDF or TXT file.") | |
| document_data = loader.load() | |
| text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) | |
| all_splits = text_splitter.split_documents(document_data) | |
| persist_directory = 'db' | |
| model_name = "sentence-transformers/all-MiniLM-L6-v2" | |
| model_kwargs = {'device': 'cpu'} | |
| embedding = HuggingFaceEmbeddings(model_name=model_name, model_kwargs=model_kwargs) | |
| vectordb = Chroma.from_documents(documents=all_splits, embedding=embedding, persist_directory=persist_directory) | |
| sys_prompt = """ | |
| You can only answer questions that are relevant to the content of the document. | |
| If the question that user asks is not relevant to the content of the document, you should say "I don't know". | |
| Do not provide any information that is not in the document. | |
| Do not provide any information by yourself. | |
| You should use sentences from the document to answer the questions. | |
| """ | |
| instruction = """CONTEXT:\n\n {context}\n\nQuestion: {question}""" | |
| prompt_template = f"[INST] <<SYS>>\n{sys_prompt}\n<</SYS>>\n\n{instruction}[/INST]" | |
| llama_prompt = PromptTemplate( | |
| template=prompt_template, input_variables=["context", "question"] | |
| ) | |
| chain_type_kwargs = {"prompt": llama_prompt} | |
| retriever = vectordb.as_retriever() | |
| return retriever, chain_type_kwargs | |
| def setup_qa(file_path, model_name, api_key): | |
| retriever, chain_type_kwargs = setup(file_path) | |
| llm = ChatGroq( | |
| groq_api_key=api_key, | |
| model_name=model_name, | |
| ) | |
| qa = RetrievalQA.from_chain_type( | |
| llm=llm, | |
| chain_type="stuff", | |
| retriever=retriever, | |
| chain_type_kwargs=chain_type_kwargs, | |
| verbose=True | |
| ) | |
| return qa | |
| def chat_with_models(file, model_a, model_b, api_key, history_a, history_b, question): | |
| if file is None: | |
| return history_a + [("請上傳文件。", None)], history_b + [("請上傳文件。", None)], "" | |
| if not api_key: | |
| return history_a + [("請輸入 API 金鑰。", None)], history_b + [("請輸入 API 金鑰。", None)], "" | |
| file_path = file.name | |
| _, extension = os.path.splitext(file_path) | |
| if extension.lower() not in ['.pdf', '.txt']: | |
| error_message = "只能上傳PDF或TXT檔案，不接受其他的格式。" | |
| return history_a + [(error_message, None)], history_b + [(error_message, None)], "" | |
| try: | |
| qa_a = setup_qa(file_path, model_a, api_key) | |
| qa_b = setup_qa(file_path, model_b, api_key) | |
| response_a = qa_a.invoke(question) | |
| response_b = qa_b.invoke(question) | |
| history_a.append((question, response_a["result"])) | |
| history_b.append((question, response_b["result"])) | |
| return history_a, history_b, "" | |
| except Exception as e: | |
| error_message = f"遇到錯誤：{str(e)}" | |
| return history_a + [(error_message, None)], history_b + [(error_message, None)], "" | |
| def load_or_create_df(): | |
| if os.path.exists(CSV_PATH): | |
| return pd.read_csv(CSV_PATH) | |
| else: | |
| return pd.DataFrame(columns=['Model A', 'Model B', 'Evaluation', 'Count']) | |
| def record_evaluation(df, model_a, model_b, evaluation): | |
| new_row = pd.DataFrame({ | |
| 'Model A': [model_a], | |
| 'Model B': [model_b], | |
| 'Evaluation': [evaluation], | |
| 'Count': [1] | |
| }) | |
| updated_df = pd.concat([df, new_row], ignore_index=True) | |
| updated_df.to_csv(CSV_PATH, index=False) | |
| return updated_df | |
| def update_statistics(df): | |
| stats = df.groupby(['Model A', 'Model B', 'Evaluation'])['Count'].sum().reset_index() | |
| fig = go.Figure(data=[ | |
| go.Bar(name=row['Evaluation'], | |
| x=[f"{row['Model A']} vs {row['Model B']}"], | |
| y=[row['Count']]) | |
| for _, row in stats.iterrows() | |
| ]) | |
| fig.update_layout(barmode='group', title='Model Comparison Statistics') | |
| return fig | |
| def evaluate(df, model_a, model_b, evaluation): | |
| updated_df = record_evaluation(df, model_a, model_b, evaluation) | |
| fig = update_statistics(updated_df) | |
| return updated_df, updated_df, fig | |
| models = ["llama3-70b-8192", "mixtral-8x7b-32768", "llama3-8b-8192", "gemma-7b-it"] | |
| def create_demo(): | |
| with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| gr.Markdown("# Choose three models to compare(With KnowledgeGraphRAG)") | |
| with gr.Row(): | |
| model_a = gr.Dropdown(choices=models, label="Model A", value=models[0]) | |
| model_b = gr.Dropdown(choices=models, label="Model B", value=models[-1]) | |
| api_key = gr.Textbox(label="Enter your Groq API Key", type="password") | |
| file_input = gr.File(label="Upload PDF or TXT file") | |
| with gr.Row(): | |
| chat_a = gr.Chatbot(label="Model A") | |
| chat_b = gr.Chatbot(label="Model B") | |
| question = gr.Textbox(label="Enter your prompt and press ENTER") | |
| send_btn = gr.Button("Send") | |
| with gr.Row(): | |
| a_better = gr.Button("A is better") | |
| b_better = gr.Button("B is better") | |
| tie = gr.Button("Tie") | |
| both_bad = gr.Button("Both are bad") | |
| evaluation_df = gr.State(load_or_create_df()) | |
| evaluation_table = gr.Dataframe(label="Evaluation Records") | |
| statistics_plot = gr.Plot(label="Evaluation Statistics") | |
| send_btn.click( | |
| fn=chat_with_models, | |
| inputs=[file_input, model_a, model_b, api_key, chat_a, chat_b, question], | |
| outputs=[chat_a, chat_b, question], | |
| ) | |
| def create_evaluate_fn(eval_type): | |
| def evaluate_with_type(df, model_a, model_b): | |
| return evaluate(df, model_a, model_b, eval_type) | |
| return evaluate_with_type | |
| for btn, eval_type in [(a_better, "A is better"), (b_better, "B is better"), (tie, "Tie"), (both_bad, "Both are bad")]: | |
| btn.click( | |
| fn=create_evaluate_fn(eval_type), | |
| inputs=[evaluation_df, model_a, model_b], | |
| outputs=[evaluation_df, evaluation_table, statistics_plot], | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| print(f"CSV file is located at: {os.path.abspath(CSV_PATH)}") | |
| demo = create_demo() | |
| demo.launch(share=True) | |
| else: | |
| demo = create_demo() | |