Inicio / Artículos / Automatización de la entrada de transacciones en FastAPI con un asistente de IA LangGraph

LangGraph LangChain FastAPI PostgreSQL LLM Groq

Automatización de la entrada de transacciones en FastAPI con un asistente de IA LangGraph

Aprende cómo crear un asistente de IA orquestado por LangGraph que analice el lenguaje natural en transacciones estructuradas y las escriba en una base de datos PostgreSQL a través de FastAPI.

5736 palabras

Cualquiera que haya intentado hacer un seguimiento de los gastos diarios a través de un formulario web sabe lo tedioso que puede ser. Registrar una compra de un café por 5 dólares no debería requerir llenar múltiples campos, pero eso es exactamente lo que ocurre en muchas aplicaciones de seguimiento financiero desarrolladas con FastAPI y PostgreSQL, donde cada transacción —por pequeña que sea— tiene que ingresarse manualmente.

Imagínese crear una aplicación de este tipo, donde cada transacción, sea simple o complicada, tenga que pasar por un formulario. Esto funciona bien para registros ocasionales, pero se vuelve agotador cuando es necesario registrar varias transacciones de una sola vez.

En ese punto, surge una pregunta natural: ¿y si todo el proceso pudiera automatizarse? ¿Y si, en lugar de llenar un formulario, pudiera simplemente decirle a un asistente “Hoy gasté 5 dólares en café en el restaurante” y dejar que él se encargue del resto?

Es aquí donde LangGraph resulta útil.

Al utilizar LangGraph, puedes crear un asistente de IA que tome una descripción en lenguaje sencillo de lo que ocurrió y la convierta en una transacción debidamente registrada en tu nombre.

Introducción

Este artículo explica los conceptos fundamentales detrás de LangGraph y su ecosistema asociado, y luego profundiza en cómo se puede incorporar un asistente de IA a una aplicación FastAPI mediante LangGraph para automatizar la entrada de transacciones.

LangGraph

LangGraph proviene del equipo detrás de LangChain, y es una herramienta de código abierto para armar y gestionar flujos de trabajo de agentes de IA a través de estructuras de grafos. Con ella, describes un proceso como una colección de “nodos” y “aristas”, lo que mantiene el comportamiento complejo de los agentes organizado, escalable y más fácil de controlar.

Antes de adentrarnos más en LangGraph, es útil comprender primero LangChain, ya que LangGraph se basa en él.

LangChain

LangChain es una herramienta, también de código abierto, para crear aplicaciones impulsadas por modelos de lenguaje grandes. Su función principal es ofrecer a los desarrolladores un puente entre un LLM y recursos externos —fuentes de datos, herramientas y pasos del flujo de trabajo— para que un sistema pueda realizar razonamientos en múltiples pasos y tareas automatizadas, en lugar de limitarse a un único intercambio de preguntas y respuestas aislado.

Propósito: Está diseñado para desarrollar aplicaciones de IA que necesitan concatenar varios pasos: por ejemplo, manejar la entrada del usuario, recuperar información relevante y generar una respuesta.

Estructura: LangChain se basa en “cadenas”, que son secuencias ordenadas de operaciones donde la salida de cada paso se utiliza como entrada para el siguiente. Esto permite descomponer la lógica compleja en partes más pequeñas y manejables.

Aplicaciones: Los casos de uso típicos incluyen chatbots, tareas de razonamiento multi-paso, recuperación y resumen de documentos, así como la conexión de LLMs con herramientas o APIs externas.

LangGraph (continuación)

En términos sencillos, LangGraph organiza las llamadas a los LLM en flujos de trabajo en forma de gráfico, lo que permite un razonamiento multi-paso flexible e incluso paralelo, en lugar de una secuencia estrictamente lineal.

Propósito: Permite desarrollar aplicaciones de IA en las que la lógica puede ramificarse, formar bucles o ejecutar pasos en paralelo, superando así las limitaciones de una simple cadena secuencial.

Estructura: LangGraph representa las operaciones como “nodos” y el flujo de datos entre ellos como “aristas”. La salida de un único nodo puede alimentar a varios nodos siguientes, lo que permite rutas de decisión dinámicas.

Aplicaciones: LangGraph es ideal para coordinar múltiples agentes, crear pipelines de decisión complejos, automatizar tareas que involucran lógica condicional y orquestar varios LLMs o herramientas al mismo tiempo.

¿Qué es un gráfico en LangGraph?

En general, un gráfico es una estructura de datos no lineal compuesta por “vértices” (nodos) y “aristas” (las conexiones entre ellos) que capturan las relaciones entre objetos.

En LangGraph, específicamente, esta estructura de gráfico se utiliza para crear flujos de trabajo cíclicos y con estado, donde la IA puede tomar decisiones, volver a pasos anteriores o bifurcarse en rutas diferentes según los resultados intermedios.

LangChain vs. LangGraph

Arquitectura del proyecto

(Punto de extremo FastAPI + LangGraph + creación y persistencia de transacciones)

El problema

Antes de que LangGraph se introdujera en la aplicación financiera, crear una transacción significaba llamar directamente al punto de extremo /transactions/add, con una carga que se veía así:

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

Para llegar a ese punto fueron necesarias dos llamadas previas a la API: una para obtener la lista de categories y otra para obtener las payment_options, solo con el fin de conseguir los IDs requeridos para el payload. En otras palabras, crear una sola transacción era un proceso de tres pasos, y además lento.

La solución

La solución consistió en dejar que un asistente de IA se encargara de los tres pasos, mientras que el usuario solo necesitaba describir, en lenguaje cotidiano, qué hizo con su dinero. Teniendo ese objetivo en mente, así es como está estructurada la implementación.

La arquitectura de tres pasos

Punto de extremo FastAPI
Orquestador LangGraph
Banco de datos PostgreSQL

1. Punto de extremo FastAPI

El usuario envía una solicitud al endpoint de FastAPI /assistance/transaction-entry, con un payload que contiene un mensaje que describe la transacción.

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

2. Orquestador LangGraph

El orquestador está construido como un grafo, donde cada nodo representa una operación y cada arista representa el flujo de datos entre las operaciones.

El primer nodo, el LLM Analizador, recibe el mensaje del usuario y verifica si contiene todo lo necesario para registrar una transacción: ya sea ingreso o gasto, la cantidad involucrada, el propósito de la transacción, el método de pago utilizado, etc.

Si el mensaje ya contiene todos los detalles requeridos, el LLM lo convierte en datos estructurados, por ejemplo:

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

Estos datos estructurados luego se envían al nodo Database Writer, después de que se eliminen los dos campos de marcador (is_complete y missing_info_message). El nodo Database Writer llama al método create_transaction(), el cual registra la transacción correspondiente al usuario en la base de datos.

Pero, ¿qué sucede si el mensaje carece de algunos detalles? Considere un mensaje como este:

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

Aquí, el modo de pago no se especifica. En este caso, los datos extraídos por el LLM incluirán valores de marcador ya definidos, como estos:

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

Dado que el marcador is_complete es False aquí, el mensaje missing_info_message se dirige al otro nodo conectado al LLM Analyzer: el nodo Clarification. Este camino solo se activa cuando is_complete tiene como valor False.

El nodo Clarification recibe el missing_info_message y llama al método ask_again(), el cual devuelve dicho mensaje como respuesta a la solicitud original de FastAPI. Esto marca el final de la ejecución del grafo en esta ejecución: la salida que recibe el usuario es simplemente una petición para que proporcione el detalle faltante, en este caso el modo de pago.

Supongamos que el usuario responde luego con la información faltante, por ejemplo:

{
  message: "UPI"
}

Esta respuesta hace que el grafo del orquestador se inicialice nuevamente, y sigue la misma secuencia de pasos que antes.

La diferencia clave en esta segunda pasada es que no se pierde ninguna de la información anterior: el historial de conversaciones se conserva cada vez que el LLM extrae datos (este mecanismo de persistencia se explica con más detalle más adelante en el artículo). Dado que ahora está disponible payment_option y se combina con los valores capturados anteriormente, is_complete pasa a ser True, y los datos finalizados y filtrados se transmiten al nodo Database Writer, de la siguiente manera:

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

El nodo Database Writer toma entonces el control con estos datos filtrados en mano. Recuerde que cuando se creaba manualmente una entrada de transacción, eran necesarias dos llamadas API adicionales: una para categories y otra para payment_options, solo para obtener sus IDs respectivos antes de poder crear la transacción. El mismo problema se presenta aquí: los datos filtrados contienen los valores de texto reales de la categoría y la opción de pago, no sus IDs en la base de datos, y esta no aceptará valores sin procesar para estos campos.

Para resolverlo, el nodo Database Writer debe consultar la base de datos para buscar los registros correspondientes de categoría y opción de pago según los valores presentes en los datos filtrados.

Dado que el proyecto utiliza FastAPI junto con SQLAlchemy, estas búsquedas se implementan como consultas de SQLAlchemy.

Para `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

Brevemente, esta lógica: Ejecuta una consulta SELECT para verificar si la categoría referenciada en data.category ya existe. Si existe, el ID de la categoría reemplaza el valor en data.category. Si no existe, se crea un nuevo registro de categoría y se utiliza su ID recién generado en su lugar.

El mismo patrón se aplica a `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

Una vez que se han determinado tanto el ID de la categoría como el ID de la opción de pago, el objeto de datos se actualiza completamente y está listo para su inserción, con este aspecto:

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

Con estos datos finalizados, el nodo Database Writer llama al método create_transaction(), que realmente persiste el registro de la transacción en la base de datos.

3. Base de datos PostgreSQL

Esto representa la última etapa de la arquitectura, donde los datos finalizados entregados por el nodo Database Writer se escriben en la tabla transactions.

La estructura resultante de la tabla transactions se ve así:

La implementación

Una vez cubierta la arquitectura, es momento de analizar los detalles reales de implementación para construir este orquestador con LangGraph. Tenga en cuenta que el orden seguido aquí no refleja exactamente el recorrido arquitectónico mencionado anteriormente. En su lugar, la implementación está organizada de la siguiente manera:

Orquestador LangGraph
Banco de datos PostgreSQL
Punto final FastAPI

1. Orquestador LangGraph

El propio orquestador se encuentra dentro de src/assistance/graph.py. Este archivo es el responsable de configurar el LLM, definir los nodos del grafo, conectarlos entre sí y, finalmente, compilar todo en un grafo ejecutable.

Tal como se mencionó anteriormente, este orquestador está compuesto por tres nodos: el Analyzer LLM, el Database Writer y el nodo de Clarification.

Nodo Analyzer LLM (Groq)

Este nodo es, en esencia, un modelo de lenguaje cuya función es determinar la intención del usuario y verificar si el mensaje contiene todos los detalles necesarios y correctos. En lugar de crear un modelo personalizado desde cero, este proyecto depende de Groq para realizar esta tarea compleja.

¿Qué es Groq?

Groq es un framework de código abierto para Python diseñado para trabajar con datos estructurados en forma de grafo. Ofrece a los desarrolladores una forma expresiva de consultar, filtrar y agregar información almacenada como grafos, y es especialmente adecuado para grandes conjuntos de datos en formato grafo, como redes sociales, grafos de conocimiento o motores de recomendación, tal como se describe en un artículo de GeekForGeeks sobre la API de Groq.

A través de la API alojada por Groq, puedes enviar solicitudes a modelos abiertos ampliamente utilizados — openai/gpt-oss-120b es el que se utiliza en este proyecto — y recibir respuestas que suelen llegar notablemente más rápido que las que obtendrías normalmente de otros proveedores que ofrecen modelos similares.

¿Por qué Groq?

Groq fue elegido en lugar de alternativas como ChatOpenAI o ChatAnthropic por varias razones:

Rapidez: Groq se basa en hardware diseñado específicamente llamado LPUs (Unidades de Procesamiento del Lenguaje), en lugar de las GPUs de las que dependen la mayoría de los otros proveedores, lo que permite una inferencia rápida.
Nivel gratuito útil: El nivel gratuito de Groq es lo suficientemente generoso como para respaldar un proyecto individual o orientado al aprendizaje sin generar costos significativos de API durante las pruebas.
Compatibilidad inmediata a través de LangChain: la clase langchain_groq.ChatGroq se integra con LangChain y LangGraph de la misma manera que lo harían ChatOpenAI o ChatAnthropic. Eso significa que cambiar a otro proveedor más adelante no requeriría rehacer la lógica del grafo, solo sustituir el cliente.

Cómo obtener una clave API de Groq

Groq permite generar claves API gratuitas para fines de desarrollo. Aquí está cómo obtener una:

Vaya a https://console.groq.com e inicie sesión o regístrate.
Seleccione la opción Claves API en la barra de navegación.
Elija Crear clave API.
Aparecerá un formulario que solicitará un nombre (este proyecto utilizó transaction-assistant) y un período de vencimiento para la clave. Envíe el formulario una vez completado.
La clave se muestra solo una vez, inmediatamente después de su creación; así que asegúrese de copiarla de inmediato.

Una vez creada, todas sus claves aparecerán en la lista principal de esa página.

Usar la clave API de Groq dentro del código FastAPI

Agregue la clave API de Groq a su archivo .env en la raíz del proyecto, junto con las demás variables de entorno:

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

Existen varios métodos para cargar variables de entorno en los módulos que las necesitan. Este proyecto utiliza una clase dedicada a la configuración:

Defina una clase Settings dentro 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()

Luego importe ese objeto de configuración donde sea necesario:

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.

Configuración de LLM

Antes de configurar el LLM, instale LangGraph y LangChain junto con la integración de Groq:

pip install -U langgraph langchain langchain-groq

A continuación, se crea una instancia del cliente de Groq y se configura con un modelo específico:

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)

Aquí, ChatGroq actúa como el wrapper de LangChain alrededor de los modelos de chat de Groq, permitiéndole interactuar con ellos a través de la interfaz estándar de LangChain en lugar de crear solicitudes HTTP manualmente.

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

Este fragmento crea la instancia del cliente Groq mencionada anteriormente, configurada con un modelo seleccionado y un valor de temperatura bajo, además de autenticarse mediante la clave API obtenida de los parámetros del entorno.

La temperatura es un parámetro que, por lo general, varía entre 0 y 1, y determina cuán aleatorias o creativas son las respuestas del modelo. Un valor más alto, como 0.8, genera resultados más variados y creativos, mientras que un valor más bajo, como 0.2, mantiene las respuestas más consistentes y predecibles. En este proyecto se establece temperature = 0.2.

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

Este código envuelve al LLM de tal manera que, en lugar de devolver texto plano, genera un objeto de Python que se ajusta exactamente a ExtractedTransactionSchema. Internamente, esto se logra instruyendo al modelo para que genere una salida que coincida con el esquema y, a continuación, analizando y validando automáticamente dicha salida, lo que elimina la necesidad de interpretar manualmente el texto bruto del modelo.

El propio ExtractedTransactionSchema está definido dentro de 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

Tenga en cuenta que en este punto el LLM aún no se ha llamado; este paso solo define la estructura que debe tener la salida una vez que se invoque.

Estado del grafo

El estado del grafo representa la estructura de datos que fluye a través del mismo y se actualiza en cada paso. Piénselo como la memoria de trabajo del orquestador: alberga toda la información que el grafo registra y modifica a medida que se ejecutan las distintas etapas. Para este asistente de transacciones, el estado del grafo se define de la siguiente manera:

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

Analicemos qué hace realmente este código:

from pydantic import BaseModel, Field

Pydantic es la biblioteca de validación de datos que se utiliza aquí. BaseModel es la clase padre a la que se extiende al definir una estructura organizada y verificada en cuanto a tipos, como GraphState. Field permite adjuntar metadatos —descripciones, valores por defecto, etc.— a cada atributo individual.

from typing import Annotated, List, Optional
import operator

Estas son las herramientas de tipado de Python. Optional indica que un campo puede estar vacío y contener None. List asigna a un atributo el tipo de lista de elementos. Annotated, combinado con operator.add, es lo que le indica a LangGraph “cuando un nodo devuelve un nuevo valor para este campo, agréguelo a lo que ya existe en lugar de reemplazarlo”. Ese es el mecanismo que permite que conversation_history crezca a lo largo de las interacciones en lugar de borrarse con cada nuevo mensaje.

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: el mensaje más reciente que el usuario envió en esta llamada específica.
conversation_history: el historial completo de mensajes anteriores, acumulado turno a turno en lugar de ser sobrescrito.

extracted: se llena una vez que el LLM ha extraído los datos estructurados de la transacción del diálogo. Inicialmente es None porque al inicio de la ejecución aún no se ha extraído nada.

final_response: el mensaje que finalmente se envía al usuario, ya sea una confirmación de que la transacción fue registrada o una pregunta adicional para solicitar más detalles.

El prompt de extracción

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

Esta es la instrucción literal que se le proporciona al LLM en lenguaje natural: especifica qué campos buscar, qué hacer cuando falta algo y cómo debe estructurarse la respuesta. Dado que structured_llm ya aplica el esquema a nivel de salida, la función del prompt consiste principalmente en guiar el razonamiento del modelo: decidir qué significa “completo”, cómo formular una pregunta de aclaración, etc., mientras que el esquema se encarga del formato.

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

La función extractor realiza lo siguiente:

Fusiona todos los mensajes anteriores con el actual para que el LLM tenga el contexto completo.
Envía ese texto fusionado al LLM.
Recibe de vuelta un objeto ExtractedTransactionSchema estructurado.

Devuelve un diccionario con las actualizaciones de estado: los datos recién extraídos junto con el mensaje actual, que LangGraph integra automáticamente en el historial gracias al comportamiento de operator.add configurado en ese campo.

La decisión

route_after_extraction

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

Esta función no realiza ningún procesamiento real; su única tarea es tomar una decisión. Dependiendo de si el LLM marcó los datos extraídos como completos, devuelve una cadena que indica al grafo qué nodo debe ejecutarse a continuación. Puedes considerarla como la lógica de ramificación en el diagrama de flujo: el grafo examina el valor devuelto por esta función y sigue la ruta correspondiente, ya sea dirigiéndose a create_transaction para registrar la transacción o a ask_again para solicitar más información.

Nodo escritor de base de datos

create_transaction_node

Este nodo es responsable de escribir los datos de la transacción completada en la base de datos correspondiente al usuario adecuado. La función create_transaction_node que implementa el nodo DB Writer se ve así:

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}

Es mucho para asimilar de una sola vez, así que analicémoslo pieza por pieza.

from langchain_core.runnables import RunnableConfig

Un tipo que representa al objeto config pasado a cualquier nodo. Existe únicamente como una indicación de tipo, por lo que cualquiera que lea la firma de create_transaction_node comprende inmediatamente qué forma tiene config.

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

Estas son las importaciones habituales de SQLAlchemy necesarias para capturar errores de la base de datos y para definir el tipo de la sesión de base de datos asíncrona utilizada para comunicarse con PostgreSQL.

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

Esto incorpora la lógica de creación de transacciones que ya se utiliza en otras partes de la aplicación, junto con su esquema de entrada. Reutilizarla significa que el asistente crea transacciones a través del mismo camino de código que la API CRUD regular, en lugar de duplicar esa lógica aquí.

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

Se trata de componentes auxiliares que se utilizan para convertir los nombres de categorías y opciones de pago extraídos por el LLM en filas e IDs reales en la base de datos, creando nuevos registros cuando aún no existen.

Nota: la lógica de búsqueda/creación para categories y payment_options se fusionó en una función auxiliar genérica, get_or_create, ya que ambos modelos necesitaban esencialmente el mismo comportamiento.

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

La función create_transaction_node solo se ejecuta una vez que se confirma que los datos extraídos están completos. Está declarada como async porque realiza operaciones reales en la base de datos, y necesita config junto con state para poder acceder a la sesión de base de datos activa y al usuario conectado. Estos dos valores provienen de la ruta API y no del LLM ni del estado de la conversación, ya que pertenecen a la solicitud específica y no al diálogo en curso.

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

Esto extrae la sesión, el usuario y los datos de la transacción obtenidos.

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

Dado que el LLM solo extrajo los nombres de la categoría y del método de pago, como “Comestibles” o “UPI”, y no sus IDs en la base de datos, este paso verifica si ya existe una fila correspondiente para el usuario actual. Si no, se crea una. En cualquier caso, se devuelve el ID correspondiente.

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

La carga útil se organiza con la misma estructura que espera la lógica existente para crear transacciones, y luego se pasa a esa misma función del controlador, reutilizando la lógica de la aplicación existente en lugar de volver a escribirla. A continuación, se confirma la transacción en la base de datos para persistir el cambio.

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

Si algo falla en la capa de base de datos, todos los cambios parciales se deshacen y se devuelve un mensaje de error amigable en lugar de permitir que la solicitud falle. Esto evita situaciones como tener una categoría recién creada pero sin la transacción correspondiente.

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

En caso de éxito, se genera un mensaje de confirmación legible por humanos y se devuelve como actualización del estado.

Nodo de aclaración

ask_again

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

Este es un camino de respaldo simple. Como se describió anteriormente, este nodo solo se ejecuta cuando los datos extraídos tienen is_complete establecido en False, acompañado de un mensaje útil en missing_info_message.

Dentro de ask_again_node, la función recibe state, lo que le permite acceder a state.extracted.is_complete y state.extracted.missing_info_message.

En resumen, cada vez que falta información, este nodo simplemente reenvía la pregunta de aclaración que el LLM ya generó durante la extracción, para que el usuario sepa exactamente qué proporcionar a continuación.

Construcción del grafo

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

Esto crea un nuevo constructor de grafo y le indica que cada nodo del grafo leerá y escribirá en un objeto con la estructura de 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)

Cada función se registra aquí como un nodo con nombre, esencialmente un paso etiquetado, dentro del grafo.

graph_builder.add_edge(START, "extractor")

Esto establece el punto de entrada: cada ejecución del grafo comienza en el nodo extractor.

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

Aquí es donde tiene lugar la toma de decisiones. Una vez que extractor termina, LangGraph llama a route_after_extraction para determinar el siguiente paso. Cualquier cadena que devuelva, ya sea create_transaction o ask_again, se busca en este mapeo, el cual conecta cada cadena de decisión con el nodo real al que debe dirigirse.

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

Las dos ramas posibles terminan la ejecución del grafo una vez que se completan, llegando al END por cualquiera de los caminos.

Compilación con memoria

from langgraph.checkpoint.memory import MemorySaver

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

Llamar a compile() convierte la definición del grafo en algo ejecutable. Al pasar checkpointer=memory, se activa el mecanismo de persistencia de estado descrito anteriormente, de modo que al invocar nuevamente el grafo con el mismo thread_id se reanuda la conversación desde donde se interrumpió en lugar de reiniciarla.

El código final

Con esto, la capa de orquestación está completa. Aquí está el archivo finalizado (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 datos PostgreSQL

La interacción con la base de datos en esta etapa ya se maneja dentro de create_transaction_node, tratado anteriormente, donde los datos de la transacción finalizados se escriben en la tabla de transacciones.

3. Punto de extremo 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"]}

Los clientes llaman a este endpoint (/assistance/transaction-entry) e incluyen un mensaje en el cuerpo de la solicitud que describe de qué se trataba la transacción.

Analicemos qué hace cada parte.

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

Esto configura una ruta POST en /transaction-entry. Al establecer status_code=status.HTTP_201_CREATED se indica a FastAPI qué código de estado devolver por defecto en caso de éxito. 201 es el código convencional para “se creó un recurso nuevo”, lo cual es apropiado aquí ya que una llamada exitosa genera una nueva fila de transacción.

Construcción de la configuración del grafo:

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

Esto crea el objeto config que se pasa a la invocación del grafo. config contiene valores específicos de la solicitud que no deberían formar parte del estado persistente de la conversación.

"thread_id": str(user.id): este valor es el que utiliza el checkpointer de LangGraph para determinar cuál historial de conversación recuperar y actualizar. Al asociarlo con el ID del usuario autenticado, cada usuario obtiene automáticamente un hilo aislado y persistente, de modo que la entrada de transacción inconclusa de un usuario nunca puede afectar a la del otro. Se convierte en cadena de texto porque el checkpointer espera que thread_id sea una cadena, mientras que user.id suele ser un UUID.
"session" y "user": estos se transmiten para que create_transaction_node, ejecutado dentro del grafo, tenga acceso a la sesión de base de datos en tiempo real y a los detalles sobre quién realiza la solicitud.

Invocar el grafo:

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

Esta es la línea que realmente activa la ejecución. ainvoke es el equivalente asíncrono de ejecutar el grafo; utilizar el invoke síncrono en su lugar bloquearía el bucle de eventos, lo cual es importante aquí porque create_transaction_node realiza operaciones de base de datos asíncronas internamente.

El primer argumento, {"user_input": payload.message}, representa el estado inicial para esta ejecución. Solo es necesario proporcionar explícitamente user_input; los demás campos de GraphState (conversation_history, extracted, final_response) ya vienen con valores predeterminados o se llenan a medida que la ejecución avanza por el grafo. Cuando un thread_id existente ya cuenta con historial guardado, LangGraph integra esta nueva entrada en ese estado almacenado en lugar de comenzar desde cero.
config=config proporciona todo lo preparado en el paso anterior: thread_id para localizar el estado adecuado, además de session y user para el nodo encargado de la escritura en la base de datos.

La palabra clave await suspende esta corutina hasta que el grafo finalice su ejecución completa, ya que ainvoke devuelve una corutina que debe ser esperada antes de que el resultado sea utilizable.

Lo que quiera que regrese como result es el estado final GraphState, representado como un diccionario que refleja la ejecución del grafo, independientemente de si terminó en create_transaction o en ask_again.

Devolución de la respuesta:

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

La ruta finaliza devolviendo un simple diccionario que contiene únicamente el texto de la respuesta final. FastAPI se encarga de convertir esto en un payload JSON para el cliente, generando algo similar a lo siguiente:

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

Este texto es idéntico a lo que se generó anteriormente dentro de create_transaction_node o ask_again_node. La ruta en sí misma no sabe qué rama se ejecutó realmente; simplemente reenvía lo que terminó en final_response.

Conclusión

Trabajar con este asistente de transacciones destaca algo que los tutoriales suelen pasar por alto: la parte difícil al implementar una función de IA no es solicitar una respuesta al modelo, sino asegurarse de que esa respuesta se comporte de manera segura una vez interactúe con un sistema real. Redactar las instrucciones es lo sencillo. El verdadero esfuerzo de ingeniería se invierte en esquemas que obliguen a una salida estructurada, gráficos condicionales que decidan entre guardar los datos o solicitar aclaraciones, y un estado que se transmita correctamente a lo largo de varias interacciones.

LangGraph resultó adecuado para este proyecto específicamente porque el flujo de trabajo requería la toma real de decisiones en lugar de un proceso simple de entrada a salida. Si su funcionalidad solo necesita un flujo lineal, una llamada directa a un LLM o una cadena de LangChain probablemente sean la herramienta más sencilla y apropiada. Pero una vez que la lógica de su IA necesita ramificarse, conservar memoria o detenerse para recopilar más información antes de continuar, una estructura basada en grafos deja de parecer una complejidad innecesaria y pasa a ser la forma más sensata de modelar ese flujo.

Referencias

Lecturas relacionadas

LangChain vs LangGraph: Elegir entre cadenas y grafos con estado — Aprenda cómo los bloques de construcción lineales de LangChain difieren de los flujos de trabajo con estado y ramificaciones de LangGraph, y cómo decidir cuál se adapta mejor a su aplicación de IA.
Crear un agente de IA desde cero: patrones, ReAct y LangGraph — Conozca los conceptos fundamentales detrás de los agentes de IA: planificación, uso de herramientas, reflexión y el patrón ReAct—y cómo LangChain y LangGraph se integran en la creación manual de uno.

Todos los artículos Enviar el brief