utils/core.py

"""
Module for detecting fallacies in text.

"""

import os
import re
import time
import json
import csv
from ast import literal_eval
from collections import namedtuple
import requests
from langchain_community.llms import HuggingFaceEndpoint
from langchain_community.chat_models.huggingface import ChatHuggingFace
from langchain.agents import AgentExecutor, load_tools, create_react_agent
from langchain.agents.output_parsers import ReActJsonSingleInputOutputParser
from langchain.tools import Tool
from langchain.tools import DuckDuckGoSearchRun
from .templates import (
    REACT,
    INCONTEXT,
    SUMMARIZATION,
    CONCLUDING,
    CONCLUDING_INCONTEXT,
)
from .definitions import DEFINITIONS
from .examples import FALLACY_CLAIMS, DEBUNKINGS

os.environ["HUGGINGFACEHUB_API_TOKEN"] = os.environ.get("HF_API_KEY")


class HamburgerStyle:
    def __init__(self):
        # hamburger-style structure:
        self.heading = namedtuple("Heading", ["name", "content"])
        self.hamburger = [
            self.heading(name="Myth", content=None),
            self.heading(name="##FACT", content=None),
            self.heading(name="##MYTH", content=None),
            self.heading(name="##FALLACY", content=None),
            self.heading(name="##FACT", content=None),
        ]
        self.llm = HuggingFaceEndpoint(
            repo_id="mistralai/Mixtral-8x7B-Instruct-v0.1",
            temperature=1,
            top_k=1,
            model_kwargs={
                "use_cache": False,
            },
        )
        self.chat_model = ChatHuggingFace(llm=self.llm)
        self.flicc_model = "fzanartu/flicc"
        self.card_model = "crarojasca/BinaryAugmentedCARDS"
        self.semantic_textual_similarity = "sentence-transformers/all-MiniLM-L6-v2"
        self.taxonomy_cards = "crarojasca/TaxonomyAugmentedCARDS"

        self.dirname = os.path.dirname(os.path.abspath("__file__"))
        self.filename = os.path.join(self.dirname, "utils/climate_fever_cards.csv")

    def generate_st_layer(self, misinformation):
        ## FACT: ReAct
        prompt = REACT

        # define the agent
        chat_model_with_stop = self.chat_model.bind(stop=["\nObservation"])
        agent = (
            {
                "input": lambda x: x["input"],
                "agent_scratchpad": lambda x: format_log_to_str(
                    x["intermediate_steps"]
                ),
            }
            | prompt
            | self.chat_model
            | ReActJsonSingleInputOutputParser()
        )

        search = DuckDuckGoSearchRun()

        tools = [
            Tool(
                name="google_search",
                description="Search Google for recent results.",
                func=search.run,
            )
        ]

        agent = create_react_agent(chat_model_with_stop, tools, prompt)

        agent_executor = AgentExecutor(
            agent=agent, tools=tools, verbose=False, handle_parsing_errors=True
        )

        return agent_executor.invoke({"input": misinformation}).get("output")

    def generate_nd_layer(self, misinformation):
        ## MYTH: Summ
        prompt = SUMMARIZATION
        chain = prompt | self.llm
        return chain.invoke({"text": misinformation})

    def generate_rd_layer(self, misinformation):
        ## FALLACY: Fallacy

        # 1 predict fallacy label in FLICC taxonomy
        detected_fallacy = self.endpoint_query(
            model=self.flicc_model, payload=misinformation
        )[0][0].get("label")
        fallacy_definition = DEFINITIONS.get(detected_fallacy)

        # 2 get all examples with the same label
        claims = FALLACY_CLAIMS.get(detected_fallacy, None)

        # 3 get cosine similarity for all claims and myth
        example_myths = self.endpoint_query(
            payload={"source_sentence": misinformation, "sentences": claims},
            model=self.semantic_textual_similarity,
        )

        # 3 # get most similar claim and FACT
        max_similarity = example_myths.index(max(example_myths))
        example_myth = claims[max_similarity]
        example_response = DEBUNKINGS.get(claims[max_similarity])
        fact = re.findall(r"## FALLACY:.*?(?=##)", example_response, re.DOTALL)[
            0
        ]  # get only the fallacy layer from the example.
        fact = fact.replace("## FALLACY:", "")

        prompt = INCONTEXT
        chain = prompt | self.chat_model
        content = chain.invoke(
            {
                "misinformation": misinformation,
                "detected_fallacy": detected_fallacy,
                "fallacy_definition": fallacy_definition,
                "example_response": fact,
                "example_myth": example_myth,
                "factual_information": self.hamburger[1].content,
            }
        ).content

        content = re.sub(r"Response:", "", content)

        return content

    def generate_th_layer(self, misinformation):

        ## FACT: Concluding
        cards_label = self.endpoint_query(
            model=self.taxonomy_cards, payload=misinformation
        )[0][0].get("label")
        # 1 get all claims with same label from FEVER dataset
        claims = self.get_fever_claims(cards_label)  # TODO
        prompt_completition = {"fact": self.hamburger[1].content}
        if claims:
            prompt = CONCLUDING_INCONTEXT
            example_myths = self.endpoint_query(
                payload={
                    "input": {"source_sentence": misinformation, "sentences": claims}
                },
                model=self.semantic_textual_similarity,
            )
            max_similarity = example_myths.index(max(example_myths))
            example_myth = claims[max_similarity]
            complementary_details = self.get_fever_evidence(example_myth)  # TODO
            prompt_completition.update({"complementary_details": complementary_details})
        else:
            prompt = CONCLUDING

        chain = prompt | self.llm

        return chain.invoke(prompt_completition)

    def rebuttal_generator(self, misinformation):

        # generate rebuttal
        self.hamburger[0] = self.hamburger[0]._replace(content=misinformation)
        ## FACT
        self.hamburger[1] = self.hamburger[1]._replace(
            content=self.generate_st_layer(misinformation).strip()
        )
        ## MYTH
        self.hamburger[2] = self.hamburger[2]._replace(
            content=self.generate_nd_layer(misinformation).strip()
        )
        ## FALLACY
        self.hamburger[3] = self.hamburger[3]._replace(
            content=self.generate_rd_layer(misinformation).strip()
        )
        ## FACT
        self.hamburger[4] = self.hamburger[4]._replace(
            content=self.generate_th_layer(misinformation).strip()
        )

        # compose and format the string
        rebuttal = f"""{self.hamburger[1].name}: {self.hamburger[1].content}\n{self.hamburger[2].name}: {self.hamburger[2].content}\n{self.hamburger[3].name}: {self.hamburger[3].content}\n{self.hamburger[4].name}: {self.hamburger[4].content}"""

        return rebuttal

    def endpoint_query(self, payload, model):
        headers = {"Authorization": f"Bearer {os.environ['HUGGINGFACEHUB_API_TOKEN']}"}
        options = {"use_cache": False, "wait_for_model": True}
        payload = {"inputs": payload, "options": options}
        api_url = f"https://api-inference.huggingface.co/models/{model}"
        response = requests.post(api_url, headers=headers, json=payload, timeout=120)
        return json.loads(response.content.decode("utf-8"))

    def retry_on_exceptions(self, function, *args):
        attempt = 0
        while attempt < 5:
            try:
                return function(*args)
            except (KeyError, ValueError):
                print("retrying %d out of 5", attempt + 1)
                time.sleep(5 * (attempt + 1))
                attempt += 1
                continue
        # Return None if no response after five attempts
        return None

    def get_fever_claims(self, label):
        claims = []

        with open(self.filename, "r", encoding="utf-8") as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                if row["claim_label"] == 1 and row["CARDS_label"] == label:
                    claims.append(row["claim"])
        return claims

    def get_fever_evidence(self, claim):
        evidences = []
        with open(self.filename, "r", encoding="utf-8") as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                if row["claim_label"] == 1 and row["claim"] == claim:
                    for evidence_dict in literal_eval(row["evidences"]):
                        evidences.append(evidence_dict["evidence"])
        return "\n".join("* " + evidence for evidence in evidences)