admin-valentin/awesome-llm-apps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

FINAL CHANGE

Browse source
This commit is contained in:
Madhu 2025-04-13 22:37:08 +05:30
parent 9975e7eea2
commit 71fcb45624
1 changed files with 68 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

169
ai_agent_tutorials/ai_financial_coach_agent/ai_financial_coach_agent.py
View file

@ -2,7 +2,7 @@ import streamlit as st
import pandas as pd import pandas as pd
import plotly.express as px import plotly.express as px
import plotly.graph_objects as go import plotly.graph_objects as go
from typing import Dict, List, Optional, Tuple, Any from typing import Dict, List, Optional, Any
import os import os
import asyncio import asyncio
from datetime import datetime from datetime import datetime
@ -13,14 +13,10 @@ from pydantic import BaseModel, Field
import csv import csv
from io import StringIO from io import StringIO
from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.agents.invocation_context import InvocationContext from google.adk.sessions import InMemorySessionService
from google.adk.events import Event, EventActions
from google.adk.sessions import InMemorySessionService, Session
from google.adk.runners import Runner from google.adk.runners import Runner
from google.genai import types 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) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -91,9 +87,15 @@ class DebtReduction(BaseModel):
recommendations: Optional[List[DebtRecommendation]] = Field(None, description="Recommendations for debt reduction") recommendations: Optional[List[DebtRecommendation]] = Field(None, description="Recommendations for debt reduction")
load_dotenv() load_dotenv()
GEMINI_API_KEY = os.getenv("GOOGLE_API_KEY") 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: class FinanceAdvisorSystem:
def __init__(self): def __init__(self):
self.session_service = InMemorySessionService() self.session_service = InMemorySessionService()
@ -219,12 +221,10 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns
state=initial_state state=initial_state
) )
transactions = session.state.get("transactions") if session.state.get("transactions"):
if transactions:
self._preprocess_transactions(session) self._preprocess_transactions(session)
manual_expenses = session.state.get("manual_expenses") if session.state.get("manual_expenses"):
if manual_expenses:
self._preprocess_manual_expenses(session) self._preprocess_manual_expenses(session)
default_results = self._create_default_results(financial_data) 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 = {} results = {}
for key in ["budget_analysis", "savings_strategy", "debt_reduction"]: for key in ["budget_analysis", "savings_strategy", "debt_reduction"]:
value = updated_session.state.get(key) value = updated_session.state.get(key)
if value is not None: results[key] = parse_json_safely(value, default_results[key]) if value else default_results[key]
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]
return results return results
@ -291,96 +273,81 @@ IMPORTANT: Store your final plan in state['debt_reduction'] and ensure it aligns
df = pd.DataFrame(transactions) df = pd.DataFrame(transactions)
if 'Date' in df.columns: if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date']) df['Date'] = pd.to_datetime(df['Date']).dt.strftime('%Y-%m-%d')
df['Month'] = df['Date'].dt.month
df['Year'] = df['Date'].dt.year
if 'Category' in df.columns and 'Amount' in df.columns: if 'Category' in df.columns and 'Amount' in df.columns:
category_spending = df.groupby('Category')['Amount'].sum().to_dict() category_spending = df.groupby('Category')['Amount'].sum().to_dict()
session.state["category_spending"] = category_spending session.state["category_spending"] = category_spending
total_spending = df['Amount'].sum() session.state["total_spending"] = df['Amount'].sum()
session.state["total_spending"] = total_spending
def _preprocess_manual_expenses(self, session): def _preprocess_manual_expenses(self, session):
manual_expenses = session.state.get("manual_expenses", {}) manual_expenses = session.state.get("manual_expenses", {})
if not manual_expenses: if not manual_expenses:
return return
total_manual_spending = sum(manual_expenses.values()) session.state.update({
session.state["total_manual_spending"] = total_manual_spending "total_manual_spending": sum(manual_expenses.values()),
session.state["manual_category_spending"] = manual_expenses "manual_category_spending": manual_expenses
})
def _create_default_results(self, financial_data: Dict[str, Any]) -> Dict[str, Any]: def _create_default_results(self, financial_data: Dict[str, Any]) -> Dict[str, Any]:
monthly_income = financial_data.get("monthly_income", 0) monthly_income = financial_data.get("monthly_income", 0)
expenses = {} expenses = financial_data.get("manual_expenses", {})
if financial_data.get("manual_expenses"): if not expenses and financial_data.get("transactions"):
expenses = financial_data.get("manual_expenses") expenses = {}
elif financial_data.get("transactions"): for transaction in financial_data["transactions"]:
for transaction in financial_data.get("transactions", []):
category = transaction.get("Category", "Uncategorized") category = transaction.get("Category", "Uncategorized")
amount = transaction.get("Amount", 0) amount = transaction.get("Amount", 0)
if category in expenses: expenses[category] = expenses.get(category, 0) + amount
expenses[category] += amount
else:
expenses[category] = amount
total_expenses = sum(expenses.values()) 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 { return {
"budget_analysis": default_budget, "budget_analysis": {
"savings_strategy": default_savings, "total_expenses": total_expenses,
"debt_reduction": default_debt "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]): def display_budget_analysis(analysis: Dict[str, Any]):