jaminy/SocialScience

1"""2Benefits Eligibility System3"""4 5import os6import json7import uuid8import re9from typing import Dict, List, Tuple10import gradio as gr11from openai import OpenAI12from dotenv import load_dotenv13from langchain_core.tools import tool14 15load_dotenv()16client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))17session_states = {}18 19# =========================================================================20# PLAIN FUNCTIONS (Used by Code)21# =========================================================================22 23def _extract_user_information(text: str) -> dict:24    """Extract structured information"""25    profile = {26        "income": None,27        "location": None,28        "ward": None,29        "family_size": None,30        "children_ages": []31    }32    33    text_lower = text.lower()34    35    # Income36    if match := re.search(r'\$?(\d{1,3}(?:,\d{3})*)\s*(?:per|/)?\s*(?:year|annually)', text_lower):37        profile['income'] = float(match.group(1).replace(',', ''))38    elif match := re.search(r'\$?(\d{1,3}(?:,\d{3})*)\s*(?:per|/)?\s*month', text_lower):39        profile['income'] = float(match.group(1).replace(',', '')) * 1240    elif match := re.search(r'(?:earning|making)\s*\$?(\d{1,3}(?:,\d{3})*)', text_lower):41        profile['income'] = float(match.group(1).replace(',', ''))42    43    # Location44    if any(x in text_lower for x in ['washington', 'dc', 'd.c.']):45        profile['location'] = 'Washington DC'46        if ward_match := re.search(r'ward\s*(\d)', text_lower):47            profile['ward'] = int(ward_match.group(1))48    49    # Family size50    if match := re.search(r'(\d+)\s*(?:kids|children)', text_lower):51        profile['family_size'] = int(match.group(1)) + 152    elif match := re.search(r'family\s*of\s*(\d+)', text_lower):53        profile['family_size'] = int(match.group(1))54    55    # Children ages56    for match in re.finditer(r'(?:ages?|aged)\s*(\d+)', text_lower):57        age = int(match.group(1))58        if 0 <= age <= 18:59            profile['children_ages'].append(age)60    61    return profile62 63 64def _calculate_snap_eligibility(income: float, family_size: int) -> dict:65    """Calculate SNAP eligibility"""66    FPL_BASE, FPL_INCREMENT = 15060, 538067    68    fpl = FPL_BASE + (family_size - 1) * FPL_INCREMENT69    threshold = fpl * 1.3070    eligible = income <= threshold71    72    benefits = {1: 291, 2: 535, 3: 766, 4: 973, 5: 1155}73    benefit = benefits.get(family_size, 1155)74    75    return {76        "program": "SNAP",77        "eligible": eligible,78        "estimated_monthly_benefit": round(benefit * 0.7) if eligible else 0,79        "income_threshold": round(threshold),80        "explanation": f"Income ${income:,.0f} vs ${threshold:,.0f} threshold (130% FPL)"81    }82 83 84def _calculate_medicaid_eligibility(income: float, family_size: int) -> dict:85    """Calculate Medicaid eligibility"""86    FPL_BASE, FPL_INCREMENT = 15060, 538087    88    fpl = FPL_BASE + (family_size - 1) * FPL_INCREMENT89    threshold = fpl * 2.1690    eligible = income <= threshold91    92    return {93        "program": "Medicaid",94        "eligible": eligible,95        "income_threshold": round(threshold),96        "explanation": f"DC Medicaid at 216% FPL: ${threshold:,.0f}"97    }98 99 100# =========================================================================101# @tool DECORATED VERSIONS (For AI Discovery)102# =========================================================================103 104@tool105def extract_user_information(text: str) -> dict:106    """Extract structured information from user's natural language.107    108    Args:109        text: User's description110    Returns:111        Structured profile dictionary112    """113    return _extract_user_information(text)114 115 116@tool117def calculate_snap_eligibility(income: float, family_size: int) -> dict:118    """Calculate SNAP eligibility using 130% FPL threshold.119    120    Args:121        income: Annual income in dollars122        family_size: Number of people in household123    Returns:124        Eligibility result with estimated benefit125    """126    return _calculate_snap_eligibility(income, family_size)127 128 129@tool130def calculate_medicaid_eligibility(income: float, family_size: int) -> dict:131    """Calculate Medicaid eligibility. DC uses 216% FPL.132    133    Args:134        income: Annual income135        family_size: Household size136    Returns:137        Eligibility result138    """139    return _calculate_medicaid_eligibility(income, family_size)140 141 142# =========================================================================143# AGENTS144# =========================================================================145 146class IntakeAgent:147    def process(self, user_query: str) -> Dict:148        # Use plain functions149        profile = _extract_user_information(user_query)150        151        # OpenAI call152        response = client.chat.completions.create(153            model="gpt-4-turbo-preview",154            temperature=0.7,155            messages=[156                {"role": "system", "content": "You are an empathetic intake specialist."},157                {"role": "user", "content": f"User: {user_query}\n\nProfile: {json.dumps(profile)}\n\nRespond warmly. Ask ONE follow-up if missing income, location, or family_size."}158            ]159        )160        161        return {162            "profile": profile,163            "response": response.choices[0].message.content,164            "ready": profile.get("income") and profile.get("family_size")165        }166 167 168class EligibilityAgent:169    def process(self, profile: Dict) -> Dict:170        results = []171        172        income = profile.get("income", 0)173        family_size = profile.get("family_size", 1)174        175        if income and family_size:176            results.append(_calculate_snap_eligibility(income, family_size))177            results.append(_calculate_medicaid_eligibility(income, family_size))178        179        response = client.chat.completions.create(180            model="gpt-4-turbo-preview",181            temperature=0.3,182            messages=[183                {"role": "system", "content": "Summarize eligibility encouragingly."},184                {"role": "user", "content": f"Results:\n{json.dumps(results, indent=2)}\n\nFocus on qualified programs."}185            ]186        )187        188        qualified = [r["program"] for r in results if r.get("eligible")]189        190        return {191            "qualified": qualified,192            "response": response.choices[0].message.content193        }194 195 196class ApplicationAgent:197    def process(self, qualified: List[str], profile: Dict) -> Dict:198        response = client.chat.completions.create(199            model="gpt-4-turbo-preview",200            temperature=0.5,201            messages=[202                {"role": "system", "content": "Create application guidance."},203                {"role": "user", "content": f"Programs: {qualified}\n\nCreate action plan:\n**THIS WEEK**: Apply online at dc.gov/access\nPhone: (202) 727-5355\nDocuments: Photo ID, Proof of income"}204            ]205        )206        207        return {"response": response.choices[0].message.content}208 209 210# =========================================================================211# MAIN SYSTEM212# =========================================================================213 214class BenefitsSystem:215    def __init__(self):216        self.intake = IntakeAgent()217        self.eligibility = EligibilityAgent()218        self.application = ApplicationAgent()219    220    def process(self, session_id: str, message: str) -> Tuple[str, Dict]:221        if session_id not in session_states:222            session_states[session_id] = {223                "stage": "intake",224                "intake_done": False,225                "eligibility_done": False226            }227        228        state = session_states[session_id]229        responses = []230        231        if not state["intake_done"]:232            result = self.intake.process(message)233            responses.append(("** Intake**", result["response"]))234            state["profile"] = result["profile"]235            236            if result["ready"]:237                state["intake_done"] = True238            else:239                return result["response"], state240        241        if state["intake_done"] and not state["eligibility_done"]:242            result = self.eligibility.process(state["profile"])243            responses.append(("** Eligibility**", result["response"]))244            state["qualified"] = result["qualified"]245            state["eligibility_done"] = True246        247        if state["eligibility_done"]:248            result = self.application.process(state["qualified"], state["profile"])249            responses.append(("** Application**", result["response"]))250            state["stage"] = "complete"251        252        final = "\n\n---\n\n".join([f"{t}\n\n{c}" for t, c in responses]) if len(responses) > 1 else responses[0][1]253        return final, state254 255 256# =========================================================================257# GRADIO UI - FIXED: Proper message format for chatbot258# =========================================================================259 260system = BenefitsSystem()261 262with gr.Blocks(title="Benefits System") as demo:263    gr.Markdown("#  Benefits Eligibility System\n\n**With @tool Decorators**")264    265    with gr.Row():266        with gr.Column(scale=2):267            # IMPORTANT: Gradio Chatbot expects list of tuples: [(user_msg, bot_msg), ...]268            chatbot = gr.Chatbot(269                label="Conversation",270                height=600271            )272            273            with gr.Row():274                msg = gr.Textbox(275                    label="Your situation",276                    placeholder="Single mom, 2 kids ages 4 and 7, $32k/year in DC",277                    lines=3,278                    scale=4279                )280                submit = gr.Button("Send", variant="primary", scale=1)281            clear = gr.Button(" New")282        283        with gr.Column(scale=1):284            session_id_display = gr.Textbox(label="Session", value="Not started", interactive=False)285            stage = gr.Textbox(label="Stage", value="Intake", interactive=False)286    287    session_state = gr.State(value=None)288    289    def process_message(message, chat_history, sess_id):290        """Process user message - returns properly formatted chat history"""291        if not message.strip():292            return "", chat_history, sess_id, "Intake"293        294        if not sess_id:295            sess_id = str(uuid.uuid4())296        297        try:298            # Process through system299            response, state = system.process(sess_id, message)300            301            # IMPORTANT: Gradio 4+ expects messages as dictionaries with 'role' and 'content'302            new_history = chat_history + [303                {"role": "user", "content": message},304                {"role": "assistant", "content": response}305            ]306            307            return "", new_history, sess_id, state["stage"].title()308            309        except Exception as e:310            error_msg = f"⚠️ Error: {str(e)}"311            new_history = chat_history + [312                {"role": "user", "content": message},313                {"role": "assistant", "content": error_msg}314            ]315            return "", new_history, sess_id, "Error"316    317    def clear_chat():318        """Clear conversation"""319        new_sess_id = str(uuid.uuid4())320        return [], new_sess_id, "Intake"321    322    # Event handlers323    msg.submit(324        process_message,325        inputs=[msg, chatbot, session_state],326        outputs=[msg, chatbot, session_state, stage]327    )328    329    submit.click(330        process_message,331        inputs=[msg, chatbot, session_state],332        outputs=[msg, chatbot, session_state, stage]333    )334    335    clear.click(336        clear_chat,337        outputs=[chatbot, session_state, stage]338    )339    340    session_state.change(341        lambda x: x[:8] + "..." if x else "Not started",342        inputs=[session_state],343        outputs=[session_id_display]344    )345    346    gr.Markdown("**Disclaimer**: AI guidance. Verify at [benefits.gov](https://benefits.gov)")347 348if __name__ == "__main__":349    if not os.getenv("OPENAI_API_KEY"):350        print("⚠️ OPENAI_API_KEY not found!")351    else:352        print(" Starting Benefits Eligibility System...")353        demo.launch(share=False, server_name="0.0.0.0")