Spaces:
Running
Running
| from typing import List, Literal, Tuple, TypedDict | |
| import os | |
| import gradio as gr | |
| try: | |
| from common import org_search_component as oss | |
| from formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions | |
| from services import RfpRecommend, RfpFeedback | |
| except ImportError: | |
| from ..common import org_search_component as oss | |
| from .formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions | |
| from .services import RfpRecommend, RfpFeedback | |
| api = RfpRecommend() | |
| reporting = RfpFeedback() | |
| class LoggedComponents(TypedDict): | |
| recommendations: gr.components.Component | |
| ratings: List[gr.components.Component] | |
| correctness: gr.components.Component | |
| sufficiency: gr.components.Component | |
| comments: gr.components.Component | |
| email: gr.components.Component | |
| def single_recommendation_response( | |
| item_number: int, | |
| rec_type: Literal["RFP"] = "RFP" | |
| ) -> gr.Radio: | |
| """Generates a radio button group to provide feedback for single recommendation indexed by `item_number`. | |
| Since the index values start from `0` we add `1` to indicate the ordinal value in the info text. | |
| Parameters | |
| ---------- | |
| item_number : int | |
| Recommendation index starting from 0 | |
| Returns | |
| ------- | |
| gr.Radio | |
| """ | |
| ordinal = str(item_number + 1) | |
| suffix = "th" | |
| if ordinal.endswith('1') and not ordinal.endswith('11'): | |
| suffix = "st" | |
| elif ordinal.endswith('2') and not ordinal.endswith('12'): | |
| suffix = "nd" | |
| elif ordinal.endswith('3') and not ordinal.endswith('13'): | |
| suffix = "rd" | |
| elem = gr.Radio( | |
| choices=[ | |
| "Not relevant and not useful", | |
| "Relevant but not useful", | |
| "Relevant and useful" | |
| ], | |
| label=f"Recommendation #{ordinal}", | |
| info=f"Evaluate the {ordinal}{suffix} {rec_type} (if applicable)" | |
| ) | |
| return elem | |
| def recommend_invoke(recipient: gr.State): | |
| response = api(candid_entity_id=recipient[0]) | |
| output = [] | |
| for rfp in (response.get("recommendations", []) or []): | |
| output.append([ | |
| rfp["funder_id"], | |
| rfp["funder_name"], | |
| rfp["funder_address"], | |
| rfp["amount"], | |
| ( | |
| f"<a href='{rfp['application_url']}' target='_blank' rel='noopener noreferrer'>" | |
| f"{rfp['application_url']}</a>" | |
| ), | |
| rfp["deadline"], | |
| rfp["description"], | |
| parse_pcs_descriptions(rfp["taxonomy"]), | |
| parse_geo_descriptions(rfp["area_served"]) | |
| ]) | |
| if len(output) == 0: | |
| raise gr.Error("No relevant RFPs were found, please try again in the future as new RFPs become available.") | |
| return output, process_reasons(response.get("meta", {}) or {}), response | |
| def build_recommender() -> Tuple[LoggedComponents, gr.Blocks]: | |
| with gr.Blocks(theme=gr.themes.Soft(), title="RFP recommendations") as demo: | |
| gr.Markdown( | |
| """ | |
| <h1>RFP recommendations</h1> | |
| <p>Receive recommendations for funding opportunities relevant to your work.</p> | |
| <p> | |
| Please read the <a | |
| href='https://info.candid.org/rfp-recommendation-guide' | |
| target="_blank" | |
| rel="noopener noreferrer" | |
| >guide</a> to get started. | |
| </p> | |
| <hr> | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| _, selected_org_state = oss.render() | |
| with gr.Row(): | |
| recommend = gr.Button("Get recommendations", scale=5, variant="primary") | |
| with gr.Row(): | |
| with gr.Accordion(label="Parameters used for recommendations", open=False): | |
| reasons_output = gr.DataFrame( | |
| col_count=3, | |
| headers=["Reason category", "Reason value", "Reason description"], | |
| interactive=False | |
| ) | |
| rec_outputs = gr.DataFrame( | |
| label="Recommended RFPs", | |
| type="array", | |
| headers=[ | |
| "Funder ID", "Name", "Address", | |
| "Amount", "Application URL", "Deadline", | |
| "Description", "About", "Where" | |
| ], | |
| col_count=(9, "fixed"), | |
| datatype=[ | |
| "number", "str", "str", | |
| "str", "markdown", "date", | |
| "str", "markdown", "markdown" | |
| ], | |
| wrap=True, | |
| max_height=1000, | |
| column_widths=[ | |
| "5%", "10%", "20%", | |
| "5", "15%", "5%", | |
| "10%", "10%", "20%" | |
| ], | |
| interactive=False | |
| ) | |
| recommendations_json = gr.JSON(label="Recommended RFPs JSON", visible=False) | |
| # pylint: disable=no-member | |
| recommend.click( | |
| fn=recommend_invoke, | |
| inputs=[selected_org_state], | |
| outputs=[rec_outputs, reasons_output, recommendations_json] | |
| ) | |
| logged = LoggedComponents( | |
| recommendations=recommendations_json | |
| ) | |
| return logged, demo | |
| def build_feedback( | |
| components: LoggedComponents, | |
| N: int = 5, | |
| rec_type: Literal["RFP"] = "RFP", | |
| ) -> gr.Blocks: | |
| def handle_feedback(*args): | |
| try: | |
| reporting( | |
| recommendation_data=args[0], | |
| ratings=list(args[1: (N + 1)]), | |
| info_is_correct=args[N + 1], | |
| info_is_sufficient=args[N + 2], | |
| comments=args[N + 3], | |
| email=args[N + 4] | |
| ) | |
| gr.Info("Thank you for providing feedback!") | |
| except Exception as ex: | |
| if hasattr(ex, "response"): | |
| error_msg = ex.response.json().get("response", {}).get("error") | |
| raise gr.Error(f"Failed to submit feedback: {error_msg}") | |
| raise gr.Error("Failed to submit feedback") | |
| feedback_components = [] | |
| with gr.Blocks(theme=gr.themes.Soft(), title="Candid AI demo") as demo: | |
| gr.Markdown(""" | |
| <h1>Help us improve this tool with your valuable feedback</h1> | |
| Please provide feedback for the recommendations on the previous tab. | |
| It is not required to provide feedback on all recommendations before submitting. | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| with gr.Group(): | |
| for i in range(N): | |
| f = single_recommendation_response(i, rec_type=rec_type) | |
| feedback_components.append(f) | |
| if "ratings" not in components: | |
| components["ratings"] = [f] | |
| else: | |
| components["ratings"].append(f) | |
| correctness = gr.Radio( | |
| choices=["True", "False"], | |
| label="Information is correct?", | |
| info="Are the displayed RFP details correct?" | |
| ) | |
| sufficiency = gr.Radio( | |
| choices=["True", "False"], | |
| label="Sufficient data?", | |
| info="Is enough RFP data available to provide meaningful recommendations?" | |
| ) | |
| comment = gr.Textbox(label="Additional comments (optional)", lines=4) | |
| email = gr.Textbox(label="Your email (optional)", lines=1) | |
| components["correctness"] = correctness | |
| components["sufficiency"] = sufficiency | |
| components["comments"] = comment | |
| components["email"] = email | |
| with gr.Row(): | |
| submit = gr.Button("Submit Feedback", variant='primary', scale=5) | |
| gr.ClearButton(components=feedback_components, variant="stop") | |
| # pylint: disable=no-member | |
| submit.click( | |
| fn=handle_feedback, | |
| inputs=[comp for k, cl in components.items() for comp in (cl if isinstance(cl, list) else [cl])], | |
| outputs=None, | |
| show_api=False, | |
| api_name=False, | |
| preprocess=False, | |
| ) | |
| return demo | |
| def build_demo(): | |
| logger, recommender = build_recommender() | |
| feedback = build_feedback(logger) | |
| return gr.TabbedInterface( | |
| interface_list=[recommender, feedback], | |
| tab_names=["RFP recommendations", "Feedback"], | |
| title="Candid's RFP recommendation engine", | |
| theme=gr.themes.Soft() | |
| ) | |
| if __name__ == '__main__': | |
| app = build_demo() | |
| app.queue(max_size=5).launch( | |
| show_api=False, | |
| auth=[ | |
| (os.getenv("APP_USERNAME"), os.getenv("APP_PASSWORD")), | |
| (os.getenv("APP_PUBLIC_USERNAME"), os.getenv("APP_PUBLIC_PASSWORD")), | |
| ], | |
| auth_message="Login to Candid's letter of intent demo", | |
| ssr_mode=False | |
| ) | |