Spaces:
Running
Running
| 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 |