Startseite / Artikel / Automatisierung der Transaktionseingabe in FastAPI mit einem LangGraph-KI-Assistenten

LangGraph LangChain FastAPI PostgreSQL LLM Groq

Automatisierung der Transaktionseingabe in FastAPI mit einem LangGraph-KI-Assistenten

Lernen Sie, wie man einen von LangGraph gesteuerten KI-Assistenten erstellt, der natürliche Sprache in strukturierte Transaktionen analysiert und diese über FastAPI in eine PostgreSQL-Datenbank schreibt.

5736 Wörter

Jeder, der versucht hat, seine täglichen Ausgaben über ein Webformular zu verfolgen, weiß, wie mühsam das sein kann. Die Erfassung eines Kaffeekaufs im Wert von 5 Dollar sollte nicht erfordern, dass man durch mehrere Felder navigieren muss – doch genau das passiert in vielen Finanzverfolgungs-Apps, die mit FastAPI und PostgreSQL entwickelt wurden, wo jede Transaktion – egal wie klein – manuell eingegeben werden muss.

Stellen Sie sich vor, Sie entwickeln eine solche App, bei der jede Transaktion, egal ob einfach oder kompliziert, über ein Formular abgewickelt werden muss. Das funktioniert gut für gelegentliche Einträge, wird aber anstrengend, sobald man mehrere Transaktionen auf einmal erfassen muss.

Dann stellt sich eine natürliche Frage: Was wäre, wenn der gesamte Prozess automatisiert werden könnte? Was wäre, wenn man anstelle des Ausfüllens eines Formulars einfach einem Assistenten sagen könnte: „Ich habe heute 5 Dollar für Kaffee im Restaurant ausgegeben“ und der Rest würde dann von ihm erledigt?

Darum wird LangGraph nützlich.

Mit LangGraph können Sie einen KI-Assistenten erstellen, der eine Beschreibung des Geschehens in Alltagssprache entgegennimmt und diese für Sie in eine ordnungsgemäß erfasste Transaktion umwandelt.

Einführung

In diesem Artikel werden die grundlegenden Konzepte hinter LangGraph und seinem Umfeld erläutert, anschließend wird detailliert gezeigt, wie ein KI-Assistent mithilfe von LangGraph in eine FastAPI-Anwendung integriert werden kann, um den Eingang von Transaktionen zu automatisieren.

LangGraph

LangGraph stammt vom Team hinter LangChain und ist ein Open-Source-Toolkit zur Erstellung und Verwaltung von KI-Agenten-Arbeitsabläufen über Graphstrukturen. Damit beschreiben Sie einen Prozess als eine Sammlung von „Knoten“ und „Kanten“, wodurch komplexes Agentenverhalten organisiert, skalierbar und leichter steuerbar wird.

Bevor man sich weiter mit LangGraph beschäftigt, ist es hilfreich, zunächst LangChain zu verstehen, da LangGraph darauf aufbaut.

LangChain

LangChain ist ein ebenfalls open-source basierendes Werkzeugset zur Erstellung von Anwendungen, die von großen Sprachmodellen angetrieben werden. Seine Hauptaufgabe besteht darin, Entwicklern eine Brücke zwischen einem LLM und externen Ressourcen – Datenquellen, Tools sowie Arbeitsablaufschritten – zu bieten, damit ein System mehrstufiges Denken und automatisierte Aufgaben ausführen kann, anstatt nur auf einen einzigen isolierten Prompt-Antwort-Austausch angewiesen zu sein.

Zweck: Es ist dafür konzipiert, KI-Anwendungen zu entwickeln, die mehrere Schritte miteinander verknüpfen müssen – beispielsweise die Verarbeitung von Benutzereingaben, das Abrufen relevanter Informationen und die Erstellung einer Antwort.

Struktur: LangChain basiert auf „Ketten“, also geordneten Abfolgen von Operationen, bei denen die Ausgabe eines Schritts als Eingabe für den nächsten Schritt dient. Dadurch kann komplexe Logik in kleinere, handhabbare Teile aufgeteilt werden.

Anwendungen: Typische Anwendungsfälle umfassen Chatbots, mehrschrittige Schlussfolgerungsaufgaben, Dokumentensuche und -zusammenfassung sowie die Verbindung von LLMs mit externen Tools oder APIs.

LangGraph (Fortsetzung)

Einfach ausgedrückt ordnet LangGraph Aufrufe von LLMs in grafenförmige Arbeitsabläufe, wodurch flexible und sogar parallele mehrschrittige Schlussfolgerungen möglich sind – anstelle einer streng linearen Abfolge.

Zweck: Es ermöglicht KI-Anwendungen, bei denen die Logik verzweigen, in Schleifen geraten oder Schritte parallel ausführen kann – über das hinaus, was eine einfache sequenzielle Kette ausdrücken kann.

Struktur: LangGraph stellt Operationen als „Knoten“ dar und den Datenfluss zwischen ihnen als „Kanten“. Die Ausgabe eines einzelnen Knotens kann mehrere nachgeschaltete Knoten versorgen, was dynamische Entscheidungspfade ermöglicht.

Anwendungen: LangGraph eignet sich hervorragend zur Koordination mehrerer Agenten, zum Aufbau komplexer Entscheidungspipelines, zur Automatisierung von Aufgaben mit bedingter Logik sowie zum gleichzeitigen Orchestrieren mehrerer LLMs oder Tools.

Was ist ein Graph in LangGraph?

Ein Graph ist im Allgemeinen eine nicht-lineare Datenstruktur, die aus „Knoten“ (Vertices) und „Kanten“ (den Verbindungen zwischen ihnen) besteht und Beziehungen zwischen Objekten darstellt.

In LangGraph wird diese Graphenstruktur insbesondere verwendet, um zustandsbasierte, zyklische Workflows zu erstellen – solche, in denen die KI Entscheidungen treffen kann, zu früheren Schritten zurückkehren oder je nach Zwischenergebnissen in unterschiedliche Pfade abbiegen kann.

LangChain vs. LangGraph

Architektur des Projekts

(FastAPI-Endpunkt + LangGraph + Erstellung und Speicherung von Transaktionen)

Das Problem

Vor der Einführung von LangGraph in der Finanzanwendung bedeutete die Erstellung einer Transaktion, direkt den /transactions/add-Endpunkt aufzurufen, wobei die übermittelten Daten wie folgt aussahen:

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

Damit zu gelangen, waren zwei vorherige API-Aufrufe notwendig – einer, um die Liste der categories abzurufen, und ein weiterer, um die payment_options zu holen – nur um die für den Payload benötigten IDs zu erhalten. Mit anderen Worten war die Erstellung einer einzigen Transaktion ein dreistufiger Prozess, und dazu noch ein langsamer.

Die Lösung

Die Lösung bestand darin, einen KI-Assistenten damit zu beauftragen, alle drei Schritte zu übernehmen, während der Benutzer lediglich in alltäglicher Sprache beschreiben muss, was er mit seinem Geld gemacht hat. Mit diesem Ziel vor Augen ist hier die Struktur der Implementierung dargestellt.

Die dreistufige Architektur

FastAPI-Endpunkt
LangGraph-Orchestrator
PostgreSQL-Datenbank

1. FastAPI-Endpunkt

Der Benutzer sendet eine Anfrage an den FastAPI-Endpunkt /assistance/transaction-entry mit einem Payload, der eine Nachricht enthält, die die Transaktion beschreibt.

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

2. LangGraph Orchestrator

Der Orchestrator wird als Graph aufgebaut, wobei jeder Knoten eine Operation darstellt und jede Kante den Datenfluss zwischen den Operationen repräsentiert.

Der erste Knoten, der Analyzer LLM, erhält die Nachricht des Benutzers und prüft, ob sie alle notwendigen Informationen zur Erfassung einer Transaktion enthält – sei es ein Einkommen oder eine Ausgabe, der betroffene Betrag, der Zweck der Transaktion, die verwendete Zahlungsmethode usw.

Falls die Nachricht bereits alle erforderlichen Details enthält, wandelt der LLM sie in strukturierte Daten um, zum Beispiel:

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

Diese strukturierten Daten gelangen anschließend zum Node „Database Writer“, nachdem die beiden Flag-Felder (is_complete und missing_info_message) entfernt wurden. Der Node „Database Writer“ ruft die Methode create_transaction() auf, welche die Transaktion im Zusammenhang mit dem Benutzer in der Datenbank speichert.

Aber was passiert, wenn die Nachricht einige Details fehlt? Betrachten wir eine solche Nachricht:

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

Hier ist der Zahlungsmodus nicht angegeben. In diesem Fall wird die von dem LLM extrahierte Datenmenge Flag-Werte enthalten, wie zum Beispiel:

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

Da das Flag is_complete hier auf False gesetzt ist, wird die Nachricht missing_info_message an den anderen mit dem Analyzer LLM verbundenen Node weitergeleitet: den Node „Clarification“. Dieser Pfad wird nur ausgelöst, wenn is_complete den Wert False hat.

Der Clarification-Node erhält die missing_info_message und ruft die ask_again()-Methode auf, welche diese Nachricht als Antwort auf die ursprüngliche FastAPI-Anfrage zurückgibt. Damit endet die Ausführung des Graphen für diesen Lauf – die vom Benutzer empfangene Ausgabe ist lediglich eine Aufforderung, den fehlenden Detailangaben nachzukommen, in diesem Fall dem Zahlungsmodus.

Nehmen wir an, der Benutzer antwortet anschließend mit den fehlenden Informationen, zum Beispiel:

{
  message: "UPI"
}

Diese Antwort veranlasst dazu, dass der Orchestrator-Graph erneut initialisiert wird, und er durchläuft dieselbe Abfolge von Schritten wie zuvor.

Der entscheidende Unterschied bei diesem zweiten Durchlauf ist, dass keine der früheren Informationen verloren geht – die Konversationshistorie wird jedes Mal beibehalten, wenn Daten vom LLM extrahiert werden (dieser Persistenzmechanismus wird später im Artikel ausführlicher erläutert). Da payment_option nun verfügbar ist und mit den zuvor erfassten Werten kombiniert wird, wechselt is_complete auf True, und die finalisierten, gefilterten Daten werden an den Database Writer-Node weitergeleitet, wie folgt:

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

Der Database Writer-Node übernimmt anschließend mit diesen gefilterten Daten. Denken Sie daran, dass bei manueller Erstellung einer Transaktionseintragung zwei zusätzliche API-Aufrufe erforderlich waren – einer für categories und einer für payment_options – nur um die jeweiligen IDs zu ermitteln, bevor die Transaktion selbst erstellt werden konnte. Dasselbe Problem tritt hier auf: Die gefilterten Daten enthalten die tatsächlichen Textwerte für Kategorie und Zahlungsoption, nicht ihre Datenbank-IDs, und die Datenbank akzeptiert keine Rohwerte für diese Felder.

Um dieses Problem zu lösen, muss der Database Writer-Node die Datenbank abfragen, um anhand der im gefilterten Datensatz enthaltenen Werte die entsprechenden Kategorie- und Zahlungsoptionseinträge zu finden.

Da das Projekt auf FastAPI zusammen mit SQLAlchemy setzt, werden diese Abfragen als SQLAlchemy-Abfragen implementiert.

Für `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

Kurz gesagt lautet diese Logik: Es wird eine SELECT-Anfrage ausgeführt, um zu überprüfen, ob die in data.category referenzierte Kategorie bereits existiert. Falls ja, wird die ID der Kategorie den Wert in data.category ersetzen. Falls nein, wird ein neuer Kategorierekord erstellt, und dessen neu generierte ID wird stattdessen verwendet.

Das gleiche Muster gilt für `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

Sobald sowohl die Kategorien-ID als auch die Zahlungsoptionen-ID ermittelt wurden, wird das Datenelement vollständig aktualisiert und ist für die Einfügung bereit, wie folgt:

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

Mit diesen finalisierten Daten ruft der Database Writer-Node die create_transaction()-Methode auf, welche den Transaktionsrekord tatsächlich in der Datenbank persistiert.

3. PostgreSQL-Datenbank

Dies stellt die letzte Phase der Architektur dar, in der die endgültigen Daten, die vom Database Writer-Node übergeben werden, in die transactions-Tabelle geschrieben werden.

Die resultierende Struktur der transactions-Tabelle sieht wie folgt aus:

Die Implementierung

Nun, da die Architektur erläutert wurde, ist es an der Zeit, die konkreten Implementierungsdetails für den Aufbau dieses Orchestrators mit LangGraph zu besprechen. Beachten Sie, dass die hier folgende Reihenfolge nicht genau der oben beschriebenen architektonischen Darstellung entspricht. Stattdessen ist die Implementierung wie folgt strukturiert:

LangGraph Orchestrator
PostgreSQL-Datenbank
FastAPI-Endpunkt

1. LangGraph Orchestrator

Der Orchestrator selbst befindet sich in src/assistance/graph.py. Diese Datei ist dafür verantwortlich, das LLM einzurichten, die Knoten des Graphen zu definieren, die Verbindungen zwischen diesen Knoten herzustellen und schließlich alles in einen ausführbaren Graphen zu kompilieren.

Wie bereits erwähnt, bestehen aus diesem Orchestrator drei Knoten: der Analyzer LLM, der Database Writer und der Clarification-Knoten.

Analyzer LLM-Knoten (Groq)

Der Knoten ist im Grunde ein Sprachmodell, dessen Aufgabe es ist, die Absicht des Benutzers zu ermitteln und zu überprüfen, ob die Nachricht alle erforderlichen und korrekten Informationen enthält. Anstatt ein eigenes Modell von Grund auf zu entwickeln, setzt dieses Projekt auf Groq, um die schwierigen Aufgaben zu bewältigen.

Was ist Groq?

Groq ist ein Open-Source-Python-Framework, das für die Verarbeitung grafisch strukturierter Daten entwickelt wurde. Es bietet Entwicklern eine leistungsstarke Möglichkeit, Informationen, die als Graphen gespeichert sind, abzufragen, zu filtern und zusammenzufassen. Es eignet sich hervorragend für große Graphen-Datensätze, wie beispielsweise soziale Netzwerke, Wissensgraphen oder Empfehlungssysteme, wie in einem Artikel von GeekForGeeks zur Groq-API beschrieben.

Mit der gehosteten API von Groq können Sie Anfragen an weit verbreitete Open-Modelle senden – openai/gpt-oss-120b wird in diesem Projekt verwendet – und Antworten erhalten, die in der Regel deutlich schneller eintreffen als bei anderen Anbietern mit vergleichbaren Modellen.

Warum Groq?

Groq wurde aus mehreren Gründen gegenüber Alternativen wie ChatOpenAI oder ChatAnthropic gewählt:

Geschwindigkeit: Groq setzt auf speziell entwickelte Hardware namens LPUs (Language Processing Units) anstelle der GPUs, auf die die meisten anderen Anbieter angewiesen sind, was eine schnelle Auswertung ermöglicht.
Nutzbare kostenlose Version: Die kostenlose Version von Groq ist ausreichend großzügig, um Einzelprojekte oder lernorientierte Projekte zu unterstützen, ohne während des Experimentierens erhebliche API-Kosten zu verursachen.
Kompatibilität durch LangChain: Die Klasse langchain_groq.ChatGroq integriert sich genauso wie ChatOpenAI oder ChatAnthropic in LangChain und LangGraph. Das bedeutet, dass ein Wechsel zu einem anderen Anbieter später keinen Neuaufbau der Graphiklogik erfordert – es genügt, den Client auszutauschen.

So erhalten Sie einen Groq API-Schlüssel

Groq ermöglicht es Ihnen, kostenlose API-Schlüssel für die Entwicklung zu erstellen. So erhalten Sie einen:

Gehen Sie zu https://console.groq.com und melden Sie sich entweder an oder registrieren Sie sich.
Wählen Sie die Option „API Keys“ in der Navigationsleiste aus.
wählen Sie „Create API Key“ aus.
Es erscheint ein Formular, in dem nach einem Namen (in diesem Projekt wurde transaction-assistant verwendet) sowie einer Gültigkeitsdauer für den Schlüssel gefragt wird. Füllen Sie das Formular aus und senden Sie es ab.
Der Schlüssel wird nur einmal angezeigt, unmittelbar nach der Erstellung – kopieren Sie ihn daher sofort.

Sobald er erstellt wurde, erscheinen alle Ihre Schlüssel in der Hauptliste auf dieser Seite.

Nutzung des Groq API-Schlüssels im FastAPI-Code

Fügen Sie den Groq API-Schlüssel zu Ihrer .env-Datei im Projektverzeichnis hinzu, zusammen mit Ihren anderen Umgebungsvariablen:

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

Es gibt mehrere Ansätze, um Umgebungsvariablen in die Module zu laden, die sie benötigen. Dieses Projekt verwendet eine spezielle Einstellungsklasse:

Definieren Sie eine Settings-Klasse in 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()

Führen Sie anschließend diese Einstellungsobjekte dort ein, wo sie benötigt werden:

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-Einrichtung

Vor der Konfiguration des LLM müssen LangGraph und LangChain zusammen mit der Groq-Integration installiert werden: pip install -U langgraph langchain langchain-groq

Daraufhin wird eine Instanz des Groq-Clients erstellt und mit einem bestimmten Modell konfiguriert:

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)

Hier dient ChatGroq als Wrapper von LangChain um Groqs Chat-Modelle und ermöglicht es Ihnen, über die Standard-Schnittstelle von LangChain mit ihnen zu interagieren, anstatt manuell HTTP-Anfragen zu erstellen.

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

Dieser Codeausschnitt erstellt die oben erwähnte Groq-Client-Instanz, die mit einem ausgewählten Modell sowie einem niedrigen Temperaturwert konfiguriert ist und mithilfe der aus den Umgebungs-Einstellungen abgerufenen API-Schlüssel authentifiziert wird.

Die Temperatur ist ein Parameter, der in der Regel zwischen 0 und 1 liegt und bestimmt, wie zufällig oder kreativ die Antworten eines Modells sind. Ein höherer Wert, wie zum Beispiel 0,8, führt zu vielfältigeren und kreativeren Ergebnissen, während ein niedrigerer Wert, wie zum Beispiel 0,2, zu präziseren und vorhersehbareren Antworten führt. In diesem Projekt wird temperature = 0,2 festgelegt.

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

Dieser Code umhüllt den LLM so, dass anstelle von reinem Text ein Python-Objekt erzeugt wird, das exakt der ExtractedTransactionSchema entspricht. Intern wird dies erreicht, indem dem Modell angegeben wird, Ausgabe im Einklang mit der Schema-Struktur zu generieren, diese anschließend automatisch解析iert und überprüft wird – wodurch eine manuelle Interpretation des rohen Textes des Modells entfällt.

Die ExtractedTransactionSchema selbst ist in src/assistance/schema.py definiert:

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

Bemerken Sie, dass der LLM zu diesem Zeitpunkt noch nicht tatsächlich aufgerufen wurde – dieser Schritt definiert lediglich die Form, die die Ausgabe nach ihrem Aufruf annehmen muss.

Graphenzustand

Der Graphenzustand repräsentiert die Datenstruktur, die durch den Graphen fließt und bei jeder Aktualisierung verändert wird. Man kann ihn sich als die Arbeitsmemorie des Orchestralisten vorstellen: Er enthält alle Informationen, die der Graph verfolgt und während der Ausführung Schritt für Schritt modifiziert. Für diesen Transaktionsassistenten ist der Graphenzustand wie folgt definiert:

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

Lassen Sie uns aufschlüsseln, was dieser Code tatsächlich tut:

from pydantic import BaseModel, Field

Pydantic ist die hier verwendete Bibliothek zur Datenvalidierung. BaseModel ist die Elternklasse, die man erweitert, um eine strukturierte, typgeprüfte Form wie GraphState zu definieren. Field ermöglicht es, Metadaten – Beschreibungen, Standardwerte usw. – zu jedem einzelnen Attribut hinzuzufügen.

from typing import Annotated, List, Optional
import operator

Dies sind Pythons Typisierungshilfen. Optional signalisiert, dass ein Feld leer sein kann und None enthält. List kennzeichnet ein Attribut als Liste von Elementen. Annotated, in Kombination mit operator.add, teilt LangGraph mit „wenn ein Knoten einen neuen Wert für dieses Feld zurückgibt, soll dieser anstelle des Bestehenden hinzugefügt werden.“ Das ist der Mechanismus, der es conversation_history ermöglicht, im Laufe mehrerer Nachrichtenaustausche zu wachsen, anstatt bei jeder neuen Nachricht überschrieben zu werden.

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: die neueste Nachricht, die der Benutzer bei dieser spezifischen Aufrufung gesendet hat.
conversation_history: die vollständige Liste aller vorherigen Nachrichten, die turnusweise angesammelt werden und nicht überschrieben werden.

extracted: wird ausgefüllt, sobald das LLM strukturierte Transaktionsdaten aus dem Gespräch extrahiert hat. Anfangs ist es None, da zu Beginn der Ausführung noch nichts extrahiert wurde.

final_response: die Nachricht, die schließlich an den Benutzer zurückgesendet wird – entweder eine Bestätigung, dass die Transaktion aufgezeichnet wurde, oder eine Nachfrage nach weiteren Details.

Der Extraktions-Aufforderungstext

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

Das ist die wörtliche Anweisung, die in natürlicher Sprache an das LLM übergeben wird – sie legt fest, nach welchen Feldern gesucht werden soll, was zu tun ist, wenn etwas fehlt, und wie die Antwort strukturiert sein muss. Da structured_llm bereits auf Ausgabenebene das Schema durchsetzt, besteht die Aufgabe der Anweisung hauptsächlich darin, das Denken des Modells zu lenken: zu entscheiden, was „vollständig“ bedeutet, wie eine klärende Frage formuliert werden soll usw., während das Schema für die Formatierung sorgt.

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

Funktion extractor führt Folgendes aus:

Fügt jede vorherige Nachricht zur aktuellen hinzu, damit das LLM den vollständigen Kontext erhält.
Gibt diesen zusammengeführten Text an das LLM weiter.
Erhält anschließend ein strukturiertes Objekt vom Typ ExtractedTransactionSchema zurück.

Gibt ein Dictionary mit Zustandsaktualisierungen zurück – die frisch extrahierten Daten zusammen mit der aktuellen Nachricht, die LangGraph aufgrund des auf diesem Feld konfigurierten operator.add-Verhaltens automatisch in die Historie aufnimmt.

Die Entscheidung

route_after_extraction

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

Diese Funktion führt keine tatsächliche Verarbeitung durch – ihre einzige Aufgabe besteht darin, eine Entscheidung zu treffen. Je nachdem, ob das LLM die extrahierten Daten als vollständig markiert hat, gibt sie einen String zurück, der dem Graphen mitteilt, welcher Knoten als Nächstes ausgeführt werden soll. Man kann dies als die Verzweigungslogik im Flussdiagramm betrachten: Der Graph prüft den Rückgabewert dieser Funktion und folgt dem entsprechenden Pfad, entweder zu create_transaction, um die Transaktion zu erfassen, oder zu ask_again, um weitere Informationen anzufragen.

Datenbank-Schreibknoten

create_transaction_node

Dieser Knoten ist dafür verantwortlich, die abgeschlossenen Transaktionsdaten unter Verwendung des entsprechenden Benutzers in die Datenbank zu schreiben. Die Funktion create_transaction_node, die den DB Writer-Knoten implementiert, sieht wie folgt aus:

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}

Das ist viel auf einmal zu verarbeiten, also gehen wir es Schritt für Schritt durch.

from langchain_core.runnables import RunnableConfig

Ein Typ, der den in jeden Knoten übergebenen config-Objekt ersetzt. Er existiert ausschließlich als Typhinweis, sodass jeder, der die Signatur von create_transaction_node liest, sofort versteht, welche Struktur config hat.

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

Das sind die üblichen SQLAlchemy-Importe, die benötigt werden, um Datenbankfehler aufzufangen sowie den asynchronen Datenbanksitzungstyp zu definieren, der zur Kommunikation mit PostgreSQL verwendet wird.

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

Dies bezieht die bereits anderswo in der Anwendung verwendete Transaktionserstellungslogik sowie ihr Eingabeschema mit ein. Durch die Wiederverwendung erstellt der Assistent Transaktionen über genau denselben Codepfad wie die reguläre CRUD-API, anstatt diese Logik hier zu duplizieren.

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

Es handelt sich um Hilfsteile, die dazu dienen, die von der LLM extrahierten Kategorie- und Zahlungsoptionennamen in tatsächliche Zeilen und IDs in der Datenbank umzuwandeln sowie neue Einträge zu erstellen, falls diese noch nicht vorhanden sind.

Hinweis: Die Abfrages-/Erstellungslogik für categories und payment_options wurde in einen einheitlichen Hilfsfunktion get_or_create zusammengeführt, da beide Modelle im Grunde dasselbe Verhalten benötigten.

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

Die Funktion create_transaction_node wird nur ausgeführt, nachdem die extrahierten Daten als vollständig bestätigt wurden. Sie ist als async deklariert, da sie tatsächliche Datenbankoperationen durchführt. Zudem werden config sowie state übergeben, damit sie auf die aktive Datenbanksession und den angemeldeten Benutzer zugreifen kann. Diese beiden Werte stammen von der API-Route und nicht vom LLM oder vom Gesprächszustand, da sie zur spezifischen Anfrage gehören und nicht zum laufenden Dialog.

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

Dies holt die Session, den Benutzer sowie die extrahierten Transaktionsdaten ab.

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

Weil die LLM nur die Namen der Kategorie und der Zahlungsmethode wie „Lebensmittel“ oder „UPI“ extrahiert hat und nicht deren Datenbank-IDs, prüft dieser Schritt, ob bereits eine passende Zeile für den aktuellen Benutzer existiert. Falls nicht, wird eine erstellt. In jedem Fall wird die entsprechende ID zurückgegeben.

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

Die Payload wird in derselben Struktur zusammengestellt, wie sie von der vorhandenen Transaktionserstellungslogik erwartet wird, und anschließend an dieselbe Controller-Funktion übergeben, wodurch die bestehende Anwendungslogik wiederverwendet statt neu geschrieben wird. Die Datenbanktransaktion wird anschließend bestätigt, um die Änderung dauerhaft zu speichern.

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

Falls etwas auf der Datenbankebene fehlschlägt, werden alle teilweisen Änderungen rückgängig gemacht und eine freundliche Fehlermeldung zurückgegeben, anstatt den Anfragen einen Abbruch zu ermöglichen. Dadurch wird verhindert, dass beispielsweise eine neu erstellte Kategorie ohne entsprechende Transaktion entsteht.

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

Im Erfolgsfall wird eine für Menschen lesbare Bestätigungsmitteilung erstellt und als Aktualisierung des Zustands zurückgegeben.

Klärungsknoten

ask_again

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

Das ist ein einfacher Ersatzweg. Wie bereits beschrieben, wird dieser Knoten nur ausgeführt, wenn für die extrahierten Daten der Wert von is_complete auf False gesetzt ist, zusammen mit einer nützlichen Nachricht in missing_info_message.

Innen in ask_again_node erhält die Funktion den state, wodurch sie Zugriff auf state.extracted.is_complete sowie state.extracted.missing_info_message hat.

Kurz gesagt: Immer dann, wenn Informationen fehlen, leitet dieser Knoten einfach die bereits von dem LLM während der Extraktion erzeugte Nachfrageseite weiter, sodass der Benutzer genau weiß, was als Nächstes bereitgestellt werden muss.

Aufbau des Graphen

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

Dies erstellt einen neuen Graphenbauer und weist ihn an, dass jeder Knoten im Graphen von einem Objekt der Art GraphState lesen und darauf schreiben soll.

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

Jede Funktion wird hier als benannter Knoten registriert, im Grunde ein beschrifteter Schritt innerhalb des Graphen.

graph_builder.add_edge(START, "extractor")

Dies legt den Eingangspunkt fest: Jede Ausführung des Graphen beginnt am extractor-Knoten.

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

Hier findet die Zweigung statt. Sobald extractor abgeschlossen ist, ruft LangGraph route_after_extraction auf, um den nächsten Schritt zu bestimmen. Unabhängig davon, welche Zeichenkette zurückgegeben wird – entweder create_transaction oder ask_again – wird diese in dieser Zuordnung gesucht, die jede Entscheidungszeichenkette mit dem tatsächlichen Knoten verbindet, zu dem übergegangen werden soll.

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

Sowohl die möglichen Zweige beenden den Ablauf des Graphen, sobald sie abgeschlossen sind, wobei auf jedem Weg der Punkt END erreicht wird.

Kompilieren mit Speicher

from langgraph.checkpoint.memory import MemorySaver

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

Durch Aufruf von compile() wird die Graphendefinition in etwas Umsetzbares umgewandelt. Die Angabe von checkpointer=memory aktiviert den zuvor beschriebenen Mechanismus zur Zustandsbeibehaltung, sodass ein erneuter Aufruf des Graphen mit derselben thread_id die Ausführung an dem Punkt fortsetzt, an dem sie aufgehört hat, anstatt neu zu starten.

Der endgültige Code

Dadurch ist die Orchestrierungsschicht abgeschlossen. Hier ist die fertige Datei (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-Datenbank

Die Datenbankinteraktion in dieser Phase wird bereits innerhalb von create_transaction_node abgewickelt, wie zuvor erläutert, wobei die finalisierten Transaktionsdaten in die Transaktions-Tabelle geschrieben werden.

3. FastAPI-Endpunkt

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

Kunden rufen diesen Endpunkt (/assistance/transaction-entry) auf und fügen im Anfragekörper eine Nachricht hinzu, die beschreibt, worum es bei der Transaktion ging.

Lassen Sie uns untersuchen, was jeder Bestandteil bewirkt.

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

Dies erstellt einen POST-Pfad bei /transaction-entry. Die Einstellung status_code=status.HTTP_201_CREATED weist FastAPI an, bei Erfolg standardmäßig welchen Statuscode zurückzugeben. 201 ist der übliche Code für „eine neue Ressource wurde erstellt“, was hier passt, da ein erfolgreicher Aufruf zu einer neuen Transaktionszeile führt.

Zusammenstellung der Graph-Konfiguration:

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

Dies erstellt das config-Objekt, das an die Aufrufung des Graphen übergeben wird. config enthält anfragespezifische Werte, die nicht im persistenten Gesprächszustand selbst gespeichert werden sollten.

"thread_id": str(user.id): Dieser Wert wird vom Checkpointer von LangGraph verwendet, um herauszufinden, wessen Konversationshistorie abgerufen und aktualisiert werden soll. Durch Verknüpfung mit der ID des authentifizierten Benutzers erhält jeder Benutzer automatisch einen isolierten, persistenten Thread, sodass eine unvollendete Transaktionseintragung eines Benutzers niemals in die eines anderen übergehen kann. Der Wert wird in einen String umgewandelt, weil der Checkpointer thread_id als String erwartet, während user.id in der Regel ein UUID ist.
"session" und "user": Diese Werte werden weitergeleitet, damit create_transaction_node, das innerhalb des Graphen ausgeführt wird, Zugriff auf die aktuelle Datenbanksession sowie auf Informationen darüber hat, wer die Anfrage stellt.

Aufruf des Graphen:

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

Dies ist die Zeile, die tatsächlich die Ausführung auslöst. ainvoke ist das asynchrone Gegenstück zum Ausführen des Graphen; der Einsatz des synchronen invoke würde den Event-Loop blockieren, was hier wichtig ist, da create_transaction_node intern asynchrone Datenbankoperationen durchführt.

Das erste Argument, {"user_input": payload.message}, repräsentiert den Startzustand für diesen Ablauf. Nur user_input muss explizit bereitgestellt werden; die übrigen GraphState-Felder (conversation_history, extracted, final_response) erhalten entweder Standardwerte oder werden im Laufe der Ausführung durch das Graph aktualisiert. Wenn bereits ein vorhandener thread_id eine gespeicherte Historie enthält, fügt LangGraph diese neue Eingabe zu diesem gespeicherten Zustand hinzu, anstatt von vorne anzufangen.
config=config liefert alles, was im vorherigen Schritt vorbereitet wurde: thread_id, um den richtigen Zustand zu finden, sowie session und user für den Knoten, der für die Schreibvorgänge in der Datenbank zuständig ist.

Das Schlüsselwort await pausiert diese Koroutine, bis der Graph seine vollständige Ausführung abgeschlossen hat, da ainvoke eine Koroutine zurückgibt, die vor der Verwendung des Ergebnisses abgewartet werden muss.

Was auch immer als result zurückkommt, ist der endgültige GraphState, der als Dictionary dargestellt wird und die Ausführung des Graphs widerspiegelt – unabhängig davon, ob er bei create_transaction oder bei ask_again beendet wurde.

Ausliefern der Antwort:

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

Die Route beendet sich durch die Rückgabe eines einfachen Dictionarys, das lediglich den endgültigen Antworttext enthält. FastAPI kümmert sich darum, diesen in einen JSON-Payload für den Client umzuwandeln, wodurch etwas in der Art entsteht:

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

Dieser Text ist identisch mit dem, was zuvor in create_transaction_node oder ask_again_node generiert wurde. Der Pfad selbst bleibt unabhängig davon, welche Abzweigung tatsächlich ausgeführt wurde; er leitet einfach das Ergebnis, das in final_response landet, weiter.

Fazit

Durch die Arbeit mit diesem Transaktionsassistenten wird deutlich, was in Schulungen oft übersehen wird: Der schwierige Teil bei der Einführung einer KI-Funktion besteht nicht darin, ein Modell zur Antwort aufzufordern, sondern darin sicherzustellen, dass diese Antwort im Umgang mit einem echten System sicher funktioniert. Das Erstellen von Anfragen ist der einfachere Teil. Der eigentliche ingenieurtechnische Aufwand geht in Schemata, die eine strukturierte Ausgabe erzwingen, in bedingte Graphen, die entscheiden, ob Daten gespeichert oder weitere Klarstellungen eingeholt werden sollen, sowie in einen Zustand, der korrekt über mehrere Gesprächsrunden hinweg weitergegeben wird.

LangGraph war für dieses Projekt besonders sinnvoll, weil der Arbeitsablauf tatsächliches Entscheidungsfinden erforderte und nicht nur einen einzigen Durchlauf von Eingang zu Ausgang. Wenn Ihre Funktion lediglich einen geradlinigen Ablauf benötigt, sind ein einfacher Aufruf eines LLMs oder eine LangChain-Kette wahrscheinlich das einfachere und geeignetere Werkzeug. Sobald jedoch die KI-Logik abzweigen, Erinnerungen speichern oder pausieren muss, um vor dem Weitermachen weitere Informationen zu sammeln, wirkt eine grafbasierte Struktur nicht mehr wie unnötige Komplexität, sondern ist die sinnvollste Methode, diesen Ablauf zu modellieren.

Referenzen

Zusätzliche Literatur

LangChain vs LangGraph: Die Wahl zwischen Ketten und zustandsbehafteten Graphen — Erfahren Sie, wie sich LangChains lineare Bausteine von LangGraphs zustandsbehafteten, verzweigten Arbeitsabläufen unterscheiden und wie Sie entscheiden können, welches Konzept für Ihre KI-Anwendung am besten geeignet ist.
Ein KI-Agenten von Grund auf erstellen: Muster, ReAct und LangGraph — Lernen Sie die grundlegenden Konzepte hinter KI-Agenten – Planung, Werkzeugnutzung, Reflexion sowie das ReAct-Muster – und wie LangChain und LangGraph bei der manuellen Erstellung eines solchen Agenten helfen.

Alle Artikel Brief senden