From 60f12d13750e0cfd71b605ff892848fc1c663c19 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 00:56:50 +0530 Subject: [PATCH 1/6] Add AI agent tutorials for Google ADK --- ai_agent_tutorials/ai_google_adk/.env | 1 + ai_agent_tutorials/ai_google_adk/README.md | 1 + .../ai_google_adk/google_adk.py | 456 ++++++++++++++++++ .../ai_google_adk/requirements.txt | 8 + 4 files changed, 466 insertions(+) create mode 100644 ai_agent_tutorials/ai_google_adk/.env create mode 100644 ai_agent_tutorials/ai_google_adk/README.md create mode 100644 ai_agent_tutorials/ai_google_adk/google_adk.py create mode 100644 ai_agent_tutorials/ai_google_adk/requirements.txt diff --git a/ai_agent_tutorials/ai_google_adk/.env b/ai_agent_tutorials/ai_google_adk/.env new file mode 100644 index 0000000..b8d7980 --- /dev/null +++ b/ai_agent_tutorials/ai_google_adk/.env @@ -0,0 +1 @@ +GOOGLE_API_KEY= \ No newline at end of file diff --git a/ai_agent_tutorials/ai_google_adk/README.md b/ai_agent_tutorials/ai_google_adk/README.md new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/ai_agent_tutorials/ai_google_adk/README.md @@ -0,0 +1 @@ + diff --git a/ai_agent_tutorials/ai_google_adk/google_adk.py b/ai_agent_tutorials/ai_google_adk/google_adk.py new file mode 100644 index 0000000..b12ad18 --- /dev/null +++ b/ai_agent_tutorials/ai_google_adk/google_adk.py @@ -0,0 +1,456 @@ +import streamlit as st +import pandas as pd +import plotly.express as px +import plotly.graph_objects as go +from typing import Dict, List, Optional, Tuple, Any, AsyncGenerator +import os +import asyncio +from datetime import datetime +from dotenv import load_dotenv + +from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent +from google.adk.agents.invocation_context import InvocationContext +from google.adk.events import Event, EventActions +from google.adk.sessions import InMemorySessionService, Session + +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + +# Get API key from environment +GEMINI_API_KEY = os.getenv("GOOGLE_API_KEY") +if not GEMINI_API_KEY: + raise ValueError("GOOGLE_API_KEY environment variable not set") + +class FinanceAdvisorSystem: + def __init__(self): + """Initialize the finance advisor system with specialized agents""" + # Budget Analysis Agent + self.budget_analysis_agent = LlmAgent( + name="BudgetAnalysisAgent", + model="gemini-2.0-flash-exp", + description="Analyzes financial data to categorize spending patterns and recommend budget improvements", + instruction="""You are a Budget Analysis Agent specialized in reviewing financial transactions and expenses. + +Your tasks: +1. Analyze income, transactions, and expenses +2. Categorize spending into logical groups +3. Identify spending patterns and trends +4. Suggest specific areas where spending could be reduced +5. Provide actionable recommendations with potential savings amounts + +Consider: +- Number of dependants when evaluating household expenses +- Typical spending ratios for the income level +- Essential vs discretionary spending +- Seasonal spending patterns if data spans multiple months""", + output_key="budget_analysis" + ) + + # Savings Strategy Agent + self.savings_strategy_agent = LlmAgent( + name="SavingsStrategyAgent", + model="gemini-2.0-flash-exp", + description="Recommends optimal savings strategies based on income, expenses, and financial goals", + instruction="""You are a Savings Strategy Agent specialized in creating personalized savings plans. + +Your tasks: +1. Recommend savings strategies based on income and expenses +2. Calculate optimal emergency fund size based on expenses and dependants +3. Suggest appropriate savings allocation across different purposes +4. Recommend practical automation techniques for saving consistently + +Consider: +- Risk factors based on job stability and dependants +- Balancing immediate needs with long-term financial health +- Progressive savings rates as discretionary income increases +- Multiple savings goals (emergency, retirement, specific purchases)""", + output_key="savings_strategy" + ) + + # Debt Reduction Agent + self.debt_reduction_agent = LlmAgent( + name="DebtReductionAgent", + model="gemini-2.0-flash-exp", + description="Creates optimized debt payoff plans to minimize interest paid and time to debt freedom", + instruction="""You are a Debt Reduction Agent specialized in creating debt payoff strategies. + +Your tasks: +1. Analyze debts by interest rate, balance, and minimum payments +2. Create prioritized debt payoff plans (avalanche and snowball methods) +3. Calculate total interest paid and time to debt freedom for each approach +4. Suggest debt consolidation or refinancing opportunities when beneficial +5. Provide specific recommendations to accelerate debt payoff + +Consider: +- Cash flow and budget constraints from the budget analysis +- Psychological factors (quick wins vs mathematical optimization) +- Interest savings potential +- Credit utilization and credit score impact""", + output_key="debt_reduction" + ) + + # Coordinator Agent - Orchestrates the specialized agents + self.coordinator_agent = SequentialAgent( + name="FinanceCoordinatorAgent", + description="Coordinates specialized finance agents to provide comprehensive financial advice", + sub_agents=[ + self.budget_analysis_agent, + self.savings_strategy_agent, + self.debt_reduction_agent + ] + ) + + async def analyze_finances(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: + """Process financial data through the agent system and return comprehensive analysis""" + # Prepare the session context + session = Session() + + # Store financial data in session state for agents to access + session.state.update({ + "monthly_income": financial_data.get("monthly_income", 0), + "dependants": financial_data.get("dependants", 0), + "transactions": financial_data.get("transactions", []), + "manual_expenses": financial_data.get("manual_expenses", {}), + "debts": financial_data.get("debts", []) + }) + + # Preprocess transaction data if available + if financial_data.get("transactions"): + self._preprocess_transactions(session) + + # Initialize preprocessing for manual expenses if provided + if financial_data.get("manual_expenses"): + self._preprocess_manual_expenses(session) + + # Set up the invocation context + context = InvocationContext(session=session, user_input="Analyze financial data") + + # Run the coordinator agent which will execute all sub-agents in sequence + async for event in self.coordinator_agent.run(context): + # We could process events here if needed + pass + + # Collect results from session state + results = { + "budget_analysis": session.state.get("budget_analysis", {}), + "savings_strategy": session.state.get("savings_strategy", {}), + "debt_reduction": session.state.get("debt_reduction", {}) + } + + return results + + def _preprocess_transactions(self, session): + """Preprocess transaction data for easier analysis by the agents""" + transactions = session.state.get("transactions", []) + + if not transactions: + return + + # Convert list of transactions to DataFrame for analysis + df = pd.DataFrame(transactions) + + # Basic preprocessing + if 'Date' in df.columns: + df['Date'] = pd.to_datetime(df['Date']) + df['Month'] = df['Date'].dt.month + df['Year'] = df['Date'].dt.year + + # Calculate spending by category + if 'Category' in df.columns and 'Amount' in df.columns: + category_spending = df.groupby('Category')['Amount'].sum().to_dict() + session.state["category_spending"] = category_spending + + # Total spending + total_spending = df['Amount'].sum() + session.state["total_spending"] = total_spending + + def _preprocess_manual_expenses(self, session): + """Process manually entered expenses""" + manual_expenses = session.state.get("manual_expenses", {}) + + if not manual_expenses: + return + + # Calculate total spending from manual entries + total_manual_spending = sum(manual_expenses.values()) + session.state["total_manual_spending"] = total_manual_spending + + # Store categorized spending directly + session.state["manual_category_spending"] = manual_expenses + +def display_budget_analysis(analysis: Dict[str, Any]): + """Display budget analysis results""" + # Display spending breakdown + if "spending_categories" in analysis: + st.subheader("Spending by Category") + fig = px.pie( + values=[cat["amount"] for cat in analysis["spending_categories"]], + names=[cat["category"] for cat in analysis["spending_categories"]], + title="Your Spending Breakdown" + ) + st.plotly_chart(fig) + + # Display income vs expenses + if "total_expenses" in analysis: + st.subheader("Income vs. Expenses") + income = analysis["monthly_income"] + expenses = analysis["total_expenses"] + surplus_deficit = income - expenses + + fig = go.Figure() + fig.add_trace(go.Bar(x=["Income", "Expenses"], + y=[income, expenses], + marker_color=["green", "red"])) + fig.update_layout(title="Monthly Income vs. Expenses") + st.plotly_chart(fig) + + st.metric("Monthly Surplus/Deficit", + f"${surplus_deficit:.2f}", + delta=f"{surplus_deficit:.2f}") + + # Display spending reduction recommendations + if "recommendations" in analysis: + st.subheader("Spending Reduction Recommendations") + for rec in analysis["recommendations"]: + st.markdown(f"**{rec['category']}**: {rec['recommendation']}") + if "potential_savings" in rec: + st.metric(f"Potential Monthly Savings", f"${rec['potential_savings']:.2f}") + +def display_savings_strategy(strategy: Dict[str, Any]): + """Display savings strategy results""" + st.subheader("Savings Recommendations") + + # Emergency Fund + if "emergency_fund" in strategy: + ef = strategy["emergency_fund"] + st.markdown(f"### Emergency Fund") + st.markdown(f"**Recommended Size**: ${ef['recommended_amount']:.2f}") + st.markdown(f"**Current Status**: {ef['current_status']}") + + # Progress bar + if "current_amount" in ef and "recommended_amount" in ef: + progress = ef["current_amount"] / ef["recommended_amount"] + st.progress(min(progress, 1.0)) + st.markdown(f"${ef['current_amount']:.2f} of ${ef['recommended_amount']:.2f}") + + # Savings Recommendations + if "recommendations" in strategy: + st.markdown("### Recommended Savings Allocations") + for rec in strategy["recommendations"]: + st.markdown(f"**{rec['category']}**: ${rec['amount']:.2f}/month") + st.markdown(f"_{rec['rationale']}_") + + # Automation Techniques + if "automation_techniques" in strategy: + st.markdown("### Automation Techniques") + for technique in strategy["automation_techniques"]: + st.markdown(f"**{technique['name']}**: {technique['description']}") + +def display_debt_reduction(plan: Dict[str, Any]): + """Display debt reduction plan results""" + # Total Debt Overview + if "total_debt" in plan: + st.metric("Total Debt", f"${plan['total_debt']:.2f}") + + # Debt Breakdown + if "debts" in plan: + st.subheader("Your Debts") + debt_df = pd.DataFrame(plan["debts"]) + st.dataframe(debt_df) + + # Debt visualization + fig = px.bar(debt_df, x="name", y="amount", color="interest_rate", + labels={"name": "Debt", "amount": "Amount ($)", "interest_rate": "Interest Rate (%)"}, + title="Debt Breakdown") + st.plotly_chart(fig) + + # Payoff Plans + if "payoff_plans" in plan: + st.subheader("Debt Payoff Plans") + tabs = st.tabs(["Avalanche Method", "Snowball Method", "Comparison"]) + + with tabs[0]: + st.markdown("### Avalanche Method (Highest Interest First)") + if "avalanche" in plan["payoff_plans"]: + avalanche = plan["payoff_plans"]["avalanche"] + st.markdown(f"**Total Interest Paid**: ${avalanche['total_interest']:.2f}") + st.markdown(f"**Time to Debt Freedom**: {avalanche['months_to_payoff']} months") + + if "monthly_payment" in avalanche: + st.markdown(f"**Recommended Monthly Payment**: ${avalanche['monthly_payment']:.2f}") + + if "schedule" in avalanche: + st.markdown("#### Payoff Schedule") + schedule_df = pd.DataFrame(avalanche["schedule"]) + st.dataframe(schedule_df) + + with tabs[1]: + st.markdown("### Snowball Method (Smallest Balance First)") + if "snowball" in plan["payoff_plans"]: + snowball = plan["payoff_plans"]["snowball"] + st.markdown(f"**Total Interest Paid**: ${snowball['total_interest']:.2f}") + st.markdown(f"**Time to Debt Freedom**: {snowball['months_to_payoff']} months") + + if "monthly_payment" in snowball: + st.markdown(f"**Recommended Monthly Payment**: ${snowball['monthly_payment']:.2f}") + + if "schedule" in snowball: + st.markdown("#### Payoff Schedule") + schedule_df = pd.DataFrame(snowball["schedule"]) + st.dataframe(schedule_df) + + with tabs[2]: + st.markdown("### Method Comparison") + if "avalanche" in plan["payoff_plans"] and "snowball" in plan["payoff_plans"]: + avalanche = plan["payoff_plans"]["avalanche"] + snowball = plan["payoff_plans"]["snowball"] + + comparison_data = { + "Method": ["Avalanche", "Snowball"], + "Total Interest": [avalanche["total_interest"], snowball["total_interest"]], + "Months to Payoff": [avalanche["months_to_payoff"], snowball["months_to_payoff"]] + } + comparison_df = pd.DataFrame(comparison_data) + + st.dataframe(comparison_df) + + fig = go.Figure(data=[ + go.Bar(name="Total Interest", x=comparison_df["Method"], y=comparison_df["Total Interest"]), + go.Bar(name="Months to Payoff", x=comparison_df["Method"], y=comparison_df["Months to Payoff"]) + ]) + fig.update_layout(barmode='group', title="Debt Payoff Method Comparison") + st.plotly_chart(fig) + + # Recommendations + if "recommendations" in plan: + st.subheader("Debt Reduction Recommendations") + for rec in plan["recommendations"]: + st.markdown(f"**{rec['title']}**: {rec['description']}") + if "impact" in rec: + st.markdown(f"_Impact: {rec['impact']}_") + +def main(): + st.set_page_config(page_title="AI Personal Finance Coach", layout="wide") + + # Check if we have the API key + if not os.getenv("GOOGLE_API_KEY"): + st.error(""" + GOOGLE_API_KEY not found in environment variables. + Please create a .env file with your Google API key: + ``` + GOOGLE_API_KEY=your_api_key_here + ``` + """) + return + + st.title("AI Personal Finance Coach") + st.subheader("Get personalized financial advice from AI agents") + + # Sidebar for user inputs + with st.sidebar: + st.header("Your Financial Information") + + # Monthly Income + monthly_income = st.number_input("Monthly Income ($)", min_value=0.0, step=100.0, value=3000.0) + + # Number of Dependants + dependants = st.number_input("Number of Dependants", min_value=0, step=1, value=0) + + # Transaction data upload + st.subheader("Upload Transaction Data") + st.write("Upload a CSV with columns: Date, Category, Amount") + transaction_file = st.file_uploader("Upload CSV of transactions", type=["csv"]) + + # Manual expense entry option + st.subheader("Or Enter Expenses Manually") + use_manual_expenses = st.checkbox("Enter expenses manually") + + manual_expenses = {} + if use_manual_expenses: + categories = ["Housing", "Utilities", "Food", "Transportation", "Healthcare", + "Entertainment", "Personal", "Savings", "Other"] + for category in categories: + manual_expenses[category] = st.number_input(f"{category} ($)", min_value=0.0, step=50.0, value=0.0) + + # Debt Information + st.subheader("Debt Information") + num_debts = st.number_input("Number of Debts", min_value=0, max_value=10, step=1, value=0) + + debts = [] + for i in range(num_debts): + st.markdown(f"**Debt #{i+1}**") + debt_name = st.text_input(f"Debt Name #{i+1}", value=f"Debt {i+1}") + debt_amount = st.number_input(f"Amount ${i+1}", min_value=0.0, step=100.0, value=1000.0) + interest_rate = st.number_input(f"Interest Rate (%) #{i+1}", min_value=0.0, max_value=100.0, step=0.1, value=5.0) + min_payment = st.number_input(f"Minimum Monthly Payment #{i+1}", min_value=0.0, step=10.0, value=50.0) + + debts.append({ + "name": debt_name, + "amount": debt_amount, + "interest_rate": interest_rate, + "min_payment": min_payment + }) + + analyze_button = st.button("Analyze My Finances") + + # Main content area + transactions_df = None + if transaction_file is not None: + transactions_df = pd.read_csv(transaction_file) + st.subheader("Your Transaction Data") + st.dataframe(transactions_df) + + if use_manual_expenses and manual_expenses: + st.subheader("Your Manual Expenses") + manual_df = pd.DataFrame({ + 'Category': list(manual_expenses.keys()), + 'Amount': list(manual_expenses.values()) + }) + st.dataframe(manual_df) + + # Prepare data for agent analysis + financial_data = { + "monthly_income": monthly_income, + "dependants": dependants, + "transactions": transactions_df.to_dict('records') if transactions_df is not None else None, + "manual_expenses": manual_expenses if use_manual_expenses else None, + "debts": debts + } + + # When analyze button is clicked, run agent analysis + if analyze_button: + with st.spinner("AI agents are analyzing your financial data..."): + # Create finance advisor system + finance_system = FinanceAdvisorSystem() + + # Run analysis + results = asyncio.run(finance_system.analyze_finances(financial_data)) + + # Display results in tabs + tabs = st.tabs(["Budget Analysis", "Savings Strategy", "Debt Reduction"]) + + with tabs[0]: + st.subheader("Budget Analysis") + if "budget_analysis" in results: + display_budget_analysis(results["budget_analysis"]) + else: + st.write("No budget analysis available.") + + with tabs[1]: + st.subheader("Savings Strategy") + if "savings_strategy" in results: + display_savings_strategy(results["savings_strategy"]) + else: + st.write("No savings strategy available.") + + with tabs[2]: + st.subheader("Debt Reduction Plan") + if "debt_reduction" in results: + display_debt_reduction(results["debt_reduction"]) + else: + st.write("No debt reduction plan available.") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_google_adk/requirements.txt b/ai_agent_tutorials/ai_google_adk/requirements.txt new file mode 100644 index 0000000..b642de4 --- /dev/null +++ b/ai_agent_tutorials/ai_google_adk/requirements.txt @@ -0,0 +1,8 @@ +google-adk==0.4.0 +streamlit==1.31.0 +pandas==2.1.1 +matplotlib==3.8.0 +numpy==1.26.0 +python-dotenv==1.0.0 +plotly==5.18.0 +asyncio==3.4.3 From c162a659375599ddee3fafc4071d2098b26c8518 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 21:22:19 +0530 Subject: [PATCH 2/6] Working Script --- .../ai_google_adk/google_adk.py | 737 ++++++++++++++---- 1 file changed, 587 insertions(+), 150 deletions(-) diff --git a/ai_agent_tutorials/ai_google_adk/google_adk.py b/ai_agent_tutorials/ai_google_adk/google_adk.py index b12ad18..6fdf357 100644 --- a/ai_agent_tutorials/ai_google_adk/google_adk.py +++ b/ai_agent_tutorials/ai_google_adk/google_adk.py @@ -7,13 +7,89 @@ import os import asyncio from datetime import datetime from dotenv import load_dotenv +import json +import logging +from pydantic import BaseModel, Field from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent from google.adk.agents.invocation_context import InvocationContext from google.adk.events import Event, EventActions from google.adk.sessions import InMemorySessionService, Session +from google.adk.runners import Runner +from google.genai import types +from google.adk.agents.callback_context import CallbackContext +from google.adk.models import LlmResponse, LlmRequest -from dotenv import load_dotenv +# Set up logging +logging.basicConfig(level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + +# Constants for session management +APP_NAME = "finance_advisor" +USER_ID = "default_user" + +# Define Pydantic models for output schemas +class SpendingCategory(BaseModel): + category: str = Field(..., description="Expense category name") + amount: float = Field(..., description="Amount spent in this category") + percentage: Optional[float] = Field(None, description="Percentage of total spending") + +class SpendingRecommendation(BaseModel): + category: str = Field(..., description="Category for recommendation") + recommendation: str = Field(..., description="Recommendation details") + potential_savings: Optional[float] = Field(None, description="Estimated monthly savings") + +class BudgetAnalysis(BaseModel): + total_expenses: float = Field(..., description="Total monthly expenses") + monthly_income: Optional[float] = Field(None, description="Monthly income") + spending_categories: List[SpendingCategory] = Field(..., description="Breakdown of spending by category") + recommendations: List[SpendingRecommendation] = Field(..., description="Spending recommendations") + +class EmergencyFund(BaseModel): + recommended_amount: float = Field(..., description="Recommended emergency fund size") + current_amount: Optional[float] = Field(None, description="Current emergency fund (if any)") + current_status: str = Field(..., description="Status assessment of emergency fund") + +class SavingsRecommendation(BaseModel): + category: str = Field(..., description="Savings category") + amount: float = Field(..., description="Recommended monthly amount") + rationale: Optional[str] = Field(None, description="Explanation for this recommendation") + +class AutomationTechnique(BaseModel): + name: str = Field(..., description="Name of automation technique") + description: str = Field(..., description="Details of how to implement") + +class SavingsStrategy(BaseModel): + emergency_fund: EmergencyFund = Field(..., description="Emergency fund recommendation") + recommendations: List[SavingsRecommendation] = Field(..., description="Savings allocation recommendations") + automation_techniques: Optional[List[AutomationTechnique]] = Field(None, description="Automation techniques to help save") + +class Debt(BaseModel): + name: str = Field(..., description="Name of debt") + amount: float = Field(..., description="Current balance") + interest_rate: float = Field(..., description="Annual interest rate (%)") + min_payment: Optional[float] = Field(None, description="Minimum monthly payment") + +class PayoffPlan(BaseModel): + total_interest: float = Field(..., description="Total interest paid") + months_to_payoff: int = Field(..., description="Months until debt-free") + monthly_payment: Optional[float] = Field(None, description="Recommended monthly payment") + +class PayoffPlans(BaseModel): + avalanche: PayoffPlan = Field(..., description="Highest interest first method") + snowball: PayoffPlan = Field(..., description="Smallest balance first method") + +class DebtRecommendation(BaseModel): + title: str = Field(..., description="Title of recommendation") + description: str = Field(..., description="Details of recommendation") + impact: Optional[str] = Field(None, description="Expected impact of this action") + +class DebtReduction(BaseModel): + total_debt: float = Field(..., description="Total debt amount") + debts: List[Debt] = Field(..., description="List of all debts") + payoff_plans: PayoffPlans = Field(..., description="Debt payoff strategies") + recommendations: Optional[List[DebtRecommendation]] = Field(None, description="Recommendations for debt reduction") # Load environment variables load_dotenv() @@ -24,27 +100,45 @@ if not GEMINI_API_KEY: raise ValueError("GOOGLE_API_KEY environment variable not set") class FinanceAdvisorSystem: + """Main class to manage finance advisor agents""" + def __init__(self): """Initialize the finance advisor system with specialized agents""" + # Initialize session service + self.session_service = InMemorySessionService() + # Budget Analysis Agent self.budget_analysis_agent = LlmAgent( name="BudgetAnalysisAgent", model="gemini-2.0-flash-exp", description="Analyzes financial data to categorize spending patterns and recommend budget improvements", instruction="""You are a Budget Analysis Agent specialized in reviewing financial transactions and expenses. +You are the first agent in a sequence of three financial advisor agents. Your tasks: -1. Analyze income, transactions, and expenses -2. Categorize spending into logical groups -3. Identify spending patterns and trends -4. Suggest specific areas where spending could be reduced -5. Provide actionable recommendations with potential savings amounts +1. Analyze income, transactions, and expenses in detail +2. Categorize spending into logical groups with clear breakdown +3. Identify spending patterns and trends across categories +4. Suggest specific areas where spending could be reduced with concrete suggestions +5. Provide actionable recommendations with specific, quantified potential savings amounts Consider: - Number of dependants when evaluating household expenses -- Typical spending ratios for the income level -- Essential vs discretionary spending -- Seasonal spending patterns if data spans multiple months""", +- Typical spending ratios for the income level (housing 30%, food 15%, etc.) +- Essential vs discretionary spending with clear separation +- Seasonal spending patterns if data spans multiple months + +For spending categories, include ALL expenses from the user's data, ensure percentages add up to 100%, +and make sure every expense is categorized. + +For recommendations: +- Provide at least 3-5 specific, actionable recommendations with estimated savings +- Explain the reasoning behind each recommendation +- Consider the impact on quality of life and long-term financial health +- Suggest specific implementation steps for each recommendation + +IMPORTANT: Store your analysis in state['budget_analysis'] for use by subsequent agents.""", + output_schema=BudgetAnalysis, output_key="budget_analysis" ) @@ -54,40 +148,51 @@ Consider: model="gemini-2.0-flash-exp", description="Recommends optimal savings strategies based on income, expenses, and financial goals", instruction="""You are a Savings Strategy Agent specialized in creating personalized savings plans. +You are the second agent in the sequence. READ the budget analysis from state['budget_analysis'] first. Your tasks: -1. Recommend savings strategies based on income and expenses -2. Calculate optimal emergency fund size based on expenses and dependants -3. Suggest appropriate savings allocation across different purposes -4. Recommend practical automation techniques for saving consistently +1. Review the budget analysis results from state['budget_analysis'] +2. Recommend comprehensive savings strategies based on the analysis +3. Calculate optimal emergency fund size based on expenses and dependants +4. Suggest appropriate savings allocation across different purposes +5. Recommend practical automation techniques for saving consistently Consider: - Risk factors based on job stability and dependants - Balancing immediate needs with long-term financial health - Progressive savings rates as discretionary income increases -- Multiple savings goals (emergency, retirement, specific purchases)""", +- Multiple savings goals (emergency, retirement, specific purchases) +- Areas of potential savings identified in the budget analysis + +IMPORTANT: Store your strategy in state['savings_strategy'] for use by the Debt Reduction Agent.""", + output_schema=SavingsStrategy, output_key="savings_strategy" ) # Debt Reduction Agent self.debt_reduction_agent = LlmAgent( name="DebtReductionAgent", - model="gemini-2.0-flash-exp", + model="gemini-2.0-flash-exp", description="Creates optimized debt payoff plans to minimize interest paid and time to debt freedom", instruction="""You are a Debt Reduction Agent specialized in creating debt payoff strategies. +You are the final agent in the sequence. READ both state['budget_analysis'] and state['savings_strategy'] first. Your tasks: -1. Analyze debts by interest rate, balance, and minimum payments -2. Create prioritized debt payoff plans (avalanche and snowball methods) -3. Calculate total interest paid and time to debt freedom for each approach -4. Suggest debt consolidation or refinancing opportunities when beneficial -5. Provide specific recommendations to accelerate debt payoff +1. Review both budget analysis and savings strategy from the state +2. Analyze debts by interest rate, balance, and minimum payments +3. Create prioritized debt payoff plans (avalanche and snowball methods) +4. Calculate total interest paid and time to debt freedom +5. Suggest debt consolidation or refinancing opportunities +6. Provide specific recommendations to accelerate debt payoff Consider: -- Cash flow and budget constraints from the budget analysis +- Cash flow constraints from the budget analysis +- Emergency fund and savings goals from the savings strategy - Psychological factors (quick wins vs mathematical optimization) -- Interest savings potential -- Credit utilization and credit score impact""", +- Credit score impact and improvement opportunities + +IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns with the previous analyses.""", + output_schema=DebtReduction, output_key="debt_reduction" ) @@ -101,45 +206,206 @@ Consider: self.debt_reduction_agent ] ) + + # Add debug callbacks to monitor agent behavior and state flow + self._add_debug_callbacks() + + # Create a runner for the coordinator agent + self.runner = Runner( + agent=self.coordinator_agent, + app_name=APP_NAME, + session_service=self.session_service + ) + + def _add_debug_callbacks(self): + """Add debug callbacks to agents to track execution and state flow""" + logger.info("=== Registering Callbacks ===") + for agent in [self.budget_analysis_agent, self.savings_strategy_agent, self.debt_reduction_agent]: + logger.info(f"Adding callbacks to agent: {agent.name}") + agent.before_model_callback = self._simple_before_model_callback + agent.after_model_callback = self._simple_after_model_callback + # Verify callback registration + logger.info(f"Callbacks registered - Before: {agent.before_model_callback.__name__}, After: {agent.after_model_callback.__name__}") + + def _simple_before_model_callback(self, callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]: + """Simple debug callback before model call""" + agent_name = callback_context.agent_name + logger.info(f"=== Before Model Callback ({agent_name}) ===") + # Log arguments excluding 'self' + args_log = {k: v for k, v in locals().items() if k != 'self'} + logger.info(f"({agent_name}) Callback args: {args_log}") + logger.info(f"({agent_name}) Callback context type: {type(callback_context)}") + logger.info(f"({agent_name}) LLM request type: {type(llm_request)}") + if hasattr(callback_context, 'state'): + logger.info(f"({agent_name}) Current state available") + return None + + def _simple_after_model_callback(self, callback_context: CallbackContext, llm_response: LlmResponse) -> Optional[LlmResponse]: + """Simple debug callback after model call""" + agent_name = callback_context.agent_name + logger.info(f"=== After Model Callback ({agent_name}) ===") + # Log arguments excluding 'self' + args_log = {k: v for k, v in locals().items() if k != 'self'} + logger.info(f"({agent_name}) Callback args: {args_log}") + logger.info(f"({agent_name}) Callback context type: {type(callback_context)}") + logger.info(f"({agent_name}) LLM response type: {type(llm_response)}") + # llm_request is not expected here based on the error + if hasattr(callback_context, 'state'): + logger.info(f"({agent_name}) Updated state available") + return None async def analyze_finances(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: """Process financial data through the agent system and return comprehensive analysis""" - # Prepare the session context - session = Session() + session_id = f"finance_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + logger.info(f"Starting finance analysis with session_id: {session_id}") - # Store financial data in session state for agents to access - session.state.update({ - "monthly_income": financial_data.get("monthly_income", 0), - "dependants": financial_data.get("dependants", 0), - "transactions": financial_data.get("transactions", []), - "manual_expenses": financial_data.get("manual_expenses", {}), - "debts": financial_data.get("debts", []) - }) - - # Preprocess transaction data if available - if financial_data.get("transactions"): - self._preprocess_transactions(session) - - # Initialize preprocessing for manual expenses if provided - if financial_data.get("manual_expenses"): - self._preprocess_manual_expenses(session) - - # Set up the invocation context - context = InvocationContext(session=session, user_input="Analyze financial data") - - # Run the coordinator agent which will execute all sub-agents in sequence - async for event in self.coordinator_agent.run(context): - # We could process events here if needed - pass - - # Collect results from session state - results = { - "budget_analysis": session.state.get("budget_analysis", {}), - "savings_strategy": session.state.get("savings_strategy", {}), - "debt_reduction": session.state.get("debt_reduction", {}) - } - - return results + try: + # Create a new session with required parameters + initial_state = { + "monthly_income": financial_data.get("monthly_income", 0), + "dependants": financial_data.get("dependants", 0), + "transactions": financial_data.get("transactions", []), + "manual_expenses": financial_data.get("manual_expenses", {}), + "debts": financial_data.get("debts", []) + } + + session = self.session_service.create_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id, + state=initial_state + ) + + # Log initial state + logger.info(f"Created session with initial state items: {list(initial_state.keys())}") + + # Preprocess transaction data if available + transactions = session.state.get("transactions") + if transactions: + self._preprocess_transactions(session) + + # Initialize preprocessing for manual expenses if provided + manual_expenses = session.state.get("manual_expenses") + if manual_expenses: + self._preprocess_manual_expenses(session) + + # Create default results + default_results = self._create_default_results(financial_data) + + # Create user message content + user_content = types.Content( + role='user', + parts=[types.Part(text=json.dumps(financial_data))] + ) + + logger.info("Running coordinator agent") + + # Run the analysis through the coordinator agent + event_count = 0 + current_agent = None + async for event in self.runner.run_async( + user_id=USER_ID, + session_id=session_id, + new_message=user_content + ): + event_count += 1 + # --- DETAILED EVENT LOGGING --- + logger.info(f"-- RAW EVENT {event_count} START --") + logger.info(f"Event Author: {event.author}") + logger.info(f"Event ID: {event.id}") + logger.info(f"Invocation ID: {event.invocation_id}") + logger.info(f"Is Final Response Flag: {event.is_final_response()}") + if event.content: + logger.info(f"Event Content: {str(event.content)[:500]}...") # Log content snippet + if hasattr(event, 'actions') and event.actions: + logger.info(f"Event Actions: {event.actions}") + logger.info(f"-- RAW EVENT {event_count} END --") + # --- END DETAILED EVENT LOGGING --- + + # Original logging logic below + logger.info(f"Event {event_count}: author={event.author}") + + if event.author != current_agent: + current_agent = event.author + logger.info(f"Agent execution changed to: {current_agent}") + + if event.content and event.content.parts: + part = event.content.parts[0] + if hasattr(part, 'text') and part.text: + logger.info(f"Text content: {part.text[:100]}...") + + if hasattr(event, 'actions') and event.actions: + if hasattr(event.actions, 'state_delta') and event.actions.state_delta: + state_delta = event.actions.state_delta + logger.info(f"State delta received: {state_delta}") + + # Check for final response *only* from the coordinator agent + if event.is_final_response() and event.author == self.coordinator_agent.name: + logger.warning(f"Event {event_count} from COORDINATOR ({event.author}) flagged as FINAL. Breaking loop.") + if event.content and event.content.parts: + part = event.content.parts[0] + if hasattr(part, 'text') and part.text: + logger.info(f"Final response text: {part.text[:100]}...") + break + elif event.is_final_response(): + # Log but don't break if a sub-agent marks as final + logger.info(f"Event {event_count} from sub-agent {event.author} flagged as FINAL, but continuing sequence.") + + # Get the updated session + logger.info("Retrieving updated session") + updated_session = self.session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id + ) + + # Process agent outputs from state + results = {} + + # Process each agent output + for key in ["budget_analysis", "savings_strategy", "debt_reduction"]: + value = updated_session.state.get(key) + if value is not None: + logger.info(f"Found {key} in state: type={type(value)}") + + if value == "": + logger.warning(f"{key} is empty in state, using default") + results[key] = default_results[key] + continue + + if isinstance(value, str): + try: + parsed_value = json.loads(value) + results[key] = parsed_value + logger.info(f"Successfully parsed {key} as JSON") + except json.JSONDecodeError: + logger.warning(f"Could not parse {key} as JSON, using as is: {value[:100]}...") + if key in default_results: + results[key] = default_results[key] + else: + results[key] = value + else: + results[key] = value + else: + logger.warning(f"{key} not found in session state, using default") + results[key] = default_results[key] + + return results + + except Exception as e: + logger.exception(f"Error during finance analysis: {str(e)}") + raise + finally: + # Clean up the session + try: + self.session_service.delete_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id + ) + logger.info(f"Cleaned up session: {session_id}") + except Exception as e: + logger.warning(f"Failed to clean up session: {e}") def _preprocess_transactions(self, session): """Preprocess transaction data for easier analysis by the agents""" @@ -180,8 +446,108 @@ Consider: # Store categorized spending directly session.state["manual_category_spending"] = manual_expenses + def _create_default_results(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: + """Create default results in case agent execution fails""" + monthly_income = financial_data.get("monthly_income", 0) + expenses = {} + + # Extract expenses from manual entries or transactions + if financial_data.get("manual_expenses"): + expenses = financial_data.get("manual_expenses") + elif financial_data.get("transactions"): + # Simplified aggregation of transactions + for transaction in financial_data.get("transactions", []): + category = transaction.get("Category", "Uncategorized") + amount = transaction.get("Amount", 0) + if category in expenses: + expenses[category] += amount + else: + expenses[category] = amount + + total_expenses = sum(expenses.values()) + + # Create default budget analysis + default_budget = { + "total_expenses": total_expenses, + "monthly_income": monthly_income, + "spending_categories": [ + {"category": cat, "amount": amt, "percentage": (amt / total_expenses * 100) if total_expenses > 0 else 0} + for cat, amt in expenses.items() + ], + "recommendations": [ + {"category": "General", "recommendation": "Consider reviewing your expenses carefully", "potential_savings": total_expenses * 0.1} + ] + } + + # Create default savings strategy + default_savings = { + "emergency_fund": { + "recommended_amount": total_expenses * 6, + "current_amount": 0, + "current_status": "Not started" + }, + "recommendations": [ + {"category": "Emergency Fund", "amount": total_expenses * 0.1, "rationale": "Build emergency fund first"}, + {"category": "Retirement", "amount": monthly_income * 0.15, "rationale": "Long-term savings"} + ], + "automation_techniques": [ + {"name": "Automatic Transfer", "description": "Set up automatic transfers on payday"} + ] + } + + # Create default debt reduction + default_debts = financial_data.get("debts", []) + total_debt = sum(debt.get("amount", 0) for debt in default_debts) + + default_debt = { + "total_debt": total_debt, + "debts": default_debts, + "payoff_plans": { + "avalanche": { + "total_interest": total_debt * 0.2, + "months_to_payoff": 24, + "monthly_payment": total_debt / 24 + }, + "snowball": { + "total_interest": total_debt * 0.25, + "months_to_payoff": 24, + "monthly_payment": total_debt / 24 + } + }, + "recommendations": [ + {"title": "Increase Payments", "description": "Increase your monthly payments", "impact": "Reduces total interest paid"} + ] + } + + return { + "budget_analysis": default_budget, + "savings_strategy": default_savings, + "debt_reduction": default_debt + } + def display_budget_analysis(analysis: Dict[str, Any]): """Display budget analysis results""" + logger.info(f"Displaying budget analysis, type: {type(analysis)}") + + # Ensure we have a dictionary + if isinstance(analysis, str): + logger.info(f"Budget analysis is a string, attempting to parse as JSON") + try: + analysis = json.loads(analysis) + logger.info("Successfully parsed budget analysis from JSON string") + except json.JSONDecodeError as e: + logger.error(f"Failed to parse budget analysis results: {e}") + logger.error(f"First 200 chars of analysis: {analysis[:200]}") + st.error("Failed to parse budget analysis results") + return + + if not isinstance(analysis, dict): + logger.error(f"Invalid budget analysis format: {type(analysis)}") + st.error("Invalid budget analysis format") + return + + logger.info(f"Budget analysis keys: {list(analysis.keys())}") + # Display spending breakdown if "spending_categories" in analysis: st.subheader("Spending by Category") @@ -195,7 +561,7 @@ def display_budget_analysis(analysis: Dict[str, Any]): # Display income vs expenses if "total_expenses" in analysis: st.subheader("Income vs. Expenses") - income = analysis["monthly_income"] + income = analysis.get("monthly_income", 0) expenses = analysis["total_expenses"] surplus_deficit = income - expenses @@ -220,6 +586,18 @@ def display_budget_analysis(analysis: Dict[str, Any]): def display_savings_strategy(strategy: Dict[str, Any]): """Display savings strategy results""" + # Ensure we have a dictionary + if isinstance(strategy, str): + try: + strategy = json.loads(strategy) + except json.JSONDecodeError: + st.error("Failed to parse savings strategy results") + return + + if not isinstance(strategy, dict): + st.error("Invalid savings strategy format") + return + st.subheader("Savings Recommendations") # Emergency Fund @@ -250,6 +628,18 @@ def display_savings_strategy(strategy: Dict[str, Any]): def display_debt_reduction(plan: Dict[str, Any]): """Display debt reduction plan results""" + # Ensure we have a dictionary + if isinstance(plan, str): + try: + plan = json.loads(plan) + except json.JSONDecodeError: + st.error("Failed to parse debt reduction results") + return + + if not isinstance(plan, dict): + st.error("Invalid debt reduction format") + return + # Total Debt Overview if "total_debt" in plan: st.metric("Total Debt", f"${plan['total_debt']:.2f}") @@ -336,6 +726,7 @@ def main(): # Check if we have the API key if not os.getenv("GOOGLE_API_KEY"): + logger.error("GOOGLE_API_KEY environment variable not set") st.error(""" GOOGLE_API_KEY not found in environment variables. Please create a .env file with your Google API key: @@ -345,112 +736,158 @@ def main(): """) return - st.title("AI Personal Finance Coach") + st.title("📊 AI Personal Finance Coach") st.subheader("Get personalized financial advice from AI agents") + st.markdown("---") - # Sidebar for user inputs - with st.sidebar: - st.header("Your Financial Information") - - # Monthly Income - monthly_income = st.number_input("Monthly Income ($)", min_value=0.0, step=100.0, value=3000.0) - - # Number of Dependants - dependants = st.number_input("Number of Dependants", min_value=0, step=1, value=0) - - # Transaction data upload - st.subheader("Upload Transaction Data") - st.write("Upload a CSV with columns: Date, Category, Amount") - transaction_file = st.file_uploader("Upload CSV of transactions", type=["csv"]) - - # Manual expense entry option - st.subheader("Or Enter Expenses Manually") - use_manual_expenses = st.checkbox("Enter expenses manually") + # --- Input Section --- + st.header("Step 1: Enter Your Financial Information") + + col1, col2 = st.columns(2) + + with col1: + st.subheader("Income & Dependants") + monthly_income = st.number_input("Monthly Income ($)", min_value=0.0, step=100.0, value=3000.0, key="income") + dependants = st.number_input("Number of Dependants", min_value=0, step=1, value=0, key="dependants") + + with col2: + st.subheader("Expense Data") + expense_option = st.radio( + "How do you want to enter expenses?", + ("Upload CSV Transactions", "Enter Manually"), + key="expense_option" + ) + transaction_file = None manual_expenses = {} - if use_manual_expenses: + use_manual_expenses = False + transactions_df = None + + if expense_option == "Upload CSV Transactions": + st.write("Upload a CSV with columns: Date, Category, Amount") + transaction_file = st.file_uploader("Upload CSV of transactions", type=["csv"], key="transaction_file") + if transaction_file is not None: + try: + transactions_df = pd.read_csv(transaction_file) + st.success("Transaction file uploaded successfully!") + # Optional: Display small preview + # st.dataframe(transactions_df.head(3)) + except Exception as e: + st.error(f"Error reading CSV: {e}") + transactions_df = None # Ensure df is None if error + else: + use_manual_expenses = True + st.write("Enter monthly expenses by category:") categories = ["Housing", "Utilities", "Food", "Transportation", "Healthcare", "Entertainment", "Personal", "Savings", "Other"] - for category in categories: - manual_expenses[category] = st.number_input(f"{category} ($)", min_value=0.0, step=50.0, value=0.0) - - # Debt Information - st.subheader("Debt Information") - num_debts = st.number_input("Number of Debts", min_value=0, max_value=10, step=1, value=0) - - debts = [] + # Use columns for better manual entry layout + exp_col1, exp_col2 = st.columns(2) + for i, category in enumerate(categories): + col = exp_col1 if i < (len(categories) + 1) // 2 else exp_col2 + manual_expenses[category] = col.number_input(f"{category} ($)", min_value=0.0, step=50.0, value=0.0, key=f"manual_{category}") + # Display manual entries for confirmation + if any(manual_expenses.values()): + st.write("Entered Manual Expenses:") + manual_df_disp = pd.DataFrame({ + 'Category': list(manual_expenses.keys()), + 'Amount': list(manual_expenses.values()) + }) + st.dataframe(manual_df_disp[manual_df_disp['Amount'] > 0]) + + + st.subheader("Debt Information") + num_debts = st.number_input("Number of Debts", min_value=0, max_value=10, step=1, value=0, key="num_debts") + + debts = [] + if num_debts > 0: + debt_cols = st.columns(num_debts) for i in range(num_debts): - st.markdown(f"**Debt #{i+1}**") - debt_name = st.text_input(f"Debt Name #{i+1}", value=f"Debt {i+1}") - debt_amount = st.number_input(f"Amount ${i+1}", min_value=0.0, step=100.0, value=1000.0) - interest_rate = st.number_input(f"Interest Rate (%) #{i+1}", min_value=0.0, max_value=100.0, step=0.1, value=5.0) - min_payment = st.number_input(f"Minimum Monthly Payment #{i+1}", min_value=0.0, step=10.0, value=50.0) - - debts.append({ - "name": debt_name, - "amount": debt_amount, - "interest_rate": interest_rate, - "min_payment": min_payment - }) + with debt_cols[i]: + st.markdown(f"**Debt #{i+1}**") + debt_name = st.text_input(f"Name", value=f"Debt {i+1}", key=f"debt_name_{i}") + debt_amount = st.number_input(f"Amount $", min_value=0.01, step=100.0, value=1000.0, key=f"debt_amount_{i}") + interest_rate = st.number_input(f"Interest Rate (%)", min_value=0.0, max_value=100.0, step=0.1, value=5.0, key=f"debt_rate_{i}") + min_payment = st.number_input(f"Min. Payment $", min_value=0.0, step=10.0, value=50.0, key=f"debt_min_payment_{i}") + + debts.append({ + "name": debt_name, + "amount": debt_amount, + "interest_rate": interest_rate, + "min_payment": min_payment + }) - analyze_button = st.button("Analyze My Finances") + st.markdown("---") + analyze_button = st.button("Analyze My Finances", key="analyze_button") + st.markdown("---") - # Main content area - transactions_df = None - if transaction_file is not None: - transactions_df = pd.read_csv(transaction_file) - st.subheader("Your Transaction Data") - st.dataframe(transactions_df) - - if use_manual_expenses and manual_expenses: - st.subheader("Your Manual Expenses") - manual_df = pd.DataFrame({ - 'Category': list(manual_expenses.keys()), - 'Amount': list(manual_expenses.values()) - }) - st.dataframe(manual_df) - - # Prepare data for agent analysis - financial_data = { - "monthly_income": monthly_income, - "dependants": dependants, - "transactions": transactions_df.to_dict('records') if transactions_df is not None else None, - "manual_expenses": manual_expenses if use_manual_expenses else None, - "debts": debts - } - - # When analyze button is clicked, run agent analysis + # --- Results Section --- if analyze_button: - with st.spinner("AI agents are analyzing your financial data..."): + # Validate inputs before proceeding + if expense_option == "Upload CSV Transactions" and transactions_df is None: + st.error("Please upload a valid transaction CSV file or choose manual entry.") + return + if use_manual_expenses and not any(manual_expenses.values()): + st.warning("No manual expenses entered. Analysis might be limited.") + # Optionally proceed or return, depending on desired behavior + + st.header("Step 2: Financial Analysis Results") + with st.spinner("AI agents are analyzing your financial data..."): + # Prepare data for agent analysis + financial_data = { + "monthly_income": monthly_income, + "dependants": dependants, + "transactions": transactions_df.to_dict('records') if transactions_df is not None else None, + "manual_expenses": manual_expenses if use_manual_expenses else None, + "debts": debts + } + # Create finance advisor system finance_system = FinanceAdvisorSystem() # Run analysis - results = asyncio.run(finance_system.analyze_finances(financial_data)) + logger.info("Starting financial analysis") + results = None + try: + results = asyncio.run(finance_system.analyze_finances(financial_data)) + logger.info(f"Analysis complete, results keys: {list(results.keys())}") + + # Log the types of each result + for key, value in results.items(): + logger.info(f"Result '{key}' is type: {type(value)}") + # if value: # Avoid logging large outputs unless needed + # preview = str(value)[:100] + "..." if len(str(value)) > 100 else str(value) + # logger.info(f"Preview of {key}: {preview}") + except Exception as e: + logger.exception(f"Error in financial analysis: {e}") + st.error(f"An error occurred during analysis: {str(e)}") + # results remains None - # Display results in tabs - tabs = st.tabs(["Budget Analysis", "Savings Strategy", "Debt Reduction"]) - - with tabs[0]: - st.subheader("Budget Analysis") - if "budget_analysis" in results: - display_budget_analysis(results["budget_analysis"]) - else: - st.write("No budget analysis available.") - - with tabs[1]: - st.subheader("Savings Strategy") - if "savings_strategy" in results: - display_savings_strategy(results["savings_strategy"]) - else: - st.write("No savings strategy available.") - - with tabs[2]: - st.subheader("Debt Reduction Plan") - if "debt_reduction" in results: - display_debt_reduction(results["debt_reduction"]) - else: - st.write("No debt reduction plan available.") + # Display results if analysis was successful + if results: + tabs = st.tabs(["💰 Budget Analysis", "📈 Savings Strategy", "đŸ’ŗ Debt Reduction"]) + + with tabs[0]: + st.subheader("Budget Analysis") + if "budget_analysis" in results and results["budget_analysis"]: + display_budget_analysis(results["budget_analysis"]) + else: + st.write("No budget analysis available or analysis failed.") + + with tabs[1]: + st.subheader("Savings Strategy") + if "savings_strategy" in results and results["savings_strategy"]: + display_savings_strategy(results["savings_strategy"]) + else: + st.write("No savings strategy available or analysis failed.") + + with tabs[2]: + st.subheader("Debt Reduction Plan") + if "debt_reduction" in results and results["debt_reduction"]: + display_debt_reduction(results["debt_reduction"]) + else: + st.write("No debt reduction plan available or analysis failed.") + else: + st.error("Financial analysis could not be completed.") if __name__ == "__main__": main() \ No newline at end of file From 4f4ad11932d5f52cf1d271aa88d1ad29309ca87b Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 21:23:47 +0530 Subject: [PATCH 3/6] rename changes --- .../{ai_google_adk => ai_financial_coach_agent}/.env | 0 .../{ai_google_adk => ai_financial_coach_agent}/README.md | 0 .../ai_financial_coach_agent.py} | 0 .../{ai_google_adk => ai_financial_coach_agent}/requirements.txt | 0 4 files changed, 0 insertions(+), 0 deletions(-) rename ai_agent_tutorials/{ai_google_adk => ai_financial_coach_agent}/.env (100%) rename ai_agent_tutorials/{ai_google_adk => ai_financial_coach_agent}/README.md (100%) rename ai_agent_tutorials/{ai_google_adk/google_adk.py => ai_financial_coach_agent/ai_financial_coach_agent.py} (100%) rename ai_agent_tutorials/{ai_google_adk => ai_financial_coach_agent}/requirements.txt (100%) diff --git a/ai_agent_tutorials/ai_google_adk/.env b/ai_agent_tutorials/ai_financial_coach_agent/.env similarity index 100% rename from ai_agent_tutorials/ai_google_adk/.env rename to ai_agent_tutorials/ai_financial_coach_agent/.env diff --git a/ai_agent_tutorials/ai_google_adk/README.md b/ai_agent_tutorials/ai_financial_coach_agent/README.md similarity index 100% rename from ai_agent_tutorials/ai_google_adk/README.md rename to ai_agent_tutorials/ai_financial_coach_agent/README.md diff --git a/ai_agent_tutorials/ai_google_adk/google_adk.py b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py similarity index 100% rename from ai_agent_tutorials/ai_google_adk/google_adk.py rename to ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py diff --git a/ai_agent_tutorials/ai_google_adk/requirements.txt b/ai_agent_tutorials/ai_financial_coach_agent/requirements.txt similarity index 100% rename from ai_agent_tutorials/ai_google_adk/requirements.txt rename to ai_agent_tutorials/ai_financial_coach_agent/requirements.txt From ac3bc19a5295c8c3d0df211d1dbc4f93a2ae6590 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 21:45:05 +0530 Subject: [PATCH 4/6] made the script shorter and cleaner - yet to add csv functionality --- .../ai_financial_coach_agent.py | 295 +++--------------- 1 file changed, 43 insertions(+), 252 deletions(-) diff --git a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py index 6fdf357..43b4f6c 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py +++ b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py @@ -20,16 +20,13 @@ from google.genai import types from google.adk.agents.callback_context import CallbackContext from google.adk.models import LlmResponse, LlmRequest -# Set up logging -logging.basicConfig(level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -# Constants for session management APP_NAME = "finance_advisor" USER_ID = "default_user" -# Define Pydantic models for output schemas +# Pydantic models for output schemas class SpendingCategory(BaseModel): category: str = Field(..., description="Expense category name") amount: float = Field(..., description="Amount spent in this category") @@ -91,23 +88,14 @@ class DebtReduction(BaseModel): payoff_plans: PayoffPlans = Field(..., description="Debt payoff strategies") recommendations: Optional[List[DebtRecommendation]] = Field(None, description="Recommendations for debt reduction") -# Load environment variables load_dotenv() -# Get API key from environment GEMINI_API_KEY = os.getenv("GOOGLE_API_KEY") -if not GEMINI_API_KEY: - raise ValueError("GOOGLE_API_KEY environment variable not set") class FinanceAdvisorSystem: - """Main class to manage finance advisor agents""" - def __init__(self): - """Initialize the finance advisor system with specialized agents""" - # Initialize session service self.session_service = InMemorySessionService() - # Budget Analysis Agent self.budget_analysis_agent = LlmAgent( name="BudgetAnalysisAgent", model="gemini-2.0-flash-exp", @@ -142,7 +130,6 @@ IMPORTANT: Store your analysis in state['budget_analysis'] for use by subsequent output_key="budget_analysis" ) - # Savings Strategy Agent self.savings_strategy_agent = LlmAgent( name="SavingsStrategyAgent", model="gemini-2.0-flash-exp", @@ -169,7 +156,6 @@ IMPORTANT: Store your strategy in state['savings_strategy'] for use by the Debt output_key="savings_strategy" ) - # Debt Reduction Agent self.debt_reduction_agent = LlmAgent( name="DebtReductionAgent", model="gemini-2.0-flash-exp", @@ -196,7 +182,6 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns output_key="debt_reduction" ) - # Coordinator Agent - Orchestrates the specialized agents self.coordinator_agent = SequentialAgent( name="FinanceCoordinatorAgent", description="Coordinates specialized finance agents to provide comprehensive financial advice", @@ -207,60 +192,16 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns ] ) - # Add debug callbacks to monitor agent behavior and state flow - self._add_debug_callbacks() - - # Create a runner for the coordinator agent self.runner = Runner( agent=self.coordinator_agent, app_name=APP_NAME, session_service=self.session_service ) - - def _add_debug_callbacks(self): - """Add debug callbacks to agents to track execution and state flow""" - logger.info("=== Registering Callbacks ===") - for agent in [self.budget_analysis_agent, self.savings_strategy_agent, self.debt_reduction_agent]: - logger.info(f"Adding callbacks to agent: {agent.name}") - agent.before_model_callback = self._simple_before_model_callback - agent.after_model_callback = self._simple_after_model_callback - # Verify callback registration - logger.info(f"Callbacks registered - Before: {agent.before_model_callback.__name__}, After: {agent.after_model_callback.__name__}") - - def _simple_before_model_callback(self, callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]: - """Simple debug callback before model call""" - agent_name = callback_context.agent_name - logger.info(f"=== Before Model Callback ({agent_name}) ===") - # Log arguments excluding 'self' - args_log = {k: v for k, v in locals().items() if k != 'self'} - logger.info(f"({agent_name}) Callback args: {args_log}") - logger.info(f"({agent_name}) Callback context type: {type(callback_context)}") - logger.info(f"({agent_name}) LLM request type: {type(llm_request)}") - if hasattr(callback_context, 'state'): - logger.info(f"({agent_name}) Current state available") - return None - - def _simple_after_model_callback(self, callback_context: CallbackContext, llm_response: LlmResponse) -> Optional[LlmResponse]: - """Simple debug callback after model call""" - agent_name = callback_context.agent_name - logger.info(f"=== After Model Callback ({agent_name}) ===") - # Log arguments excluding 'self' - args_log = {k: v for k, v in locals().items() if k != 'self'} - logger.info(f"({agent_name}) Callback args: {args_log}") - logger.info(f"({agent_name}) Callback context type: {type(callback_context)}") - logger.info(f"({agent_name}) LLM response type: {type(llm_response)}") - # llm_request is not expected here based on the error - if hasattr(callback_context, 'state'): - logger.info(f"({agent_name}) Updated state available") - return None - + async def analyze_finances(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: - """Process financial data through the agent system and return comprehensive analysis""" session_id = f"finance_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" - logger.info(f"Starting finance analysis with session_id: {session_id}") try: - # Create a new session with required parameters initial_state = { "monthly_income": financial_data.get("monthly_income", 0), "dependants": financial_data.get("dependants", 0), @@ -276,100 +217,41 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns state=initial_state ) - # Log initial state - logger.info(f"Created session with initial state items: {list(initial_state.keys())}") - - # Preprocess transaction data if available transactions = session.state.get("transactions") if transactions: self._preprocess_transactions(session) - # Initialize preprocessing for manual expenses if provided manual_expenses = session.state.get("manual_expenses") if manual_expenses: self._preprocess_manual_expenses(session) - # Create default results default_results = self._create_default_results(financial_data) - # Create user message content user_content = types.Content( role='user', parts=[types.Part(text=json.dumps(financial_data))] ) - logger.info("Running coordinator agent") - - # Run the analysis through the coordinator agent - event_count = 0 - current_agent = None async for event in self.runner.run_async( user_id=USER_ID, session_id=session_id, new_message=user_content ): - event_count += 1 - # --- DETAILED EVENT LOGGING --- - logger.info(f"-- RAW EVENT {event_count} START --") - logger.info(f"Event Author: {event.author}") - logger.info(f"Event ID: {event.id}") - logger.info(f"Invocation ID: {event.invocation_id}") - logger.info(f"Is Final Response Flag: {event.is_final_response()}") - if event.content: - logger.info(f"Event Content: {str(event.content)[:500]}...") # Log content snippet - if hasattr(event, 'actions') and event.actions: - logger.info(f"Event Actions: {event.actions}") - logger.info(f"-- RAW EVENT {event_count} END --") - # --- END DETAILED EVENT LOGGING --- - - # Original logging logic below - logger.info(f"Event {event_count}: author={event.author}") - - if event.author != current_agent: - current_agent = event.author - logger.info(f"Agent execution changed to: {current_agent}") - - if event.content and event.content.parts: - part = event.content.parts[0] - if hasattr(part, 'text') and part.text: - logger.info(f"Text content: {part.text[:100]}...") - - if hasattr(event, 'actions') and event.actions: - if hasattr(event.actions, 'state_delta') and event.actions.state_delta: - state_delta = event.actions.state_delta - logger.info(f"State delta received: {state_delta}") - - # Check for final response *only* from the coordinator agent if event.is_final_response() and event.author == self.coordinator_agent.name: - logger.warning(f"Event {event_count} from COORDINATOR ({event.author}) flagged as FINAL. Breaking loop.") - if event.content and event.content.parts: - part = event.content.parts[0] - if hasattr(part, 'text') and part.text: - logger.info(f"Final response text: {part.text[:100]}...") break - elif event.is_final_response(): - # Log but don't break if a sub-agent marks as final - logger.info(f"Event {event_count} from sub-agent {event.author} flagged as FINAL, but continuing sequence.") - # Get the updated session - logger.info("Retrieving updated session") updated_session = self.session_service.get_session( app_name=APP_NAME, user_id=USER_ID, session_id=session_id ) - # Process agent outputs from state results = {} - # Process each agent output for key in ["budget_analysis", "savings_strategy", "debt_reduction"]: value = updated_session.state.get(key) if value is not None: - logger.info(f"Found {key} in state: type={type(value)}") - if value == "": - logger.warning(f"{key} is empty in state, using default") results[key] = default_results[key] continue @@ -377,9 +259,7 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns try: parsed_value = json.loads(value) results[key] = parsed_value - logger.info(f"Successfully parsed {key} as JSON") except json.JSONDecodeError: - logger.warning(f"Could not parse {key} as JSON, using as is: {value[:100]}...") if key in default_results: results[key] = default_results[key] else: @@ -387,7 +267,6 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns else: results[key] = value else: - logger.warning(f"{key} not found in session state, using default") results[key] = default_results[key] return results @@ -396,66 +275,46 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns logger.exception(f"Error during finance analysis: {str(e)}") raise finally: - # Clean up the session - try: - self.session_service.delete_session( - app_name=APP_NAME, - user_id=USER_ID, - session_id=session_id - ) - logger.info(f"Cleaned up session: {session_id}") - except Exception as e: - logger.warning(f"Failed to clean up session: {e}") + self.session_service.delete_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id + ) def _preprocess_transactions(self, session): - """Preprocess transaction data for easier analysis by the agents""" transactions = session.state.get("transactions", []) - if not transactions: return - # Convert list of transactions to DataFrame for analysis df = pd.DataFrame(transactions) - # Basic preprocessing if 'Date' in df.columns: df['Date'] = pd.to_datetime(df['Date']) df['Month'] = df['Date'].dt.month df['Year'] = df['Date'].dt.year - # Calculate spending by category if 'Category' in df.columns and 'Amount' in df.columns: category_spending = df.groupby('Category')['Amount'].sum().to_dict() session.state["category_spending"] = category_spending - - # Total spending total_spending = df['Amount'].sum() session.state["total_spending"] = total_spending def _preprocess_manual_expenses(self, session): - """Process manually entered expenses""" manual_expenses = session.state.get("manual_expenses", {}) - if not manual_expenses: return - # Calculate total spending from manual entries total_manual_spending = sum(manual_expenses.values()) session.state["total_manual_spending"] = total_manual_spending - - # Store categorized spending directly session.state["manual_category_spending"] = manual_expenses def _create_default_results(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: - """Create default results in case agent execution fails""" monthly_income = financial_data.get("monthly_income", 0) expenses = {} - # Extract expenses from manual entries or transactions if financial_data.get("manual_expenses"): expenses = financial_data.get("manual_expenses") elif financial_data.get("transactions"): - # Simplified aggregation of transactions for transaction in financial_data.get("transactions", []): category = transaction.get("Category", "Uncategorized") amount = transaction.get("Amount", 0) @@ -466,7 +325,6 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns total_expenses = sum(expenses.values()) - # Create default budget analysis default_budget = { "total_expenses": total_expenses, "monthly_income": monthly_income, @@ -479,7 +337,6 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns ] } - # Create default savings strategy default_savings = { "emergency_fund": { "recommended_amount": total_expenses * 6, @@ -495,7 +352,6 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns ] } - # Create default debt reduction default_debts = financial_data.get("debts", []) total_debt = sum(debt.get("amount", 0) for debt in default_debts) @@ -526,29 +382,17 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns } def display_budget_analysis(analysis: Dict[str, Any]): - """Display budget analysis results""" - logger.info(f"Displaying budget analysis, type: {type(analysis)}") - - # Ensure we have a dictionary if isinstance(analysis, str): - logger.info(f"Budget analysis is a string, attempting to parse as JSON") try: analysis = json.loads(analysis) - logger.info("Successfully parsed budget analysis from JSON string") - except json.JSONDecodeError as e: - logger.error(f"Failed to parse budget analysis results: {e}") - logger.error(f"First 200 chars of analysis: {analysis[:200]}") + except json.JSONDecodeError: st.error("Failed to parse budget analysis results") return if not isinstance(analysis, dict): - logger.error(f"Invalid budget analysis format: {type(analysis)}") st.error("Invalid budget analysis format") return - logger.info(f"Budget analysis keys: {list(analysis.keys())}") - - # Display spending breakdown if "spending_categories" in analysis: st.subheader("Spending by Category") fig = px.pie( @@ -558,7 +402,6 @@ def display_budget_analysis(analysis: Dict[str, Any]): ) st.plotly_chart(fig) - # Display income vs expenses if "total_expenses" in analysis: st.subheader("Income vs. Expenses") income = analysis.get("monthly_income", 0) @@ -576,7 +419,6 @@ def display_budget_analysis(analysis: Dict[str, Any]): f"${surplus_deficit:.2f}", delta=f"{surplus_deficit:.2f}") - # Display spending reduction recommendations if "recommendations" in analysis: st.subheader("Spending Reduction Recommendations") for rec in analysis["recommendations"]: @@ -585,8 +427,6 @@ def display_budget_analysis(analysis: Dict[str, Any]): st.metric(f"Potential Monthly Savings", f"${rec['potential_savings']:.2f}") def display_savings_strategy(strategy: Dict[str, Any]): - """Display savings strategy results""" - # Ensure we have a dictionary if isinstance(strategy, str): try: strategy = json.loads(strategy) @@ -600,35 +440,29 @@ def display_savings_strategy(strategy: Dict[str, Any]): st.subheader("Savings Recommendations") - # Emergency Fund if "emergency_fund" in strategy: ef = strategy["emergency_fund"] st.markdown(f"### Emergency Fund") st.markdown(f"**Recommended Size**: ${ef['recommended_amount']:.2f}") st.markdown(f"**Current Status**: {ef['current_status']}") - # Progress bar if "current_amount" in ef and "recommended_amount" in ef: progress = ef["current_amount"] / ef["recommended_amount"] st.progress(min(progress, 1.0)) st.markdown(f"${ef['current_amount']:.2f} of ${ef['recommended_amount']:.2f}") - # Savings Recommendations if "recommendations" in strategy: st.markdown("### Recommended Savings Allocations") for rec in strategy["recommendations"]: st.markdown(f"**{rec['category']}**: ${rec['amount']:.2f}/month") st.markdown(f"_{rec['rationale']}_") - # Automation Techniques if "automation_techniques" in strategy: st.markdown("### Automation Techniques") for technique in strategy["automation_techniques"]: st.markdown(f"**{technique['name']}**: {technique['description']}") def display_debt_reduction(plan: Dict[str, Any]): - """Display debt reduction plan results""" - # Ensure we have a dictionary if isinstance(plan, str): try: plan = json.loads(plan) @@ -640,23 +474,19 @@ def display_debt_reduction(plan: Dict[str, Any]): st.error("Invalid debt reduction format") return - # Total Debt Overview if "total_debt" in plan: st.metric("Total Debt", f"${plan['total_debt']:.2f}") - # Debt Breakdown if "debts" in plan: st.subheader("Your Debts") debt_df = pd.DataFrame(plan["debts"]) st.dataframe(debt_df) - # Debt visualization fig = px.bar(debt_df, x="name", y="amount", color="interest_rate", labels={"name": "Debt", "amount": "Amount ($)", "interest_rate": "Interest Rate (%)"}, title="Debt Breakdown") st.plotly_chart(fig) - # Payoff Plans if "payoff_plans" in plan: st.subheader("Debt Payoff Plans") tabs = st.tabs(["Avalanche Method", "Snowball Method", "Comparison"]) @@ -670,11 +500,6 @@ def display_debt_reduction(plan: Dict[str, Any]): if "monthly_payment" in avalanche: st.markdown(f"**Recommended Monthly Payment**: ${avalanche['monthly_payment']:.2f}") - - if "schedule" in avalanche: - st.markdown("#### Payoff Schedule") - schedule_df = pd.DataFrame(avalanche["schedule"]) - st.dataframe(schedule_df) with tabs[1]: st.markdown("### Snowball Method (Smallest Balance First)") @@ -685,11 +510,6 @@ def display_debt_reduction(plan: Dict[str, Any]): if "monthly_payment" in snowball: st.markdown(f"**Recommended Monthly Payment**: ${snowball['monthly_payment']:.2f}") - - if "schedule" in snowball: - st.markdown("#### Payoff Schedule") - schedule_df = pd.DataFrame(snowball["schedule"]) - st.dataframe(schedule_df) with tabs[2]: st.markdown("### Method Comparison") @@ -713,7 +533,6 @@ def display_debt_reduction(plan: Dict[str, Any]): fig.update_layout(barmode='group', title="Debt Payoff Method Comparison") st.plotly_chart(fig) - # Recommendations if "recommendations" in plan: st.subheader("Debt Reduction Recommendations") for rec in plan["recommendations"]: @@ -724,24 +543,22 @@ def display_debt_reduction(plan: Dict[str, Any]): def main(): st.set_page_config(page_title="AI Personal Finance Coach", layout="wide") - # Check if we have the API key - if not os.getenv("GOOGLE_API_KEY"): - logger.error("GOOGLE_API_KEY environment variable not set") - st.error(""" - GOOGLE_API_KEY not found in environment variables. - Please create a .env file with your Google API key: - ``` - GOOGLE_API_KEY=your_api_key_here - ``` - """) + # Sidebar with API key info + with st.sidebar: + st.info("📝 Please ensure you have your Gemini API key in the .env file:\n```\nGOOGLE_API_KEY=your_api_key_here\n```") + st.caption("This application uses Google's Gemini AI to provide personalized financial advice.") + + if not GEMINI_API_KEY: + st.error("GOOGLE_API_KEY not found in environment variables. Please add it to your .env file.") return st.title("📊 AI Personal Finance Coach") st.subheader("Get personalized financial advice from AI agents") + st.info("This tool analyzes your financial data and provides tailored recommendations for budgeting, savings, and debt management.") st.markdown("---") - # --- Input Section --- st.header("Step 1: Enter Your Financial Information") + st.caption("All data is processed locally and not stored anywhere.") col1, col2 = st.columns(2) @@ -770,22 +587,18 @@ def main(): try: transactions_df = pd.read_csv(transaction_file) st.success("Transaction file uploaded successfully!") - # Optional: Display small preview - # st.dataframe(transactions_df.head(3)) except Exception as e: st.error(f"Error reading CSV: {e}") - transactions_df = None # Ensure df is None if error + transactions_df = None else: use_manual_expenses = True st.write("Enter monthly expenses by category:") categories = ["Housing", "Utilities", "Food", "Transportation", "Healthcare", "Entertainment", "Personal", "Savings", "Other"] - # Use columns for better manual entry layout exp_col1, exp_col2 = st.columns(2) for i, category in enumerate(categories): col = exp_col1 if i < (len(categories) + 1) // 2 else exp_col2 manual_expenses[category] = col.number_input(f"{category} ($)", min_value=0.0, step=50.0, value=0.0, key=f"manual_{category}") - # Display manual entries for confirmation if any(manual_expenses.values()): st.write("Entered Manual Expenses:") manual_df_disp = pd.DataFrame({ @@ -794,8 +607,8 @@ def main(): }) st.dataframe(manual_df_disp[manual_df_disp['Amount'] > 0]) - st.subheader("Debt Information") + st.info("Enter your debts to get personalized payoff strategies.") num_debts = st.number_input("Number of Debts", min_value=0, max_value=10, step=1, value=0, key="num_debts") debts = [] @@ -815,24 +628,20 @@ def main(): "interest_rate": interest_rate, "min_payment": min_payment }) - + st.markdown("---") analyze_button = st.button("Analyze My Finances", key="analyze_button") st.markdown("---") - # --- Results Section --- if analyze_button: - # Validate inputs before proceeding if expense_option == "Upload CSV Transactions" and transactions_df is None: st.error("Please upload a valid transaction CSV file or choose manual entry.") return if use_manual_expenses and not any(manual_expenses.values()): st.warning("No manual expenses entered. Analysis might be limited.") - # Optionally proceed or return, depending on desired behavior st.header("Step 2: Financial Analysis Results") with st.spinner("AI agents are analyzing your financial data..."): - # Prepare data for agent analysis financial_data = { "monthly_income": monthly_income, "dependants": dependants, @@ -841,53 +650,35 @@ def main(): "debts": debts } - # Create finance advisor system finance_system = FinanceAdvisorSystem() - # Run analysis - logger.info("Starting financial analysis") - results = None try: results = asyncio.run(finance_system.analyze_finances(financial_data)) - logger.info(f"Analysis complete, results keys: {list(results.keys())}") - # Log the types of each result - for key, value in results.items(): - logger.info(f"Result '{key}' is type: {type(value)}") - # if value: # Avoid logging large outputs unless needed - # preview = str(value)[:100] + "..." if len(str(value)) > 100 else str(value) - # logger.info(f"Preview of {key}: {preview}") + tabs = st.tabs(["💰 Budget Analysis", "📈 Savings Strategy", "đŸ’ŗ Debt Reduction"]) + + with tabs[0]: + st.subheader("Budget Analysis") + if "budget_analysis" in results and results["budget_analysis"]: + display_budget_analysis(results["budget_analysis"]) + else: + st.write("No budget analysis available.") + + with tabs[1]: + st.subheader("Savings Strategy") + if "savings_strategy" in results and results["savings_strategy"]: + display_savings_strategy(results["savings_strategy"]) + else: + st.write("No savings strategy available.") + + with tabs[2]: + st.subheader("Debt Reduction Plan") + if "debt_reduction" in results and results["debt_reduction"]: + display_debt_reduction(results["debt_reduction"]) + else: + st.write("No debt reduction plan available.") except Exception as e: - logger.exception(f"Error in financial analysis: {e}") st.error(f"An error occurred during analysis: {str(e)}") - # results remains None - - # Display results if analysis was successful - if results: - tabs = st.tabs(["💰 Budget Analysis", "📈 Savings Strategy", "đŸ’ŗ Debt Reduction"]) - - with tabs[0]: - st.subheader("Budget Analysis") - if "budget_analysis" in results and results["budget_analysis"]: - display_budget_analysis(results["budget_analysis"]) - else: - st.write("No budget analysis available or analysis failed.") - - with tabs[1]: - st.subheader("Savings Strategy") - if "savings_strategy" in results and results["savings_strategy"]: - display_savings_strategy(results["savings_strategy"]) - else: - st.write("No savings strategy available or analysis failed.") - - with tabs[2]: - st.subheader("Debt Reduction Plan") - if "debt_reduction" in results and results["debt_reduction"]: - display_debt_reduction(results["debt_reduction"]) - else: - st.write("No debt reduction plan available or analysis failed.") - else: - st.error("Financial analysis could not be completed.") if __name__ == "__main__": main() \ No newline at end of file From 9975e7eea278614b212e497392c1839ff9b2ea74 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 22:10:33 +0530 Subject: [PATCH 5/6] completed the script - readme --- .../ai_financial_coach_agent/.env | 2 +- .../ai_financial_coach_agent/README.md | 78 +++ .../ai_financial_coach_agent.py | 565 ++++++++++++++---- .../ai_financial_coach_agent/requirements.txt | 4 +- 4 files changed, 520 insertions(+), 129 deletions(-) diff --git a/ai_agent_tutorials/ai_financial_coach_agent/.env b/ai_agent_tutorials/ai_financial_coach_agent/.env index b8d7980..bc02262 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/.env +++ b/ai_agent_tutorials/ai_financial_coach_agent/.env @@ -1 +1 @@ -GOOGLE_API_KEY= \ No newline at end of file +GOOGLE_API_KEY=your_gemini_api_key_here \ No newline at end of file diff --git a/ai_agent_tutorials/ai_financial_coach_agent/README.md b/ai_agent_tutorials/ai_financial_coach_agent/README.md index 8b13789..c73671e 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/README.md +++ b/ai_agent_tutorials/ai_financial_coach_agent/README.md @@ -1 +1,79 @@ +# AI Financial Coach Agent with Google ADK 💰 +The **AI Financial Coach** is a personalized financial advisor powered by Google's ADK (Agent Development Kit) framework. This app provides comprehensive financial analysis and recommendations based on user inputs including income, expenses, debts, and financial goals. + +## Features + +- **Multi-Agent Financial Analysis System** + - Budget Analysis Agent: Analyzes spending patterns and recommends optimizations + - Savings Strategy Agent: Creates personalized savings plans and emergency fund strategies + - Debt Reduction Agent: Develops optimized debt payoff strategies using avalanche and snowball methods + +- **Expense Analysis**: + - Supports both CSV upload and manual expense entry + - CSV transaction analysis with date, category, and amount tracking + - Visual breakdown of spending by category + - Automated expense categorization and pattern detection + +- **Savings Recommendations**: + - Emergency fund sizing and building strategies + - Custom savings allocations across different goals + - Practical automation techniques for consistent saving + - Progress tracking and milestone recommendations + +- **Debt Management**: + - Multiple debt handling with interest rate optimization + - Comparison between avalanche and snowball methods + - Visual debt payoff timeline and interest savings analysis + - Actionable debt reduction recommendations + +- **Interactive Visualizations**: + - Pie charts for expense breakdown + - Bar charts for income vs. expenses + - Debt comparison graphs + - Progress tracking metrics + + +## How to Run + +Follow the steps below to set up and run the application: + +1. **Get API Key**: + - Get a free Gemini API Key from Google AI Studio: https://aistudio.google.com/apikey + - Create a `.env` file in the project root and add your API key: + ``` + GOOGLE_API_KEY=your_api_key_here + ``` + +2. **Clone the Repository**: + ```bash + git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git + cd awesome-llm-apps/ai_agent_tutorials/ai_financial_coach_agent + ``` + +3. **Install Dependencies**: + ```bash + pip install -r requirements.txt + ``` + +4. **Run the Streamlit App**: + ```bash + streamlit run ai_financial_coach_agent.py + ``` + +## CSV File Format + +The application accepts CSV files with the following required columns: +- `Date`: Transaction date in YYYY-MM-DD format +- `Category`: Expense category +- `Amount`: Transaction amount (supports currency symbols and comma formatting) + +Example: +```csv +Date,Category,Amount +2024-01-01,Housing,1200.00 +2024-01-02,Food,150.50 +2024-01-03,Transportation,45.00 +``` + +A template CSV file can be downloaded directly from the application's sidebar. diff --git a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py index 43b4f6c..42d93d2 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py +++ b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py @@ -2,7 +2,7 @@ import streamlit as st import pandas as pd import plotly.express as px import plotly.graph_objects as go -from typing import Dict, List, Optional, Tuple, Any, AsyncGenerator +from typing import Dict, List, Optional, Tuple, Any import os import asyncio from datetime import datetime @@ -10,6 +10,8 @@ from dotenv import load_dotenv import json import logging from pydantic import BaseModel, Field +import csv +from io import StringIO from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent from google.adk.agents.invocation_context import InvocationContext @@ -540,145 +542,456 @@ def display_debt_reduction(plan: Dict[str, Any]): if "impact" in rec: st.markdown(f"_Impact: {rec['impact']}_") -def main(): - st.set_page_config(page_title="AI Personal Finance Coach", layout="wide") +def parse_csv_transactions(file_content) -> List[Dict[str, Any]]: + """Parse CSV file content into a list of transactions""" + try: + # Read CSV content + df = pd.read_csv(StringIO(file_content.decode('utf-8'))) + + # Validate required columns + required_columns = ['Date', 'Category', 'Amount'] + missing_columns = [col for col in required_columns if col not in df.columns] + + if missing_columns: + raise ValueError(f"Missing required columns: {', '.join(missing_columns)}") + + # Convert date strings to datetime and then to string format YYYY-MM-DD + df['Date'] = pd.to_datetime(df['Date']).dt.strftime('%Y-%m-%d') + + # Convert amount strings to float, handling currency symbols and commas + df['Amount'] = df['Amount'].replace('[\$,]', '', regex=True).astype(float) + + # Group by category and calculate totals + category_totals = df.groupby('Category')['Amount'].sum().reset_index() + + # Convert to list of dictionaries + transactions = df.to_dict('records') + + return { + 'transactions': transactions, + 'category_totals': category_totals.to_dict('records') + } + except Exception as e: + raise ValueError(f"Error parsing CSV file: {str(e)}") + +def validate_csv_format(file) -> bool: + """Validate CSV file format and content""" + try: + content = file.read().decode('utf-8') + dialect = csv.Sniffer().sniff(content) + has_header = csv.Sniffer().has_header(content) + file.seek(0) # Reset file pointer + + if not has_header: + return False, "CSV file must have headers" + + df = pd.read_csv(StringIO(content)) + required_columns = ['Date', 'Category', 'Amount'] + missing_columns = [col for col in required_columns if col not in df.columns] + + if missing_columns: + return False, f"Missing required columns: {', '.join(missing_columns)}" + + # Validate date format + try: + pd.to_datetime(df['Date']) + except: + return False, "Invalid date format in Date column" + + # Validate amount format (should be numeric after removing currency symbols) + try: + df['Amount'].replace('[\$,]', '', regex=True).astype(float) + except: + return False, "Invalid amount format in Amount column" + + return True, "CSV format is valid" + except Exception as e: + return False, f"Invalid CSV format: {str(e)}" + +def display_csv_preview(df: pd.DataFrame): + """Display a preview of the CSV data with basic statistics""" + st.subheader("CSV Data Preview") - # Sidebar with API key info + # Show basic statistics + total_transactions = len(df) + total_amount = df['Amount'].sum() + + # Convert dates for display + df_dates = pd.to_datetime(df['Date']) + date_range = f"{df_dates.min().strftime('%Y-%m-%d')} to {df_dates.max().strftime('%Y-%m-%d')}" + + col1, col2, col3 = st.columns(3) + with col1: + st.metric("Total Transactions", total_transactions) + with col2: + st.metric("Total Amount", f"${total_amount:,.2f}") + with col3: + st.metric("Date Range", date_range) + + # Show category breakdown + st.subheader("Spending by Category") + category_totals = df.groupby('Category')['Amount'].agg(['sum', 'count']).reset_index() + category_totals.columns = ['Category', 'Total Amount', 'Transaction Count'] + st.dataframe(category_totals) + + # Show sample transactions + st.subheader("Sample Transactions") + st.dataframe(df.head()) + +def main(): + st.set_page_config( + page_title="AI Financial Coach with Google ADK", + layout="wide", + initial_sidebar_state="expanded" + ) + + # Sidebar with API key info and CSV template with st.sidebar: + st.title("🔑 Setup & Templates") st.info("📝 Please ensure you have your Gemini API key in the .env file:\n```\nGOOGLE_API_KEY=your_api_key_here\n```") - st.caption("This application uses Google's Gemini AI to provide personalized financial advice.") + st.caption("This application uses Google's ADK (Agent Development Kit) and Gemini AI to provide personalized financial advice.") + + st.divider() + + # Add CSV template download + st.subheader("📊 CSV Template") + st.markdown(""" + Download the template CSV file with the required format: + - Date (YYYY-MM-DD) + - Category + - Amount (numeric) + """) + + # Create sample CSV content + sample_csv = """Date,Category,Amount +2024-01-01,Housing,1200.00 +2024-01-02,Food,150.50 +2024-01-03,Transportation,45.00""" + + st.download_button( + label="đŸ“Ĩ Download CSV Template", + data=sample_csv, + file_name="expense_template.csv", + mime="text/csv" + ) if not GEMINI_API_KEY: - st.error("GOOGLE_API_KEY not found in environment variables. Please add it to your .env file.") + st.error("🔑 GOOGLE_API_KEY not found in environment variables. Please add it to your .env file.") return - st.title("📊 AI Personal Finance Coach") - st.subheader("Get personalized financial advice from AI agents") - st.info("This tool analyzes your financial data and provides tailored recommendations for budgeting, savings, and debt management.") - st.markdown("---") + # Main content + st.title("📊 AI Financial Coach with Google ADK") + st.caption("Powered by Google's Agent Development Kit (ADK) and Gemini AI") + st.info("This tool analyzes your financial data and provides tailored recommendations for budgeting, savings, and debt management using multiple specialized AI agents.") + st.divider() - st.header("Step 1: Enter Your Financial Information") - st.caption("All data is processed locally and not stored anywhere.") + # Create tabs for different sections + input_tab, about_tab = st.tabs(["đŸ’ŧ Financial Information", "ℹī¸ About"]) - col1, col2 = st.columns(2) - - with col1: - st.subheader("Income & Dependants") - monthly_income = st.number_input("Monthly Income ($)", min_value=0.0, step=100.0, value=3000.0, key="income") - dependants = st.number_input("Number of Dependants", min_value=0, step=1, value=0, key="dependants") - - with col2: - st.subheader("Expense Data") - expense_option = st.radio( - "How do you want to enter expenses?", - ("Upload CSV Transactions", "Enter Manually"), - key="expense_option" - ) + with input_tab: + st.header("Enter Your Financial Information") + st.caption("All data is processed locally and not stored anywhere.") - transaction_file = None - manual_expenses = {} - use_manual_expenses = False - transactions_df = None + # Income and Dependants section in a container + with st.container(): + st.subheader("💰 Income & Household") + income_col, dependants_col = st.columns([2, 1]) + with income_col: + monthly_income = st.number_input( + "Monthly Income ($)", + min_value=0.0, + step=100.0, + value=3000.0, + key="income", + help="Enter your total monthly income after taxes" + ) + with dependants_col: + dependants = st.number_input( + "Number of Dependants", + min_value=0, + step=1, + value=0, + key="dependants", + help="Include all dependants in your household" + ) + + st.divider() + + # Expenses section + with st.container(): + st.subheader("đŸ’ŗ Expenses") + expense_option = st.radio( + "How would you like to enter your expenses?", + ("📤 Upload CSV Transactions", "✍ī¸ Enter Manually"), + key="expense_option", + horizontal=True + ) + + transaction_file = None + manual_expenses = {} + use_manual_expenses = False + transactions_df = None - if expense_option == "Upload CSV Transactions": - st.write("Upload a CSV with columns: Date, Category, Amount") - transaction_file = st.file_uploader("Upload CSV of transactions", type=["csv"], key="transaction_file") - if transaction_file is not None: + if expense_option == "📤 Upload CSV Transactions": + col1, col2 = st.columns([2, 1]) + with col1: + st.markdown(""" + #### Upload your transaction data + Your CSV file should have these columns: + - 📅 Date (YYYY-MM-DD) + - 📝 Category + - 💲 Amount + """) + + transaction_file = st.file_uploader( + "Choose your CSV file", + type=["csv"], + key="transaction_file", + help="Upload a CSV file containing your transactions" + ) + + if transaction_file is not None: + # Validate CSV format + is_valid, message = validate_csv_format(transaction_file) + + if is_valid: + try: + # Parse CSV content + transaction_file.seek(0) + file_content = transaction_file.read() + parsed_data = parse_csv_transactions(file_content) + + # Create DataFrame + transactions_df = pd.DataFrame(parsed_data['transactions']) + + # Display preview + display_csv_preview(transactions_df) + + st.success("✅ Transaction file uploaded and validated successfully!") + except Exception as e: + st.error(f"❌ Error processing CSV file: {str(e)}") + transactions_df = None + else: + st.error(message) + transactions_df = None + else: + use_manual_expenses = True + st.markdown("#### Enter your monthly expenses by category") + + # Define expense categories with emojis + categories = [ + ("🏠 Housing", "Housing"), + ("🔌 Utilities", "Utilities"), + ("đŸŊī¸ Food", "Food"), + ("🚗 Transportation", "Transportation"), + ("đŸĨ Healthcare", "Healthcare"), + ("🎭 Entertainment", "Entertainment"), + ("👤 Personal", "Personal"), + ("💰 Savings", "Savings"), + ("đŸ“Ļ Other", "Other") + ] + + # Create three columns for better layout + col1, col2, col3 = st.columns(3) + cols = [col1, col2, col3] + + # Distribute categories across columns + for i, (emoji_cat, cat) in enumerate(categories): + with cols[i % 3]: + manual_expenses[cat] = st.number_input( + emoji_cat, + min_value=0.0, + step=50.0, + value=0.0, + key=f"manual_{cat}", + help=f"Enter your monthly {cat.lower()} expenses" + ) + + if any(manual_expenses.values()): + st.markdown("#### 📊 Summary of Entered Expenses") + manual_df_disp = pd.DataFrame({ + 'Category': list(manual_expenses.keys()), + 'Amount': list(manual_expenses.values()) + }) + manual_df_disp = manual_df_disp[manual_df_disp['Amount'] > 0] + if not manual_df_disp.empty: + col1, col2 = st.columns([2, 1]) + with col1: + st.dataframe( + manual_df_disp, + column_config={ + "Category": "Category", + "Amount": st.column_config.NumberColumn( + "Amount", + format="$%.2f" + ) + }, + hide_index=True + ) + with col2: + st.metric( + "Total Monthly Expenses", + f"${manual_df_disp['Amount'].sum():,.2f}" + ) + + st.divider() + + # Debt Information section + with st.container(): + st.subheader("đŸĻ Debt Information") + st.info("Enter your debts to get personalized payoff strategies using both avalanche and snowball methods.") + + num_debts = st.number_input( + "How many debts do you have?", + min_value=0, + max_value=10, + step=1, + value=0, + key="num_debts" + ) + + debts = [] + if num_debts > 0: + # Create columns for debts + cols = st.columns(min(num_debts, 3)) # Max 3 columns per row + for i in range(num_debts): + col_idx = i % 3 + with cols[col_idx]: + st.markdown(f"##### Debt #{i+1}") + debt_name = st.text_input( + "Name", + value=f"Debt {i+1}", + key=f"debt_name_{i}", + help="Enter a name for this debt (e.g., Credit Card, Student Loan)" + ) + debt_amount = st.number_input( + "Amount ($)", + min_value=0.01, + step=100.0, + value=1000.0, + key=f"debt_amount_{i}", + help="Enter the current balance of this debt" + ) + interest_rate = st.number_input( + "Interest Rate (%)", + min_value=0.0, + max_value=100.0, + step=0.1, + value=5.0, + key=f"debt_rate_{i}", + help="Enter the annual interest rate" + ) + min_payment = st.number_input( + "Minimum Payment ($)", + min_value=0.0, + step=10.0, + value=50.0, + key=f"debt_min_payment_{i}", + help="Enter the minimum monthly payment required" + ) + + debts.append({ + "name": debt_name, + "amount": debt_amount, + "interest_rate": interest_rate, + "min_payment": min_payment + }) + + if col_idx == 2 or i == num_debts - 1: # Add spacing after every 3 debts or last debt + st.markdown("---") + + st.divider() + + # Analysis button + col1, col2, col3 = st.columns([1, 2, 1]) + with col2: + analyze_button = st.button( + "🔄 Analyze My Finances", + key="analyze_button", + use_container_width=True, + help="Click to get your personalized financial analysis" + ) + + if analyze_button: + if expense_option == "Upload CSV Transactions" and transactions_df is None: + st.error("Please upload a valid transaction CSV file or choose manual entry.") + return + if use_manual_expenses and not any(manual_expenses.values()): + st.warning("No manual expenses entered. Analysis might be limited.") + + st.header("Financial Analysis Results") + with st.spinner("🤖 AI agents are analyzing your financial data..."): + financial_data = { + "monthly_income": monthly_income, + "dependants": dependants, + "transactions": transactions_df.to_dict('records') if transactions_df is not None else None, + "manual_expenses": manual_expenses if use_manual_expenses else None, + "debts": debts + } + + finance_system = FinanceAdvisorSystem() + try: - transactions_df = pd.read_csv(transaction_file) - st.success("Transaction file uploaded successfully!") + results = asyncio.run(finance_system.analyze_finances(financial_data)) + + tabs = st.tabs(["💰 Budget Analysis", "📈 Savings Strategy", "đŸ’ŗ Debt Reduction"]) + + with tabs[0]: + st.subheader("Budget Analysis") + if "budget_analysis" in results and results["budget_analysis"]: + display_budget_analysis(results["budget_analysis"]) + else: + st.write("No budget analysis available.") + + with tabs[1]: + st.subheader("Savings Strategy") + if "savings_strategy" in results and results["savings_strategy"]: + display_savings_strategy(results["savings_strategy"]) + else: + st.write("No savings strategy available.") + + with tabs[2]: + st.subheader("Debt Reduction Plan") + if "debt_reduction" in results and results["debt_reduction"]: + display_debt_reduction(results["debt_reduction"]) + else: + st.write("No debt reduction plan available.") except Exception as e: - st.error(f"Error reading CSV: {e}") - transactions_df = None - else: - use_manual_expenses = True - st.write("Enter monthly expenses by category:") - categories = ["Housing", "Utilities", "Food", "Transportation", "Healthcare", - "Entertainment", "Personal", "Savings", "Other"] - exp_col1, exp_col2 = st.columns(2) - for i, category in enumerate(categories): - col = exp_col1 if i < (len(categories) + 1) // 2 else exp_col2 - manual_expenses[category] = col.number_input(f"{category} ($)", min_value=0.0, step=50.0, value=0.0, key=f"manual_{category}") - if any(manual_expenses.values()): - st.write("Entered Manual Expenses:") - manual_df_disp = pd.DataFrame({ - 'Category': list(manual_expenses.keys()), - 'Amount': list(manual_expenses.values()) - }) - st.dataframe(manual_df_disp[manual_df_disp['Amount'] > 0]) - - st.subheader("Debt Information") - st.info("Enter your debts to get personalized payoff strategies.") - num_debts = st.number_input("Number of Debts", min_value=0, max_value=10, step=1, value=0, key="num_debts") + st.error(f"An error occurred during analysis: {str(e)}") - debts = [] - if num_debts > 0: - debt_cols = st.columns(num_debts) - for i in range(num_debts): - with debt_cols[i]: - st.markdown(f"**Debt #{i+1}**") - debt_name = st.text_input(f"Name", value=f"Debt {i+1}", key=f"debt_name_{i}") - debt_amount = st.number_input(f"Amount $", min_value=0.01, step=100.0, value=1000.0, key=f"debt_amount_{i}") - interest_rate = st.number_input(f"Interest Rate (%)", min_value=0.0, max_value=100.0, step=0.1, value=5.0, key=f"debt_rate_{i}") - min_payment = st.number_input(f"Min. Payment $", min_value=0.0, step=10.0, value=50.0, key=f"debt_min_payment_{i}") - - debts.append({ - "name": debt_name, - "amount": debt_amount, - "interest_rate": interest_rate, - "min_payment": min_payment - }) - - st.markdown("---") - analyze_button = st.button("Analyze My Finances", key="analyze_button") - st.markdown("---") - - if analyze_button: - if expense_option == "Upload CSV Transactions" and transactions_df is None: - st.error("Please upload a valid transaction CSV file or choose manual entry.") - return - if use_manual_expenses and not any(manual_expenses.values()): - st.warning("No manual expenses entered. Analysis might be limited.") - - st.header("Step 2: Financial Analysis Results") - with st.spinner("AI agents are analyzing your financial data..."): - financial_data = { - "monthly_income": monthly_income, - "dependants": dependants, - "transactions": transactions_df.to_dict('records') if transactions_df is not None else None, - "manual_expenses": manual_expenses if use_manual_expenses else None, - "debts": debts - } - - finance_system = FinanceAdvisorSystem() - - try: - results = asyncio.run(finance_system.analyze_finances(financial_data)) - - tabs = st.tabs(["💰 Budget Analysis", "📈 Savings Strategy", "đŸ’ŗ Debt Reduction"]) - - with tabs[0]: - st.subheader("Budget Analysis") - if "budget_analysis" in results and results["budget_analysis"]: - display_budget_analysis(results["budget_analysis"]) - else: - st.write("No budget analysis available.") - - with tabs[1]: - st.subheader("Savings Strategy") - if "savings_strategy" in results and results["savings_strategy"]: - display_savings_strategy(results["savings_strategy"]) - else: - st.write("No savings strategy available.") - - with tabs[2]: - st.subheader("Debt Reduction Plan") - if "debt_reduction" in results and results["debt_reduction"]: - display_debt_reduction(results["debt_reduction"]) - else: - st.write("No debt reduction plan available.") - except Exception as e: - st.error(f"An error occurred during analysis: {str(e)}") + with about_tab: + st.markdown(""" + ### About AI Financial Coach + + This application uses Google's Agent Development Kit (ADK) to provide comprehensive financial analysis and advice through multiple specialized AI agents: + + 1. **🔍 Budget Analysis Agent** + - Analyzes spending patterns + - Identifies areas for cost reduction + - Provides actionable recommendations + + 2. **💰 Savings Strategy Agent** + - Creates personalized savings plans + - Calculates emergency fund requirements + - Suggests automation techniques + + 3. **đŸ’ŗ Debt Reduction Agent** + - Develops optimal debt payoff strategies + - Compares different repayment methods + - Provides actionable debt reduction tips + + ### Privacy & Security + + - All data is processed locally + - No financial information is stored or transmitted + - Secure API communication with Google's services + + ### Need Help? + + For support or questions: + - Check the [documentation](https://github.com/Shubhamsaboo/awesome-llm-apps) + - Report issues on [GitHub](https://github.com/Shubhamsaboo/awesome-llm-apps/issues) + """) if __name__ == "__main__": main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_financial_coach_agent/requirements.txt b/ai_agent_tutorials/ai_financial_coach_agent/requirements.txt index b642de4..2f6d591 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/requirements.txt +++ b/ai_agent_tutorials/ai_financial_coach_agent/requirements.txt @@ -1,5 +1,5 @@ -google-adk==0.4.0 -streamlit==1.31.0 +google-adk==0.1.0 +streamlit pandas==2.1.1 matplotlib==3.8.0 numpy==1.26.0 From 71fcb45624222095facb9c998d2bb0de66bbe3c5 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 13 Apr 2025 22:37:08 +0530 Subject: [PATCH 6/6] FINAL CHANGE --- .../ai_financial_coach_agent.py | 169 +++++++----------- 1 file changed, 68 insertions(+), 101 deletions(-) diff --git a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py index 42d93d2..2e7c361 100644 --- a/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py +++ b/ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py @@ -2,7 +2,7 @@ import streamlit as st import pandas as pd import plotly.express as px import plotly.graph_objects as go -from typing import Dict, List, Optional, Tuple, Any +from typing import Dict, List, Optional, Any import os import asyncio from datetime import datetime @@ -13,14 +13,10 @@ from pydantic import BaseModel, Field import csv from io import StringIO -from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent -from google.adk.agents.invocation_context import InvocationContext -from google.adk.events import Event, EventActions -from google.adk.sessions import InMemorySessionService, Session +from google.adk.agents import LlmAgent, SequentialAgent +from google.adk.sessions import InMemorySessionService from google.adk.runners import Runner from google.genai import types -from google.adk.agents.callback_context import CallbackContext -from google.adk.models import LlmResponse, LlmRequest logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -91,9 +87,15 @@ class DebtReduction(BaseModel): recommendations: Optional[List[DebtRecommendation]] = Field(None, description="Recommendations for debt reduction") load_dotenv() - GEMINI_API_KEY = os.getenv("GOOGLE_API_KEY") +def parse_json_safely(data: str, default_value: Any = None) -> Any: + """Safely parse JSON data with error handling""" + try: + return json.loads(data) if isinstance(data, str) else data + except json.JSONDecodeError: + return default_value + class FinanceAdvisorSystem: def __init__(self): self.session_service = InMemorySessionService() @@ -219,12 +221,10 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns state=initial_state ) - transactions = session.state.get("transactions") - if transactions: + if session.state.get("transactions"): self._preprocess_transactions(session) - manual_expenses = session.state.get("manual_expenses") - if manual_expenses: + if session.state.get("manual_expenses"): self._preprocess_manual_expenses(session) default_results = self._create_default_results(financial_data) @@ -249,27 +249,9 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns ) results = {} - for key in ["budget_analysis", "savings_strategy", "debt_reduction"]: value = updated_session.state.get(key) - if value is not None: - if value == "": - results[key] = default_results[key] - continue - - if isinstance(value, str): - try: - parsed_value = json.loads(value) - results[key] = parsed_value - except json.JSONDecodeError: - if key in default_results: - results[key] = default_results[key] - else: - results[key] = value - else: - results[key] = value - else: - results[key] = default_results[key] + results[key] = parse_json_safely(value, default_results[key]) if value else default_results[key] return results @@ -291,96 +273,81 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns df = pd.DataFrame(transactions) if 'Date' in df.columns: - df['Date'] = pd.to_datetime(df['Date']) - df['Month'] = df['Date'].dt.month - df['Year'] = df['Date'].dt.year + df['Date'] = pd.to_datetime(df['Date']).dt.strftime('%Y-%m-%d') if 'Category' in df.columns and 'Amount' in df.columns: category_spending = df.groupby('Category')['Amount'].sum().to_dict() session.state["category_spending"] = category_spending - total_spending = df['Amount'].sum() - session.state["total_spending"] = total_spending + session.state["total_spending"] = df['Amount'].sum() def _preprocess_manual_expenses(self, session): manual_expenses = session.state.get("manual_expenses", {}) if not manual_expenses: return - total_manual_spending = sum(manual_expenses.values()) - session.state["total_manual_spending"] = total_manual_spending - session.state["manual_category_spending"] = manual_expenses + session.state.update({ + "total_manual_spending": sum(manual_expenses.values()), + "manual_category_spending": manual_expenses + }) def _create_default_results(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: monthly_income = financial_data.get("monthly_income", 0) - expenses = {} + expenses = financial_data.get("manual_expenses", {}) - if financial_data.get("manual_expenses"): - expenses = financial_data.get("manual_expenses") - elif financial_data.get("transactions"): - for transaction in financial_data.get("transactions", []): + if not expenses and financial_data.get("transactions"): + expenses = {} + for transaction in financial_data["transactions"]: category = transaction.get("Category", "Uncategorized") amount = transaction.get("Amount", 0) - if category in expenses: - expenses[category] += amount - else: - expenses[category] = amount + expenses[category] = expenses.get(category, 0) + amount total_expenses = sum(expenses.values()) - default_budget = { - "total_expenses": total_expenses, - "monthly_income": monthly_income, - "spending_categories": [ - {"category": cat, "amount": amt, "percentage": (amt / total_expenses * 100) if total_expenses > 0 else 0} - for cat, amt in expenses.items() - ], - "recommendations": [ - {"category": "General", "recommendation": "Consider reviewing your expenses carefully", "potential_savings": total_expenses * 0.1} - ] - } - - default_savings = { - "emergency_fund": { - "recommended_amount": total_expenses * 6, - "current_amount": 0, - "current_status": "Not started" - }, - "recommendations": [ - {"category": "Emergency Fund", "amount": total_expenses * 0.1, "rationale": "Build emergency fund first"}, - {"category": "Retirement", "amount": monthly_income * 0.15, "rationale": "Long-term savings"} - ], - "automation_techniques": [ - {"name": "Automatic Transfer", "description": "Set up automatic transfers on payday"} - ] - } - - default_debts = financial_data.get("debts", []) - total_debt = sum(debt.get("amount", 0) for debt in default_debts) - - default_debt = { - "total_debt": total_debt, - "debts": default_debts, - "payoff_plans": { - "avalanche": { - "total_interest": total_debt * 0.2, - "months_to_payoff": 24, - "monthly_payment": total_debt / 24 - }, - "snowball": { - "total_interest": total_debt * 0.25, - "months_to_payoff": 24, - "monthly_payment": total_debt / 24 - } - }, - "recommendations": [ - {"title": "Increase Payments", "description": "Increase your monthly payments", "impact": "Reduces total interest paid"} - ] - } - return { - "budget_analysis": default_budget, - "savings_strategy": default_savings, - "debt_reduction": default_debt + "budget_analysis": { + "total_expenses": total_expenses, + "monthly_income": monthly_income, + "spending_categories": [ + {"category": cat, "amount": amt, "percentage": (amt / total_expenses * 100) if total_expenses > 0 else 0} + for cat, amt in expenses.items() + ], + "recommendations": [ + {"category": "General", "recommendation": "Consider reviewing your expenses carefully", "potential_savings": total_expenses * 0.1} + ] + }, + "savings_strategy": { + "emergency_fund": { + "recommended_amount": total_expenses * 6, + "current_amount": 0, + "current_status": "Not started" + }, + "recommendations": [ + {"category": "Emergency Fund", "amount": total_expenses * 0.1, "rationale": "Build emergency fund first"}, + {"category": "Retirement", "amount": monthly_income * 0.15, "rationale": "Long-term savings"} + ], + "automation_techniques": [ + {"name": "Automatic Transfer", "description": "Set up automatic transfers on payday"} + ] + }, + "debt_reduction": { + "total_debt": sum(debt.get("amount", 0) for debt in financial_data.get("debts", [])), + "debts": financial_data.get("debts", []), + "payoff_plans": { + "avalanche": { + "total_interest": sum(debt.get("amount", 0) for debt in financial_data.get("debts", [])) * 0.2, + "months_to_payoff": 24, + "monthly_payment": sum(debt.get("amount", 0) for debt in financial_data.get("debts", [])) / 24 + }, + "snowball": { + "total_interest": sum(debt.get("amount", 0) for debt in financial_data.get("debts", [])) * 0.25, + "months_to_payoff": 24, + "monthly_payment": sum(debt.get("amount", 0) for debt in financial_data.get("debts", [])) / 24 + } + }, + "recommendations": [ + {"title": "Increase Payments", "description": "Increase your monthly payments", "impact": "Reduces total interest paid"} + ] + } } def display_budget_analysis(analysis: Dict[str, Any]):