samirsaci

async def handle_error(state: AgentState) -> AgentState:
    err = state.get("error", "Unknown error")
    logger.error(f"[BudgetGraph] handle_error: {err}")
    html = (
        "<b>Budget Summary</b><br>"
        "<ul><li>There was an error while processing your request.</li></ul>"
        f"<b>Details</b><br>{err}"
    )
    return {"html_summary": html}

samirsaci

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import logging
from app.utils.functions.budagent_parser import EmailParser
from app.utils.functions.budagent_functions import BudgetPlanInterpreter
from app.utils.config_loader import load_config

logger = logging.getLogger(__name__)
config = load_config()

samirsaci

import logging
import requests
from langchain.chat_models import init_chat_model
from app.utils.functions.budagent_runner import run_budget_api
from app.models.budagent_models import LaunchParamsBudget
from app.utils.config_loader import load_config

logger = logging.getLogger(__name__)
config = load_config()

samirsaci

import os, logging, httpx
import logging
from app.models.budagent_models import LaunchParamsBudget
from app.utils.config_loader import load_config
logger = logging.getLogger(__name__)
API = os.getenv("API_URL")
LAUNCH = f"{API}/budget/launch_budget"

async def run_budget_api(params: LaunchParamsBudget, 
                         session_id: str = "test_agent") -> dict:

samirsaci

{
'objective': 'Return On Investment', 
'budget_year1': 1250000, 
'budget_year2': 1500000, 
'budget_year3': 1750000, 
'set_min_budget': True, 
'min_budget_objective': 'Sustainability', 
'min_budget_perc': 20.0
}

samirsaci

budget_agent:
  system_prompt_parser:  |
    You are a budget planning analyst for LogiGreen.
    Your task is to extract structured input parameters from emails 
    requesting budget optimization runs.
    Fields to return (match exactly the schema):
    - budget_year1: integer (annual cap for Year 1)
    - budget_year2: integer (annual cap for Year 2)
    - budget_year3: integer (annual cap for Year 3)
    - set_min_budget: boolean (true/false)

samirsaci

import logging
from langchain.chat_models import init_chat_model
from app.utils.config_loader import load_config
from app.models.budagent_models import LaunchParamsBudget

logger = logging.getLogger(__name__)
config = load_config()
class EmailParser:
    def __init__(self, model_name: str | None = None):
        model_name = config.get("budget_agent", {}).get(

samirsaci

"budget_results": {
"total_budget": 4500000,
"total_budget_application": 8147425,
"total_turnover": 128880000,
"expected_roi": 715071,
"n_projects": 58,
"n_management_objective": 7,
"roi": 1024051,
"allocated_budget": 4183750,
"allocated_year1": 1247450,

samirsaci

from pydantic import BaseModel
from typing import Optional

class LaunchParamsBudget(BaseModel):
    budget_year1: int = 1250000
    budget_year2: int = 1500000
    budget_year3: int = 1750000
    set_min_budget: bool = False
    min_budget_objective: Optional[str] = 'Sustainability'
    min_budget_perc: float = 20

samirsaci

{
"params": 
{
"objective": 
"Return On Investment",
"budget_year1": 
1250000,
"budget_year2": 
1500000,