Spaces:
Running
Running
| from langchain_openai import OpenAIEmbeddings | |
| from langchain_openai import ChatOpenAI | |
| from langchain_chroma import Chroma | |
| import chromadb | |
| from chromadb.config import Settings | |
| from langchain_core.prompts import ChatPromptTemplate | |
| from typing import List | |
| from langchain_core.documents import Document | |
| from langchain_core.runnables import RunnablePassthrough | |
| from langchain_core.output_parsers import XMLOutputParser | |
| import gradio as gr | |
| import pandas as pd | |
| import logging | |
| from langchain_core.exceptions import OutputParserException | |
| from langchain_core.prompts import MessagesPlaceholder | |
| import os | |
| import uuid | |
| import hashlib | |
| from langchain_community.chat_message_histories import ChatMessageHistory | |
| from langchain_core.chat_history import BaseChatMessageHistory | |
| from langchain_core.messages import ( | |
| AIMessage, | |
| HumanMessage, | |
| SystemMessage, | |
| ToolMessage, | |
| trim_messages, | |
| ) | |
| # Constants | |
| PERSIST_DIRECTORY = "chroma_store" | |
| K_VALUE = 5 | |
| store = {} | |
| xml_system = """You are a helpful AI assistant for the FORTIFIED program, a voluntary initiative for construction | |
| and re-roofing designed to strengthen homes and commercial buildings against severe weather threats, including | |
| high winds, hail, hurricanes, and tornadoes. Your users may include homeowners, insurance agents, realtors, | |
| design professionals and construction professionals. When given a user question, you should consult the provided | |
| technical and general documents containing FORTIFIED standards, instructions, and program information to deliver | |
| accurate and relevant responses. If the information necessary to answer the inquiry is not available within these | |
| documents, simply respond: ‘There is no information relevant to your inquiry in our current resources. | |
| Please contact FORTIFIED customer support for further assistance.’ | |
| Remember, you must return both an answer and citations. A citation consists of a VERBATIM quote that | |
| justifies the answer and the ID and also Source Name of the quote article. Return a citation for every quote across all articles | |
| that justify the answer. Use the following format for your final output: | |
| <cited_answer> | |
| <answer></answer> | |
| <citations> | |
| <citation><source_id></source_id><source></source><quote></quote></citation> | |
| <citation><source_id></source_id><source></source><quote></quote></citation> | |
| ... | |
| </citations> | |
| </cited_answer> | |
| Here are the articles:{context}""" | |
| xml_prompt = ChatPromptTemplate.from_messages([("system", xml_system), ("human", "{input}")]) | |
| llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) | |
| contextualize_q_system_prompt = ( | |
| "Given a chat history and the latest user question " | |
| "which might reference context in the chat history, " | |
| "formulate a standalone question which can be understood " | |
| "without the chat history. Do NOT answer the question, " | |
| "just reformulate it if needed and otherwise return it as is." | |
| ) | |
| contextualize_q_prompt = ChatPromptTemplate.from_messages( | |
| [ | |
| ("system", contextualize_q_system_prompt), | |
| MessagesPlaceholder("chat_history"), | |
| ("human", "{input}"), | |
| ] | |
| ) | |
| runnable = contextualize_q_prompt | llm | |
| def format_docs_xml(docs: List[Document]) -> str: | |
| formatted_docs = [ | |
| f"<source id=\"{i}\">\n<source>{doc.metadata['source']}</source>\n<article_snippet>{doc.page_content}</article_snippet>\n</source>" | |
| for i, doc in enumerate(docs) | |
| ] | |
| return f"\n\n<sources>\n{chr(10).join(formatted_docs)}\n</sources>" | |
| rag_chain_from_docs = ( | |
| RunnablePassthrough.assign(context=(lambda x: format_docs_xml(x["context"]))) | |
| | xml_prompt | |
| | llm | |
| | XMLOutputParser() | |
| ) | |
| settings = Settings(persist_directory=PERSIST_DIRECTORY) | |
| vectordb = Chroma(embedding_function=OpenAIEmbeddings(), persist_directory=PERSIST_DIRECTORY) | |
| retriever = vectordb.as_retriever(search_type="mmr", search_kwargs={"k": K_VALUE}) | |
| retrieve_docs = (lambda x: x["input"]) | retriever | |
| chain = RunnablePassthrough.assign(context=retrieve_docs).assign( | |
| answer=rag_chain_from_docs | |
| ) | |
| def get_article_info(df, file_name): | |
| if ".pdf" in file_name: | |
| title = df[df["FileName"] == file_name]["Title"].iloc[0] | |
| link = df[df["FileName"] == file_name]["Link"].iloc[0] | |
| else: | |
| title = df[df["Link"] == file_name]["Title"].iloc[0] | |
| link = file_name | |
| return title, link | |
| df = pd.read_csv("articles_db.csv") | |
| def vectordb_search(query): | |
| titles, links = [], [] | |
| question_search = retriever.invoke(query) | |
| for item in question_search: | |
| edited_item = item.metadata["source"].replace("Articles/", "") | |
| title, link = get_article_info(df, edited_item) | |
| if title not in titles: | |
| titles.append(title) | |
| if link not in links: | |
| links.append(link) | |
| return "\n".join([f"- [{title}]({link})" for title, link in zip(titles, links)]) | |
| def get_session_history(session_id: str) -> BaseChatMessageHistory: | |
| if session_id not in store: | |
| store[session_id] = ChatMessageHistory() | |
| return store[session_id] | |
| def generate_unique_string(): | |
| # Get the MAC address | |
| mac = uuid.getnode() | |
| # Convert the MAC address to a hexadecimal string | |
| mac_str = f'{mac:012x}' | |
| # Get the CPU info as a unique identifier for the machine | |
| try: | |
| with open('/proc/cpuinfo') as f: | |
| cpu_info = f.read() | |
| except FileNotFoundError: | |
| cpu_info = str(uuid.getnode()) # Fallback to MAC if CPU info is not accessible | |
| # Combine the MAC and CPU info, and hash it for uniqueness | |
| unique_string = hashlib.sha256((mac_str + cpu_info).encode()).hexdigest() | |
| return unique_string | |
| def llm_response(query, session_id): | |
| titles, links, res_titles, res_links = [], [], [], [] | |
| print(session_id) | |
| unique_id = session_id | |
| config = {"configurable": {"thread_id": unique_id }} | |
| try: | |
| # modified_query = runnable.invoke({"input": query, "chat_history": history.messages}).content | |
| filtered_history = trim_messages(get_session_history(unique_id).messages, strategy="last", token_counter=len, max_tokens=5, | |
| start_on="human", end_on=("human", "tool"), include_system=True,) | |
| modified_query = runnable.invoke({"input": query, "chat_history": filtered_history}).content | |
| result = chain.invoke({"input": modified_query}, config=config) | |
| answer = result['answer']['cited_answer'][0]["answer"] | |
| history = get_session_history(unique_id) | |
| history.add_user_message(modified_query) | |
| history.add_ai_message(answer) | |
| citations = result['answer']['cited_answer'][1]['citations'] | |
| for citation in citations: | |
| edited_item = citation['citation'][1]["source"].replace("Articles/", "") | |
| title, link = get_article_info(df, edited_item) | |
| if title not in titles: | |
| titles.append(title) | |
| if link not in links: | |
| links.append(link) | |
| question_search = retriever.invoke(query) | |
| for res_item in question_search: | |
| edited_item = res_item.metadata["source"].replace("Articles/", "") | |
| res_title, res_link = get_article_info(df, edited_item) | |
| if res_title not in res_titles: | |
| res_titles.append(res_title) | |
| if res_link not in res_links: | |
| res_links.append(res_link) | |
| # Build the answer with superscript citations | |
| answer_with_citations = f"{answer}" | |
| for i, (title, link) in enumerate(zip(titles, links), start=1): | |
| answer_with_citations += f" <sup>[[{i}]({link})]</sup> " # Append superscript citation numbers to the answer text | |
| # Build the references section with clickable links | |
| citations_section = "\n\nCitations:\n" + "\n".join( | |
| [f"[{i}]: [{title}]({link})" for i, (title, link) in enumerate(zip(titles, links), start=1)] | |
| ) | |
| # Combine answer and citations for final markdown output | |
| markdown_list = f"{answer_with_citations}{citations_section}" | |
| markdown_list += f"\n\n\nHere is a list of resources that can provide more information about your inquiry:\n" | |
| markdown_list += "\n".join( | |
| [f"- [{res_title}]({res_link})" for res_title, res_link in zip(res_titles, res_links)]) | |
| except OutputParserException: | |
| markdown_list = "There is no information relevant to your inquiry in my current resources. Please contact [FORTIFIED customer support](https://fortifiedhome.org/contact/) for further assistance." | |
| return markdown_list | |
| with gr.Blocks() as demo: | |
| gr.Markdown("## FORTIFIED AI Assistant!") | |
| with gr.Row(visible=True) as role_selector: | |
| with gr.Column(scale=4): | |
| pass | |
| with gr.Column(scale=2): | |
| gr.Markdown("### Choose your role:") | |
| homeowner = gr.Button("Homeowner", variant="primary") | |
| design_pro = gr.Button("Design Professionals", variant="primary") | |
| evaluator = gr.Button("Evaluator", variant="primary") | |
| with gr.Column(scale=4): | |
| pass | |
| with gr.Row(visible=False) as chat_container: | |
| with gr.Column(): | |
| session_id = gr.Textbox(label="Session ID", visible=False) | |
| gr.Markdown( | |
| "### I'll try to answer any questions related to FORTIFIED program. Tell me what's on your mind?") | |
| chatbot = gr.Chatbot(type="messages", height=400) | |
| msg = gr.Textbox(label="Hit the Enter to send your question", placeholder="What's on your mind?", show_copy_button=True) | |
| send = gr.Button("Send") | |
| def user(user_message, history: list): | |
| return "", history + [{"role": "user", "content": user_message}] | |
| def bot(history: list, session_id_i): | |
| bot_message = llm_response(history[-1]['content'], session_id_i) | |
| history.append({"role": "assistant", "content": ""}) | |
| for character in bot_message: | |
| history[-1]['content'] += character | |
| yield history | |
| msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then( | |
| bot, [chatbot, session_id], chatbot | |
| ) | |
| send.click(user, [msg, chatbot], [msg, chatbot], queue=False).then( | |
| bot, [chatbot, session_id], chatbot | |
| ) | |
| def start(): | |
| unique_id = uuid.uuid4() | |
| return { | |
| chat_container: gr.update(visible=True), | |
| role_selector: gr.update(visible=False), | |
| session_id: gr.update(value=unique_id), | |
| } | |
| homeowner.click(start, [], [chat_container, role_selector, session_id]) | |
| design_pro.click(start, [], [chat_container, role_selector,session_id]) | |
| evaluator.click(start, [], [chat_container, role_selector, session_id]) | |
| if __name__ == "__main__": | |
| demo.launch() |