Automating Transaction Entry in FastAPI with a LangGraph AI Assistant

Anyone who has tried to track daily spending through a web form knows how tedious it gets. Logging a $5 coffee purchase should not require navigating multiple fields, yet that is exactly what happens in many finance-tracking apps built with FastAPI and PostgreSQL, where every transaction — however small — has to be entered manually.

Imagine building such an app, where each transaction, simple or complicated, needs to go through a form. This works fine for occasional entries, but becomes exhausting once you need to record several transactions in one sitting.

At that point, a natural question arises: what if the whole process could be automated? What if, instead of filling out a form, you could simply tell an assistant "I spent $5 on coffee at the restaurant today" and have it handle the rest?

This is where LangGraph becomes useful.

Using LangGraph, you can build an AI assistant that takes a plain-language description of what happened and turns it into a properly recorded transaction on your behalf.

Introduction

This article walks through the core concepts behind LangGraph and its surrounding ecosystem, then goes deep into how an AI assistant can be added to a FastAPI application using LangGraph to automate transaction entry.

LangGraph

LangGraph comes from the team behind LangChain, and it's an open-source toolkit for putting together and managing AI agent workflows through graph structures. With it, you describe a process as a collection of "nodes" and "edges," which keeps complicated agent behavior organized, scalable, and easier to control.

Before diving further into LangGraph, it helps to first understand LangChain, since LangGraph builds on top of it.

LangChain

LangChain is a toolkit, also open-source, for putting together applications powered by large language models. Its main job is to give developers a bridge between an LLM and outside resources — data sources, tools, and workflow steps — so that a system can carry out multi-step reasoning and automated tasks rather than a single isolated prompt-response exchange.

Purpose: It's designed for building AI applications that need to chain several steps together — for example, handling user input, retrieving relevant information, and generating a response.

Structure: LangChain relies on "chains," which are ordered sequences of operations where each step's output feeds into the next step's input. This lets you decompose complicated logic into smaller, manageable pieces.

Applications: Typical use cases include chatbots, multi-step reasoning tasks, document retrieval and summarization, and connecting LLMs to external tools or APIs.

LangGraph (contd.)

Put simply, LangGraph arranges LLM calls into graph-shaped workflows, which supports flexible and even parallel multi-step reasoning rather than a strictly linear sequence.

Purpose: It enables AI applications where the logic can branch, loop, or execute steps in parallel, going beyond what a simple sequential chain can express.

Structure: LangGraph represents operations as "nodes" and the flow of data between them as "edges." A single node's output can feed several downstream nodes, which allows for dynamic decision paths.

Applications: LangGraph is well suited for coordinating multiple agents, building complex decision pipelines, automating tasks that involve conditional logic, and orchestrating several LLMs or tools at once.

What is a Graph in LangGraph?

A graph, in general, is a non-linear data structure composed of "vertices" (nodes) and "edges" (the connections between them) that capture relationships among objects.

In LangGraph specifically, this graph structure is used to build stateful, cyclical workflows — ones where the AI can make decisions, loop back to earlier steps, or branch into different paths depending on intermediate results.

LangChain vs. LangGraph

Architecture of the Project

(FastAPI endpoint + LangGraph + transaction creation and persistence)

The Problem

Before LangGraph was introduced into the finance application, creating a transaction meant calling the /transactions/add endpoint directly, with a payload that looked like this:

{
    title: "Coffee for Rosy",
    type: "expense",
    amount: 5,
    note: "Paid $5 to Rosy for coffee",
    category_id: "5bc22126-5982-4500-9e74-71c9c089f0c8",
    payment_option_id: "07c5d180-fa4d-4435-aa04-b54ef436eca1"
}

Reaching that point required two prior API calls — one to fetch the list of categories and another to fetch payment_options — just to obtain the IDs needed for the payload. In other words, creating a single transaction was a three-step process, and a slow one at that.

The Solution

The fix was to let an AI assistant take care of all three steps, while the user only needs to describe, in ordinary language, what they did with their money. With that goal in mind, here is how the implementation is structured.

The Three-Steps Architecture

1. FastAPI Endpoint
2. LangGraph Orchestrator
3. PostgreSQL Database

1. FastAPI Endpoint

The user sends a request to the FastAPI endpoint /assistance/transaction-entry, with a payload that contains a message describing the transaction.

{
  message: "Sent $5 to Rosy for Coffee through cash."
}

2. LangGraph Orchestrator

The orchestrator is built as a graph, where each node represents an operation and each edge represents the flow of data between operations.

The first node, the Analyzer LLM, receives the user's message and checks whether it contains everything needed to record a transaction — whether it's income or an expense, the amount involved, the purpose of the transaction, the payment method used, and so on.

If the message already contains all the required details, the LLM converts it into structured data, for example:

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category: "coffee",
    payment_option: "cash",
    payment_type: "Cash",
    is_complete: True,  // Flag
    missing_info_message: None  // Flag
}

This structured data then moves to the Database Writer node, after the two flag fields (is_complete and missing_info_message) are stripped out. The Database Writer node calls the create_transaction() method, which records the transaction against the user in the database.

But what happens if the message is missing some detail? Consider a message like:

{
  message: "Sent $5 to Rosy for Coffee." // payment mode is not specified
}

Here, the payment mode is left unspecified. In this case, the data extracted by the LLM will include populated flag values, such as:

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee",
    category: "coffee",
    payment_option: None,
    payment_type: None,
    is_complete: False,  // Flag
    missing_info_message: "Please enter the payment mode used for this expense." // Flag
}

Since the is_complete flag is False here, the missing_info_message gets routed to the other node connected to the Analyzer LLM: the Clarification node. This path is only triggered when is_complete evaluates to False.

The Clarification node receives the missing_info_message and calls the ask_again() method, which returns that message as the response to the original FastAPI request. This marks the end of the graph's execution for this run — the output the user receives is simply a prompt asking for the missing detail, in this case the payment mode.

Suppose the user then replies with the missing information, for example:

{
  message: "UPI"
}

This reply causes the orchestrator graph to be initialized again, and it goes through the same sequence of steps as before.

The key difference in this second pass is that none of the earlier information is lost — the conversation history is preserved each time data is extracted by the LLM (this persistence mechanism is covered in more detail later in the article). Because the payment_option is now available and combined with the previously captured values, is_complete switches to True, and the finalized, filtered data is passed along to the Database Writer node, looking like this:

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category: "coffee",
    payment_option: "UPI",
    payment_type: "Digital"
}
// Flags removed.

The Database Writer node then takes over with this filtered data in hand. Recall that when a transaction entry was created manually, it took two additional API calls — one for categories and one for payment_options — just to resolve their respective IDs before the transaction itself could be created. The same problem shows up here: the filtered data holds the actual text values for category and payment option, not their database IDs, and the database won't accept raw values for these fields.

To resolve this, the Database Writer node has to query the database to look up the matching category and payment option records based on the values present in the filtered data.

Since the project relies on FastAPI together with SQLAlchemy, these lookups are implemented as SQLAlchemy queries.

For categories:

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

# Run a select query to check if the category in data.category exists or not.
stmt = select(CategoriesModel).where(
  CategoriesModel.user_id == user_id,
  func.lower(getattr(CategoriesModel, name)) == data.category.lower()
)

# Execute the query.
result = await session.execute(stmt)

# If category exists, assign its ID to data.category.
existing_category = resule.scalar_one_or_none()
if existing_category:
  data.category = existing_category.id

# If category doens't exits, create a new category and save it to database.
new_category = CategoriesModel(**{name: data.category, "user_id": user_id})
session.add(new_row)

try:
  await session.flush()
except IntegrityError:
  # In case another concurrent request created it first,
  # we need to roll back and fetch it again.
  await session.rollback()

  result = await session.execute(stmt)
  existing_category = result.scalar_one_or_none()
  if existing_category:
      data.category = existing_category.id
  raise

In short, this logic: Runs a select query to check whether the category referenced in data.category already exists. If it does, the category's ID replaces the value in data.category. If it doesn't, a new category record gets created, and its newly generated ID is used instead.

The same pattern applies to payment_options:

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

# Run a select query to check if the peyment_option in data.payment_option exists or not.
stmt = select(PaymentOptionsModel).where(
  PaymentOptionsModel.user_id == user_id,
  func.lower(getattr(PaymentOptionsModel, name)) == data.payment_option.lower()
)

# Execute the query.
result = await session.execute(stmt)

# If payment_option exists, assign its ID to data.payment_option.
existing_option = resule.scalar_one_or_none()
if existing_option:
  data.payment_option = existing_option.id

# If payment_option doesn't exits, create a new payment_option and save it to database.
new_option = PaymentOptionsModel(**{name: data.payment_option, "user_id": user_id})
session.add(new_row)

try:
  await session.flush()
except IntegrityError:
  # In case another concurrent request created it first,
  # we need to roll back and fetch it again.
  await session.rollback()

  result = await session.execute(stmt)
  existing_option = result.scalar_one_or_none()
  if existing_option:
      data.payment_option = existing_option.id
  raise

Once both the category ID and payment option ID have been resolved, the data object is fully updated and ready for insertion, looking like this:

{
    title: "Coffee",
    type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category_id: "5bc22126-5982-4500-9e74-71c9c089f0c8",
    payment_option_id: "07c5d180-fa4d-4435-aa04-b54ef436eca1"
}

With this finalized data, the Database Writer node calls the create_transaction() method, which actually persists the transaction record into the database.

3. PostgreSQL Database

This represents the last stage of the architecture, where the finalized data handed off by the Database Writer node gets written into the transactions table.

The resulting structure of the transactions table looks like this:

The Implementation

With the architecture covered, it's time to walk through the actual implementation details of building this orchestrator with LangGraph. Note that the order followed here doesn't exactly mirror the architectural walkthrough above. Instead, the implementation is organized as follows:

1. LangGraph Orchestrator
2. PostgreSQL Database
3. FastAPI Endpoint

1. LangGraph Orchestrator

The orchestrator itself lives inside src/assistance/graph.py. This file is responsible for setting up the LLM, defining the nodes of the graph, wiring the connections between those nodes, and finally compiling everything into a runnable graph.

As mentioned earlier, three nodes make up this orchestrator: the Analyzer LLM, the Database Writer, and the Clarification node.

Analyzer LLM Node (Groq)

This node is essentially a language model whose job is to figure out the user's intent and verify whether the message contains all the required, correct details. Rather than building a custom model from scratch, this project relies on Groq to handle the heavy lifting.

What is Groq?

Groq is an open-source Python framework built for working with graph-structured data. It gives developers an expressive way to query, filter, and aggregate information stored as graphs, and it's well suited to large graph datasets, including things like social networks, knowledge graphs, or recommendation engines, as described in a GeekForGeeks writeup on the Groq API.

Through Groq's hosted API, you can send prompts to widely used open models — openai/gpt-oss-120b being the one used in this project — and receive responses that tend to arrive markedly quicker than what you'd typically get from other providers serving comparable models.

Why Groq?

Groq was chosen over alternatives such as ChatOpenAI or ChatAnthropic for a few reasons:

• Speed: Groq relies on purpose-built hardware called LPUs (Language Processing Units) rather than the GPUs most other providers depend on, which translates into fast inference.
• A usable free tier: Groq's free tier is generous enough to support a solo or learning-oriented project without generating meaningful API costs during experimentation.
• Drop-in compatibility through LangChain: the langchain_groq.ChatGroq class integrates with LangChain and LangGraph the same way ChatOpenAI or ChatAnthropic would. That means switching to a different provider down the line wouldn't require reworking the graph logic — just swapping out the client.

How to get a Groq API Key

Groq lets you generate free API keys for development. Here's how to obtain one:

1. Go to https://console.groq.com and either log in or sign up.
2. Select the API Keys option from the navigation bar.
3. Choose Create API Key.
4. A form will appear asking for a name (this project used transaction-assistant) and an expiration period for the key. Submit the form once filled in.
5. The key is displayed only once, immediately after creation — so make sure to copy it right away.

Once created, all your keys will show up in the main list on that page.

Using the Groq API Key inside the FastAPI code

Add the Groq API key to your .env file at the project root, alongside your other environment variables:

GROQ_API_KEY = "gsk_***************************************DyxM"

There are several approaches to loading environment variables into the modules that need them. This project uses a dedicated settings class:

Define a Settings class inside src/utils/settings.py:

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # Configure connection with the .env file
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    # ... Other Variables ...
    GROQ_API_KEY: str


settings = Settings()

Then import that settings object wherever it's needed:

from src.utils.settings import settings

# After importing, the object settings can be used as
# "settings.GROQ_API_KEY" to access the environment variable for Groq API Key.

LLM Setup

Before configuring the LLM, install LangGraph and LangChain along with the Groq integration: pip install -U langgraph langchain langchain-groq

An instance of the Groq client is then created and configured with a specific model:

from langchain_groq import ChatGroq
from src.assitance.schema import ExtractedTransactionSchema
from src.utils.settings import settings

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

Here, ChatGroq acts as LangChain's wrapper around Groq's chat models, letting you interact with them through LangChain's standard interface instead of manually crafting HTTP requests.

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

This snippet builds the Groq client instance mentioned above, configured with a chosen model and a low temperature value, and authenticated using the API key pulled from the environment settings.

Temperature is a parameter, typically ranging from 0 to 1, that governs how random or inventive a model's responses are. A higher value, such as 0.8, pushes the output toward more varied and creative results, while a lower value, such as 0.2, keeps responses tighter and more predictable. This project sets temperature = 0.2.

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

This code wraps the LLM so that, instead of returning plain text, it produces a Python object that conforms exactly to ExtractedTransactionSchema. Internally, this is achieved by instructing the model to generate output matching the schema, then parsing and validating that output automatically — eliminating the need to manually interpret the model's raw text.

The ExtractedTransactionSchema itself is defined inside src/assistance/schema.py:

from typing import Optional
from pydantic import BaseModel, Field


class ExtractedTransactionSchema(BaseModel):
    is_complete: bool = Field(
        description="True only if title, type, amount, category, and payment method were all found.")
    missing_info_message: Optional[str] = Field(
        default=None, description="A polite clarifying question listing listing exactly what's missing. Must be null if is_complete is True")
    title: str
    transaction_type: str = Field(description="'income' or 'expense'")
    amount: float
    category: str
    payment_option: str = Field(
        description="e.g. 'UPI', 'Cash', 'HDFC Credit Card'")
    payment_type: str = Field(
        description="Broad classification of the payment_option, one of: 'Cash', 'Card', 'Digital', 'Bank Transfer', 'Other'"
    )
    note: str

Note that at this point the LLM hasn't actually been called yet — this step only defines the shape the output must take once it is invoked.

Graph State

The graph state represents the data structure that flows through and gets updated across the graph. Think of it as the orchestrator's working memory: it holds every piece of information the graph tracks and modifies as execution proceeds through each step. For this transaction assistant, the graph state is defined as follows:

from pydantic import BaseModel, Field
from typing import Annotated, List, Optional
import operator
from src.assitance.schema import ExtractedTransactionSchema

class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None

Let's break down what this code actually does:

from pydantic import BaseModel, Field

Pydantic is the data validation library being used here. BaseModel is the parent class you extend when defining a structured, type-checked shape like GraphState. Field allows you to attach metadata — descriptions, defaults, and so on — to each individual attribute.

from typing import Annotated, List, Optional
import operator

These are Python's typing utilities. Optional signals that a field may be empty and hold None. List types an attribute as a list of items. Annotated, paired with operator.add, is what tells LangGraph "when a node returns a new value for this field, append it to what's already there instead of replacing it." That's the mechanism that lets conversation_history grow across turns rather than getting wiped out on every new message.

class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None

• user_input: the most recent message the user sent in for this particular invocation.
• conversation_history: the full backlog of prior messages, accumulated turn by turn instead of being overwritten.
• extracted: populated once the LLM has pulled structured transaction data out of the conversation. It starts out as None because nothing has been extracted yet at the start of the run.
• final_response: the message that eventually goes back to the user — either a confirmation that the transaction was recorded, or a follow-up question asking for more detail.

The Extraction Prompt

from langchain_core.prompts import PromptTemplate

EXTRACTION_PROMPT = PromptTemplate(
    template="""
        You are a financial assistant extracting transaction details.

        Below is the conversation so far (it may span multiple messages, where later
        messages answer questions raised by earlier ones). Treat it as one combined input.

        Required fields: title, transaction_type (income/expense), amount, category, payment_option.

        If title is missing, add one based on the context of the message.
        If anything required is missing, except title, set is_complete to False and write a short, polite
        clarifying question in missing_info_message asking only for what's missing.

        If everything is present, set is_complete to True, leave missing_info_message null,
        and fill in all fields. Always copy the user's original message into `note`.

        Conversation so far:
        {user_input}
        """,
    input_variables=["user_input"]
)

This is the literal instruction handed to the LLM in natural language — it spells out which fields to look for, what to do when something's missing, and how the response should be structured. Since structured_llm already enforces the schema at the output level, the prompt's job is mostly to steer the model's reasoning: deciding what "complete" means, how to word a clarifying question, and so on, while the schema takes care of formatting.

Extractor

def extractor(state: GraphState):
    full_conversation = "\n".join(
        state.conversation_history + [state.user_input])

    prompt = EXTRACTION_PROMPT.format(user_input=full_conversation)
    result: ExtractedTransactionSchema = structured_llm.invoke(prompt)

    return {"extracted": result, "conversation_history": [state.user_input]}

The extractor function does the following:

• Merges every previous message with the current one so the LLM sees the full context.
• Passes that merged text to the LLM.
• Receives back a structured ExtractedTransactionSchema object.
• Returns a dictionary of state updates — the freshly extracted data plus the current message, which LangGraph automatically folds into the history thanks to the operator.add behavior configured on that field.

The Decision

route_after_extraction

def route_after_extraction(state: GraphState):
    return "create_transaction" if state.extracted.is_complete else "ask_again"

This function performs no actual processing — its only job is to make a decision. Depending on whether the LLM flagged the extracted data as complete, it returns a string that tells the graph which node should run next. You can think of it as the branching logic in the flowchart: the graph inspects this function's return value and follows the corresponding path, either heading to create_transaction to record the transaction, or to ask_again to request more information.

Database Writer Node

create_transaction_node

This node is responsible for writing the completed transaction data to the database against the appropriate user. The function create_transaction_node implementing the DB Writer node looks like this:

from langchain_core.runnables import RunnableConfig
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema
from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel
from src.utils.db_helper import get_or_create

async def create_transaction_node(state: GraphState, config: RunnableConfig):
    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted

    try:
        category_id = await get_or_create(
            session, CategoriesModel, user.id, data.category,
            extra_defaults={"color": get_deterministic_color(data.category)}
        )
        payment_option_id = await get_or_create(
            session, PaymentOptionsModel, user.id, data.payment_option,
            extra_defaults={"payment_type": data.payment_type}
        )

        payload = TransactionCreateSchema(
            amount=data.amount,
            category_id=category_id,
            payment_option_id=payment_option_id,
            note=data.note,
            title=data.title,
            type=data.transaction_type,
        )

        await controller.create_transaction(payload, session, user)
        await session.commit()

    except SQLAlchemyError as err:
        await session.rollback()
        print(
            f"Error while creating transaction through AI assistance :: {err}")
        return {
            "final_response": "Something went wrong while saving your transaction. Please try again."
        }

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}

That's a lot to take in at once, so let's go through it piece by piece.

from langchain_core.runnables import RunnableConfig

A type standing in for the config object passed into any node. It exists purely as a type hint, so anyone reading the signature of create_transaction_node immediately understands what shape config takes.

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

These are the usual SQLAlchemy imports needed to catch database errors and to type the async database session used to communicate with PostgreSQL.

from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema

This pulls in the transaction-creation logic already used elsewhere in the app, along with its input schema. Reusing it means the assistant creates transactions through the exact same code path as the regular CRUD API, instead of duplicating that logic here.

from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel

These are supporting pieces used to resolve the category and payment-option names extracted by the LLM into actual rows and IDs in the database, creating new records when they don't already exist.

Note: the lookup/creation logic for categories and payment_options was merged into one generic helper, get_or_create, since both models needed essentially the same behavior.

async def create_transaction_node(state: GraphState, config: RunnableConfig):

The create_transaction_node function only runs once the extracted data is confirmed complete. It's declared async because it performs real database work, and it takes config alongside state so it can access the active database session and the logged-in user. Those two values come from the API route rather than from the LLM or the conversation state, since they belong to the specific request rather than to the ongoing dialogue.

    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted

This pulls out the session, the user, and the extracted transaction data.

    try:
        category_id = await get_or_create(...)
        payment_option_id = await get_or_create(...)

Because the LLM only extracted the names of the category and payment method, things like "Groceries" or "UPI", not their database IDs, this step checks whether a matching row already exists for the current user. If not, it creates one. Either way, it returns the corresponding ID.

        payload = TransactionCreateSchema(...)
        await controller.create_transaction(payload, session, user)
        await session.commit()

The payload is assembled in the same shape expected by the existing transaction-creation logic, then passed into that same controller function, reusing existing application logic instead of rewriting it. The database transaction is then committed to persist the change.

    except SQLAlchemyError as err:
        await session.rollback()
        print(...)
        return {"final_response": "Something went wrong..."}

If something fails at the database layer, any partial changes are rolled back and a friendly error message is returned instead of letting the request crash. This guards against ending up with, say, a newly created category but no corresponding transaction to go with it.

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}

On success, a human-readable confirmation message is built and returned as an update to the state.

Clarification Node

ask_again

def ask_again_node(state: GraphState):
    return {"final_response": state.extracted.missing_info_message}

This is a simple fallback path. As described earlier, this node only runs when the extracted data has is_complete set to False, accompanied by a useful message in missing_info_message.

Inside ask_again_node, the function receives state, giving it access to state.extracted.is_complete and state.extracted.missing_info_message.

In short, whenever information is missing, this node simply forwards the clarifying question the LLM already produced during extraction, so the user knows exactly what to supply next.

Assembling the Graph

from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(GraphState)

This creates a new graph builder and tells it that every node in the graph will read from and write to an object shaped like GraphState.

graph_builder.add_node("extractor", extractor)
graph_builder.add_node("create_transaction", create_transaction_node)
graph_builder.add_node("ask_again", ask_again_node)

Each function is registered here as a named node, essentially a labeled step, within the graph.

graph_builder.add_edge(START, "extractor")

This sets the entry point: every execution of the graph starts at the extractor node.

graph_builder.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "create_transaction": "create_transaction",
        "ask_again": "ask_again",
    },
)

This is where the branching happens. Once extractor finishes, LangGraph calls route_after_extraction to determine the next step. Whatever string it returns, either create_transaction or ask_again, is looked up in this mapping, which connects each decision string to the actual node it should jump to.

graph_builder.add_edge("create_transaction", END)
graph_builder.add_edge("ask_again", END)

Both possible branches terminate the graph run once they complete, with END reached along either path.

Compiling With Memory

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
assistance_graph = graph_builder.compile(checkpointer=memory)

Calling compile() turns the graph definition into something runnable. Passing checkpointer=memory wires in the state-persistence mechanism described earlier, so that invoking the graph again with the same thread_id resumes the conversation from where it left off instead of restarting it.

The Final Code

With this, the orchestration layer is complete. Here is the finished file (src/assistance/graph.py):

from langgraph.graph import StateGraph, START, END
from langchain_groq import ChatGroq
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig
from pydantic import BaseModel, Field
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated, List, Optional
import operator

from src.utils.settings import settings
from src.assitance.schema import ExtractedTransactionSchema
from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema
from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)


class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None


EXTRACTION_PROMPT = PromptTemplate(
    template="""
        You are a financial assistant extracting transaction details.

        Below is the conversation so far (it may span multiple messages, where later
        messages answer questions raised by earlier ones). Treat it as one combined input.

        Required fields: title, transaction_type (income/expense), amount, category, payment_option.

        If title is missing, add one based on the context of the message.
        If anything required is missing, except title, set is_complete to False and write a short, polite
        clarifying question in missing_info_message asking only for what's missing.

        If everything is present, set is_complete to True, leave missing_info_message null,
        and fill in all fields. Always copy the user's original message into `note`.

        Conversation so far:
        {user_input}
        """,
    input_variables=["user_input"]
)


def extractor(state: GraphState):
    full_conversation = "\n".join(
        state.conversation_history + [state.user_input])

    prompt = EXTRACTION_PROMPT.format(user_input=full_conversation)
    result: ExtractedTransactionSchema = structured_llm.invoke(prompt)


    return {"extracted": result, "conversation_history": [state.user_input]}


def route_after_extraction(state: GraphState):
    return "create_transaction" if state.extracted.is_complete else "ask_again"


async def create_transaction_node(state: GraphState, config: RunnableConfig):
    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted


    try:
        category_id = await get_or_create(
            session, CategoriesModel, user.id, data.category,
            extra_defaults={"color": get_deterministic_color(data.category)}
        )
        payment_option_id = await get_or_create(
            session, PaymentOptionsModel, user.id, data.payment_option,
            extra_defaults={"payment_type": data.payment_type}
        )

        payload = TransactionCreateSchema(
            amount=data.amount,
            category_id=category_id,
            payment_option_id=payment_option_id,
            note=data.note,
            title=data.title,
            type=data.transaction_type,
        )

        await controller.create_transaction(payload, session, user)
        await session.commit()

    except SQLAlchemyError as err:
        await session.rollback()
        return {
            "final_response": "Something went wrong while saving your transaction. Please try again."
        }

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}


def ask_again_node(state: GraphState):
    return {"final_response": state.extracted.missing_info_message}


graph_builder = StateGraph(GraphState)
graph_builder.add_node("extractor", extractor)
graph_builder.add_node("create_transaction", create_transaction_node)
graph_builder.add_node("ask_again", ask_again_node)

graph_builder.add_edge(START, "extractor")
graph_builder.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "create_transaction": "create_transaction",
        "ask_again": "ask_again",
    },
)
graph_builder.add_edge("create_transaction", END)
graph_builder.add_edge("ask_again", END)

memory = MemorySaver()
assistance_graph = graph_builder.compile(checkpointer=memory)

2. PostgreSQL Database

The database interaction for this stage is already handled inside create_transaction_node, covered earlier, where the finalized transaction data gets written into the transactions table.

3. FastAPI Endpoint

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.assitance.schema import UserMessageSchema
from src.assitance.graph import assistance_graph
from src.auth.models import UsersModel
from src.utils.db import get_db
from src.utils.auth.authentication import allow_all


assistance_routes = APIRouter(prefix="/assistance")


@assistance_routes.post("/transaction-entry", status_code=status.HTTP_201_CREATED)
async def run_transaction_assistance(payload: UserMessageSchema, session: AsyncSession = Depends(get_db), user: UsersModel = Depends(allow_all)):
    config = {"configurable": {
              "thread_id": str(user.id),
              "session": session,
              "user": user
              }
              }

    result = await assistance_graph.ainvoke(
        {"user_input": payload.message}, config=config)

    return {"response": result["final_response"]}

Clients call this endpoint (/assistance/transaction-entry) and include a message in the request body describing what the transaction was about.

Let's break down what each piece does.

@assistance_routes.post("/transaction-entry", status_code=status.HTTP_201_CREATED)

This sets up a POST route at /transaction-entry. Setting status_code=status.HTTP_201_CREATED tells FastAPI what status code to return by default on success. 201 is the conventional code for "a new resource was created," which fits here since a successful call results in a new transaction row.

Assembling the graph config:

    config = {
        "configurable": {
            "thread_id": str(user.id),
            "session": session,
            "user": user
        }
    }

This builds the config object handed to the graph invocation. config carries request-specific values that shouldn't live inside the persisted conversation state itself.

• "thread_id": str(user.id): this value is what LangGraph's checkpointer uses to figure out whose conversation history to retrieve and update. By keying it on the authenticated user's ID, each user automatically ends up with an isolated, persistent thread, so one user's half-finished transaction entry can never bleed into another's. It gets cast to a string because the checkpointer expects thread_id as a string, while user.id is typically a UUID.
• "session" and "user": these are forwarded so that create_transaction_node, running inside the graph, has access to the live database session and to details about who is making the request.

Invoking the graph:

    result = await assistance_graph.ainvoke(
        {"user_input": payload.message}, config=config)

This is the line that actually triggers execution. ainvoke is the async counterpart to running the graph, using the synchronous invoke instead would block the event loop, which matters here because create_transaction_node performs async database operations internally.

• The first argument, {"user_input": payload.message}, represents the starting state for this run. Only user_input needs to be supplied explicitly; the remaining GraphState fields (conversation_history, extracted, final_response) either come with default values or get populated as execution proceeds through the graph. When an existing thread_id already has saved history, LangGraph folds this new input into that stored state rather than beginning from scratch.
• config=config supplies everything prepared in the prior step: thread_id for locating the right state, plus session and user for the node responsible for the database write.
• The await keyword suspends this coroutine until the graph finishes its full run, since ainvoke hands back a coroutine that must be awaited before the result is usable.

Whatever comes back as result is the final GraphState, represented as a dictionary, reflecting the graph's execution regardless of whether it terminated at create_transaction or at ask_again.

Sending back the response:

    return {"response": result["final_response"]}

The route wraps up by returning a simple dictionary holding just the final response text. FastAPI takes care of converting this into a JSON payload for the client, producing something along the lines of:

{ "response": "Added expense of 450 under 'Groceries' (UPI)" }

This text is identical to whatever was assembled earlier inside either create_transaction_node or ask_again_node. The route itself stays agnostic about which branch actually ran; it simply forwards whatever ended up in final_response.

Conclusion

Working through this transaction assistant highlights something tutorials tend to gloss over: the difficult part of shipping an AI feature isn't prompting a model for a reply, it's making sure that reply behaves safely once it interacts with a real system. Writing prompts is the easy part. The real engineering effort goes into schemas that force structured output, conditional graphs that decide between committing data and asking for clarification, and state that correctly carries over across multiple turns of conversation.

LangGraph made sense for this project specifically because the workflow required actual decision-making rather than a single pass from input to output. If your feature only needs a straight-line flow, a plain LLM call or a LangChain chain is likely the simpler and more appropriate tool. But once your AI logic needs to branch, retain memory, or pause to collect more information before proceeding, a graph-based structure stops looking like unnecessary complexity and starts being the most sensible way to model that flow.

References