brainsqueeze's picture
Fixing postprocessing of editorial AI response
a72a624 verified
raw
history blame
7.23 kB
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