import os import gradio as gr from operator import itemgetter from pinecone import Pinecone from huggingface_hub import whoami from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough, RunnableLambda from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_openai import AzureChatOpenAI from langchain.prompts.prompt import PromptTemplate from langchain.memory import ConversationBufferMemory from langchain_community.vectorstores import Pinecone as PineconeVectorstore from celsius_csrd_chatbot.utils import ( make_html_source, make_pairs, _format_chat_history, _combine_documents, get_llm, init_env, ) init_env() chat_model_init = get_llm() demo_name = "ESRS_QA" hf_model = "BAAI/bge-base-en-v1.5" embeddings = HuggingFaceBgeEmbeddings( model_name=hf_model, encode_kwargs={"normalize_embeddings": True}, ) pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY")) index = pc.Index(os.getenv("PINECONE_API_INDEX")) vectorstore = PineconeVectorstore(index, embeddings, "page_content") retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) chat_model = AzureChatOpenAI() esrs_wiki = """ The Corporate Sustainability Reporting Directive (CSRD) is a mandate that requires all companies to report on their sustainability initiatives. In response to this directive, the European Sustainability Reporting Standards (ESRS) were developed. These standards are a key tool in promoting the transition to a sustainable economy within the EU, providing a structured framework for companies to disclose their sustainability initiatives. The ESRS cover a wide range of environmental, social, and governance (ESG) issues, including climate change, biodiversity, and human rights. Companies that adhere to the ESRS can provide investors with valuable insights into their sustainability impact, thereby informing investment decisions. The ESRS are designed to be highly interoperable with global reporting standards, which helps to avoid unnecessary duplication in reporting by companies. The reporting requirements based on the ESRS will be gradually implemented for different companies over time. In summary, the ESRS play a critical role in fostering sustainable finance and enabling companies to demonstrate their commitment to the green deal agenda while accessing sustainable finance. --- """ reformulation_template = """Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question, in its original language. Chat History: {chat_history} Follow Up Input: {question} Standalone question:""" CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(reformulation_template) answering_template = """ You are an ESG expert, with 20 years experience analyzing corporate sustainability reports. You are specialist in the upcoming CSRD regulation and in general with corporate sustainability disclosure requirements. {esrs_wiki} You will answer the question based on the following passages extracted from CSRD specific sustainability guidelines and reports: ``` {context} ``` Guidelines: 1. Context: You'll receive relevant excerpts from a CSRD-specific sustainability guideline or report to address a given question. 2. Relevance: Only include passages directly pertaining to the question; omit irrelevant content. 3. Facts and Figures: Prioritize factual information in your response. 4. Conciseness: Keep answers sharp and succinct, avoiding unnecessary context. 5. Focus: Address the specific question without veering into related topics. 6. Honesty: If unsure, state that you don't know rather than inventing an answer. 7. Source Attribution: When using information from a passage, mention it as [Doc i] at the end of the sentence (where 'i' represents the document number). 8. Multiple Sources: If the same content appears in multiple documents, cite them collectively (e.g., [Doc i, Doc j, Doc k]). 9. Structured Paragraphs: Instead of bullet-point summaries, compile your responses into well-structured paragraphs. 10. Method Focus: When addressing "how" questions, emphasize methods and procedures over outcomes. 11. Selective Usage: You're not obligated to use every passage; include only those relevant to the question. 12. Insufficient Information: If documents lack necessary details, indicate that you don't have enough information. Question: {question} Answer: """ ANSWER_PROMPT = ChatPromptTemplate.from_template(answering_template) DEFAULT_DOCUMENT_PROMPT = PromptTemplate.from_template(template="{page_content}") memory = ConversationBufferMemory( return_messages=True, output_key="answer", input_key="question" ) # First we add a step to load memory # This adds a "memory" key to the input object loaded_memory = RunnablePassthrough.assign( chat_history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"), ) # Now we calculate the standalone question standalone_question = { "standalone_question": { "question": lambda x: x["question"], "chat_history": lambda x: _format_chat_history(x["chat_history"]), } | CONDENSE_QUESTION_PROMPT | chat_model | StrOutputParser(), } # Now we retrieve the documents retrieved_documents = { "docs": itemgetter("standalone_question") | retriever, "question": lambda x: x["standalone_question"], } # Now we construct the inputs for the final prompt final_inputs = { "context": lambda x: _combine_documents(x["docs"], DEFAULT_DOCUMENT_PROMPT), "question": itemgetter("question"), "esrs_wiki": lambda x: esrs_wiki, } # And finally, we do the part that returns the answers answer = { "answer": final_inputs | ANSWER_PROMPT | chat_model, "docs": itemgetter("docs"), } # And now we put it all together! final_chain = loaded_memory | standalone_question | retrieved_documents | answer async def chat( query: str, history: list = [], ): """taking a query and a message history, use a pipeline (reformulation, retriever, answering) to yield a tuple of: (messages in gradio format, messages in langchain format, source documents)""" source_string = "" gradio_format = make_pairs([a.content for a in history]) + [(query, "")] # reset memory memory.clear() for message in history: memory.chat_memory.add_message(message) inputs = {"question": query} result = final_chain.astream_log({"question": query}) reformulated_question_path_id = "/logs/AzureChatOpenAI/streamed_output_str/-" retriever_path_id = "/logs/Retriever/final_output" final_answer_path_id = "/logs/AzureChatOpenAI:2/streamed_output_str/-" async for op in result: op = op.ops[0] if op["path"] == reformulated_question_path_id: # reforulated question new_token = op["value"] # str elif op["path"] == retriever_path_id: # documents sources = op["value"]["documents"] # List[Document] source_string = "\n\n".join( [(make_html_source(i, doc)) for i, doc in enumerate(sources, 1)] ) elif op["path"] == final_answer_path_id: # final answer new_token = op["value"] # str answer_yet = gradio_format[-1][1] gradio_format[-1] = (query, answer_yet + new_token) yield "", gradio_format, history, source_string memory.save_context(inputs, {"answer": gradio_format[-1][1]}) yield "", gradio_format, memory.load_memory_variables({})["history"], source_string with open("./assets/style.css", "r") as f: css = f.read() def update_visible(oauth_token: gr.OAuthToken | None): if oauth_token is None: return { bloc_1: gr.update(visible=True), bloc_2: gr.update(visible=False), bloc_3: gr.update(visible=False), } org_names = [org["name"] for org in whoami(oauth_token.token)["orgs"]] if "ekimetrics-esrsqa" in org_names: # logged in group return { bloc_1: gr.update(visible=False), bloc_2: gr.update(visible=True), bloc_3: gr.update(visible=False), } else: # logged but not in group return { bloc_1: gr.update(visible=False), bloc_2: gr.update(visible=False), bloc_3: gr.update(visible=True), } # Set up Gradio Theme theme = gr.themes.Base( primary_hue="blue", secondary_hue="red", font=[gr.themes.GoogleFont("Poppins"), "ui-sans-serif", "system-ui", "sans-serif"], ) init_prompt = """ Hello, I am ESRS Q&A, a conversational assistant designed to help you understand the content of European Sustainability Reporting Standards (ESRS). I will answer your questions based **on the official definition of each ESRS as well as complementary guidelines**. ⚠️ Limitations *Please note that this chatbot is in an early stage phase, it is not perfect and may sometimes give irrelevant answers. If you are not satisfied with the answer, please ask a more specific question or report your feedback to help us improve the system.* What do you want to learn ? """ with gr.Blocks(title=f"{demo_name}", css=css, theme=theme) as demo: # with gr.Row(): # with gr.Column(scale=1): # login = gr.LoginButton() # with gr.Column() as bloc_1: # textbox_1 = gr.Textbox("You are not logged to Hugging Face !", show_label=False) # with gr.Column(visible=False) as bloc_3: # textbox_3 = gr.Textbox( # "You are not part of the ESRS Q&A Project, if interested ask access here : https://huggingface.co/ekimetrics-esrsqa" # ) with gr.Column(visible=True) as bloc_2: with gr.Tab("ESRS Q&A"): with gr.Row(): with gr.Column(scale=2): chatbot = gr.Chatbot( value=[(None, init_prompt)], show_copy_button=True, show_label=False, elem_id="chatbot", layout="panel", avatar_images=( None, "https://i.ibb.co/cN0czLp/celsius-logo.png", ), ) state = gr.State([]) with gr.Row(elem_id="input-message"): ask = gr.Textbox( placeholder="Ask me anything here!", show_label=False, scale=7, lines=1, interactive=True, elem_id="input-textbox", ) with gr.Column(scale=1, variant="panel", elem_id="right-panel"): with gr.Tab("Sources", elem_id="tab-citations", id=1): sources_textbox = gr.HTML( show_label=False, elem_id="sources-textbox" ) docs_textbox = gr.State("") with gr.Tab("About", elem_classes="max-height other-tabs"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("WIP") # demo.load(update_visible, inputs=None, outputs=[bloc_1, bloc_2, bloc_3]) # login.click(update_visible, inputs=[], outputs=[bloc_1, bloc_2, bloc_3]) ask.submit( fn=chat, inputs=[ ask, state, ], outputs=[ask, chatbot, state, sources_textbox], ) demo.launch( share=True, debug=True, )