Delete ask_candid/tools/recommendation.py

ask_candid/tools/recommendation.py

@@ -1,238 +0,0 @@
-import os
-
-from openai import OpenAI
-from langchain_core.prompts import ChatPromptTemplate
-import requests
-
-from ask_candid.agents.schema import AgentState, Context
-from ask_candid.base.api_base import BaseAPI
-
-class AutocodingAPI(BaseAPI):
-    def __init__(self):
-        super().__init__(
-            url=os.getenv("AUTOCODING_API_URL"),
-            headers={
-                'x-api-key': os.getenv("AUTOCODING_API_KEY"),
-                'Content-Type': 'application/json'
-            }
-        )
-
-    def __call__(self, text: str, taxonomy: str = 'pcs-v3'):
-        params = {
-            'text': text,
-            'taxonomy': taxonomy
-        }
-        return self.get(**params)
-
-class GeoAPI(BaseAPI):
-    def __init__(self):
-        super().__init__(
-            url=os.getenv("GEO_API_URL"),
-            headers={
-                'x-api-key': os.getenv("GEO_API_KEY"),
-                'Content-Type': 'application/json'
-            }
-        )
-
-    def __call__(self, text: str):
-        payload = {
-            'text': text
-        }
-        return self.post(payload=payload)
-    
-class EntitiesAPI(BaseAPI):
-    def __init__(self):
-        super().__init__(
-            url=f'{os.getenv("DOCUMENT_API_URL")}/entities',
-            headers={
-                'x-api-key': os.getenv("DOCUMENT_API_KEY"),
-                'Content-Type': 'application/json'
-            }
-        )
-
-    def __call__(self, text: str):
-        payload = {
-            'text': text
-        }
-        return self.post(payload=payload)
-    
-
-
-class FunderRecommendationAPI(BaseAPI):
-    def __init__(self):
-        super().__init__(
-            url=os.getenv("FUNDER_REC_API_URL"),
-            headers={"x-api-key": os.getenv("FUNDER_REC_API_KEY")}
-        )
-
-    def __call__(self, subjects, populations, geos):
-        params = {
-            "subjects": subjects,
-            "populations": populations,
-            "geos": geos
-        }
-        return self.get(**params)
-
-class RFPRecommendationAPI(BaseAPI):
-    def __init__(self):
-        super().__init__(
-            url= f'{os.getenv("FUNDER_REC_API_URL")}/rfp',  
-            headers={"x-api-key": os.getenv("FUNDER_REC_API_KEY")}
-        )
-
-    def __call__(self, org_id, subjects, populations, geos):
-        params = {
-            "candid_entity_id": org_id,
-            "subjects": subjects,
-            "populations": populations,
-            "geos": geos
-        }
-        return self.get(**params) 
-
-
-
-
-def detect_intent_with_llm(state: AgentState, llm) -> AgentState:
-    """Detect query intent (which type of recommendation) and update the state using the specified LLM."""
-    print("running detect intent")
-    
-    query = state["messages"][-1].content
-
-    prompt_template = ChatPromptTemplate.from_messages(
-        [
-            ("system", """
-            Please classify the following query by stating ONLY the category name: 'none', 'funder', or 'rfp'. 
-            Please answer WITHOUT any reasoning.
-            - 'none': The query does not ask for any recommendations.
-            - 'funder': The query asks for recommendations about funders, such as foundations or donors.
-            - 'rfp': The query asks for recommendations about specific Requests for Proposals (RFPs).
-            
-            Consider:
-            - If the query seeks broad, long-term funding sources or organizations, classify as 'funder'.
-            - If the query seeks specific, time-bound funding opportunities with a deadline, classify as 'rfp'.
-            - If the query does not seek any recommendations, classify as 'none'.
-            
-            Query: """),
-            ("human", f"{query}")
-        ]
-    )
-
-    chain = prompt_template | llm
-    response = chain.invoke({"query": query})
-
-    intent = response.content.strip().lower()
-    state["intent"] = intent.strip("'").strip('"')  # Remove extra quotes if necessary
-    print(state["intent"])
-    return state
-
-
-def determine_context(state: AgentState) -> AgentState:
-    print("running context")
-    query = state["messages"][-1].content
-
-    autocoding_api = AutocodingAPI()
-    entities_api = EntitiesAPI()
-
-    subject_codes, population_codes, geo_ids = [], [], []
-
-    try:
-        autocoding_response = autocoding_api(text=query)
-        returned_pcs = autocoding_response.get("data", {})
-        population_codes = [item['full_code'] for item in returned_pcs.get("population", [])]
-        subject_codes = [item['full_code'] for item in returned_pcs.get("subject", [])]
-    except Exception as e:
-        print(f"Failed to retrieve autocoding data: {e}")
-
-    try:
-        geo_response = entities_api(text=query)
-        entities = geo_response.get('entities', [])
-        geo_ids = [match['geonames_id'] for entity in entities if entity['type'] == 'geo' and 'match' in entity
-                   for match in entity['match'] if 'geonames_id' in match]
-    except Exception as e:
-        print(f"Failed to retrieve geographic data: {e}")
-
-    state["context"] = Context(
-        subject=subject_codes,
-        population=population_codes,
-        geography=geo_ids
-    )
-    return state
-
-
-
-def format_recommendations(intent, data):
-    if 'recommendations' not in data:
-        return "No recommendations available."
-
-    recommendations = data['recommendations']  
-    if not recommendations:
-        return "No recommendations found."
-
-    recommendation_texts = []
-    if intent == "funder":
-        for rec in recommendations:
-            main_sort_name = rec['funder_data']['main_sort_name']
-            profile_url = f"https://app.candid.org/profile/{rec['funder_id']}"
-            recommendation_texts.append(f"{main_sort_name} - Profile: {profile_url}")
-    elif intent == "rfp":
-        for rec in recommendations:
-            title = rec.get('title', 'N/A')
-            funder_name = rec.get('funder_name', 'N/A')
-            amount = rec.get('amount', 'Not specified')
-            description = rec.get('description', 'No description available')
-            deadline = rec.get('deadline', 'No deadline provided')
-            application_url = rec.get('application_url', 'No URL available')
-            text = (f"Title: {title}\n"
-                    f"Funder: {funder_name}\n"
-                    f"Amount: {amount}\n"
-                    f"Description: {description}\n"
-                    f"Deadline: {deadline}\n"
-                    f"Application URL: {application_url}\n")
-            recommendation_texts.append(text)
-    else:
-        return "Only funder recommendation or RFP recommendation are supported."
-
-    return "\n".join(recommendation_texts)
-
-
-
-def make_recommendation(state: AgentState) -> AgentState:
-    print("running recommendation")
-    org_id = "6908122"  # Example organization ID (Candid)
-    funder_or_rfp = state["intent"]
-
-    contexts = state["context"]
-    subject_codes = ",".join(contexts.get("subject", []))
-    population_codes = ",".join(contexts.get("population", []))
-    geo_ids = ",".join([str(geo) for geo in contexts.get("geography", [])])
-
-    recommendation_display_text = ""
-
-    try:
-        if funder_or_rfp == "funder":
-            funder_api = FunderRecommendationAPI()
-            recommendations = funder_api(subject_codes, population_codes, geo_ids)
-        elif funder_or_rfp == "rfp":
-            rfp_api = RFPRecommendationAPI()
-            recommendations = rfp_api(org_id, subject_codes, population_codes, geo_ids)
-        else:
-            recommendation_display_text = "Unknown intent. Intent 'funder' or 'rfp' expected."
-            state["recommendation"] = recommendation_display_text
-            return state
-
-        if recommendations:
-            recommendation_display_text = format_recommendations(funder_or_rfp, recommendations)
-        else:
-            recommendation_display_text = "No recommendations were found for your query. Please try refining your search criteria."
-
-    except requests.exceptions.HTTPError as e:
-        # Handle HTTP errors raised by raise_for_status()
-        print(f"HTTP error occurred: {e.response.status_code} - {e.response.reason}")
-        recommendation_display_text = "HTTP error occurred, please report this to datascience@candid.org"
-    except Exception as e:
-        # Catch-all for any other exceptions that are not HTTP errors
-        print(f"An unexpected error occurred: {str(e)}")
-        recommendation_display_text = "Unexpected error occurred, please report this to datascience@candid.org"
-
-    state["recommendation"] = recommendation_display_text
-    return state