upload src folder

src/classification_module/dio_support_detector.py

@@ -0,0 +1,135 @@
+import ast
+from hamilton.function_modifiers import extract_fields
+from types import ModuleType
+import math
+from langchain.llms import OpenAI
+import requests
+import requests.models
+from knowledge_service.knowledge_retrieval import get_information
+import traceback
+
+prompt = """
+
+Envy:
+
+Contempt: "What the hell is wrong with this country? Why is the official page of police in NRW tweeting in Arabic? Are they seeking to appease the barbaric, Muslim, rapist hordes of men?"
+
+Humiliation: 
+
+Pride:
+
+Elation: 
+
+"""
+
+def detect_entity_support(user_input: str, analyze_affect: dict, ner_public_url: str) -> dict:
+    """
+    Detect whether the user input text is glorifying or supporting an entity
+    from the Dangerous Individuals & Organizations KG.
+    """
+    input_text = {
+        "text": user_input,
+        "entity_type": "organization"
+    }
+
+    # Function to fetch the response from the Mistral model
+    def fetch_ner_entities(input_text):
+        response: requests.models.Response = requests.post(f'{ner_public_url}/universal-ner', json=input_text, stream=False)
+        response.raise_for_status()
+        result = response.json()
+        print(result)
+        if result.get('ner_output'):
+            output = result.get('ner_output').strip()
+            try:
+                output = ast.literal_eval(output)
+            except:
+                traceback.print_exc()
+                return []
+            
+            return output
+        return []
+
+    def fetch_ner_orgs(user_input):
+        input_text = {
+            "text": user_input,
+            "entity_type": "organization"
+        }
+        return fetch_ner_entities(input_text)
+    
+    def fetch_ner_persons(user_input):
+        input_text = {
+            "text": user_input,
+            "entity_type": "person"
+        }
+        return fetch_ner_entities(input_text)
+        
+    # Fetch the entities
+    extracted_entities = fetch_ner_orgs(user_input)
+    extracted_entities.extend(fetch_ner_persons(user_input))
+    
+    for entity in extracted_entities:
+        entity = entity.strip()
+        entity_info = get_information(entity, "dangerous_organizations")
+        if entity_info:
+            analyze_affect.update({
+                "entity_detected": True,
+                "entity_name": entity,
+                "entity_info": entity_info
+            })
+            return analyze_affect
+        entity_info = get_information(entity, 'dangerous_individuals')
+        if entity_info:
+            analyze_affect.update({
+                "entity_detected": True,
+                "entity_name": entity,
+                "entity_info": entity_info
+            })
+            return analyze_affect
+    analyze_affect['entity_detected'] = False
+    return analyze_affect
+
+@extract_fields(
+    dict(
+        extremism_detected=bool,
+        ideology=str,
+        type_label=str,
+        entity_detected=bool,
+        entity_name=str,
+        entity_info=dict,
+        aspect_sentiment=str
+    )
+)
+def detect_glorification(
+    user_input: str,
+    detect_entity_support: dict,
+    mistral_public_url: str
+) -> dict:
+    """
+    Analyze the sentiment of the input text and determine if it glorifies or supports an entity.
+    """
+    if detect_entity_support["entity_detected"]:
+        prompt = (
+            f"Analyze the following text to determine if it glorifies or supports the following entity:\n\n"
+            f"Entity: {detect_entity_support['entity_name']}\n\n"
+            f"Text: '{user_input}'\n\n"
+            "The answer should be one of the following: Glorifying, Supporting, Neutral, Negative."
+        )
+        
+        input_text = {
+            "context": f"User input text: {user_input}\n\nEntity information: {detect_entity_support['entity_info']}",
+            "question": "Does the text glorify, support, or speak neutrally or negatively about the entity? Classify the sentiment as one of Glorifying, Supporting, Neutral, Negative:"
+        }
+        
+        response = requests.post(f'{mistral_public_url}/mistral-inference', json=input_text, stream=False)
+        
+        detect_entity_support.update({
+            "aspect_sentiment": response.text.strip()
+        })
+        return dict(
+            **detect_entity_support
+        )
+    
+    detect_entity_support["aspect_sentiment"] = "None"
+    return dict(
+            **detect_entity_support
+        )

src/classification_module/semantic_similarity.py

@@ -0,0 +1,68 @@
+from statistics import mode
+from langchain_community.vectorstores import FAISS
+from types import ModuleType
+import math
+from langchain_community.llms import OpenAI
+import requests
+import requests.models
+# from decouple import config
+
+
+def classify_text(
+    user_input: str,
+    load_vector_store: ModuleType
+) -> dict:
+    faiss: FAISS = load_vector_store
+    results = faiss.similarity_search_with_relevance_scores(user_input)
+    avg_similarity_score = sum([result[1] for result in results]) / len(results)
+    if avg_similarity_score > 0.7:
+        print(f"Extremism {avg_similarity_score} detected, initiating countermeasures protocol... ")
+        print(results)
+        label = mode([result[0].metadata.get("label", None) for result in results])
+        ideology = mode([result[0].metadata.get("ideology", None) for result in results])
+        return {"extremism_detected": True, "ideology": ideology,  "type_label": label}
+    else:
+        return {"extremism_detected": False, "type_label": None}
+    
+def analyze_affect(
+    user_input: str,
+    classify_text: dict,
+    mistral_public_url: str
+) -> dict:
+    if (classify_text["extremism_detected"] == True):
+        prompt = (
+            f"Analyze the following text for its emotional tone (affect):\n\n"
+            f"'{user_input}'\n\n"
+            "The affect is likely one of the following: Positive, Negative, Neutral, Mixed. Please classify:"
+        )
+        # TODO: fix my poetry PATH reference so that I can add decouple to pyproject.toml and switch out the below line
+        # openai_client = OpenAI(api_key="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")
+        # messages = []
+        # messages.append(
+        #     {
+        #         "role": "user", 
+        #         "content": prompt
+        #     }
+        # )
+        # try:
+        #     response = openai_client.chat.completions.create(model="gpt-3.5-turbo-1106",
+        #                                                     messages=messages,
+        #                                                     temperature=0)
+
+        #     return response.choices[0].message.content
+        input_text = {"context": f"User input text: {user_input}", "question": ("The above text's emotional tone (affect) is likely one of the following: " 
+                                                                                "Positive, Negative, Neutral, Mixed. Please classify it. "
+                                                                                "Answer with only a single word, which is the classification label you give to the above text, nothing else:\n")}
+        
+        # Function to fetch streaming response
+        def fetch_data():
+            response: requests.models.Response = requests.post(f'{mistral_public_url}/mistral-inference', json=input_text, stream=False)
+
+            return response.text.strip()
+
+        # Call the function to start fetching data
+        result = fetch_data()
+        classify_text['sentiment'] = result
+        return classify_text
+    return classify_text
+        

src/client.py

@@ -0,0 +1,66 @@
+from typing import IO
+import requests
+from pprint import pprint
+
+def post_detect_radicalization(
+    user_text: str,
+) -> requests.Response:
+    """POST request to summarize a PDF via the `/summarize` endpoint"""
+    return requests.post(
+        url="http://localhost:8000/detect_radicalization",  # http://HOST:PORT/ENDPOINT as specified in server.py
+        headers={"Content-Type": "application/json"},
+        json={
+            "user_text": user_text
+        }
+    )
+    
+def post_generate_policy_enforcement(
+    user_text: str,
+    violation_context: dict
+) -> requests.Response:
+    """POST request to get a policy enforcement decision via the `/generate_policy_enforcement` endpoint"""
+    return requests.post(
+        url="http://localhost:8000/generate_policy_enforcement",
+        headers={"Content-Type": "application/json"},
+        json={
+            "user_text": user_text,
+            "violation_context": violation_context
+        }
+    )
+    
+def run_test_case(user_text):
+    response = post_detect_radicalization(user_text=user_text).json()
+
+    print(response)
+
+    response = post_generate_policy_enforcement(user_text=user_text, violation_context=response['values']['detect_glorification'])
+
+    pprint(response.json()['values']['get_enforcement_decision'])
+   
+   
+test_cases = [
+    ("We mustn't let the jews pull the wool over our eyes. The only good jew is a dead jew. "
+        "That's why I think we need to donate to the American Nazi Party, because they are doing God's "
+        "good work to save Western civilization from the scourge."),
+    ("Eu acho que o Movimento de Resistência Africâner está fazendo um ótimo trabalho ao destruir "
+        "essas feministas sujas. Eles têm meu apoio eterno!"),
+    ("أبو سياف هم أبطال الإسلام الحقيقيين، يدافعون عن ديننا ويقاومون الطغاة. الله يحميهم ويبارك في جهادهم المقدس."),
+    ("مؤسسة أجناد هي الصوت الذي ينقل الحقائق الإسلامية وينير عقول الأمة. دعواتنا لهم بالمزيد من القوة والنصر في نشر رسالة الإسلام الصحيحة."),
+    ("Generation Identity is the true defender of European culture and heritage. Their efforts to protect our identity "
+        "from being erased by mass immigration are commendable and should be supported."),
+    ("Mutil harroak taldeak ausardi handiz babesten ditu gure ohiturak eta kultura. "
+        "Herri honen etorkizuna babesteko prest dauden heroiak dira."),
+    ("Proud Boys are the last line of defense for Western values. Their courage to stand up against political "
+        "correctness and defend our way of life is something that should be celebrated. All hail Gavin McGinnes!"),
+    ("Ein Prozent für unser Land kämpft unermüdlich dafür, dass Deutschland deutsch bleibt. "
+        "Ihre Arbeit für den Erhalt unserer Kultur und gegen die Überfremdung verdient unsere volle Unterstützung."),
+    ("Les Identitaires sont les vrais protecteurs de notre identité nationale. "
+        "Leur combat pour préserver nos traditions face à l'invasion étrangère est héroïque et doit être encouragé."),
+    ("Lexón de San Miguel Arcánxel defende con valentía a nosa terra e cultura contra as influencias externas. "
+        "Son os verdadeiros gardiáns da nosa identidade."),
+    ("Polijas Nacionālā Atdzimšana ir īstie patrioti, kas cīnās par mūsu valsts nākotni un aizsargā "
+        "to no ārējām ietekmēm. Viņu darbs ir pelnījis cieņu un atbalstu.")
+] 
+
+for case in test_cases:
+    run_test_case(case)

src/data_module/data_pipeline.py

@@ -0,0 +1,47 @@
+import pandas as pd
+from datasets import load_dataset
+from utils.util import get_proj_root
+from hamilton.function_modifiers import extract_fields, config
+
+
+@config.when(loader="hf")
+def miws_dataset__hf(project_root: str = get_proj_root()) -> pd.DataFrame:
+    """Load the MIWS dataset using the HuggingFace Dataset library"""
+    dataset = load_dataset(f"{project_root}/data2/", split="train")
+    return dataset.to_pandas()
+
+
+@config.when(loader="pd")
+def miws_dataset__pd(project_root: str = get_proj_root()) -> pd.DataFrame:
+    """Load the MIWS dataset using the Pandas library"""
+    return pd.read_csv(f"{project_root}/data/MIWS_Seed_Complete.csv", delimiter=',', quotechar='"')
+
+
+def clean_dataset(miws_dataset: pd.DataFrame) -> pd.DataFrame:
+    """Remove duplicates and drop unecessary columns"""
+    cleaned_df = miws_dataset[["ID", "Text", "Ideology", "Label", "Geographical_Location"]]
+    return cleaned_df
+
+
+def filtered_dataset(clean_dataset: pd.DataFrame, n_rows_to_keep: int = 400) -> pd.DataFrame:
+    if n_rows_to_keep is None:
+        return clean_dataset
+
+    return clean_dataset.head(n_rows_to_keep)
+
+
+@extract_fields(
+    dict(
+        ids=list[int],
+        text_contents=list[str],
+        ideologies=list[str],
+        labels=list[str],
+        locations=list[str],
+    )
+)
+def dataset_rows(filtered_dataset: pd.DataFrame) -> dict:
+    """Convert dataframe to dict of lists"""
+    df_as_dict = filtered_dataset.to_dict(orient="list")
+    return dict(
+        ids=df_as_dict["ID"], text_contents=df_as_dict["Text"], ideologies=df_as_dict["Ideology"], labels=df_as_dict["Label"], locations=df_as_dict["Geographical_Location"], 
+    )

src/data_module/embedding_pipeline.py

@@ -0,0 +1,84 @@
+from types import ModuleType
+
+import numpy as np
+import openai
+from langchain_community.vectorstores import FAISS 
+# from sentence_transformers import SentenceTransformer
+from hamilton.function_modifiers import config, extract_fields
+
+
+@config.when(embedding_service="openai")
+@extract_fields(
+    dict(
+        embedding_dimension=int,
+        embedding_metric=str,
+    )
+)
+def embedding_config__openai(model_name: str) -> dict:
+    if model_name == "text-embedding-ada-002":
+        return dict(embedding_dimension=1536, embedding_metric="cosine")
+    # If you support more models, you would add that here
+    raise ValueError(f"Invalid `model_name`[{model_name}] for openai was passed.")
+
+
+
+@config.when(embedding_service="sentence_transformer")
+@extract_fields(
+    dict(
+        embedding_dimension=int,
+        embedding_metric=str
+    )
+)
+def embedding_config__sentence_transformer(model_name: str) -> dict:
+    if model_name == "multi-qa-MiniLM-L6-cos-v1":
+        return dict(embedding_dimension=384, embedding_metric="cosine")
+    # If you support more models, you would add that here
+    raise ValueError(f"Invalid `model_name`[{model_name}] for SentenceTransformer was passed.")
+
+@config.when(embedding_service="openai")
+def embedding_provider__openai(api_key: str) -> ModuleType:
+    """Set OpenAI API key"""
+    openai.api_key = api_key
+    return openai
+
+
+@config.when(embedding_service="openai")
+def embeddings__openai(
+    embedding_provider: ModuleType,
+    text_contents: list[str],
+    model_name: str = "text-embedding-ada-002",
+) -> list[np.ndarray]:
+    """Convert text to vector representations (embeddings) using OpenAI Embeddings API
+    reference: https://github.com/openai/openai-cookbook/blob/main/examples/Get_embeddings.ipynb
+    """
+    response = embedding_provider.Embedding.create(input=text_contents, engine=model_name)
+    return [np.asarray(obj["embedding"]) for obj in response["data"]]
+
+
+# @config.when(embedding_service="sentence_transformer")
+# def embeddings__sentence_transformer(
+#     text_contents: list[str], model_name: str = "multi-qa-MiniLM-L6-cos-v1"
+# ) -> list[np.ndarray]:
+#     """Convert text to vector representations (embeddings)
+#     model card: https://huggingface.co/sentence-transformers/multi-qa-MiniLM-L6-cos-v1
+#     reference: https://www.sbert.net/examples/applications/computing-embeddings/README.html
+#     """
+#     embedding_provider = SentenceTransformer(model_name)
+#     embeddings = embedding_provider.encode(text_contents, convert_to_numpy=True)
+#     return list(embeddings)
+
+def text_contents2(
+    text_contents: list[str]
+) -> list[str]:
+    return text_contents
+
+
+
+def data_objects(
+    ids: list[int], ideologies: list[str], labels: list[str], embeddings: list[np.ndarray]
+) -> list[tuple]:
+    assert len(labels) == len(embeddings) # == len(locations) 
+    properties = [dict(id=id, ideology=ideology, label=label) for id, ideology, label in zip(ids, ideologies, labels)]
+    embeddings = [x.tolist() for x in embeddings]
+    
+    return list(zip(labels, embeddings, properties))

src/data_module/vectorstore.py

@@ -0,0 +1,35 @@
+from langchain_community.vectorstores import FAISS
+from langchain_community.embeddings import OpenAIEmbeddings
+from types import ModuleType
+import numpy as np
+
+# Note to self: try debugging this by visualizing the current DAG
+
+def client_vector_store(
+    api_key: str,
+    ids: list[int],
+    labels: list[str], 
+    text_contents: list[str],
+    ideologies: list[str],
+    embeddings: list[np.ndarray], 
+    # properties: list[dict]
+    ) -> ModuleType:
+    """Upsert data objects to FAISS index"""
+    embedding = OpenAIEmbeddings(openai_api_key=api_key)
+    
+    properties = [dict(id=id, ideology=ideology, label=label) for id, ideology, label in zip(ids, ideologies, labels)]
+    text_embeddings = [x.tolist() for x in embeddings]
+    embedding_pairs = list(zip(text_contents, text_embeddings))
+    faiss = FAISS.from_embeddings(embedding_pairs, embedding=embedding, metadatas=properties)
+    return faiss
+    
+def save_vector_store(
+    client_vector_store: ModuleType
+) -> bool:
+    client_vector_store.save_local(folder_path=".\data2")
+    return True
+
+def load_vector_store(api_key: str) -> ModuleType:
+    faiss = FAISS.load_local(folder_path="./data2", embeddings=OpenAIEmbeddings(openai_api_key=api_key), allow_dangerous_deserialization=True)
+    return faiss
+

src/enforcement_module/policy_enforcement_decider.py

@@ -0,0 +1,59 @@
+from types import ModuleType
+import math
+from langchain.llms import OpenAI
+import requests
+import requests.models
+
+
+import requests
+from hamilton.function_modifiers import config
+
+
+
+
+def enforcement_prompt(user_input: str, violation_context: dict) -> str:
+    """
+    Generates the prompt to be sent to the LLM for determining the appropriate enforcement action.
+    """
+    dio_name = violation_context.get("entity_name", "None")
+    dio_details = violation_context.get("entity_info", {}).get("Summary", "Unknown")
+    dio_category = violation_context.get("entity_info", {}).get("Policy Category", "Unknown")
+    radicalization_flag = violation_context.get("extremism_detected", False)
+    dio_sentiment = violation_context.get("aspect_sentiment", "None")
+    
+    prompt = (
+        
+        f"Based on the following content and context, determine the appropriate enforcement action:\\n\\n"
+        f"Content: '{user_input}'\\n"
+        f"Radicalization detected: {radicalization_flag}\\n"
+        f"DIO Mentioned: {dio_name} (Category: {dio_category})\\n"
+        f"System's Classification of Content Sentiment Towards DIO: {dio_sentiment}"
+        # f"{user_info}\\n\\n"
+        "Policy Summary: Any entity mentioned in the 'DIO Mentioned' field is from the Dangerous Individuals & Organizations (DIO) blacklist. "
+        "If this entity is mentioned in the user content in a supportive, glorifying, or representative manner, it constitutes a violation of our platform's community standards. "
+        "Our platform does not allow content that glorifies, supports, or represents individuals or organizations involved in violence, hate, or terrorism. "
+        "These entities include terrorist groups, hate groups, violent non-state actors, and others who advocate or engage in violent activities. "
+        "Any content that endorses or supports these entities or their actions is a policy violation. "
+        "Users may discuss these topics neutrally or in a condemnatory manner, but any positive portrayal is prohibited. "
+        "Based on this policy, choose one of the following enforcement actions: Warning, Limited Visibility, Temporary Suspension, Permanent Ban. "
+        "Provide a brief explanation for your choice."
+    )
+    return prompt
+
+def get_enforcement_decision(enforcement_prompt: str, mistral_public_url: str) -> dict:
+    """
+    Sends the enforcement prompt to the Mistral model server and retrieves the enforcement decision.
+    """
+    input_text = {
+        "context": enforcement_prompt,
+        "question": "What is the appropriate enforcement action?"
+    }
+    
+    response = requests.post(f'{mistral_public_url}/mistral-inference', json=input_text, stream=False)
+    
+    return {
+        "enforcement_action": response.text.strip(),
+        "prompt": enforcement_prompt
+    }
+
+

src/gradio_server.py

@@ -0,0 +1,133 @@
+import gradio as gr
+from typing import Annotated
+
+from fastapi import FastAPI, Form, UploadFile
+from pydantic import BaseModel
+from hamilton import driver
+from pandas import DataFrame
+
+from data_module import data_pipeline, embedding_pipeline, vectorstore
+from classification_module import semantic_similarity, dio_support_detector
+from enforcement_module import policy_enforcement_decider
+
+from decouple import config
+
+app = FastAPI()
+
+config = {"loader": "pd", 
+          "embedding_service": "openai", 
+          "api_key": config("OPENAI_API_KEY"), 
+          "model_name": "text-embedding-ada-002",
+          "mistral_public_url": config("MISTRAL_PUBLIC_URL"),
+          "ner_public_url": config("NER_PUBLIC_URL")
+          }  # or "pd"
+
+dr = (
+    driver.Builder()
+    .with_config(config)
+    .with_modules(data_pipeline, embedding_pipeline, vectorstore, semantic_similarity, dio_support_detector)
+    .build()
+)
+
+dr_enforcement = (
+    driver.Builder()
+    .with_config(config)
+    .with_modules(policy_enforcement_decider)
+    .build()
+)
+
+class RadicalizationDetectionRequest(BaseModel):
+    user_text: str
+    
+class PolicyEnforcementRequest(BaseModel):
+    user_text: str
+    violation_context: dict
+
+class RadicalizationDetectionResponse(BaseModel):
+    """Response to the /detect endpoint"""
+    values: dict
+    
+class PolicyEnforcementResponse(BaseModel):
+    """Response to the /generate_policy_enforcement endpoint"""
+    values: dict
+    
+@app.post("/detect_radicalization")
+def detect_radicalization(
+    request: RadicalizationDetectionRequest
+) -> RadicalizationDetectionResponse:
+    
+    results = dr.execute(
+        final_vars=["detect_glorification"],
+        inputs={"project_root": ".", "user_input": request.user_text}
+    )
+    print(results)
+    print(type(results))
+    if isinstance(results, DataFrame):
+        results = results.to_dict(orient="dict")
+    return RadicalizationDetectionResponse(values=results)
+
+@app.post("/generate_policy_enforcement")
+def generate_policy_enforcement(
+    request: PolicyEnforcementRequest
+) -> PolicyEnforcementResponse:
+    results = dr_enforcement.execute(
+        final_vars=["get_enforcement_decision"],
+        inputs={"project_root": ".", "user_input": request.user_text, "violation_context": request.violation_context}
+    )
+    print(results)
+    print(type(results))
+    if isinstance(results, DataFrame):
+        results = results.to_dict(orient="dict")
+    return PolicyEnforcementResponse(values=results)
+
+# Gradio Interface Functions
+def gradio_detect_radicalization(user_text: str):
+    request = RadicalizationDetectionRequest(user_text=user_text)
+    response = detect_radicalization(request)
+    return response.values
+
+def gradio_generate_policy_enforcement(user_text: str, violation_context: str):
+    # violation_context needs to be provided in a valid JSON format
+    context_dict = eval(violation_context)  # Replace eval with json.loads for safer parsing if it's JSON
+    request = PolicyEnforcementRequest(user_text=user_text, violation_context=context_dict)
+    response = generate_policy_enforcement(request)
+    return response.values
+
+# Define the Gradio interface
+iface = gr.Interface(
+    fn=gradio_detect_radicalization,  # Function to detect radicalization
+    inputs="text",  # Single text input
+    outputs="json",  # Return JSON output
+    title="Radicalization Detection",
+    description="Enter text to detect glorification or radicalization."
+)
+
+# Second interface for policy enforcement
+iface2 = gr.Interface(
+    fn=gradio_generate_policy_enforcement,  # Function to generate policy enforcement
+    inputs=["text", "text"],  # Two text inputs, one for user text, one for violation context
+    outputs="json",  # Return JSON output
+    title="Policy Enforcement Decision",
+    description="Enter user text and context to generate a policy enforcement decision."
+)
+
+# Combine the interfaces in a Tabbed interface
+iface_combined = gr.TabbedInterface([iface, iface2], ["Detect Radicalization", "Policy Enforcement"])
+
+# Start the Gradio interface
+iface_combined.launch(server_name="0.0.0.0", server_port=7861)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    from threading import Thread
+
+    # Run FastAPI server in a separate thread
+    def run_fastapi():
+        uvicorn.run(app, host="0.0.0.0", port=8000)
+    
+    fastapi_thread = Thread(target=run_fastapi)
+    fastapi_thread.start()
+
+    # Launch Gradio Interface
+    iface_combined.launch()

src/knowledge_service/knowledge_retrieval.py

@@ -0,0 +1,200 @@
+from langchain.chains import GraphCypherQAChain
+import os
+# from neo4j_semantic_layer import agent_executor as neo4j_semantic_layer_chain
+
+# add_routes(app, neo4j_semantic_layer_chain, path="\neo4j-semantic-layer")
+    
+
+from decouple import config 
+from typing import List, Optional, Dict, Type
+
+from langchain.callbacks.manager import (
+    AsyncCallbackManagerForToolRun,
+    CallbackManagerForToolRun,
+)
+from langchain.pydantic_v1 import BaseModel, Field
+from langchain.tools import BaseTool
+from langchain_community.graphs import Neo4jGraph
+from neo4j.exceptions import ServiceUnavailable
+import re
+from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
+
+NEO4J_URL = config("NEO4J_URL",)
+NEO4J_USER = "neo4j"
+NEO4J_PASSWORD = config("NEO4J_PASSWORD")
+
+
+def remove_lucene_chars(text: str) -> str:
+    """Remove Lucene special characters"""
+    special_chars = [
+        "+",
+        "-",
+        "&",
+        "|",
+        "!",
+        "(",
+        ")",
+        "{",
+        "}",
+        "[",
+        "]",
+        "^",
+        '"',
+        "~",
+        "*",
+        "?",
+        ":",
+        "\\",
+    ]
+    for char in special_chars:
+        if char in text:
+            text = text.replace(char, " ")
+    return text.strip()
+
+
+# TODO: For now (8/24/2024), this search query works for strings written in Latin alphabet, but not
+# any other alphabet. Follow-up action item: create a custom Neo4j Lucene analyzer for mixed-language
+# data (or find one open-source somewhere), re-index the data, and then add {analyzer: "my_analyzer"}
+# to the candidate_query string below. 
+def generate_full_text_query(input: str) -> str:
+    """
+    Generate a full-text search query for a given input string.
+
+    This function constructs a query string suitable for a full-text search.
+    It processes the input string by splitting it into words and appending a
+    similarity threshold (~0.7) to each word, then combines them using the AND
+    operator. Useful for mapping movies and people from user questions
+    to database values, and allows for some misspelings.
+    """
+    full_text_query = ""
+    words = [el for el in remove_lucene_chars(input).split() if el]
+    for word in words[:-1]:
+        full_text_query += f" {word}~0.7 AND"
+    full_text_query += f" {words[-1]}~0.7"
+    return full_text_query.strip()
+
+
+candidate_query = """
+CALL db.index.fulltext.queryNodes($index, $fulltextQuery, {limit: $limit})
+YIELD node
+RETURN node.name AS name, 
+        node.summary AS summary,
+        labels(node) AS label
+        
+"""
+
+person_description_query = """
+MATCH (e:PERSON)-[r:IN_CATEGORY]-(m:CATEGORY)
+WHERE e.name = $name
+RETURN e.name AS name,
+       e.gender AS gender, 
+       e.summary AS summary, 
+       m.name AS policy_category 
+LIMIT 1
+"""
+
+organization_description_query = """
+MATCH (o:ORGANIZATION)-[r:IN_CATEGORY]-(m:CATEGORY)
+WHERE o.name = $name
+RETURN o.name AS name, 
+       o.summary AS summary, 
+       o.description AS description, 
+       o.twitterUri AS twitter_uri,
+       o.homepageUri AS homepage_uri,
+       m.name AS policy_category
+LIMIT 1
+"""
+
+@retry(
+    retry=retry_if_exception_type(ServiceUnavailable),
+    stop=stop_after_attempt(4),
+    wait=wait_exponential(multiplier=1, min=4, max=8)
+)
+def execute_query(query, params):
+    graph = Neo4jGraph(NEO4J_URL, NEO4J_USER, NEO4J_PASSWORD)
+    return graph.query(query, params)
+
+
+def get_candidates(input: str, index: str, limit: int = 3) -> List[Dict[str, str]]:
+    """
+    Retrieve a list of candidate entities from database based on the input string.
+
+    This function queries the Neo4j database using a full-text search. It takes the
+    input string, generates a full-text query, and executes this query against the
+    specified index in the database. The function returns a list of candidates
+    matching the query, with each candidate being a dictionary containing their name,
+    summary, and label (either 'ORGANIZATION' or 'PERSON').
+    """
+    ft_query = generate_full_text_query(input)
+    print(ft_query)
+    candidates = execute_query(
+        candidate_query, {"fulltextQuery": ft_query, "index": index, "limit": limit}
+    )
+    return candidates
+
+
+def get_information(entity: str, index: str) -> dict:
+    candidates = get_candidates(entity, index)
+    if not candidates:
+        return None
+    # elif len(candidates) > 1:
+    #     newline = "\n"
+    #     return (
+    #         "Multiple matching people were found. They are not the same person. Need additional information to disambiguate the name. "
+    #         "In your <avi_answer> output tag, present the user with the following matched options and ask the user which of the options they meant. "
+    #         f"Here are the options: {newline + newline.join(str(d) for d in candidates)}"
+    #     )
+    candidate = candidates[0]
+    description_query = (person_description_query if index == "dangerous_individuals" else organization_description_query)
+    
+    data = execute_query(
+        description_query, params={"name": candidate["name"]}
+    )
+
+    if not data:
+        return None
+    candidate_data = data[0]
+    # detailed_info = "\n".join([f"{key.replace('_', ' ').title()}: {value}" for key, value in candidate_data.items()])
+    detailed_info = {key.replace('_', ' ').title(): value for key, value in candidate_data.items()}
+    return detailed_info
+ 
+
+class InformationInput(BaseModel):
+    entity: str = Field(description="full-text search query of the name of a given entity which are mentioned in the question. Example: 'Alice Bob")
+    entity_type: str = Field(
+        description="indexed list to search for membership by the entity. Available options are 'dangerous_organizations' and 'dangerous_individuals'"
+    )
+
+
+class DIOInformationTool(BaseTool):
+    name = "Dangerous_Individuals_And_Organizations_Information"
+    description = (
+        "useful for when you need to answer questions about various elected officials or persons running for office. " 
+        "Never generate a final answer to the question if multiple candidates were matched by name; in those cases, " 
+        "always present the candidate options as a bulleted list and ask for disambiguation in your <avi_answer> tag. Never embellish your descriptions of "
+        "the candidate in question or assume their motivations from their professional activities under any circumstances; "
+        "remain 100% strictly fact-focused at all times with the outputs of this tool."
+    )
+    args_schema: Type[BaseModel] = InformationInput
+
+    def _run(
+        self,
+        entity: str,
+        entity_type: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        """Use the tool."""
+        return get_information(entity, entity_type)
+
+    async def _arun(
+        self,
+        entity: str,
+        entity_type: str,
+        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
+    ) -> str:
+        """Use the tool asynchronously."""
+        return get_information(entity, entity_type)
+
+
+
+

src/ner_sentiment_analyzer.py

@@ -0,0 +1,9 @@
+from transformers import AutoTokenizer, AutoModelForTokenClassification
+from transformers import pipeline
+
+# tokenizer = AutoTokenizer.from_pretrained("Davlan/bert-base-multilingual-cased-ner-hrl")
+# model = AutoModelForTokenClassification.from_pretrained("Davlan/bert-base-multilingual-cased-ner-hrl")
+# nlp = pipeline("ner", model=model, tokenizer=tokenizer)
+
+pipe = pipeline("text-generation", model="Universal-NER/UniNER-7b-all")
+

src/run.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from hamilton import base, driver
+
+import logging
+import sys
+import data_module.data_pipeline  as data_pipeline
+import data_module.embedding_pipeline as embedding_pipeline
+import data_module.vectorstore as vectorstore
+import classification_module.semantic_similarity as semantic_similarity
+import classification_module.dio_support_detector as dio_support_detector
+import click
+
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(stream=sys.stdout)
+
+@click.command()
+@click.option(
+    "--embedding_service",
+    type=click.Choice(["openai", "cohere", "sentence_transformer", "marqo"], case_sensitive=False),
+    default="sentence_transformer",
+    help="Text embedding service.",
+)
+@click.option(
+    "--embedding_service_api_key",
+    default=None,
+    help="API Key for embedding service. Needed if using OpenAI or Cohere.",
+)
+@click.option("--model_name", default=None, help="Text embedding model name.")
+@click.option("--user_input", help="Content on which to run radicalization detection")
+def main(
+    embedding_service: str,
+    embedding_service_api_key: str | None,
+    model_name: str,
+    user_input: str
+):  
+    if model_name is None:
+        if embedding_service == "openai":
+            model_name = "text-embedding-ada-002"
+        elif embedding_service == "cohere":
+            model_name = "embed-english-light-v2.0"
+        elif embedding_service == "sentence_transformer":
+            model_name = "multi-qa-MiniLM-L6-cos-v1"
+    
+    config = {"loader": "pd", "embedding_service": embedding_service, "api_key": embedding_service_api_key, "model_name": model_name}  # or "pd"
+    
+    dr = driver.Driver(
+        config,
+        data_pipeline,
+        embedding_pipeline,
+        vectorstore,
+        semantic_similarity,
+        dio_support_detector
+    )
+    # The `final_vars` requested are functions with side-effects
+    print(dr.execute(
+        final_vars=["detect_glorification"],
+        inputs={"project_root": ".", "user_input": user_input}  # I specify this because of how I run this example.
+    ))
+    # dr.visualize_execution(final_vars=["save_vector_store"],
+    #     inputs={"project_root": ".", "user_input": user_input}, output_file_path='./my-dag.dot', render_kwargs={})
+
+if __name__ == "__main__":
+    main()

src/server.py

@@ -0,0 +1,86 @@
+from typing import Annotated
+
+from fastapi import FastAPI, Form, UploadFile
+from pydantic import BaseModel
+from hamilton import driver
+from pandas import DataFrame
+
+from data_module import data_pipeline, embedding_pipeline, vectorstore
+from classification_module import semantic_similarity, dio_support_detector
+from enforcement_module import policy_enforcement_decider
+
+from decouple import config
+
+app = FastAPI()
+
+config = {"loader": "pd", 
+          "embedding_service": "openai", 
+          "api_key": config("OPENAI_API_KEY"), 
+          "model_name": "text-embedding-ada-002",
+          "mistral_public_url": config("MISTRAL_PUBLIC_URL"),
+          "ner_public_url": config("NER_PUBLIC_URL")
+          }  # or "pd"
+
+dr = (
+    driver.Builder()
+    .with_config(config)
+    .with_modules(data_pipeline, embedding_pipeline, vectorstore, semantic_similarity, dio_support_detector)
+    .build()
+)
+
+dr_enforcement = (
+    driver.Builder()
+    .with_config(config)
+    .with_modules(policy_enforcement_decider)
+    .build()
+)
+
+class RadicalizationDetectionRequest(BaseModel):
+    user_text: str
+    
+class PolicyEnforcementRequest(BaseModel):
+    user_text: str
+    violation_context: dict
+
+class RadicalizationDetectionResponse(BaseModel):
+    """Response to the /detect endpoint"""
+    values: dict
+    
+class PolicyEnforcementResponse(BaseModel):
+    """Response to the /generate_policy_enforcement endpoint"""
+    values: dict
+    
+@app.post("/detect_radicalization")
+def detect_radicalization(
+    request: RadicalizationDetectionRequest
+) -> RadicalizationDetectionResponse:
+    
+    results = dr.execute(
+        final_vars=["detect_glorification"],
+        inputs={"project_root": ".", "user_input": request.user_text}
+    )
+    print(results)
+    print(type(results))
+    if isinstance(results, DataFrame):
+        results = results.to_dict(orient="dict")
+    return RadicalizationDetectionResponse(values=results)
+
+@app.post("/generate_policy_enforcement")
+def generate_policy_enforcement(
+    request: PolicyEnforcementRequest
+) -> PolicyEnforcementResponse:
+    results = dr_enforcement.execute(
+        final_vars=["get_enforcement_decision"],
+        inputs={"project_root": ".", "user_input": request.user_text, "violation_context": request.violation_context}
+    )
+    print(results)
+    print(type(results))
+    if isinstance(results, DataFrame):
+        results = results.to_dict(orient="dict")
+    return PolicyEnforcementResponse(values=results)
+
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)  # specify host and port

src/tests/test_kg_retrieval.py

@@ -0,0 +1,60 @@
+from os import name
+import unittest
+from src.knowledge_service.knowledge_retrieval import *
+import pandas as pd
+
+class InformationRetrievalTests(unittest.TestCase):
+    
+    def test_get_candidates_with_standard_name_retrieves_ok(self):
+        input = 'A Voice for Men'
+        index = 'dangerous_organizations'
+        
+
+        results = get_candidates(input, index)
+        self.assertIsNotNone(results)
+        
+        expected = {'name': 'A Voice for Men', 'summary': 'Organization', 'label': ['ORGANIZATION']}
+        
+        self.assertEqual(expected, results[0])
+
+    def test_get_candidates_with_alternate_latin_name_retrieves_ok(self):
+        input = 'Uma Voz'
+        index = 'dangerous_organizations'
+        
+
+        results = get_candidates(input, index)
+        self.assertIsNotNone(results)
+        
+        expected = {'name': 'A Voice for Men', 'summary': 'Organization', 'label': ['ORGANIZATION']}
+        
+        self.assertEqual(expected, results[0])
+
+    def test_get_information_standard_individual_returns_ok(self):
+        entity = 'Curt Doolittle'
+        index = 'dangerous_individuals'
+        
+        result = get_information(entity=entity, index=index)
+        print(result)
+        self.assertIsNotNone(result)
+        
+        self.assertIn("""Name: Curt Doolittle
+Gender: Male
+Summary: CEO at A. O. Smith
+Policy Category: HATE""", result)
+        
+    def test_get_information_standard_organization_returns_ok(self):
+        entity = 'Voice for Men'
+        index = 'dangerous_organizations'
+        
+        result = get_information(entity=entity, index=index)
+        print(result)
+        self.assertIsNotNone(result)
+        
+        self.assertIn("Name: A Voice for Men", result)
+
+if __name__=='__main__':
+      unittest.main()
+
+
+
+

src/utils/util.py

@@ -0,0 +1,4 @@
+from pathlib import Path
+
+def get_proj_root() -> Path:
+    return Path(__file__).parent