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