brainsqueeze's picture
Fixing postprocessing of editorial AI response
a72a624 verified
from typing import List, Dict, Tuple, Union, Optional, Any
import os
import gradio as gr
import requests
import backoff
class OrganizationNotFound(Exception):
"No organization(s) found"
@backoff.on_exception(
backoff.expo,
(requests.exceptions.Timeout, requests.exceptions.ConnectionError),
max_tries=5
)
def _get_json(route: str, **request_kwargs) -> Dict[str, Any] | str:
r = requests.get(
url=f"{os.getenv('LOI_API_URL')}/{route}",
params=request_kwargs,
headers={"x-api-key": os.getenv("LOI_API_KEY")},
timeout=30
)
r.raise_for_status()
return r.json().get("response")
@backoff.on_exception(
backoff.expo,
(requests.exceptions.Timeout, requests.exceptions.ConnectionError),
max_tries=5
)
def _post_json(route: str, *, payload: Dict[str, Any]) -> Dict[str, Any] | str | int:
r = requests.post(
url=f"{os.getenv('LOI_API_URL')}/{route}",
json=payload,
headers={"x-api-key": os.getenv("LOI_API_KEY")},
timeout=30
)
r.raise_for_status()
return r.json().get("response")
def organization_pair_autofill(
recipient_name: str,
recipient_ein: str,
funder_name: str,
funder_ein: str
):
recip_match = _get_json("/organization/search", name=recipient_name, ein=recipient_ein)
if len(recip_match or []) == 0:
# raise OrganizationNotFound()
raise gr.Error("No matching recipient could be found")
gr.Info(f"{recipient_name} found, auto-filling fields...")
funder_match = _get_json("/organization/search", name=funder_name, ein=funder_ein)
if len(funder_match or []) == 0:
# raise OrganizationNotFound()
raise gr.Error("No matching funder could be found")
gr.Info(f"{funder_name} found, auto-filling fields...")
data = _get_json(
"/organization/autofill",
recipient_candid_entity_id=recip_match[0]["candid_entity_id"],
funder_candid_entity_id=funder_match[0]["candid_entity_id"],
)
return (
data.get("recipient_data", {}).get("projects_text"),
data.get("recipient_data", {}).get("capacity_text"),
data.get("recipient_data", {}).get("contact_text"),
data.get("recipient_data", {}).get("data_text"), # accomplishments
data.get("recipient_data", {}).get("mission_statement_text"),
data.get("funder_data", {}).get("mission_statement_text"),
data.get("funding_history_text"),
data.get("recipient_data", {}).get("org_data"),
recip_match[0]["candid_entity_id"],
data.get("funder_data", {}).get("org_data"),
funder_match[0]["candid_entity_id"],
)
def cost_estimator(
recipient_candid_entity_id: Union[int, str],
recipient_data: Dict[str, Any],
funder_data: Dict[str, Any],
program_desc: Optional[str] = None
) -> str:
estimate: str = _post_json(
"budget",
payload={
"recipient_candid_entity_id": recipient_candid_entity_id,
"program_description": program_desc,
"recipient_data": recipient_data,
"funder_data": funder_data,
}
)
return estimate
def identify_vague_statements(text: str) -> Tuple[List[str], List[str], List[str]]:
data = _get_json("/editorialai/vaguestatement", input_section=text)
statements, alternatives, reasons = [], [], []
for record in data:
statements.append(record["vague_statement"])
alternatives.append(record["alternative_text"])
reasons.append(record["reason"])
return statements, alternatives, reasons
def help_mission_statement(recipient_name: str, recipient_mission_info: str):
return _get_json("/writerhelper/missionstatement", info=recipient_mission_info, org_name=recipient_name)
def draft_letter(
recipient_name: str, funder_name: str, projects: str, amount: str,
recipient_mission_statement: str = "", funder_mission_statement: str = "",
project_name: str = "", project_purpose: str = "",
prior_contact: str = "", connection: str = "",
capacity: str = "", path_to_solution: str = "", recent_accomplishments: str = "",
recipient_history: str = "", mutual_history: str = "",
geo_pop_targets: str = "", project_data_stats: str = "",
desired_objectives: str = "", major_activities: str = "", key_staff: str = "", standout_features: str = "",
success_metric: str = "",
other_funding_sources: str = "",
contact_information: str = ""
):
gr.Info("Writing the letter, please scroll to the top of the page.")
opening: str = _post_json(
"/writer/opening",
payload=dict(
funder_name=funder_name,
recipient_name=recipient_name,
project_name=project_name,
project_purpose=project_purpose,
amount=amount,
prior_contact=prior_contact,
connection=connection
)
)
recipient_description: str = _post_json(
"/writer/org",
payload=dict(
opening=opening,
recipient_mission_statement=recipient_mission_statement,
capacity=capacity,
history=recipient_history,
path=path_to_solution,
accomplishment=recent_accomplishments
)
)
need_statement: str = _post_json(
"/writer/need",
payload=dict(
recipient_desc=recipient_description,
target=geo_pop_targets,
data=project_data_stats,
funder_mission_statement=funder_mission_statement
)
)
project_description: str = _post_json(
"/writer/project",
payload=dict(
need=need_statement,
projects=projects,
desired_objectives=desired_objectives,
major_activities=major_activities,
key_staff=key_staff,
stand_out=standout_features,
success=success_metric
)
)
funding_request: str = _post_json(
"/writer/fund",
payload=dict(
project_desc=project_description,
amount=amount,
funding_history=mutual_history,
other_funding=other_funding_sources
)
)
conclusion: str = _post_json(
"/writer/conclusion",
payload=dict(
funding_request=funding_request,
project_desc=project_description,
follow_up=contact_information
)
)
return (
opening,
recipient_description,
need_statement,
project_description,
funding_request,
conclusion
)
def send_feedback(
context: Dict[str, Any],
letter: Dict[str, Any],
relevance: int,
coherence: int,
fluency: int,
overall_quality: int,
comments: Optional[str] = None,
email: Optional[str] = None
) -> int:
count = _post_json(
"feedback",
payload={
"context": context,
"letter": letter,
"relevance": relevance,
"coherence": coherence,
"fluency": fluency,
"overall_quality": overall_quality,
"comments": comments,
"email": email
}
)
return count