Spaces:
Running
Running
File size: 8,512 Bytes
c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 c751e97 bea5044 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 |
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 |