Upload New File

prompting_module.py

+import re
+import requests
+
+
+prompt_tempdic = {
+    'eve_template': (
+    "You are a scientific paper classifier. Classify the following **PAPER TO CLASSIFY** "
+    "as 'participative science', 'non-participative science', or 'ambiguous', based on the the following "
+    " **DEFINITION** of participative sciences and **RELATED PAPERS** provided. Explain your reasoning:\n\n"
+    "**DEFINITION:**\n {field_def}\n\n"
+    "**PAPER TO CLASSIFY**:\nText: {doc_target}\nKeywords: {kwrd_target}\n\n"
+    "**RELATED PAPERS:**:\n{examples}\n\n"
+    "Provide your classification, confidence score, and rationale in JSON format."
+    "Confidence score is between 0 and 1, where 1 indicates very high confidence on your classification and 0 very low.\n\n"
+    "Output should follow this structure: {{'label': your choice, 'confidence': your confidence score,'rationale': your explanation}}."),
+    'lilith_template': (
+        "You are a researcher in Citizen Science/Participative Science.\n"
+        "You are provided with the output of an expert system in the form of a JSON dictionary. This includes:\n"
+        "**EXPERT LABEL**: The label assigned by the expert system,\n"
+        "**CONFIDENCE**: a numerical score indicating the system’s confidence in its classification,\n"
+        "**RATIONALE**: the reasoning provided for this classification,\n"
+        "**PAPER**: the paper that has been classified by the expert system,\n"
+        "**KEYWORDS**: the authors' keywords of the paper classified by the expert system,\n"
+        "**RELATED PAPERS**: the N closest documents to the paper being classified.\n\n"
+        
+        "Your TASK is to CRITICALLY evaluate these elements for errors, inaccuracies, and logical consistency. For this, focus on the following questions :\n"
+        "1.Is the **RATIONALE** internally consistent?\n"
+        "2.Does the **RATIONALE** explicitly use evidence from **RELATED PAPERS** to support its classification?\n"
+        "3.Are the **RELATED PAPERS** relevant and representative of participative science or citizen science?\n"
+        "4.Do the **RELATED PAPERS** provide enough evidence to support or challenge the classification?\n\n"
+
+        "{llmresult}\n{doc}\n\n{examples}\n\n"
+
+        "Provide an alternative classification as either 'participative science', 'non-participative science', or 'ambiguous'.\n"
+        "Include a confidence score between 0 and 1, where 1 indicates very high confidence on your classification and 0 very low.\n\n"
+        "The output should adhere to the following structure and use lowercase keys as shown:\n "
+        "{{'label': your alternative classification, 'confidence': your confidence score, 'rationale': your explanation}}") , 
+    
+    'asherah_template':(
+        "You are an objective mediator. Your task is to compare the outputs of two expert classification of a **PAPER TO CLASSIFY**.\n\n"
+        "You are provided with the following informations for each decision:\n"
+        "**EXPERT LABEL**: The label assigned by the system ('participative science', 'non-participative science', or 'ambiguous').\n"
+        "**CONFIDENCE**: A numerical score indicating the system’s confidence in its classification.\n"
+        "**RATIONALE**: The reasoning provided for the classification.\n\n"
+        "{llmresult1}\n"
+        "{llmresult2}\n\n"
+        "COMPARE these two outputs IMPARTIALLY based on the following criteria:\n"
+        "1. Logical consistency of each **RATIONALE**.\n"
+        "2. Use of evidence in the **RATIONALE** (e.g., relevance of **RELATED PAPERS**).\n"
+        "3. Alignment of **CONFIDENCE** scores with the strength of the **RATIONALE**.\n"
+        "4. Areas of agreement and disagreement between the two classifications.\n\n"
+        "After analyzing both outputs, provide a synthesized conclusion that includes:\n"
+        "\t- Your own classification of the **PAPER TO CLASSIFY** as either 'participative science', 'non-participative science', or 'ambiguous'.\n"
+        "\t- Your confidence score (0-1), where 1 indicates very high confidence in your conclusion.\n"
+        "-\t A rationale explaining your decision, highlighting why your classification better aligns with the evidence and reasoning provided by both systems.\n\n"
+        " {doc}\n\n{examples}\n\n"
+        "Provide an alternative classification as either 'participative science', 'non-participative science', or 'ambiguous'.\n"
+        "The output should adhere to the following structure and use lowercase keys as shown: \n"
+        "{{'label': your alternative classification, 'confidence': your confidence score, 'rationale': your explanation, 'agreements': key points of agreeemnt between the two systems, 'disagreements':key points of disagreement between the two systems}}"),
+    'casper_template':(
+        "You are a Librarian. Your task is to summarize the rationales provided by scientific experts to classify the following **SCIENTIFIC PAPER** and to mean their confidence score.\n\n"
+        "{llmresult1}\n"
+        "{llmresult2}\n\n"
+        "{doc}\n\n"
+        "Provide the **Experts Classification**, the mean of their confidence score, and your summary in JSON format."
+        "The output should adhere to the following structure and use lowercase keys as shown:\n "
+        "{{'label': **Experts Classification** 'confidence': mean of **Expert Confidence Score 1**, "
+            "and **Expert Confidence Score 2** in float format, 'rationale': your summary}}")} 
+
+
+class PromptGenerator:
+    def __init__(self,paper_tires,paper_keywords,context,prompt_templates):
+        self.doc = paper_tires
+        self.kwrd = paper_keywords
+        self.context_docs = context
+        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
+        self.template = prompt_templates
+        self.field_def = "Participative science or Citizen Science, also known as participatory science, involves the active engagement of the general public or amateur researchers in scientific projects. This approach allows non-professionals to contribute to data collection, analysis, and even project design across a large range of scientific fields. The practice is diverse, with different organizations emphasizing distinct forms of involvement, from data-gathering under professional guidance to more autonomous community-led initiatives. The core aim of citizen science is to democratize science by making it responsive to public needs and concerns, empowering communities to investigate issues directly affecting them, such as environmental or health risks. Citizen Science also plays a significant role in education, enriching public understanding of science and enhancing community engagement. It is embraced in both formal and informal educational settings, promoting scientific literacy and awareness. Technology has recently amplified this practice, enabling broader participation through online platforms and apps, facilitating large-scale projects that would otherwise be impractical for individual researchers."
+
+    
+    def inputsize_cheker(self, fullprompt):
+        if len(self.tokenizer(fullprompt)) < 20000:
+            return True
+        else:
+            raise ValueError(f'Input length exceed the model capabilities')
+
+
+    def format_prompt(self, template, **kwargs):
+        """
+        Formats the given template by replacing placeholders with keyword arguments.
+        """
+        return template.format(**kwargs)
+    
+
+    
+    def eve(self):
+        
+        examples = "\n".join(
+            f"\nExample {i+1}:\nTitle+Abstract: {doc['text']}\nKeywords: {doc['keywords']}\n"
+            f"Label: {doc['label']}\nDistance: {doc['distance']}"
+            for i, doc in enumerate(self.context_docs)
+        )
+
+        prompt = self.format_prompt(
+            self.template['eve_template'],
+            field_def = self.field_def,
+            doc_target=self.doc,
+            kwrd_target=self.kwrd,
+            examples=examples
+        )
+        return prompt
+    
+    def lilith(self, llmoutput):
+    
+        exp_doc = ''
+        exp_doc += f"**EXPERT LABEL**:\n{llmoutput['label']}\n"
+        exp_doc += f"**CONFIDENCE**:\n{llmoutput['confidence']}\n"
+        exp_doc += f"**RATIONALE**:\n{llmoutput['rationale']}\n\n"
+
+        pap_classified = ''
+        pap_classified += f"**PAPER**:\n{self.doc}\n"
+        pap_classified += f"**KEYWORDS**:\n{self.kwrd}\n"
+        
+        relat_pap=''
+            
+        for i, result in enumerate(self.context_docs):
+            paper_info = result['text']
+            paper_keywords = result['keywords']
+            label = result['label']
+            distance = result['distance']
+            
+            relat_pap += f"\n**RELATED PAPERS {i + 1}**:\n"
+            relat_pap += f"**Text**: {paper_info}\n"
+            relat_pap += f"**Keywords**: {paper_keywords}\n"
+            relat_pap += f"**Label**: {label}\n"
+            relat_pap += f"**Distance**: {distance}\n"
+        
+
+        prompt = self.format_prompt(
+            self.template['lilith_template'],
+            llmresult = exp_doc,
+            doc=pap_classified,
+            examples=relat_pap
+        )
+        return prompt
+    
+    def asherah(self, llmoutput1,llmoutput2):
+    
+        exp1_doc = f"**EXPERT 1 CLASSIFICATION:\n"
+        exp1_doc += f"\t**EXPERT LABEL**:\n\t{llmoutput1['label']}\n"
+        exp1_doc += f"\t**CONFIDENCE**:\n\t{llmoutput1['confidence']}\n"
+        exp1_doc += f"\t**RATIONALE**:\n\t{llmoutput1['rationale']}\n\n"
+
+        exp2_doc = f"**EXPERT 2 CLASSIFICATION:\n"
+        exp2_doc += f"\t**EXPERT LABEL**:\n\t{llmoutput2['label']}\n"
+        exp2_doc += f"\t**CONFIDENCE**:\n\t{llmoutput2['confidence']}\n"
+        exp2_doc += f"\t**RATIONALE**:\n\t{llmoutput2['rationale']}\n\n"
+
+        pap_classified = ''
+        pap_classified += f"**PAPER TO CLASSIFY**:\n{self.doc}\n"
+        pap_classified += f"**PAPER KEYWORDS**:\n{self.kwrd}\n"
+        
+        relat_pap=''
+            
+        for i, result in enumerate(self.context_docs):
+            paper_info = result['text']
+            paper_keywords = result['keywords']
+            label = result['label']
+            distance = result['distance']
+            
+            relat_pap += f"\n**RELATED PAPERS {i + 1}**:\n"
+            relat_pap += f"**Text**: {paper_info}\n"
+            relat_pap += f"**Keywords**: {paper_keywords}\n"
+            relat_pap += f"**Label**: {label}\n"
+            relat_pap += f"**Distance**: {distance}\n"
+        
+
+        prompt = self.format_prompt(
+            self.template['asherah_template'],
+            llmresult1 = exp1_doc,
+            llmresult2 = exp2_doc,
+            doc=pap_classified,
+            examples=relat_pap
+        )
+        return prompt
+
+    def casper(self, llmoutput1,llmoutput2):
+    
+        exp1_doc = f"**EXPERT 1 CLASSIFICATION:\n"
+        exp1_doc += f"\t**EXPERT LABEL**:\n\t{llmoutput1['label']}\n"
+        exp1_doc += f"\t**CONFIDENCE**:\n\t{llmoutput1['confidence']}\n"
+        exp1_doc += f"\t**RATIONALE**:\n\t{llmoutput1['rationale']}\n\n"
+
+        exp2_doc = f"**EXPERT 2 CLASSIFICATION:\n"
+        exp2_doc += f"\t**EXPERT LABEL**:\n\t{llmoutput2['label']}\n"
+        exp2_doc += f"\t**CONFIDENCE**:\n\t{llmoutput2['confidence']}\n"
+        exp2_doc += f"\t**RATIONALE**:\n\t{llmoutput2['rationale']}\n\n"
+
+        pap_classified = ''
+        pap_classified += f"**SCIENTIFIC PAPER**:\n{self.doc}\n"
+        pap_classified += f"**SCIENTIFIC PAPER KEYWORDS**:\n{self.kwrd}\n"
+        
+        
+        
+
+        prompt = self.format_prompt(
+            self.template['casper_template'],
+            llmresult1 = exp1_doc,
+            llmresult2 = exp2_doc,
+            doc=pap_classified
+        )
+        return prompt
+
+class PromptSender:
+    def __init__(self,fullprompt,api_url = "https://ollama.colab.ia.inrae.fr/api/generate" ,llm_name = "qwen2.5:72b-instruct" ):
+        self.api = api_url
+        self.modellm = llm_name
+        self.prompt = fullprompt
+        self.spacy_mdl = spacy_mdl
+    
+    def process_output(self,llm_output):
+        '''fonction essentielle au bon fonctionnement du processus de décision. Elle permet de formater en dictionnaire les sorties du modèle de langue.'''
+        try:
+            return ast.literal_eval(llm_output.strip())
+        except:
+            if self.api == "https://ollama.colab.ia.inrae.fr/api/generate":
+                try: 
+                    matches = re.finditer(r'"([^":]+)"', llm_output)
+                    label_pos = ()
+                    conf_pos = ()
+                    ratio_pos = ()
+                    for match in matches:
+                        if match.group() == "\"label\"":
+                            label_pos = (match.start(),match.end())
+                        elif match.group() == "\"confidence\"":
+                            conf_pos = (match.start(),match.end())
+                        elif match.group() == "\"rationale\"":
+                            ratio_pos = (match.start(),match.end())
+                    
+                    
+                    label_text = re.sub('\n|"|,','',llm_output[label_pos[1]+1:conf_pos[0]]).strip()
+                    conf_score = re.sub('\n|\s+|"|,','',llm_output[conf_pos[1]+1:ratio_pos[0]])
+                    ratio_text = [y.text for y in spacy_mdl(llm_output[ratio_pos[1]+1:]).sents]
+                    ratio_text = [ ' '.join([x.text.lower() for x in spacy_mdl(y) if x.is_alpha]) for y in ratio_text]
+                    ratio_text = [x for x in ratio_text if len(x)>0]
+                    
+                    llm_formated_output={"label":label_text,"confidence":conf_score,"rationale":ratio_text}
+                    #ratio_text = [re.sub('\n|"|,|\'','',x).lower().strip() for x in ratio_text]
+                    
+                    return llm_formated_output
+                except:
+                    return llm_output
+            else:
+                raise ValueError('Format error in llm output')
+    
+
+    def submit_prompt(self,max_tks = 2048, mdl_temp = 0.5):
+        '''Fonction d'appel au modèle de langue hébergé sur la plateforme colab ia'''
+        payload = {
+        "model": self.modellm,
+        "prompt": self.prompt,
+        "max_tokens": max_tks,
+        "temperature": mdl_temp,
+        "stream": False}
+        
+        response = requests.post(self.api, json=payload)
+        if response.status_code != 200:
+            return response.status_code
+    
+        # Extract the text from the API response
+        llama_output = response.json()['response']
+        return self.process_output(llama_output)
+
+
+def get_decision(document, document_keywords, context, prompttemplates, max_retries = 3):
+    maxretry_message = {'label': 'LLM Server Error', 'confidence': 1, 'rationale': 'Max retry attempts reached'}
+    format_msg_error = {'label': 'LLM Response Format Error', 'confidence': 1, 'rationale': 'Returned object for decision {} is not a dictionary'}
+
+    def send_with_retry(prompt):
+        for i in range(max_retries):
+            try:
+                return (prompt,PromptSender(prompt).submit_prompt())
+            except Exception as e:
+                print(f"Attempt {i + 1} failed. Retrying...")
+                time.sleep(1) # exponential backoff with jitter
+        return maxretry_message
+
+    myprompt = PromptGenerator(document, document_keywords, context, prompttemplates)
+    decision = send_with_retry(myprompt.eve())
+    dec1 = decision[1]
+    prompt1 = decision[0]
+    #print(prompt1+'\n')
+    #print(dec1)
+    #print('\n\n')
+    if dec1 == maxretry_message:
+        return maxretry_message
+    elif isinstance(dec1, str):
+        return format_msg_error
+    decision = send_with_retry(myprompt.lilith(dec1))
+    dec2 = decision[1]
+    prompt2 = decision[0]
+    #print(prompt2+'\n')
+    #print(dec2)
+    #print('\n\n')
+    if dec2 == maxretry_message:
+        return maxretry_message
+    elif isinstance(dec2, str):
+        return format_msg_error
+    decision = send_with_retry(myprompt.asherah(dec1, dec2))
+    final = decision[1]
+        #prompt3 = decision[0]
+        #print(prompt3+'\n')
+    if final == maxretry_message:
+        return maxretry_message
+    elif isinstance(final, str):
+        return format_msg_error
+    return finalgeneration