Головна / Статті / Автоматизація введення транзакцій у FastAPI за допомогою допоможника ШІ LangGraph

LangGraph LangChain FastAPI PostgreSQL LLM Groq

Автоматизація введення транзакцій у FastAPI за допомогою допоможника ШІ LangGraph

Дізнайтеся, як створити допоможника ШІ під керуванням LangGraph, який парсить природну мову у структуровані транзакції та записує їх у базу даних PostgreSQL через FastAPI.

5736 слів

Кожен, хто намагався відстежувати щоденні витрати за допомогою веб-форми, знає, наскільки це складно. Для запису покупки кави за 5 доларів не повинно знадоблятися заповнення кількох полів, проте саме це відбувається у багатьох додатках для відстеження фінансів, створених за допомогою FastAPI та PostgreSQL, де кожна транзакція — якою б незначною вона не була — мусить вводитися вручну.

Уявіть собі створення такого додатку, де кожна транзакція, проста чи складна, мусить проходити через форму. Це підходить для епізодичних записів, але стає виснажливим, коли потрібно задокументувати кілька транзакцій за один раз.

У такому випадку виникає природне запитання: а що, якщо весь цей процес можна автоматизувати? А що, якби замість заповнення форми можна було просто сказати асистентові: «Сьогодні я витратив 5 доларів на каву в ресторані», і нехай він сам займається рештою?

Саме тут стає корисним LangGraph.

Використовуючи LangGraph, ви можете створити AI-асистента, який приймає опис того, що сталось, у звичайній мові, та перетворює його на належним чином задокументовану транзакцію від вашого імені.

Вступ

У цій статті розглядаються основні концепції LangGraph та його екосистеми, а потім детально пояснюється, як можна додати AI-асистента до застосунку FastAPI з використанням LangGraph для автоматизації введення транзакцій.

LangGraph

LangGraph був створений командою, яка розробила LangChain, і це інструментарій з відкритим кодом для створення та керування робочими процесами AI-агентів за допомогою графових структур. За його допомогою ви описуєте процес як сукупність „вузлів“ та „ребер“, що дозволяє упорядкувати складну поведінку агента, зробити її масштабованою та легкіше керувати нею.

Перш ніж детальніше розглядати LangGraph, корисно спочатку зрозуміти LangChain, адже LangGraph будується на його основі.

LangChain

LangChain — це також інструментарій з відкритим кодом для створення додатків, які працюють за допомогою великих мовних моделей. Його основна функція — створити для розробників місток між LLM та зовнішніми ресурсами — джерелами даних, інструментами та кроками робочого процесу — щоб система могла виконувати багатокрокові міркування та автоматизовані завдання, а не просто обмінюватися одним ізольованим запитом та відповіддю.

Мета: Він призначений для створення додатків ШІ, яким потрібно поєднувати кілька кроків — наприклад, обробка вхідних даних користувача, отримання відповідної інформації та генерація відповіді.

Структура: LangChain базується на „ланцюгах“, які є впорядкованими послідовностями операцій, де результат кожного кроку стає вхідними даними для наступного. Це дозволяє розбивати складну логіку на менші, керовані частини.

Застосування: Типові сценарії використання включають чат-ботів, завдання з багаторівневим міркуванням, пошук та узагальнення документів, а також підключення LLM до зовнішніх інструментів чи API.

LangGraph (продовження)

Простіше кажучи, LangGraph організовує виклики LLM у робочі процеси у формі графа, що дозволяє здійснювати гнучке та навіть паралельне багаторівневе міркування замість суто лінійної послідовності.

Мета: Він дозволяє створювати AI-застосунки, у яких логіка може розгалужуватися, формувати цикли чи виконувати кроки паралельно, що виходить за межі того, що може передати простий послідовний ланцюг.

Структура: LangGraph представляє операції у вигляді „вузлів“, а потік даних між ними — у вигляді „ребер“. Вихідний сигнал одного вузла може надходити до кількох наступних вузлів, що дозволяє створювати динамічні шляхи прийняття рішень.

Застосування: LangGraph ідеально підходить для координації кількох агентів, створення складних конвеєрів прийняття рішень, автоматизації завдань, що вимагають умовної логіки, а також для одночасного керування кількома LLM чи інструментами.

Що таке граф у LangGraph?

Граф, загалом, — це нелінійна структура даних, яка складається з „вершин“ (вузлів) та „ребер“ (з’єднань між ними) та відображає взаємозв’язки між об’єктами.

У LangGraph саме ця структура графа використовується для створення станових, циклічних робочих процесів — таких, де ШІ може приймати рішення, повертатися до попередніх кроків або розгалужуватися по різних шляхах залежно від проміжних результатів.

LangChain проти LangGraph

Архітектура проєкту

(Кінцева точка FastAPI + LangGraph + створення та зберігання транзакцій)

Проблема

До того, як LangGraph було впроваджено у фінансовий додаток, для створення транзакції потрібно було безпосередньо викликати кінцеву точку /transactions/add із вантажем даних, схожим на цей:

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

Щоб досягти цього етапу, знадобилося два попередніх виклики API — один для отримання списку categories та інший для отримання payment_options — лише для того, щоб отримати ID, необхідні для передачі даних. Іншими словами, створення однієї транзакції було процесом у трьох кроках, причому досить повільним.

Рішення

Щоб вирішити цю проблему, було вирішено доручити всі три кроки штучному інтелекту-асистенту, тоді як користувачеві потрібно лише описати звичайною мовою, що він робив із своїми грошима. З огляду на цю мету ось як структуровано реалізацію.

Архітектура з трьома кроками

Кінцева точка FastAPI
Оркестратор LangGraph
База даних PostgreSQL

1. Кінцева точка FastAPI

Користувач надсилає запит до кінцевої точки FastAPI /assistance/transaction-entry із вантажем, який містить повідомлення з описом транзакції.

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

2. LangGraph Orchestrator

Оркестратор побудований у вигляді графа, де кожен вузол представляє операцію, а кожна ребро — потік даних між операціями.

Перший вузол, Analyzer LLM, отримує повідомлення користувача та перевіряє, чи містить воно всю необхідну інформацію для запису транзакції — чи йдеться про дохід чи витрату, суму, мету транзакції, спосіб оплати тощо.

Якщо повідомлення вже містить усі необхідні деталі, LLM перетворює його на структуровані дані, наприклад:

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

Ці структуровані дані потім надходять до вузла Database Writer після того, як два поля-флаги (is_complete та missing_info_message) видаляються. Вузол Database Writer викликає метод create_transaction(), який записує транзакцію щодо користувача в базі даних.

Але що станеться, якщо у повідомленні бракує певних деталей? Розглянемо повідомлення на кшталт:

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

Тут спосіб оплати не вказаний. У цьому випадку дані, отримані за допомогою LLM, будуть містити заповнені значення флагів, наприклад:

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

Оскільки флаг is_complete тут дорівнює False, значення missing_info_message направляється до іншого вузла, підключеного до Analyzer LLM – вузла Clarification. Цей шлях активується лише тоді, коли значення is_complete дорівнює False.

Вузол Clarification отримує missing_info_message та викликає метод ask_again(), який повертає це повідомлення як відповідь на початковий запит FastAPI. Це ознаменовує кінець виконання графа для цього запуску — результат, який отримує користувач, — це просто запит на надання відсутньої інформації, у цьому випадку способу оплати.

Припустимо, користувач потім надсилає відсутню інформацію, наприклад:

{
  message: "UPI"
}

Ця відповідь змушує граф оркестратора ініціалізуватися знову, і він проходить ту саму послідовність кроків, що й раніше.

Ключова відмінність у цьому другому етапі полягає у тому, що жодна з попередніх інформацій не втрачається — історія розмови зберігається щоразу, коли дані отримуються за допомогою LLM (цей механізм збереження детальніше розглядається пізніше в статті). Оскільки тепер доступний payment_option, який поєднується з раніше зафіксованими значеннями, значення is_complete змінюється на True, а остаточні, відфільтровані дані передаються до вузла Database Writer у такому вигляді:

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

Потім вузол Database Writer бере на себе обробку з цими відфільтрованими даними. Нагадаємо, що коли запис транзакції створювався вручну, було потрібно два додаткові виклики API — один для categories та один для payment_options — щоб отримати відповідні ID перед тим, як можна було створити саму транзакцію. Та сама проблема виникає й тут: у відфільтрованих даних містяться фактичні текстові значення категорій та варіантів оплати, а не їхні ID у базі даних, і база даних не прийматиме сирі значення для цих полів.

Щоб вирішити цю проблему, вузол Database Writer мусить здійснити запит до бази даних, щоб знайти відповідні записи категорій та варіантів оплати на основі значень, що містяться у відфільтрованих даних.

Оскільки проект базується на FastAPI разом із SQLAlchemy, ці пошуки реалізуються у вигляді запитів SQLAlchemy.

Для `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

Коротко кажучи, ця логіка: Виконує запит select для перевірки, чи існує вже категорія, згадана у data.category. Якщо вона існує, ID цієї категорії замінює значення у data.category. Якщо ні, створюється новий запис категорії, і замість нього використовується її новостворений ID.

Той самий патерн застосовується до `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

Як тільки ID категорії та ID варіанта оплати будуть визначені, об’єкт даних повністю оновлюється та готовий до вставки, виглядаючи ось так:

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

З цими остаточними даними вузол Database Writer викликає метод create_transaction(), який фактично зберігає запис транзакції в базі даних.

3. База даних PostgreSQL

Це останній етап архітектури, де завершені дані, надані вузлом Database Writer, записуються у таблицю transactions.

Отримана структура таблиці transactions виглядає так:

Реалізація

Після розгляду архітектури настав час детально розглянути сам процес створення цього оркестратора за допомогою LangGraph. Зауважте, що порядок, який тут наведено, не зовсім відповідає порядку розгляду архітектури вище. Натомість реалізація організована наступним чином:

LangGraph Orchestrator
База даних PostgreSQL
Кінцева точка FastAPI

1. LangGraph Orchestrator

Сам оркестратор знаходиться у файлі src/assistance/graph.py. Цей файл відповідає за налаштування моделі LLM, визначення вузлів графа, створення з’єднань між цими вузлами та, нарешті, компіляцію всього в робочий граф.

Як зазначалося раніше, цей оркестратор складається з трьох вузлів: Analyzer LLM, Database Writer та вузла Clarification.

Вузол Analyzer LLM (Groq)

Цей вузел по суті є мовною моделлю, завданням якої є визначення наміру користувача та перевірка наявності у повідомленні всіх необхідних та правильних деталей. Замість створення власної моделі з нуля цей проект використовує Groq для виконання складних завдань.

Що таке Groq?

Groq — це фреймворк на Python з відкритим кодом, створений для роботи з даними у графічній структурі. Він надає розробникам засоби для ефективного запитування, фільтрації та агрегації інформації, збереженої у вигляді графів, і чудово підходить для великих наборів даних у графічній формі, зокрема для соціальних мереж, графів знань чи систем рекомендацій, як описано у статті GeekForGeeks про API Groq.

За допомогою хостованого API Groq ви можете надсилати запити до широко використовуваних відкритих моделей — openai/gpt-oss-120b є тією, яка використовується у цьому проекті — та отримувати відповіді, які зазвичай надходять значно швидше, ніж від інших постачальників подібних моделей.

Чому саме Groq?

Гроq було обрано замість альтернатив, таких як ChatOpenAI чи ChatAnthropic, з кількох причин:

Швидкість: Groq використовує спеціально створене обладнання під назвою LPU (одиниці обробки мови), замість GPU, від яких залежать більшість інших постачальників, що забезпечує швидку обробку даних.
Корисний безкоштовний тариф: безкоштовний тариф Groq настільки щедрий, що дозволяє підтримувати індивідуальні чи навчальні проекти, не створюючи значних витрат на API під час експериментування.
Сумісність через LangChain: клас langchain_groq.ChatGroq інтегрується з LangChain та LangGraph так само, як це роблять ChatOpenAI чи ChatAnthropic. Це означає, що згодом для переходу на іншого постачальника не доведеться переробляти логіку графа — достатньо просто замінити клієнта.

Як отримати ключ API Groq

Groq дозволяє створювати безкоштовні ключі API для розробки. Ось як їх отримати:

Перейдіть на https://console.groq.com та увійдіть або зареєструйтесь.
У меню навігації виберіть опцію API Keys.
Виберіть опцію «Створити ключ API».
З’явиться форма, у якій потрібно вказати назву (для цього проекту використано transaction-assistant) та термін дії ключа. Після заповнення натисніть «Надіслати».
Ключ відображається лише один раз, відразу після створення — тож обов’язково скопіюйте його негайно.

Після створення всі ваші ключі з’являться у основному списку на цій сторінці.

Використання ключа API Groq у коді FastAPI

Додайте ключ API Groq до вашого файлу .env, розташованого у корені проекту, разом з іншими змінними середовища:

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

Існує кілька способів завантаження змінних середовища до модулів, які їх потребують. У цьому проекті використовується спеціальний клас налаштувань:

Визначте клас Settings у файлі 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()

Потім імпортуйте цей об’єкт налаштувань там, де він потрібен:

from src.utils.settings import settings

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

Налаштування LLM

Перш ніж налаштовувати LLM, встановіть LangGraph та LangChain разом із інтеграцією Groq: pip install -U langgraph langchain langchain-groq

Потім створюється екземпляр клієнта Groq та налаштовується з використанням конкретної моделі:

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)

Тут ChatGroq виступає як обгортка LangChain навколо чат-моделей Groq, що дозволяє взаємодіяти з ними через стандартний інтерфейс LangChain замість ручного формування HTTP-запитів.

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

Цей фрагмент створює згадану вище інстанцію клієнта Groq, налаштовану з обраним моделлю та значенням температури, що є низьким, а також автентифіковану за допомогою ключа API, отриманого з налаштувань середовища.

Температура — це параметр, який зазвичай коливається від 0 до 1, і визначає, наскільки випадковими чи творчими є відповіді моделі. Вище значення, наприклад 0.8, сприяє отриманню більш різноманітних та креативних результатів, тоді як нижче значення, наприклад 0.2, робить відповіді більш стабільними та передбачуваними. У цьому проекті встановлено temperature = 0.2.

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

Цей код обгортає LLM так, щоб замість повернення простого тексту він створював об’єкт Python, який повністю відповідає ExtractedTransactionSchema. Внутрішньо це досягається шляхом наказу моделі генерувати вихід, що відповідає схемі, а потім автоматичного парсингу та перевірки цього вихіду — що усуває необхідність ручної інтерпретації сирого тексту моделі.

Сама ExtractedTransactionSchema визначена всередині файлу 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

Зауважте, що на цьому етапі LLM ще не був викликаний — цей крок лише визначає форму, яку має мати вихід після його запуску.

Стан графу

Стан графа представляє структуру даних, яка проходить крізь граф та оновлюється під час його обробки. Уявіть собі це як робочу пам’ять диригента: вона зберігає всю інформацію, яку відстежує та змінює граф під час виконання кожного кроку. Для цього асистента з транзакціями стан графа визначається наступним чином:

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

Давайте розберемо, що насправді робить цей код:

from pydantic import BaseModel, Field

Pydantic — це бібліотека для перевірки даних, яка використовується тут. BaseModel — це батьківський клас, який ви розширюєте під час визначення структурованої, перевіреної за типом форми, як-от GraphState. Field дозволяє приєднувати метадані — описи, значення за замовчуванням тощо — до кожного окремого атрибута.

from typing import Annotated, List, Optional
import operator

Це інструменти типізації в Python. Optional вказує на те, що поле може бути порожнім та містити значення None. List означає, що атрибут є списком елементів. Annotated, у поєднанні з operator.add, повідомляє LangGraph: „коли вузол повертає нове значення для цього поля, додайте його до того, що вже є, замість того, щоб замінювати його“. Саме цей механізм дозволяє conversation_history рости з кожною новою реплікою, а не стиратися з кожним новим повідомленням.

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: найновіше повідомлення, яке надіслав користувач під час цього конкретного виклику.
conversation_history: повний список попередніх повідомлень, які накопичуються з кожною новою реплікою, замість того, щоб їх перезаписувати.

extracted: заповнюється після того, як ШІ витягне структуровані дані транзакції з розмови. Спочатку воно має значення None, оскільки на початку виконання ще нічого не було витягнуто.

final_response: повідомлення, яке зрештою надсилається користувачеві — це або підтвердження про те, що транзакція була записана, або додаткове запитання з проханням надати більше деталей.

Запит на витягування даних

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

Це буквальна інструкція, яка надається LLM у природній мові — вона вказує, які поля потрібно шукати, що робити, якщо щось відсутнє, та як має бути структурована відповідь. Оскільки structured_llm вже забезпечує дотримання схеми на рівні вихідних даних, завданням запиту є переважно керувати міркуваннями моделі: визначати, що означає „повність“, як сформулювати уточнююче запитання тощо, тоді як схема відповідає за форматування.

Екстрактор

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

Функція extractor виконує наступне:

Об’єднує кожне попереднє повідомлення з поточним, щоб LLM бачив повний контекст.
Передає цей об’єднаний текст LLM.
Отримує назад структурований об’єкт ExtractedTransactionSchema.

Повертає словник оновлень стану — нещодавно вилучені дані разом із поточним повідомленням, яке LangGraph автоматично включає до історії завдяки поведінці operator.add, налаштованій для цього поля.

Рішення

route_after_extraction

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

Ця функція не виконує жодної реальної обробки — її єдиною задачею є прийняття рішення. Залежно від того, чи позначив ШІ вилучені дані як повні, вона повертає рядок, який вказує графу, який вузол слід виконати далі. Його можна уявити як логіку розгалуження в діаграмі потоку: граф перевіряє значення, повернуте цією функцією, та йде відповідним шляхом — або до create_transaction для запису транзакції, або до ask_again для отримання додаткової інформації.

Вузол запису в базу даних

create_transaction_node

Цей вузол відповідає за запис даних завершеної транзакції до бази даних від імені відповідного користувача. Функція create_transaction_node, яка реалізує вузол DB Writer, виглядає так:

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}

Це занадто багато для одразу, тож розглянемо все поетапно.

from langchain_core.runnables import RunnableConfig

Тип, який представляє об’єкт config, переданий у будь-який вузол. Він існує лише як підказка щодо типу, тож будь-хто, хто читає сигнатуру create_transaction_node, одразу розуміє, яку форму має config.

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

Це звичайні імпорти SQLAlchemy, необхідні для виявлення помилок бази даних та для визначення типу асинхронної сесії бази даних, яка використовується для взаємодії з PostgreSQL.

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

Це використовує логіку створення транзакцій, яка вже застосовується в інших частинах додатку, разом із її схемою вхідних даних. Її повторне використання означає, що асистент створює транзакції через точно такий самий кодовий шлях, як і звичайний CRUD API, замість того, щоб дублювати цю логіку тут.

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

Це допоміжні елементи, які використовуються для перетворення назв категорій та варіантів оплати, отриманих за допомогою LLM, на реальні записи та ID у базі даних, створюючи нові записи, якщо їх ще немає.

Примітка: логіка пошуку/створення для categories та payment_options була об’єднана в один універсальний помічник get_or_create, оскільки обидва моделі потребували фактично однакової поведінки.

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

Функція create_transaction_node виконується лише після підтвердження повноти вилучених даних. Вона оголошена як async, тому що виконує реальні операції з базою даних, і приймає параметри config та state, щоб мати доступ до активної сесії бази даних та авторизованого користувача. Ці два значення надходять з маршруту API, а не від LLM чи стану розмови, оскільки вони належать конкретному запиту, а не до триваючого діалогу.

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

Цей код отримує сесію, користувача та вилучені дані транзакції.

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

Оскільки ШІ видобув лише назви категорії та способу оплати, такі як „Продукти“ чи „UPI“, а не їхні ID у базі даних, цей крок перевіряє, чи вже існує відповідний рядок для поточного користувача. Якщо ні, він створює його. У будь-якому разі повертається відповідний ID.

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

Пакет даних формується у тому самому форматі, який очікує логіка створення транзакцій, а потім передається у ту саму функцію-контролер, що дозволяє повторно використовувати існуючу логіку програми замість її переписування. Після цього транзакція в базі даних підтверджується для збереження змін.

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

Якщо щось зазнає невдачі на рівні бази даних, усі часткові зміни скасовуються, і повертається зрозуміле повідомлення про помилку, замість того щоб запит завершувався з помилкою. Це запобігає ситуації, коли, наприклад, створюється нова категорія, але немає відповідної транзакції.

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

У разі успіху створюється повідомлення про підтвердження, зрозуміле для людини, яке повертається як оновлення стану.

Узел уточнення

ask_again

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

Це простий запасний шлях. Як описано раніше, цей узел запускається лише тоді, коли значення is_complete для вилучених даних дорівнює False, разом із корисним повідомленням у missing_info_message.

Усередині ask_again_node функція отримує параметр state, що дає їй доступ до state.extracted.is_complete та state.extracted.missing_info_message.

Коротко кажучи, коли бракує інформації, цей вузол просто передає уточнювальне запитання, яке вже сформулював LLM під час видобутку даних, щоб користувач точно знав, що потрібно надати далі.

Складання графа

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

Це створює нового будівельника графа та повідомляє йому, що кожен вузол у графі буде читати з об’єкта формату 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")

Це встановлює точку входу: кожна експекутація графа починається з вузла extractor.

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

Саме тут відбувається розгалуження. Як тільки extractor завершує свою роботу, LangGraph викликає route_after_extraction, щоб визначити наступний крок. Будь-який рядок, який він повертає — чи то create_transaction, чи ask_again — шукається в цьому мапуванні, яке пов’язує кожен рядок рішення з конкретним вузлом, до якого потрібно перейти.

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

Обидва можливі шляхи розгалуження завершують виконання графа після свого завершення, коли досягається точка END по будь-якому з них.

Компіляція з використанням пам’яті

from langgraph.checkpoint.memory import MemorySaver

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

Виклик функції compile() перетворює визначення графа на щось, що можна запустити. Передача параметра checkpointer=memory активує механізм збереження стану, описаний раніше, тож повторний виклик графа з тим самим thread_id продовжує роботу з того місця, де вона була припинена, замість того, щоб почати все спочатку.

Остаточний код

З цим шар оркестрації є готовим. Ось завершений файл (src/assistance/graph.py):

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

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

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

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)


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


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

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

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

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

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

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


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

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


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


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


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


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

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

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

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

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


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


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

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

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

2. База даних PostgreSQL

Взаємодія з базою даних на цьому етапі вже обробляється всередині функції create_transaction_node, про яку йшлося раніше, де остаточні дані транзакції записуються у таблицю transactions.

3. Кінцева точка 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"]}

Клієнти викликають цей ендпоїнт (/assistance/transaction-entry) та додають у тіло запиту повідомлення, яке описує зміст транзакції.

Давайте розглянемо, що робить кожен елемент.

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

Це налаштовує маршрут POST за адресою /transaction-entry. Встановлення значення status_code=status.HTTP_201_CREATED вказує FastAPI, який код статусу повертати за замовчуванням у разі успіху. 201 — це стандартний код для позначення „створено новий ресурс“, що тут доречне, адже успішний виклик призводить до створення нового рядка транзакції.

Складання конфігурації графа:

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

Цей код створює об’єкт config, який передається під час виклику графа. Об’єкт config містить значення, специфічні для запиту, які не повинні зберігатися в самому стані діалогу.

"thread_id": str(user.id): це значення використовується чекпойнтером LangGraph для визначення історії розмови, яку потрібно отримати та оновити. Завдяки прив’язці до ID автентифікованого користувача кожен користувач автоматично отримує ізольовану, постійну смугу обробки даних, тож напівзавершена операція одного користувача ніколи не може вплинути на операцію іншого. Це значення перетворюється на рядок, оскільки чекпойнтер очікує thread_id у вигляді рядка, тоді як user.id зазвичай є UUID.
"session" та "user": вони передаються далі, щоб функція create_transaction_node, яка виконується всередині графа, мала доступ до поточної сесії бази даних та до інформації про того, хто подає запит.

Виклик графа:

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

Це рядок, який фактично запускає виконання. ainvoke є асинхронною аналогією до виконання графу; використання синхронного invoke призведе до блокування циклу подій, що має значення тут, оскільки create_transaction_node внутрішньо виконує асинхронні операції з базою даних.

Перший аргумент, {"user_input": payload.message}, позначає вихідний стан для цього запуску. Лише user_input потрібно вказувати явно; решта полів GraphState (conversation_history, extracted, final_response) або мають значення за замовчуванням, або заповнюються під час виконання через граф. Якщо існуючий thread_id вже має збережену історію, LangGraph об’єднує цей новий вхід із цим збереженим станом, замість того щоб починати з нуля.
config=config передає все, що було підготовлено на попередньому кроці: thread_id для пошуку відповідного стану, а також session та user для вузла, відповідального за запис у базу даних.

Ключове слово await призупиняє цей корутину до тих пір, доки граф не завершить свою повну роботу, оскільки ainvoke повертає корутину, яку необхідно очекати, перш ніж результат стане придатним для використання.

Те, що повертається як result, — це кінцевий стан GraphState, представлений у вигляді словника, який відображає процес виконання графа, незалежно від того, чи він завершився у функції create_transaction, чи у функції ask_again.

Надсилання відповіді назад:

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

Шлях завершується поверненням простого словника, який містить лише текст кінцевої відповіді. FastAPI бере на себе перетворення цього тексту на JSON-дані для клієнта, отримуючи щось на кшталт:

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

Цей текст ідентичний тому, що було сформовано раніше всередині функцій create_transaction_node або ask_again_node. Сам маршрут не знає, яка саме гілка була виконана; він просто передає те, що опинилося у змінній final_response.

Висновок

Робота з цим асистентом для транзакцій підкреслює те, що навчальні матеріали зазвичай ігнорують: складною частиною впровадження функції ШІ є не запит у модель на відповідь, а забезпечення того, щоб ця відповідь безпечно функціонувала під час взаємодії з реальною системою. Складання запитів — це проста частина. Справжні інженерні зусилля витрачаються на схеми, які забезпечують структурований вихід, умовні графи, що вирішують, чи потрібно зберегти дані чи запросити уточнення, а також на стан, який правильно передається протягом кількох етапів розмови.

LangGraph виявився ідеальним вибором для цього проекту саме тому, що робочий процес вимагав справжнього прийняття рішень, а не просто однократного переходу від вхідних даних до результату. Якщо ваша функція потребує простого лінійного процесу, то звичайний виклик LLM або ланцюг LangChain, ймовірно, є простішим та більш підходящим інструментом. Але коли логіка ШІ потребує розгалужень, зберігання інформації чи паузи для збору додаткових даних перед подальшою роботою, структура на основі графа вже не здається зайвою складністю, а стає найбільш розумним способом моделювання такого процесу.

Посилання

Пов’язана література

LangChain проти LangGraph: вибір між ланцюгами та графами зі станом — Дізнайтеся, чим відрізняються лінійні елементи будови в LangChain від робочих процесів зі станом та гілками в LangGraph, а також як вирішити, що краще підходить для вашого AI-застосунку.
Створення AI-агента з нуля: шаблони, ReAct та LangGraph — Ознайомтесь з основними концепціями AI-агентів — плануванням, використанням інструментів, рефлексією та шаблоном ReAct — а також з тим, як LangChain та LangGraph допомагають у ручній створці такого агента.

Усі статті Надіслати бриф

Вступ

LangGraph

LangChain

LangGraph (продовження)

Що таке граф у LangGraph?

LangChain проти LangGraph

Архітектура проєкту

Проблема

Рішення

Архітектура з трьома кроками

1. Кінцева точка FastAPI

2. LangGraph Orchestrator

Для categories:

Той самий патерн застосовується до payment_options:

3. База даних PostgreSQL

Реалізація

1. LangGraph Orchestrator

Вузол Analyzer LLM (Groq)

Що таке Groq?

Чому саме Groq?

Як отримати ключ API Groq

Використання ключа API Groq у коді FastAPI

Налаштування LLM

Стан графу

Запит на витягування даних

Екстрактор

Рішення

Вузол запису в базу даних

Узел уточнення

Складання графа

Компіляція з використанням пам’яті

Остаточний код

2. База даних PostgreSQL

3. Кінцева точка FastAPI

Складання конфігурації графа:

Виклик графа:

Надсилання відповіді назад:

Висновок

Посилання

Пов’язана література

Для `categories`:

Той самий патерн застосовується до `payment_options`: