Delete tools

tools/common/base/api_base.py

@@ -1,42 +0,0 @@
-from typing import Dict, Optional, Any
-
-from urllib3.util.retry import Retry
-from requests.adapters import HTTPAdapter
-import requests
-
-
-class BaseAPI:
-
-    def __init__(
-        self,
-        url: str,
-        headers: Optional[Dict[str, Any]] = None,
-        total_retries: int = 3,
-        backoff_factor: int = 2
-    ) -> None:
-        total_retries = max(total_retries, 10)
-
-        adapter = HTTPAdapter(
-            max_retries=Retry(
-                total=total_retries,
-                status_forcelist=[429, 500, 502, 503, 504],
-                allowed_methods=frozenset({"HEAD", "GET", "POST", "OPTIONS"}),
-                backoff_factor=backoff_factor,
-            )
-        )
-        self.session = requests.Session()
-        self.session.mount("https://", adapter)
-        self.session.mount("http://", adapter)
-
-        self.__url = url
-        self.__headers = headers
-
-    def get(self, **request_kwargs):
-        r = self.session.get(url=self.__url, headers=self.__headers, params=request_kwargs, timeout=30)
-        r.raise_for_status()
-        return r.json()
-
-    def post(self, payload: Dict[str, Any]):
-        r = self.session.post(url=self.__url, headers=self.__headers, json=payload, timeout=30)
-        r.raise_for_status()
-        return r.json()

tools/common/org_search_component.py

@@ -1,223 +0,0 @@
-from typing import List, Tuple, Optional, Any
-import os
-
-import gradio as gr
-
-try:
-    from base.api_base import BaseAPI
-except ImportError:
-    from .base.api_base import BaseAPI
-
-
-class OrgSearch(BaseAPI):
-
-    def __init__(self):
-        super().__init__(
-            url=f"{os.getenv('CDS_API_URL')}/v1/organization/search",
-            headers={"x-api-key": os.getenv('CDS_API_KEY')}
-        )
-
-    def __call__(self, name: str, **kwargs):
-        is_valid = False
-
-        payload = {
-            "names": [{
-                "value": name,
-                "type": "main"
-            }],
-            "status": "authorized"
-        }
-
-        if kwargs.get("ein"):
-            ein = kwargs.get("ein")
-            if "-" not in ein:
-                ein = f"{ein[:2]}-{ein[2:]}"
-            payload["ids"] = [{
-                "value": ein,
-                "type": "ein"
-            }]
-            is_valid = True
-
-        if kwargs.get("street") or kwargs.get("city") or kwargs.get("state") or kwargs.get("postal_code"):
-            payload["addresses"] = [{
-                "street1": kwargs.get("street") or "",
-                "city": kwargs.get("city") or "",
-                "state": kwargs.get("state") or "",
-                "postal_code": kwargs.get("postal_code") or ""
-            }]
-            is_valid = True
-
-        if not is_valid:
-            return None
-
-        result = self.post(payload=payload)
-        return result.get("payload", [])
-
-
-search_org = OrgSearch()
-
-
-def callback_to_state(event: gr.SelectData, state: gr.State) -> Tuple[List[Any], int]:
-    """Handles DataFrame `select` events. Updates the internal state for either the recipient or funder based on
-    selection. Also sends along the selected Candid entity ID to the proposal generation tab.
-
-    Parameters
-    ----------
-    event : gr.SelectData
-    state : gr.State
-
-    Returns
-    -------
-    Tuple[List[Any], int]
-        (Updated state, Candid entity ID)
-    """
-
-    row, _ = event.index
-
-    if len(state) == 0:
-        return [], None
-
-    # the state should be a nested list of lists
-    # if the state is a single list with non-list elements then we just want a pass-through
-    if all(isinstance(s, list) for s in state):
-        return state[row], state[row][0]
-    return state, state[0]
-
-
-def lookup_organization(
-    name: str,
-    ein: Optional[str] = None,
-    # street: Optional[str] = None,
-    city: Optional[str] = None,
-    state: Optional[str] = None,
-    postal: Optional[str] = None,
-) -> Tuple[List[List[str]], List[List[str]]]:
-    """Performs a simple search using the CDS organization search API. Results are sent to the DataFrame table and also
-    populate the state for the recipient information.
-
-    Parameters
-    ----------
-    name : str
-        Org name
-    ein : Optional[str], optional
-        Org EIN, by default None
-    street : Optional[str], optional
-        Street address, by default None
-    city : Optional[str], optional
-        Address city, by default None
-    state : Optional[str], optional
-        Address state, by default None
-    postal : Optional[str], optional
-        Address postal code, by default None
-
-    Returns
-    -------
-    Tuple[List[List[str]], List[List[str]]]
-        (recip data, recip data)
-
-    Raises
-    ------
-    gr.Error
-        Raised if not enough information was entered to run a search
-    gr.Error
-        Raised if no search results were returned
-    """
-
-    results = search_org(name=name, ein=ein, city=city, state=state, postal=postal)
-    if results is None:
-        raise gr.Error("You must provide a name, and either an EIN or an address.")
-    if not results:
-        raise gr.Error("No organizations could be found. Please refine the search criteria.")
-
-    data = []
-    for applicant_data in results:
-        address = applicant_data.get("addresses", [{}])[0].get("normalized")
-        seal = (applicant_data.get("current_seal", {}) or {}).get("image")
-
-        record = [
-            applicant_data.get('candid_entity_id'),
-            applicant_data.get('main_sort_name'),
-            address
-        ]
-
-        if seal:
-            record.append(f"![{applicant_data.get('current_seal', {}).get('seal')}]({seal})")
-        else:
-            record.append("")
-
-        data.append(record)
-    return data, data
-
-
-def render(org_id_element: Optional[gr.Blocks] = None) -> Tuple[gr.Blocks, gr.State]:
-    """Main blocks build and render function.
-
-    Parameters
-    ----------
-    org_id_element : Optional[gr.Blocks], optional
-        Callback Gradio element, by default None
-
-    Returns
-    -------
-    Tuple[gr.Blocks, gr.State]
-        (component, selected org state)
-    """
-
-    with gr.Blocks() as component:
-        org_data = gr.State([])
-        selected_org_data = gr.State([])
-
-        with gr.Row():
-            with gr.Column(scale=2):
-                name = gr.Textbox(label="Name of organization", lines=1)
-                ein = gr.Textbox(label="EIN of organization", lines=1)
-            with gr.Column(scale=3):
-                with gr.Group():
-                    with gr.Row():
-                        with gr.Column():
-                            # street = gr.Textbox(label="Street address", lines=1)
-                            city = gr.Textbox(label="City", lines=1)
-                        with gr.Column():
-                            state = gr.Textbox(label="State/province", lines=1)
-                            postal = gr.Textbox(label="Postal code", lines=1)
-
-        search_button = gr.Button("Find organization", variant="primary")
-        org_info = gr.DataFrame(
-            label="Organizations",
-            type="array",
-            headers=["Candid ID", "Name", "Address", "Seal"],
-            col_count=(4, "fixed"),
-            datatype=["number", "str", "str", "markdown"],
-            wrap=True,
-            column_widths=["20%", "30%", "30%", "20%"]
-        )
-
-        if org_id_element is None:
-            org_id_element = gr.Textbox(label="Selected Candid entity ID", lines=1)
-
-        # pylint: disable=no-member
-        search_button.click(
-            fn=lambda name, ein, city, state, postal: lookup_organization(
-                name=name,
-                ein=ein,
-                # street=street,
-                city=city,
-                state=state,
-                postal=postal
-            ),
-            # inputs=[name, ein, street, city, state, postal],
-            inputs=[name, ein, city, state, postal],
-            outputs=[org_info, org_data],
-            api_name=False,
-            show_api=False
-        )
-
-        # pylint: disable=no-member
-        org_info.select(
-            fn=callback_to_state,
-            inputs=org_data,
-            outputs=[selected_org_data, org_id_element],
-            api_name=False,
-            show_api=False
-        )
-    return component, selected_org_data

tools/config.py

@@ -1,5 +0,0 @@
-import os
-CDS_API = {
-    'CDS_API_URL': os.getenv('CDS_API_URL'),
-    'CDS_API_KEY': os.getenv('CDS_API_KEY')
-}

tools/org_seach.py

@@ -1,188 +0,0 @@
-from typing import List
-import re
-
-from fuzzywuzzy import fuzz
-
-from langchain.output_parsers.openai_tools import JsonOutputToolsParser
-from langchain_openai.chat_models import ChatOpenAI
-from langchain_core.runnables import RunnableSequence
-from langchain_core.prompts import ChatPromptTemplate
-from pydantic import BaseModel
-
-try:
-    from common.org_search_component import OrgSearch
-except ImportError:
-    from ...common.org_search_component import OrgSearch
-
-search = OrgSearch()
-
-
-class OrganizationNames(BaseModel):
-    orgnames: List[str]
-
-
-def extract_org_links_from_chatbot(chatbot_output: str):
-    """
-    Extracts a list of organization names from the provided text.
-
-    Args:
-        chatbot_output (str):The chatbot output containing organization names and other content.
-
-    Returns:
-        list: A list of organization names extracted from the text.
-
-    Raises:
-        ValueError: If parsing fails or if an unexpected output format is received.
-    """
-    prompt = """Extract only the names of officially recognized organizations, foundations, and government entities from the text below. Do not include any entries that contain descriptions, regional identifiers, or explanations within parentheses or following the name. Strictly exclude databases, resources, crowdfunding platforms, and general terms. Provide the output only in the specified JSON format.
-
-    input text below:
-    
-        ```{chatbot_output}``
-
-    output format:
-    {{
-    'orgnames' : [list of organization names without any additional descriptions or identifiers]
-    }}
-
-    """
-
-    try:
-        parser = JsonOutputToolsParser()
-        llm = ChatOpenAI(model="gpt-4o").bind_tools([OrganizationNames])
-        prompt = ChatPromptTemplate.from_template(prompt)
-        chain = RunnableSequence(prompt, llm, parser)
-
-        # Run the chain with the input data
-        result = chain.invoke({"chatbot_output": chatbot_output})
-
-        # Extract the organization names from the output
-        output_list = result[0]["args"].get("orgnames", [])
-
-        # Validate output format
-        if not isinstance(output_list, list):
-            raise ValueError("Unexpected output format: 'orgnames' should be a list")
-
-        return output_list
-
-    except Exception as e:
-        # Log or print the error as needed for debugging
-        print(f"text does not have any organization: {e}")
-        return []
-
-
-def is_similar(name: str, list_of_dict: list, threshold: int = 80):
-    """
-    Returns True if `name` is similar to any names in `list_of_dict` based on a similarity threshold.
-    """
-    try:
-        for item in list_of_dict:
-            try:
-                # Attempt to calculate similarity score
-                similarity = fuzz.ratio(name.lower(), item["name"].lower())
-                if similarity >= threshold:
-                    return True
-            except KeyError:
-                # Handle cases where 'name' key might be missing in dictionary
-                print(f"KeyError: Missing 'name' key in dictionary item {item}")
-                continue
-            except AttributeError:
-                # Handle non-string name values in dictionary items
-                print(f"AttributeError: Non-string 'name' in dictionary item {item}")
-                continue
-    except TypeError as e:
-        # Handle cases where input types are incorrect
-        print(f"TypeError: {e}")
-        return False
-
-    return False
-
-
-def generate_org_link_dict(org_names_list: list):
-    """
-    Maps organization names to their Candid profile URLs if available.
-
-    For each organization in `output_list`, this function attempts to retrieve a matching profile
-    using `search_org`. If a similar name is found and a Candid entity ID is available, it constructs
-    a profile URL. If no ID or similar match is found, or if an error occurs, it assigns an empty string.
-
-    Args:
-        output_list (list): List of organization names (str) to retrieve Candid profile links for.
-
-    Returns:
-        dict: Dictionary with organization names as keys and Candid profile URLs or empty strings as values.
-
-    Example:
-        get_org_link(['New York-Presbyterian Hospital'])
-        # {'New York-Presbyterian Hospital': 'https://app.candid.org/profile/6915255'}
-    """
-    link_dict = {}
-
-    for org in org_names_list:
-        try:
-            # Attempt to retrieve organization data
-            response = search(org)
-
-            # Check if there is a valid response and if names are similar
-            if response and is_similar(org, response[0].get("names", "")):
-                # Try to get the Candid entity ID and construct the URL
-                candid_entity_id = response[0].get("candid_entity_id")
-                if candid_entity_id:
-                    link_dict[org] = (
-                        f"https://app.candid.org/profile/{candid_entity_id}"
-                    )
-                else:
-                    link_dict[org] = ""  # No ID found, set empty string
-            else:
-                link_dict[org] = ""  # No similar match found
-
-        except KeyError as e:
-            # Handle missing keys in the response dictionary
-            print(f"KeyError encountered for organization '{org}': {e}")
-            link_dict[org] = ""
-
-        except Exception as e:
-            # Catch any other unexpected errors
-
-            print(f"An error occurred for organization '{org}': {e}")
-            link_dict[org] = ""
-
-    return link_dict
-
-
-def embed_org_links_in_text(input_text: str, org_link_dict: dict):
-    """
-    Replaces organization names in `text` with links from `link_dict` and appends a Candid info message.
-
-    Args:
-        text (str): The text containing organization names.
-        link_dict (dict): Mapping of organization names to URLs.
-
-    Returns:
-        str: Updated text with linked organization names and an appended Candid message.
-    """
-    try:
-        for org_name, url in org_link_dict.items():
-            if url:  # Only proceed if the URL is not empty
-                regex_pattern = re.compile(re.escape(org_name))
-                input_text = regex_pattern.sub(
-                    repl=f"<a href={url} target='_blank' rel='noreferrer' class='candid-org-link'>{org_name}</a>",
-                    string=input_text
-                )
-
-        # Append Candid information message at the end
-        input_text += "<p class='candid-app-link'> Visit <a href=https://app.candid.org/ target='_blank' rel='noreferrer' class='candid-org-link'>Candid</a> to get nonprofit information you need.</p>"
-
-    except TypeError as e:
-        print(f"TypeError encountered: {e}")
-        return input_text
-
-    except re.error as e:
-        print(f"Regex error encountered for '{org_name}': {e}")
-        return input_text
-
-    except Exception as e:
-        print(f"Unexpected error: {e}")
-        return input_text
-
-    return input_text

tools/org_search_component.py

@@ -1,223 +0,0 @@
-from typing import List, Tuple, Optional, Any
-import os
-
-import gradio as gr
-
-try:
-    from base.api_base import BaseAPI
-except ImportError:
-    from .base.api_base import BaseAPI
-
-
-class OrgSearch(BaseAPI):
-
-    def __init__(self):
-        super().__init__(
-            url=f"{os.getenv('CDS_API_URL')}/v1/organization/search",
-            headers={"x-api-key": os.getenv('CDS_API_KEY')}
-        )
-
-    def __call__(self, name: str, **kwargs):
-        is_valid = False
-
-        payload = {
-            "names": [{
-                "value": name,
-                "type": "main"
-            }],
-            "status": "authorized"
-        }
-
-        if kwargs.get("ein"):
-            ein = kwargs.get("ein")
-            if "-" not in ein:
-                ein = f"{ein[:2]}-{ein[2:]}"
-            payload["ids"] = [{
-                "value": ein,
-                "type": "ein"
-            }]
-            is_valid = True
-
-        if kwargs.get("street") or kwargs.get("city") or kwargs.get("state") or kwargs.get("postal_code"):
-            payload["addresses"] = [{
-                "street1": kwargs.get("street") or "",
-                "city": kwargs.get("city") or "",
-                "state": kwargs.get("state") or "",
-                "postal_code": kwargs.get("postal_code") or ""
-            }]
-            is_valid = True
-
-        if not is_valid:
-            return None
-
-        result = self.post(payload=payload)
-        return result.get("payload", [])
-
-
-search_org = OrgSearch()
-
-
-def callback_to_state(event: gr.SelectData, state: gr.State) -> Tuple[List[Any], int]:
-    """Handles DataFrame `select` events. Updates the internal state for either the recipient or funder based on
-    selection. Also sends along the selected Candid entity ID to the proposal generation tab.
-
-    Parameters
-    ----------
-    event : gr.SelectData
-    state : gr.State
-
-    Returns
-    -------
-    Tuple[List[Any], int]
-        (Updated state, Candid entity ID)
-    """
-
-    row, _ = event.index
-
-    if len(state) == 0:
-        return [], None
-
-    # the state should be a nested list of lists
-    # if the state is a single list with non-list elements then we just want a pass-through
-    if all(isinstance(s, list) for s in state):
-        return state[row], state[row][0]
-    return state, state[0]
-
-
-def lookup_organization(
-    name: str,
-    ein: Optional[str] = None,
-    # street: Optional[str] = None,
-    city: Optional[str] = None,
-    state: Optional[str] = None,
-    postal: Optional[str] = None,
-) -> Tuple[List[List[str]], List[List[str]]]:
-    """Performs a simple search using the CDS organization search API. Results are sent to the DataFrame table and also
-    populate the state for the recipient information.
-
-    Parameters
-    ----------
-    name : str
-        Org name
-    ein : Optional[str], optional
-        Org EIN, by default None
-    street : Optional[str], optional
-        Street address, by default None
-    city : Optional[str], optional
-        Address city, by default None
-    state : Optional[str], optional
-        Address state, by default None
-    postal : Optional[str], optional
-        Address postal code, by default None
-
-    Returns
-    -------
-    Tuple[List[List[str]], List[List[str]]]
-        (recip data, recip data)
-
-    Raises
-    ------
-    gr.Error
-        Raised if not enough information was entered to run a search
-    gr.Error
-        Raised if no search results were returned
-    """
-
-    results = search_org(name=name, ein=ein, city=city, state=state, postal=postal)
-    if results is None:
-        raise gr.Error("You must provide a name, and either an EIN or an address.")
-    if not results:
-        raise gr.Error("No organizations could be found. Please refine the search criteria.")
-
-    data = []
-    for applicant_data in results:
-        address = applicant_data.get("addresses", [{}])[0].get("normalized")
-        seal = (applicant_data.get("current_seal", {}) or {}).get("image")
-
-        record = [
-            applicant_data.get('candid_entity_id'),
-            applicant_data.get('main_sort_name'),
-            address
-        ]
-
-        if seal:
-            record.append(f"![{applicant_data.get('current_seal', {}).get('seal')}]({seal})")
-        else:
-            record.append("")
-
-        data.append(record)
-    return data, data
-
-
-def render(org_id_element: Optional[gr.Blocks] = None) -> Tuple[gr.Blocks, gr.State]:
-    """Main blocks build and render function.
-
-    Parameters
-    ----------
-    org_id_element : Optional[gr.Blocks], optional
-        Callback Gradio element, by default None
-
-    Returns
-    -------
-    Tuple[gr.Blocks, gr.State]
-        (component, selected org state)
-    """
-
-    with gr.Blocks() as component:
-        org_data = gr.State([])
-        selected_org_data = gr.State([])
-
-        with gr.Row():
-            with gr.Column(scale=2):
-                name = gr.Textbox(label="Name of organization", lines=1)
-                ein = gr.Textbox(label="EIN of organization", lines=1)
-            with gr.Column(scale=3):
-                with gr.Group():
-                    with gr.Row():
-                        with gr.Column():
-                            # street = gr.Textbox(label="Street address", lines=1)
-                            city = gr.Textbox(label="City", lines=1)
-                        with gr.Column():
-                            state = gr.Textbox(label="State/province", lines=1)
-                            postal = gr.Textbox(label="Postal code", lines=1)
-
-        search_button = gr.Button("Find organization", variant="primary")
-        org_info = gr.DataFrame(
-            label="Organizations",
-            type="array",
-            headers=["Candid ID", "Name", "Address", "Seal"],
-            col_count=(4, "fixed"),
-            datatype=["number", "str", "str", "markdown"],
-            wrap=True,
-            column_widths=["20%", "30%", "30%", "20%"]
-        )
-
-        if org_id_element is None:
-            org_id_element = gr.Textbox(label="Selected Candid entity ID", lines=1)
-
-        # pylint: disable=no-member
-        search_button.click(
-            fn=lambda name, ein, city, state, postal: lookup_organization(
-                name=name,
-                ein=ein,
-                # street=street,
-                city=city,
-                state=state,
-                postal=postal
-            ),
-            # inputs=[name, ein, street, city, state, postal],
-            inputs=[name, ein, city, state, postal],
-            outputs=[org_info, org_data],
-            api_name=False,
-            show_api=False
-        )
-
-        # pylint: disable=no-member
-        org_info.select(
-            fn=callback_to_state,
-            inputs=org_data,
-            outputs=[selected_org_data, org_id_element],
-            api_name=False,
-            show_api=False
-        )
-    return component, selected_org_data

tools/question_reformulation.py

@@ -1,44 +0,0 @@
-from langchain_core.prompts import ChatPromptTemplate
-from langchain_core.output_parsers import StrOutputParser
-
-
-def reformulate_question_using_history(state, llm):
-    """
-    Transform the query to produce a better query with details from previous messages.
-
-    Args:
-        state (messages): The current state
-        llm: LLM to use
-    Returns:
-        dict: The updated state with re-phrased question and original user_input for UI
-    """
-    print("---REFORMULATE THE USER INPUT---")
-    messages = state["messages"]
-    question = messages[-1].content
-
-    if len(messages) > 1:
-        contextualize_q_system_prompt = """Given a chat history and the latest user input \
-        which might reference context in the chat history, formulate a standalone input \
-        which can be understood without the chat history.
-        Chat history:
-        \n ------- \n
-        {chat_history} 
-        \n ------- \n
-        User input:
-        \n ------- \n
-        {question} 
-        \n ------- \n
-        Do NOT answer the question, \
-        just reformulate it if needed and otherwise return it as is.
-        """
-
-        contextualize_q_prompt = ChatPromptTemplate([
-            ("system", contextualize_q_system_prompt),
-            ("human", question),
-        ])
-
-        rag_chain = contextualize_q_prompt | llm | StrOutputParser()
-        new_question = rag_chain.invoke({"chat_history": messages, "question": question})
-        print(f"user asked: '{question}', agent reformulated the question basing on the chat history: {new_question}")
-        return {"messages": [new_question], "user_input" : question}
-    return {"messages": [question], "user_input" : question}