How to Build an AI Budget-Planning Optimizer for Your 2026 CAPEX Review: LangGraph, FastAPI, and n8n

was the UI for complex budget planning, with an AI agent handling the optimisation behind the scenes?

If I could just send ‘Run 2026 budget with XXX caps and Y% on Sustainability’ and get a decision back, that would change our budget review.

That’s how a logistics VP framed the problem when we discussed automating portfolio selection for their annual CAPEX approval cycle.

Because companies invest in expensive logistics equipment, directors receive CAPEX applications from operations teams for short and long-term projects.

This is a complex exercise as you need to balance between return on investment and long-term strategy.

As a data scientist, how can we support this decision-making process?

Using linear programming, we can help decide which projects to fund to maximise ROI while respecting multi-year budget constraints.

In this article, we will build an AI Agent for Budget Planning that turns email requests into an optimised CAPEX portfolio.

This AI workflow is built using n8n for orchestration and LangGraph to create a reasoning agent connected to a FastAPI microservice that executes a linear programming model.

We will review the architecture and explore the results using a real request for budget optimisation with a constraint of a minimum 20% allocation for sustainability projects.

Budget Planning Optimisation with Python

Problem Statement: Operational Budget Planning

We are supporting the APAC director of a large Third-Party Logistics Service Provider (3PL) based in Singapore.

Their job is to manage warehousing and transportation operations for other companies in four countries in the Asia Pacific.

We are talking about operations supporting 48 customers grouped in over eight market verticals (Luxury, Cosmetics …).

For instance, they manage a 10,000 sqm warehouse for a large fast-fashion retailer in Beijing, delivering to 50 stores across North China.

Warehouse Manager: We need 150k euros for a new conveyor belt that will increase our receiving productivity by 20%.

Our director receives a list of projects that require capital expenditure (CAPEX) from his 17 warehouse managers across the APAC region.

For each project, the CAPEX application includes a brief description (e.g., renting 500 sqm), a three-year cost profile (Year 1: €115k; Year 2: €120k; Year 3: €150k), and the expected Return On Investment (e.g., +€50k).

These projects can also bring additional benefits:

• Business Development: unlock new revenue (e.g., capacity for a new customer or product line)
• Sustainability (CO₂ Reduction): lower emissions via energy-efficient equipment or layout changes
• Digital Transformation: improve data visibility, automate repetitive tasks, and enable AI-driven decisions
• Operational Excellence: increase throughput, reduce defects and rework, shorten changeovers, and stabilise processes.
• HSE (Health, Safety & Environment): reduce incident risk and insurance exposure with safer equipment
• CSR (Corporate Social Responsibility): strengthen community and workforce initiatives (e.g., training, accessibility)

For instance, a warehouse automation project can reduce packaging use (sustainability), decrease operator strain (CSR), and accelerate digital transformation.

Some of the additional benefits are linked to the top management guidelines that our director needs to follow.

APAC Director: “How should I allocate my budget of XX M€ to maximise the ROI while respecting my top management guidelines?“

As the director receives over 50 projects each session, we proposed to build a linear programming module to determine the optimal selection, taking into account external constraints.

As inputs, we get the available budget and the management objectives, along with the spreadsheet of CAPEX applications.

This will be part of the inputs of our solution.

FastAPI Microservice: 0–1 Mixed-Integer Optimiser for CAPEX Budget Planning

To address this problem, we utilised the modelling framework for Linear Programming (LP) and Integer Programming (IP) problems provided by the PuLP library of Python.

The solution is packaged in a FastAPI microservice deployed on the cloud.

Decision Variables

For each project i, we define a binary value to inform if we allocate a budget or not:

Objective Function

The objective is to maximise the total return on investment of the portfolio of projects we selected:

Constraints

As we cannot spend more than what we have allocated per year, we must consider the budget constraints for the next three years.

APAC Director: Our CEO want us to invest 20% of our budget on projects supporting our sustainability roadmap.

Moreover, we may also have constraints on the minimum budget for specific management objectives.

In the example above, we ensure that the total investment for sustainability projects is equal to or greater than S_min.

Now that we have defined our model, we can package it as a FastAPI microservice using the code shared in this article.

This overall workflow will accept a specific schema defined using Pydantic for validation and defaults.

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
    
class EmailRequest(BaseModel):
    email_text: str

LaunchParamsBudget captures the optimisation inputs:

• The three annual budget caps (budget_year1/2/3 in euros)
• The optional minimum-allocation rule toggled by set_min_budget with its target management objective min_budget_objective and the required share min_budget_perc in %.

We need to ensure that the Agent respects this schema; if not, it will be unable to query the API.

What can we expect as outputs?

"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,
"allocated_year2": 1421150,
"allocated_year3": 1515150,
"percentage_allocated": 92.97,
"projects_positive_roi": 38,
"projects_negative_roi": 20,
"n_selected_projects": 35,
"sustainable_projects": 21,
"business_dev_projects": 17,
"hse_projects": 17,
"digital_projects": 16,
"CSR_projects": 4,
"excellence_projects": 35
}

The agent will receive the details of the allocated budget, including the annual split, along with information on the selected projects.

With these results and a proper system prompt, our agent can provide a concise summary of the budget allocation to the VP of Logistics.

Building an Agentic Workflow with LangGraph and n8

End-to-End Workflow Description

Let us assume that our Asia Pacific Director received a spreadsheet with all the CAPEX applications.

The idea is to receive the hypotheses regarding the budget amount and constraints in plain English via email, along with the spreadsheet attached.

Based on this email, the workflow should automatically select projects for optimal budget allocation.

But also, the director expects a concise explanation of the choices made to build this optimal “investment portfolio”.

We can build an automated workflow using LangGraph and n8n for orchestration to perform these tasks.

This will be a workflow in 4 steps:

• Step 1: Using the n8n Gmail node, we send the email body and the spreadsheet attached to the FastAPI Backend
• Step 2: An AI Agent Parser will collect the parameters from the email and call the other API for Budget Planning
• Step 3: The outputs will be sent to an AI Agent Summarizer that will use them to generate the summary
• Step 4: The user receives a reply by email with the summary via a Gmail node in n8n

The core LangGraph agentic workflow will be implemented within a FastAPI microservice, which will be wrapped in an n8n workflow for orchestration.

The first two nodes on the top will extract the content of the spreadsheet and upload it to the API.

Then, we extract the email content in the Extract Email Body node and send it to the agent endpoint using the node Query AI Agent.

The outputs of the API include:

• A detailed explanation of the optimal budget allocation that are sent by email using the Reply node
• A JSON with the allocation by project used by the node Update Allocation to add the ✅ and ❌ in the spreadsheet

Ultimately, our director receives a comprehensive analysis of the optimal portfolio, which includes three distinct sections.

A Budget Summary provides details on the allocated budget and the return on investment.

The Portfolio Composition details the number of projects awarded per management objective.

Finally, the agent concludes with a Recommendation based on its understanding of the objective.

Let us see how we built the core LangGraph Budget Agent, which parses an email to return a complete analysis.

Budget Planning AI Agent with LangGraph

The code shared in this section has been greatly simplified for concision.

Before creating the graph, let us build the different blocks:

• EmailParser is used to parse the email body received from the n8n HTTP node to extract the parameters of the Budget API
• BudgetPlanInterpreter will call the API, retrieve the results and generate the summary

For your information, I have used this approach with standalone blocks because we reuse them in other workflows that combine multiple agents.

Block 1: AI Agent Parser

Let us build the block for this agent that we will call AI Agent Parser:

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(
                "model_name", "anthropic:claude-3-7-sonnet-latest"
            )
        self.llm = init_chat_model(model_name)
        self.params: dict | None = None
        self.raw_email: str | None = None

    def parse(self, content: str | None = None) -> dict:
        """
        Parse email content to extract parameters for API
        """
        content = content.strip()
        system_prompt = config.get("budget_agent", {}).get(
            "system_prompt_parser", {}
        )
        structured_llm = self.llm.with_structured_output(LaunchParamsBudget)
        result = structured_llm.invoke(
            [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": content},
            ]
        )
        payload = result.model_dump()
        self.params = payload
        logger.info(f"[BudgetAgent] Parsed params: {self.params}")
        return self.params

We include the parameters that will be used for the Graph State variables.

This block EmailParser turns a plain-text email body into typed, schema-valid parameters for our Budget Planning API using an LLM.

1. On initialisation, we load the chat model from the config and build a LangChain chat model.
2. The function parse() takes the raw email content sent by n8n HTTP node and the system prompt to invoke the model with structured outputs defined with the LaunchParamsBudget Pydantic schema.

The EmailParser system_prompt stored in the YAML config file (minimal version for concision):

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)
    - min_budget_objective: string (e.g., "Sustainability")
    - min_budget_perc: number (percentage between 0 and 100)

    Output ONLY these fields; no extra keys.

It includes a list of fields to parse, along with concise explanations and strict format rules.

The output looks like this:

{
'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
}

This will be sent to the second agent for call tooling (our other FastAPI Microservice) and results interpretation.

Block 2: Block for Tool Calling and Interpretation

Two blocks will be used to call the Budget Planning API and to process the API’s output.

As I will be using this Budget Planning API for multiple agents, I created a standalone function to call it.

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:
    payload = {
        "objective": params.objective,
        "budget_year1": params.budget_year1,
        "budget_year2": params.budget_year2,
        "budget_year3": params.budget_year3,
        "set_min_budget": params.set_min_budget,
        "min_budget_objective": params.min_budget_objective,
        "min_budget_perc": params.min_budget_perc,
    }
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(5, read=30)) as c:
            r = await c.post(LAUNCH, 
                             json=payload, 
                             headers={"session_id": session_id})
            r.raise_for_status()
            return r.json()
    except httpx.HTTPError as e:
        logger.error("[BudgetAgent]: API call failed: %s", e)
        code = getattr(e.response, "status_code", "")
        return {"error": f"{code} {e}"}

We can now call this function in the block BudgetPlanInterpreter defined below.

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()

class BudgetPlanInterpreter:
    def __init__(self, 
                model_name: str = "anthropic:claude-3-7-sonnet-latest", 
                session_id: str = "test_agent"):
        self.llm = init_chat_model(model_name)
        self.session_id = session_id
        self.api_result: dict | None = None 
        self.html_summary: str | None = None

    async def run_plan(self, params: dict) -> dict:
        """Run budget planning using FastAPI Microservice API"""
        try:
            launch_params = LaunchParamsBudget(**params) 
            self.api_result = await run_budget_api(launch_params, self.session_id)
            return self.api_result
        except Exception as e:
            logger.error(f"[BudgetAgent] Direct API call failed: {e}")
            return {"error": str(e)}

    async def interpret(self, params: dict, api_output: dict | None = None) -> str:
        """Interpret API budget planning results into HTML summary for the Director"""
        if api_output is None:
            if not self.api_result:
                raise ValueError("No API result available.")
            api_output = self.api_result

        messages = [
            {
                "role": "system",
                "content": config["budget_agent"]["system_prompt_tool"]
            },
            {
                "role": "user",
                "content": f"Input parameters: {params}n
                             nModel results: {api_output}"
            }
        ]

        reply = self.llm.invoke(messages)
        self.html_summary = reply.content

        logger.info("[BudgetPlanAgent] Generated HTML summary")
        return self.html_summary

    def get_summary(self) -> str:
        if not self.html_summary:
            raise ValueError("No summary available.")
        return self.html_summary

It runs the budget optimiser and turns its JSON into an “executive summary” in HTML format.

• We use run_plan() to call the FastAPI microservice utilising the function run_budget_api(s) to retrieve the results
• The function interpret() generates the analysis based on the API’s outputs, following the instructions of a system prompt.

We now have our three foundation blocks that can be used to build a graph with nodes and conditional edges.

LangGraph Builder with nodes and conditional edges

Now that we have the three blocks, we can build our graph.

This is a small experimental LangGraph state machine that includes the three steps (parsing, running, interpretation) with error handling at every step.

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()

class AgentState(TypedDict, total=False):
    email_text: str
    params: dict
    budget_results: dict
    html_summary: str
    error: str
    session_id: str

This block is fundamental as it defines the shared state passed between LangGraph nodes

• session_id: included in the API call
• email_text: raw email body received from the n8n node
• params: structured inputs parsed from the email by EmailParser
• budget_results: JSON output of the FastAPI microservice
• html_summary: the analysis (in HTML format) generated by interpret() based on the output budget_results
• error: storing an error message if needed

We should now define the functions of each node that will receive the current AgentState and return a partial update.

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

Function : handle_error(state)

• Returns an error message in HTML format, which will be sent to the Gmail node in n8n to notify the user

In this case, for the support team, it is more practical as the user just has to forward the email.

Note: In the production version, we have added a run_id to help track the issues in the logs.

async def parse_email(state: AgentState) -> AgentState:
    try:
        parser = EmailParser(model_name=config["budget_agent"]["model_name"])
        params = parser.parse(state["email_text"])

        if not params or ("error" in params and params["error"]):
            return {"error": f"Parse failed: {params.get('error', 'unknown')}"}

        return {"params": params}
    except Exception as e:
        logger.exception("[BudgetGraph] parse_email crashed")
        return {"error": f"Parse exception: {e}"}

Function 2: parse_email(state)

• Uses EmailParser to transform the email body received into parameters for Budget Planning in JSON format
• On success: returns {"params": …} used to call the FastAPI Microservice

async def run_budget(state: AgentState) -> AgentState:
    if "error" in state:
        return {}
    try:
        interpreter = BudgetPlanInterpreter(
            model_name=config["budget_agent"]["model_name"],
            session_id=state.get("session_id", 
                                 config["budget_agent"]["session_id"]))
        results = await interpreter.run_plan(state["params"])

        if "error" in results:
            return {"error": f"Budget run failed: {results['error']}"}

        return {"budget_results": results, "interpreter": interpreter}
    except Exception as e:
        logger.exception("[BudgetGraph] run_budget crashed")
        return {"error": f"Budget exception: {e}"}

Function 3: run_budget(state)

• Uses BudgetPlanInterpreter to call the function run_plan that will execute the budget optimisation via the FastAPI microservice
• On success: returns the output of the optimiser in JSON format as budget_results

This output can be used to generate the summary of the budget allocation.

async def summarize(state: AgentState) -> AgentState:
    if "error" in state:
        return {}
    try:
        interpreter = state.get("interpreter") or BudgetPlanInterpreter(
            model_name=config["budget_agent"]["model_name"],
            session_id=state.get("session_id", config["budget_agent"]["session_id"]),
        )
        html = await interpreter.interpret(state["params"], state["budget_results"])
        return {"html_summary": html}
    except Exception as e:
        logger.exception("[BudgetGraph] summarize crashed")
        return {"error": f"Summarization exception: {e}"}

Function 4: summarize(state)

• Reuses interpreter from state (or creates one), then calls interpret()
• On success: returns a concise and professional summary of the budget allocation in HTML format, ready to be sent by email {"html_summary": …}

This output html_summary is then returned by the API to the Gmail node on n8n to reply to the sender.

Now that we have all the functions, we can create the nodes and “wire” the Graph using the function build_budagent_graph() defined below:

def route_after_parse(state: AgentState) -> str:
    return "error" if "error" in state else "run"

def route_after_run(state: AgentState) -> str:
    return "error" if "error" in state else "summarize"

def build_budagent_graph():
    graph_builder = StateGraph(AgentState)

    graph_builder.add_node("parse", parse_email)
    graph_builder.add_node("run", run_budget)
    graph_builder.add_node("summarize", summarize)
    graph_builder.add_node("error", handle_error)

    graph_builder.add_edge(START, "parse")
    graph_builder.add_conditional_edges("parse", route_after_parse, {
        "run": "run",
        "error": "error",
    })
    graph_builder.add_conditional_edges("run", route_after_run, {
        "summarize": "summarize",
        "error": "error",
    })
    graph_builder.add_edge("summarize", END)
    graph_builder.add_edge("error", END)

    return graph_builder.compile()

These four nodes are connected using Routers:

• route_after_parse will direct the flow based on the output of the email parsing: if error in state → go to "error"; else → "run".
• route_after_run will direct the flow based on the output of the FastAPI Microservice calling: if error in state → go to "error"; else → "summarize".

We are nearly done!

We just need to package this in a FastAPI endpoint:

@router.post("/graph_parse_and_run")
async def graph_parse_and_run(request: EmailRequest):
    """
    Parse an email body, run Budget Planning, and return an HTML summary — orchestrated via a LangGraph StateGraph.
    """
    try:
        initial_state = {
            "email_text": request.email_text,
            "session_id": config.get("budget_agent", {}).get("session_id", "test_agent"),
        }
        final_state = await _graph.ainvoke(initial_state)

        return {
            "params": final_state.get("params"),
            "budget_results": final_state.get("budget_results"),
            "html_summary": final_state.get("html_summary"),
            "error": final_state.get("error"),
        }
    except Exception as e:
        logger.exception("[BudAgent] Graph run failed")
        raise HTTPException(status_code=500, detail=f"Graph run failed: {e}")

It will be queried by the Query Agent API node of our n8n workflow to return input parameters in params, Budget Optimiser results in budget_results and the summary generated by the Agent Interpreter in html_summary.

A fully functional AI Workflow for Budget Planning

We can now activate the workflow on n8n and test the tool with different scenarios.

What if we don’t have a minimum budget for any management objectives?

I will try to adapt the email to have set_min_budget at False.

The email has been well parsed with now set_min_budget at the value False.

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

And we can see the results in the summary:

As we could expect, the performance is better:

• Total ROI: €1,050,976 (vs. €1,024,051)
• ROI/€: €0.26 (vs. €0.24)

Conclusion

This workflow has been presented to the APAC team, who started to “play with it”.

We learned that they use it to prepare slides with different scenarios of portfolio allocation for the board meetings.

This remains a “strategic tool” that is used only a couple of times per year.

However, we plan to reuse the same architecture for more “tactical” tools that supply chain departments can use for ABC Analysis, Stock Management, or Supply Chain Optimisation, as well as human resources for Workforce Planning or business controlling teams.

Can we go beyond this simple workflow?

I am still not satisfied with the contribution of the Agentic part of the workflow.

Indeed, it’s nice to have a tool that can be triggered by an email and provide a concise summary.

However, I would like to explore the idea of having multiple agents proposing different scenarios that would compete against each other.

What would be the impact on the ROI if we increase the minimum budget of sustainability by 15%?

For instance, we can ask agents to run multiple scenarios and provide a comparative study.

We are still experimenting with various types of orchestration to determine the most efficient approach.

This will be the topic of the following articles.

Other examples of Agentic Workflows?

This is not the first time I am trying to link an optimisation tool (packaged in a FastAPI Microservice) with an Agentic Workflow.

My initial attempt was to create a Production Planning Optimisation Agent.

Like here, I packaged an optimisation algorithm in a FastAPI Microservice that I wanted to connect to an email workflow.

Unlike here, the Agentic part of the workflow was handled in n8n with two Agent nodes.

The results were quite satisfying, as you can see in the short video linked below.

The user experience was very smooth.

However, the maintenance proved to be challenging for the team.

This is why we wanted to explore the use of Python and TypeScript frameworks for the agentic workflow, like here.

What’s next? Agentic Approach of Business Planning

At our startup, LogiGreen, we are attempting (through experiments like this one) to extend beyond supply chain optimisation and encompass business decision-making.

On my roadmap, I have a tool that I have developed to help small and medium-sized company optimise their cash flow.

In another article published in Towards Data Science, I have introduced how I used Python to simulate the financial flows of a company selling renewable paper cups to coffee shops.

“We have to refuse orders as we don’t have enough cash to pay suppliers for stock replenishment.”

A close friend, who owns a small business, was complaining about cash flow issues limiting the development of his company.

I started by tackling the problem of inventory management with an optimised built-in solution using Python.

Then I enriched the model considering sales channel strategy, payment terms and many other strategic business parameters to help him find the optimal business plan to maximise profitability.

This solution is probably the next candidate in our experimentation with using agentic workflows to support business and operational decision-making.

For more information, you can have a look at this short presentation of the tool

Currently, it is packaged in a FastAPI microservice connected to a React frontend (and streamlit for the public demo).

I would like to implement an AI workflow that would

1. Take multiple scenarios (like the ones presented in the video)
2. Call the API for each scenario
3. Collect and process the results
4. Provide a comparative analysis to recommend the best decision

Basically, I would like to outsource to a single (or multiple agents) the complete study presented in the article and the video.

For that, we are exploring multiple approaches to agent orchestration.

We will share our findings in a future article. Stay tuned!

About Me

Let’s connect on Linkedin and Twitter. I am a Supply Chain Engineer who uses data analytics to improve logistics operations and reduce costs.

For consulting or advice on analytics and sustainable supply chain transformation, feel free to contact me via Logigreen Consulting.

If you are interested in Data Analytics and Supply Chain, look at my website.

Samir Saci | Data Science & Productivity