Automatisation de l’entrée des transactions dans FastAPI à l’aide d’un assistant IA LangGraph

Tout ceux qui ont essayé de suivre leurs dépenses quotidiennes via un formulaire en ligne savent à quel point c’est fastidieux. Enregistrer l’achat d’un café coûtant 5 dollars ne devrait pas nécessiter de remplir plusieurs champs, or c’est précisément ce qui se produit dans de nombreuses applications de suivi financier développées avec FastAPI et PostgreSQL, où chaque transaction — aussi petite soit-elle — doit être saisie manuellement.

Imaginez développer une telle application, où chaque transaction, simple ou complexe, doit passer par un formulaire. Cela fonctionne bien pour des entrées occasionnelles, mais devient épuisant dès qu’il faut enregistrer plusieurs transactions d’un coup.

À ce stade, une question naturelle se pose : et si tout le processus pouvait être automatisé ? Et si, au lieu de remplir un formulaire, on pouvait simplement dire à un assistant « J’ai dépensé 5 dollars en café au restaurant aujourd’hui » et le laisser s’occuper du reste ?

C’est là que LangGraph devient utile.

Avec LangGraph, vous pouvez créer un assistant IA qui prend une description en langage courant de ce qui s’est passé et la transforme en une transaction correctement enregistrée pour vous.

Introduction

Cet article présente les concepts fondamentaux de LangGraph et de son écosystème associé, puis explique en détail comment un assistant IA peut être intégré à une application FastAPI à l’aide de LangGraph afin d’automatiser la saisie des transactions.

LangGraph

LangGraph est issu de l’équipe derrière LangChain ; il s’agit d’un outil open source permettant de mettre en place et de gérer des workflows d’agents IA à l’aide de structures graphiques. Avec lui, vous décrivez un processus sous forme d’un ensemble de « nœuds » et de « arêtes », ce qui permet d’organiser, de rendre scalable et plus facile à contrôler le comportement complexe des agents.

LangChain

LangChain est un outil, également open source, permettant de créer des applications utilisant des grands modèles de langage. Son rôle principal est d’offrir aux développeurs un lien entre un LLM et des ressources externes — sources de données, outils et étapes de workflow — afin que le système puisse effectuer des raisonnements en plusieurs étapes et des tâches automatisées, plutôt que de se limiter à un échange simple de prompts et de réponses.

Objectif : Il est conçu pour développer des applications d’IA nécessitant la mise en chaîne de plusieurs étapes — par exemple, gérer les entrées des utilisateurs, récupérer des informations pertinentes et générer une réponse.

Structure : LangChain s’appuie sur des « chaînes », qui sont des séquences ordonnées d’opérations où la sortie de chaque étape alimente l’entrée de la suivante. Cela vous permet de décomposer une logique complexe en parties plus petites et plus faciles à gérer.

Applications : Les cas d’usage typiques incluent les chatbots, les tâches de raisonnement à plusieurs étapes, la récupération et la synthèse de documents, ainsi que le raccordement des LLM à des outils ou des API externes.

LangGraph (suite)

En termes simples, LangGraph organise les appels aux LLM en workflows sous forme de graphes, ce qui permet un raisonnement flexible et même parallèle à plusieurs étapes, plutôt qu’une séquence strictement linéaire.

But : Il permet des applications d’IA où la logique peut se diviser en branches, former des boucles ou exécuter des étapes en parallèle, dépassant ainsi les capacités d’une simple chaîne séquentielle.

Structure : LangGraph représente les opérations sous forme de « nœuds » et le flux de données entre eux sous forme d’« arêtes ». La sortie d’un seul nœud peut alimenter plusieurs nœuds suivants, ce qui permet des chemins de décision dynamiques.

Applications : LangGraph est particulièrement adapté à la coordination de plusieurs agents, à la création de pipelines de décision complexes, à l’automatisation de tâches impliquant une logique conditionnelle, ainsi qu’à l’orchestration de plusieurs LLM ou outils en même temps.

Qu’est-ce qu’un graph dans LangGraph ?

En général, un graph est une structure de données non linéaire composée de « sommets » (nœuds) et d’« arêtes » (les connexions entre eux), qui permettent de représenter les relations entre des objets.

Dans LangGraph en particulier, cette structure de graphe est utilisée pour créer des workflows cycliques et à état — où l’IA peut prendre des décisions, revenir à des étapes précédentes ou s’orienter vers différentes voies en fonction des résultats intermédiaires.

LangChain vs. LangGraph

Architecture du projet

(Point d’entrée FastAPI + LangGraph + création et persistance de transactions)

Le problème

Au moment où LangGraph a été intégré à l’application financière, créer une transaction signifiait appeler directement le point d’entrée /transactions/add, avec un en-tête de données ayant cette forme :

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

Pour y parvenir, il a fallu effectuer deux appels API préalables — un pour récupérer la liste des categories et un autre pour obtenir les payment_options — rien que pour acquérir les IDs nécessaires au payload. En d’autres termes, créer une seule transaction était un processus en trois étapes, et de surcroît lent.

La solution

La correction consistait à laisser un assistant IA gérer ces trois étapes, tandis que l’utilisateur n’avait qu’à décrire, en langage courant, ce qu’il avait fait de son argent. Afin d’atteindre cet objectif, voici comment l’implémentation est structurée.

L’architecture en trois étapes

1. Point d’accès FastAPI
2. Orchestrateur LangGraph
3. Base de données PostgreSQL

1. Point d’accès FastAPI

L’utilisateur envoie une demande à l’endpoint FastAPI /assistance/transaction-entry, avec un en-tête contenant un message décrivant la transaction.

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

2. LangGraph Orchestrator

L’orchestrateur est conçu sous forme de graphe, où chaque nœud représente une opération et chaque arête représente le flux de données entre les opérations.

Si le message contient déjà toutes les informations requises, l’LLM le convertit en données structurées, par exemple :

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

Ces données structurées sont ensuite transférées vers le nœud Database Writer, après que les deux champs de flag (is_complete et missing_info_message) aient été supprimés. Le nœud Database Writer appelle la méthode create_transaction(), qui enregistre la transaction relative à l’utilisateur dans la base de données.

Mais que se passe-t-il si le message manque de certains détails ? Prenons un message comme celui-ci :

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

Ici, le mode de paiement n’est pas spécifié. Dans ce cas, les données extraites par le LLM incluront des valeurs de flag renseignées, telles que :

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

Puisque le flag is_complete vaut False ici, le missing_info_message est acheminé vers l’autre nœud connecté au LLM Analyzer : le nœud Clarification. Ce parcours n’est déclenché que lorsque is_complete évalue à False.

Le nœud Clarification reçoit le missing_info_message et appelle la méthode ask_again(), qui renvoie ce message en réponse à la demande initiale FastAPI. Cela marque la fin de l’exécution du graphe pour cette exécution — le résultat que reçoit l’utilisateur n’est qu’une demande d’information manquante, dans ce cas le mode de paiement.

Supposons que l’utilisateur réponde ensuite avec les informations manquantes, par exemple :

{
  message: "UPI"
}

Cette réponse provoque une nouvelle initialisation du graphe d’orchestration, qui suit alors la même séquence d’étapes que précédemment.

La différence clé dans cette deuxième étape est qu’aucune des informations précédentes n’est perdue — l’historique de la conversation est conservé à chaque fois que des données sont extraites par le LLM (ce mécanisme de persistance est abordé en plus grand détail plus tard dans l’article). Comme payment_option est désormais disponible et combiné aux valeurs capturées précédemment, is_complete passe à True, et les données finalisées et filtrées sont transmises au nœud Database Writer, comme suit :

{
    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.

Le nœud Database Writer prend alors le relais avec ces données filtrées en main. Rappelons que lorsqu’une entrée de transaction était créée manuellement, il fallait effectuer deux appels API supplémentaires — un pour categories et un pour payment_options — rien que pour obtenir leurs IDs respectifs avant de pouvoir créer la transaction elle-même. Le même problème se pose ici : les données filtrées contiennent les valeurs textuelles réelles des catégories et des options de paiement, et non leurs IDs dans la base de données, or la base de données ne accepte pas de valeurs brutes pour ces champs.

Pour résoudre ce problème, le nœud Database Writer doit interroger la base de données afin de trouver les enregistrements correspondants des catégories et des options de paiement en se basant sur les valeurs présentes dans les données filtrées.

Puisque le projet repose sur FastAPI ainsi que sur SQLAlchemy, ces recherches sont réalisées sous forme de requêtes SQLAlchemy.

Pour les 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

En bref, cette logique : Exécute une requête SELECT pour vérifier si la catégorie référencée dans data.category existe déjà. Si c’est le cas, l’ID de la catégorie remplace la valeur dans data.category. Sinon, un nouveau enregistrement de catégorie est créé, et son ID généré est utilisé à la place.

Le même schéma s’applique aux 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

Lorsque l’ID de la catégorie et l’ID de l’option de paiement ont tous deux été déterminés, l’objet de données est entièrement mis à jour et prêt pour insertion, comme suit :

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

Avec ces données finalisées, le nœud Database Writer appelle la méthode create_transaction(), qui persiste réellement l’enregistrement de la transaction dans la base de données.

3. Base de données PostgreSQL

Ce dernier étape de l’architecture correspond à l’instant où les données finalisées, transmises par le nœud Database Writer, sont écrites dans la table transactions.

La structure résultante de la table transactions est la suivante :

L’implémentation

Puisque l’architecture a été présentée, il est temps d’examiner en détail comment construire cet orchestrateur à l’aide de LangGraph. Notez que l’ordre présenté ici ne correspond pas exactement à celui de la présentation architecturale précédente. En réalité, l’implémentation est organisée comme suit :

1. LangGraph Orchestrator
2. Base de données PostgreSQL
3. Point d’entrée FastAPI

1. LangGraph Orchestrator

L’orchestrateur lui-même se trouve dans src/assistance/graph.py. Ce fichier est chargé de configurer le LLM, de définir les nœuds du graphe, d’établir les connexions entre ces nœuds, et enfin de compiler tout cela en un graphe exécutable.

Analyzer LLM, le Database Writer et le nœud Clarification.

Nœud Analyzer LLM (Groq)

Ce nœud est essentiellement un modèle de langage dont la fonction est de déterminer l’intention de l’utilisateur et de vérifier si le message contient toutes les informations nécessaires et correctes. Plutôt que de créer un modèle personnalisé à partir de zéro, ce projet s’appuie sur Groq pour gérer cette tâche complexe.

Qu’est-ce que Groq ?

Groq est un framework Python open source conçu pour travailler avec des données structurées en graphes. Il offre aux développeurs un moyen expressif de consulter, filtrer et agréger des informations stockées sous forme de graphes, et il est particulièrement adapté aux grands ensembles de données graphiques, tels que les réseaux sociaux, les graphes de connaissances ou les moteurs de recommandation, comme cela est décrit dans un article de GeekForGeeks sur l’API Groq.

Grâce à l’API hébergée par Groq, vous pouvez envoyer des requêtes vers des modèles open source largement utilisés — openai/gpt-oss-120b étant celui utilisé dans ce projet — et recevoir des réponses qui arrivent généralement bien plus rapidement que celles obtenues habituellement auprès d’autres fournisseurs proposant des modèles similaires.

Pourquoi Groq ?

Groq a été choisi par rapport à des alternatives comme ChatOpenAI ou ChatAnthropic pour plusieurs raisons :

• Vitesse : Groq s’appuie sur du matériel spécialement conçu appelé LPU (Unités de traitement du langage) plutôt que sur les GPU dont dépendent la plupart des autres fournisseurs, ce qui permet des inférences rapides.
• Niveau gratuit utilisable : le plan gratuit de Groq est suffisamment généreux pour permettre à un projet individuel ou pédagogique de fonctionner sans générer de coûts API significatifs pendant les expérimentations.
• Compatibilité immédiate via LangChain : la classe langchain_groq.ChatGroq s’intègre à LangChain et LangGraph de la même manière que ChatOpenAI ou ChatAnthropic. Cela signifie qu’un changement de fournisseur ultérieurement ne nécessitera pas de revoir la logique du graphe — il suffira simplement de remplacer le client.

Comment obtenir une clé API Groq

Groq vous permet de générer des clés API gratuites pour le développement. Voici comment en obtenir une :

1. Allez sur https://console.groq.com et connectez-vous ou créez un compte.
2. Sélectionnez l’option « API Keys » dans la barre de navigation.
3. Choisissez « Create API Key ».
4. transaction-assistant) ainsi qu’une période d’expiration pour la clé. Soumettez le formulaire une fois les informations saisies.

Une fois créées, toutes vos clés apparaîtront dans la liste principale de cette page.

Utiliser la clé API Groq dans le code FastAPI

Ajoutez la clé API Groq à votre fichier .env situé dans le répertoire racine du projet, aux côtés des autres variables d’environnement :

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

Il existe plusieurs méthodes pour charger les variables d’environnement dans les modules qui en ont besoin. Ce projet utilise une classe dédiée aux paramètres :

Définissez une classe Settings à l’intérieur de 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()

Puis importez cet objet de configuration là où il est nécessaire :

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.

Configuration du LLM

Au préalable de configurer le LLM, installez LangGraph et LangChain ainsi que l’intégration Groq : pip install -U langgraph langchain langchain-groq

Une instance du client Groq est ensuite créée et configurée avec un modèle spécifique :

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)

Ici, ChatGroq agit comme un enveloppeur de LangChain autour des modèles de chat de Groq, vous permettant d’interagir avec eux via l’interface standard de LangChain plutôt que de créer manuellement des requêtes HTTP.

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

Ce fragment crée l’instance du client Groq mentionnée précédemment, configurée avec un modèle choisi et une valeur de température faible, et authentifiée à l’aide de la clé API extraite des paramètres d’environnement.

La température est un paramètre, généralement compris entre 0 et 1, qui détermine le degré de aléatoireté ou d’originalité des réponses du modèle. Une valeur plus élevée, comme 0,8, favorise des résultats plus variés et créatifs, tandis qu’une valeur plus basse, comme 0,2, rend les réponses plus cohérentes et prévisibles. Ce projet définit temperature = 0,2.

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

Ce code entoure l’LLM de manière à ce qu’au lieu de renvoyer du texte brut, il génère un objet Python conforme exactement au ExtractedTransactionSchema. En interne, cela s’obtient en demandant au modèle de produire une sortie correspondant au schéma, puis en analysant et en validant automatiquement cette sortie — ce qui élimine la nécessité d’interpréter manuellement le texte brut du modèle.

Le ExtractedTransactionSchema lui-même est défini dans 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

Notez que à ce stade, l’LLM n’a pas encore été appelé — cette étape définit uniquement la forme que doit prendre la sortie une fois qu’il est invoqué.

État du graphe

L’état du graphe représente la structure de données qui circule à travers le graphe et est mise à jour au fur et à mesure. Pensez-y comme à la mémoire de travail de l’orchestrateur : elle contient toutes les informations que le graphe traque et modifie au cours de l’exécution à chaque étape. Pour cet assistant transactionnel, l’état du graphe est défini comme suit :

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

Décomposons ce que fait réellement ce code :

from pydantic import BaseModel, Field

Pydantic est la bibliothèque de validation des données utilisée ici. BaseModel est la classe parente que l’on étend pour définir une structure structurée et vérifiée au niveau des types, comme GraphState. Field permet d’attacher des métadonnées — descriptions, valeurs par défaut, etc. — à chaque attribut individuel.

from typing import Annotated, List, Optional
import operator

Ces outils font partie des fonctionnalités de typage de Python. Optional indique qu’un champ peut être vide et contenir None. List définit un attribut comme une liste d’éléments. Annotated, utilisé avec operator.add, indique à LangGraph « lorsqu’un nœud renvoie une nouvelle valeur pour ce champ, l’ajouter à ce qui est déjà présent plutôt que de le remplacer ». C’est ce mécanisme qui permet à conversation_history de s’alimenter au fil des échanges sans être effacé à chaque nouvelle 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 : le dernier message envoyé par l’utilisateur pour cette invocation spécifique.
• conversation_history : l’ensemble complet des messages précédents, accumulés tour par tour sans être écrasés.

extracted : cette valeur est remplie une fois que le LLM a extrait les données structurées des transactions de la conversation. Elle commence par None car rien n’a encore été extrait au début de l’exécution.

final_response : le message qui est finalement renvoyé à l’utilisateur — soit une confirmation que la transaction a été enregistrée, soit une question complémentaire demandant plus de détails.

L’instruction d’extraction

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"]
)

Ce sont les instructions littérales transmises à l’LLM en langue naturelle — elles précisent quels champs rechercher, que faire en cas d’absence d’informations, et comment structurer la réponse. Étant donné que structured_llm impose déjà le schéma au niveau de la sortie, la tâche du prompt consiste principalement à guider le raisonnement du modèle : déterminer ce que signifie « complet », comment formuler une question de clarification, etc., tandis que le schéma s’occupe du formatage.

Extracteur

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]}

La fonction extractor effectue ce qui suit :

• Mélange chaque message précédent avec le message actuel afin que l’LLM voie le contexte complet.
• Transmet ce texte mélangé à l’LLM.
• Récupère en retour un objet ExtractedTransactionSchema structuré.

Retourne un dictionnaire des mises à jour d’état — les données fraîchement extraites ainsi que le message actuel, que LangGraph intègre automatiquement à l’historique grâce au comportement operator.add configuré pour ce champ.

La décision

route_after_extraction

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

Cette fonction ne réalise aucun traitement concret — sa seule tâche est de prendre une décision. Selon que le LLM ait marqué les données extraites comme complètes ou non, elle retourne une chaîne de caractères indiquant au graphe quel nœud doit être exécuté en suivant. On peut l’imaginer comme la logique de bifurcation dans un organigramme : le graphe examine la valeur de retour de cette fonction et emprunte le chemin correspondant, soit en se dirigeant vers create_transaction pour enregistrer la transaction, soit vers ask_again pour demander plus d’informations.

Nœud d’écriture dans la base de données

create_transaction_node

Ce nœud est chargé d’écrire les données de la transaction finalisée dans la base de données au nom de l’utilisateur approprié. La fonction create_transaction_node qui implémente le nœud DB Writer a l’aspect suivant :

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}

C’est beaucoup à assimiler d’un coup, alors examinons cela étape par étape.

from langchain_core.runnables import RunnableConfig

Un type servant de représentation pour l’objet config transmis à n’importe quel nœud. Il existe uniquement en tant que suggestion de type, de sorte que quiconque lit la signature de create_transaction_node comprend immédiatement quelle forme prend config.

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

Ces importations SQLAlchemy habituelles sont nécessaires pour détecter les erreurs de base de données ainsi que pour définir le type de la session asynchrone utilisée pour communiquer avec PostgreSQL.

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

Cela intègre la logique de création de transactions déjà utilisée ailleurs dans l’application, ainsi que son schéma d’entrées. La réutilisation de cette logique permet à l’assistant de créer des transactions via exactement le même chemin de code que l’API CRUD standard, au lieu de dupliquer cette logique ici.

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

Il s’agit d’éléments auxiliaires utilisés pour transformer les noms de catégories et d’options de paiement extraits par le LLM en lignes et identifiants réels dans la base de données, en créant de nouveaux enregistrements lorsqu’ils n’existent pas encore.

Remarque : la logique de recherche/création pour categories et payment_options a été fusionnée dans un outil générique, get_or_create, car les deux modèles nécessitaient essentiellement le même comportement.

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

La fonction create_transaction_node ne s’exécute qu’une fois que les données extraites ont été confirmées comme complètes. Elle est déclarée async car elle effectue des opérations réelles sur la base de données, et elle prend en paramètre config ainsi que state afin de pouvoir accéder à la session de base de données active et à l’utilisateur connecté. Ces deux valeurs proviennent de la route API plutôt que du LLM ou de l’état de la conversation, car elles appartiennent à la demande spécifique et non au dialogue en cours.

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

Cela récupère la session, l’utilisateur ainsi que les données de transaction extraites.

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

Puisque le LLM n’a extrait que les noms de la catégorie et du mode de paiement, tels que « Groceries » ou « UPI », et non leurs IDs dans la base de données, cette étape vérifie s’il existe déjà une ligne correspondante pour l’utilisateur actuel. Sinon, elle en crée une. Dans tous les cas, elle renvoie l’ID correspondant.

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

La charge utile est assemblée selon la même structure attendue par la logique existante de création de transactions, puis transmise à cette même fonction de contrôleur, en réutilisant ainsi la logique applicative existante au lieu de la réécrire. La transaction de base de données est ensuite confirmée pour persister le changement.

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

Si quelque chose échoue au niveau de la base de données, toutes les modifications partielles sont annulées et un message d’erreur clair est retourné, afin d’éviter que la demande ne plante. Cela permet de prévenir des situations où, par exemple, une catégorie vient d’être créée mais qu’il n’y a pas de transaction correspondante.

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

En cas de succès, un message de confirmation lisible par l’humain est généré et retourné pour mettre à jour l’état.

Nœud de clarification

ask_again

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

C’est une solution de secours simple. Comme décrit précédemment, ce nœud ne s’exécute que lorsque les données extraites ont is_complete défini sur False, accompagné d’un message utile dans missing_info_message.

Dans ask_again_node, la fonction reçoit state, ce qui lui permet d’accéder à state.extracted.is_complete et state.extracted.missing_info_message.

En bref, chaque fois que des informations manquent, ce nœud renvoie simplement la question de clarification déjà générée par le LLM lors de l’extraction, afin que l’utilisateur sache exactement quoi fournir ensuite.

Assemblage du graphe

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

Cela crée un nouveau constructeur de graphe et lui indique que chaque nœud du graphe lira dans et écrira vers un objet de type 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)

Chaque fonction est enregistrée ici en tant que nœud nommé, essentiellement une étape étiquetée, au sein du graphe.

graph_builder.add_edge(START, "extractor")

Cela définit le point d’entrée : chaque exécution du graphe commence au nœud extractor.

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

C’est ici que se produit la bifurcation. Une fois que extractor a terminé, LangGraph appelle route_after_extraction pour déterminer l’étape suivante. Quelle que soit la chaîne de caractères qu’il renvoie, que ce soit create_transaction ou ask_again, elle est recherchée dans ce tableau de correspondance, qui relie chaque chaîne de décision au nœud réel vers lequel il faut se diriger.

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

Les deux branches possibles mettent fin à l’exécution du graphe une fois qu’elles sont terminées, en atteignant END sur l’une ou l’autre voie.

Compilation avec mémoire

from langgraph.checkpoint.memory import MemorySaver

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

L’appel à compile() transforme la définition du graphe en quelque chose de pouvant être exécuté. La transmission de checkpointer=memory active le mécanisme de persistance d’état décrit précédemment, permettant ainsi à la réutilisation du graphe avec le même thread_id de reprendre la conversation là où elle s’était arrêtée au lieu de la recommencer.

Le code final

Avec cela, la couche d’orchestration est complète. Voici le fichier final (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. Base de données PostgreSQL

L’interaction avec la base de données pour cette étape est déjà gérée à l’intérieur de create_transaction_node, abordé précédemment, où les données de transaction finalisées sont écrites dans la table des transactions.

3. Point d’entrée FastAPI

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"]}

Les clients appellent cet endpoint (/assistance/transaction-entry) et incluent un message dans le corps de la requête décrivant l’objet de la transaction.

Décomposons maintenant le rôle de chaque élément.

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

Cela définit une route POST à l’adresse /transaction-entry. La valeur status_code=status.HTTP_201_CREATED indique à FastAPI quel code d’état retourner par défaut en cas de succès. Le code 201 correspond traditionnellement à « une nouvelle ressource a été créée », ce qui convient ici car une appel réussi génère une nouvelle ligne de transaction.

Assemblage de la configuration du graphe :

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

Ceci crée l’objet config qui est transmis lors de l’appel au graphe. L’objet config contient des valeurs spécifiques à la requête qui ne devraient pas figurer dans l’état de conversation persistant lui-même.

• "thread_id": str(user.id): cette valeur est utilisée par le checkpointer de LangGraph pour déterminer quel historique de conversation récupérer et mettre à jour. En la liant au ID de l’utilisateur authentifié, chaque utilisateur dispose automatiquement d’un thread isolé et persistant, de sorte qu’une entrée de transaction inachevée d’un utilisateur ne peut jamais affecter celle d’un autre. Elle est convertie en chaîne de caractères car le checkpointer attend thread_id sous forme de chaîne, tandis que user.id est généralement un UUID.
• "session" et "user": ces éléments sont transmis afin que create_transaction_node, exécuté à l’intérieur du graphe, ait accès à la session de base de données en cours et aux informations sur l’utilisateur qui effectue la demande.

Appel du graphe :

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

C’est cette ligne qui déclenche réellement l’exécution. ainvoke correspond à la version asynchrone de l’exécution du graphe ; utiliser la version synchrone invoke bloquerait le cycle d’événements, ce qui est important ici car create_transaction_node effectue en interne des opérations de base de données asynchrones.

• {"user_input": payload.message}, représente l’état départ pour cette exécution. Seul le champ user_input doit être spécifié explicitement ; les autres champs de GraphState (conversation_history, extracted, final_response) disposent soit de valeurs par défaut, soit sont remplis au fur et à mesure que l’exécution progresse dans le graphe. Lorsqu’un thread_id existant contient déjà une histoire enregistrée, LangGraph intègre cette nouvelle entrée dans cet état stocké plutôt que de commencer à zéro.
• config=config fournit tout ce qui a été préparé dans l’étape précédente : le thread_id permettant de localiser l’état approprié, ainsi que session et user pour le nœud chargé de l’écriture dans la base de données.

Le mot-clé await suspend cette coroutine jusqu’à ce que le graphe ait terminé son exécution complète, car ainvoke renvoie une coroutine qui doit être attendue avant que le résultat ne puisse être utilisé.

Ce qui est retourné sous le nom de result correspond au GraphState final, représenté sous forme de dictionnaire, reflétant l’exécution du graphe quel que soit le point où celle-ci s’est terminée, que ce soit à create_transaction ou à ask_again.

Renvoi de la réponse :

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

La route se termine en renvoyant un simple dictionnaire contenant uniquement le texte de la réponse finale. FastAPI s’occupe de convertir ce dernier en un payload JSON destiné au client, produisant quelque chose comme ceci :

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

Ce texte est identique à ce qui a été généré précédemment dans create_transaction_node ou ask_again_node. La route elle-même reste indifférente au branchement qui a réellement été exécuté ; elle transmet simplement ce qui se trouve dans final_response.

Conclusion

Travailler sur cet assistant de transactions met en évidence quelque chose que les tutoriels ont tendance à négliger : la partie difficile dans le déploiement d’une fonctionnalité basée sur l’IA n’est pas de demander une réponse au modèle, mais de s’assurer que cette réponse se comporte de manière sûre une fois qu’elle interagit avec un système réel. Rédiger des prompts est la partie facile. L’effort d’ingénierie réel concerne les schémas qui imposent une sortie structurée, les graphes conditionnels qui décident s’il faut enregistrer les données ou demander des précisions, ainsi que l’état qui doit être correctement transmis au fil de plusieurs échanges.

LangGraph s’est avéré adapté à ce projet précisément parce que le flux de travail nécessitait une véritable prise de décision plutôt qu’un simple passage du point d’entrée au point de sortie. Si votre fonctionnalité exige uniquement un flux linéaire, une simple appel à un LLM ou une chaîne LangChain constitue probablement l’outil le plus simple et le plus approprié. Cependant, dès que votre logique d’IA doit prendre des décisions divergentes, conserver de la mémoire ou s’arrêter pour collecter plus d’informations avant de continuer, une structure basée sur des graphes cesse d’apparaître comme une complexité inutile et devient alors le moyen le plus judicieux pour modéliser ce flux.

Références

Lectures complémentaires

• LangChain vs LangGraph : Choisir entre chaînes et graphes à état — Découvrez en quoi les blocs de construction linéaires de LangChain diffèrent des workflows à état et ramifiés de LangGraph, ainsi que la manière de déterminer lequel convient le mieux à votre application d’IA.
• Construire un agent d’IA de zéro : Patterns, ReAct et LangGraph — Apprenez les concepts fondamentaux des agents d’IA — planification, utilisation d’outils, réflexion et le pattern ReAct — ainsi que le rôle de LangChain et LangGraph dans la création manuelle d’un tel agent.