Unzählige KI-Unternehmen bringen derzeit agentische Business-Intelligence-Systeme auf den Markt, die Ihre Daten sofort analysieren und umfassende Erkenntnisse liefern – ohne komplexe SQL-Abfragen oder manuelle Tabellenkalkulation. Ein RAG-gestütztes Verkaufsanalysesystem kann beispielsweise Fragen wie „Was war unser meistverkauftes Produkt im letzten Quartal?“ in Sekundenschnelle statt in Stunden beantworten.
Ein RAG-gestütztes Verkaufsanalysesystem löst diese Probleme, indem es KI-Agenten einsetzt, die eigenständig denken, Entscheidungen treffen und externe Werkzeuge koordinieren können, um natürliche Sprachabfragen über Ihre gesamte Datenlandschaft zu ermöglichen. Anstatt Datenbankschemata oder Tabellenkalkulationsformeln zu lernen, zerlegt der Agent Ihre Anfrage, bestimmt die beste Datenquelle, führt eine gezielte Analyse durch und liefert umfassende, datengestützte Antworten mit korrekten Quellenangaben.
Beispielsweise könnten Sie ein anderes LLM, ein benutzerdefiniertes Dokumentformat verwenden oder die Art und Weise anpassen, wie Agenten analysieren und sich verhalten, um spezifische Geschäftsergebnisse zu erzielen. In diesem Tutorial erfahren Sie, wie Sie Ihr eigenes RAG-gestütztes Verkaufsanalysesystem mit der einheitlichen LLM-API von Novita AI, dem LangChain-Agenten-Framework und erweiterten Dokumentverarbeitungsfunktionen erstellen. Sie erschaffen ein System, das Anfragen automatisch an optimale Datenquellen weiterleitet und umsetzbare Geschäftseinblicke liefert.
Was ist ein RAG-gestütztes Verkaufsanalysesystem?
Ein RAG-gestütztes Verkaufsanalysesystem nutzt Retrieval-Augmented Generation, um präzise, fundierte Antworten auf Fragen zum Einzelhandelsverkauf zu liefern, indem es die Leistungsfähigkeit großer Sprachmodelle mit einer Abrufschicht kombiniert, die auf Ihre tatsächlichen Geschäftsdaten zugreifen kann.
Dieses System umfasst in der Regel:
- Der KI-Agent analysiert die Anfrage und entscheidet, ob eine strukturierte Datenanalyse (SQL), eine Dokumentenanalyse oder beides erforderlich ist.
- Der Agent leitet Abfragen intelligent an die am besten geeignete Datenquelle weiter – SQL-Datenbanken für quantitative Analysen oder Dokumentenspeicher für qualitative Erkenntnisse.
- Der Agent nutzt verschiedene Werkzeuge zur Informationssammlung, darunter Pandas-Agenten zur Analyse von CSV-Dateien, SQL-Agenten zum Abfragen von Datenbanken und Vektorspeicher zum Abrufen von Dokumenten. Ohne diese Werkzeuge kann der Agent keine wesentlichen Aufgaben im Workflow ausführen.
- Der Agent verarbeitet die gesammelten Informationen mit speziellen Techniken: statistische Analyse für numerische Daten und semantische Suche für Textdokumente.
- Nach der Analyse der Daten erstellt der Agent eine strukturierte Antwort mit Zusammenfassungen, Berechnungen und korrekten Quellenangaben.
Benötigte Werkzeuge
Bevor wir mit dem Bau beginnen, stellen wir die notwendigen Werkzeuge bereit.
Novita AI
Für den Bau unseres RAG-gestützten Analysesystems benötigen wir Zugang zu leistungsstarken Large Language Models (LLMs) und Embedding-Modellen. Novita AI bietet erschwingliche, leistungsstarke APIs, die über eine einzige, einheitliche Schnittstelle Zugriff auf die neuesten Large Language Models (LLMs), Embedding-Modelle und mehr bieten.
LangChain
LangChain ist ein Open-Source-Framework für die Entwicklung von Anwendungen mit LLMs. Mit LangChain können Sie einen agentischen Workflow erstellen, der Schritt für Schritt argumentiert, Werkzeuge einsetzt und mit APIs interagiert. Für unser Verkaufsanalysesystem nutzen wir LangChain, um den Analyseprozess zu strukturieren, Werkzeuge wie SQL-Agenten und Dokumentenprozessoren einzusetzen und alle Daten zu strukturierten Erkenntnissen zu synthetisieren.
Streamlit
Wir verwenden Streamlit, um die interaktive Weboberfläche zu erstellen. Es eignet sich perfekt für schnelles Prototyping und erstellt mit minimalem Code professionell aussehende Benutzeroberflächen, was das Tutorial einfach nachvollziehbar macht.
FAISS
Für den vektorbasierten Dokumentenabruf bietet FAISS schnelle Ähnlichkeitssuche, sodass wir schnell relevante Dokumentabschnitte basierend auf Benutzeranfragen finden können.
SQLAlchemy & PyMySQL
Diese kümmern sich um unsere SQL-Datenbankoperationen und die MySQL-Konnektivität und ermöglichen es unserem System, strukturierte Geschäftsdaten direkt abzufragen.
pandas
Unverzichtbar für die Manipulation und Analyse von CSV-Daten. Unser System verwendet Pandas-Agenten, um komplexe Berechnungen für tabellarische Daten durchzuführen.
Systemarchitektur – Überblick
Unser RAG-System verarbeitet Benutzeranfragen intelligent über einen mehrstufigen Workflow, der automatisch die beste Datenquelle und Verarbeitungsmethode für jede Frage ermittelt. Anstatt Benutzer zu zwingen, zu wissen, wo ihre Daten gespeichert sind, ermittelt das System hinter den Kulissen den optimalen Ansatz.
Abfrageverarbeitung
Der erste Schritt in unserem Arbeitsablauf mit dem Agenten ist die Abfrageverarbeitung. In diesem Fall überprüft der Agent die vom Benutzer eingegebene Abfrage und entwickelt einen maßgeschneiderten Verarbeitungsplan, der zum Kontext der Abfrage passt. Anschließend bestimmt er, gegen welche Art von Daten die Frage analysiert werden muss: strukturierte Daten, Dokumentenanalyse oder beides.
Datenquellenauswahl
Wenn ein Benutzer eine Frage stellt, verfolgt das System einen einfachen Ansatz: Es versucht zuerst, das Problem durch Abfrage der SQL-Datenbank zu lösen. Die Datenquellenauswahl ist sinnvoll, da strukturierte Daten oft die präzisesten Antworten auf quantitative Fragen liefern. Der SQL-Agent kann schnell Abfragen ausführen und exakte Zahlen mit korrekten Berechnungen zurückgeben.
Dokumentenanalyse
Wenn der SQL-Agent leere Ergebnisse zurückgibt oder keine relevanten Informationen finden kann, wechselt das System automatisch zur Dokumentenanalyse. Der Dokumentenverarbeitungspfad variiert je nach Dateityp. CSV- und Excel-Dateien werden in Pandas DataFrames geladen, um komplexe Datenanalysen durchzuführen, während PDFs, Word-Dateien und Textdokumente mit FAISS in Abschnitte unterteilt und für die semantische Suche indiziert werden.
Quellenverwaltung
Es ist wichtig, alle Quellen korrekt zu zitieren, um sicherzustellen, dass die bereitgestellten Informationen auf eine glaubwürdige Autorität zurückgeführt werden können. Dieser Schritt im Workflow hilft, Vertrauen beim Publikum aufzubauen und ermöglicht es, Behauptungen oder Daten im endgültigen Bericht zu überprüfen.
Antwortsynthese
Um die Verbindungen zwischen allen gesammelten Informationen herzustellen, fasst der Agent Erkenntnisse aus allen analysierten Datenquellen zu einer kohärenten Antwort zusammen. Ohne diese Phase des Workflows könnte die Ausgabe ein zusammenhangloser Bericht sein.
Implementierung des RAG-gestützten Analyse-Workflows
Bisher haben Sie verstanden, worum es bei dem RAG-gestützten Verkaufsanalysesystem geht. Jetzt wird es Zeit, es zu implementieren. Dazu führen wir einige wichtige Schritte durch:
Installation und Einrichtung
Bevor wir fortfahren, legen wir die vollständige Dateistruktur für unser RAG-Analysesystem fest. Sie müssen die folgenden Dateien in Ihrem Projektverzeichnis erstellen:
| rag-analytics-system/ ├── .env # Umgebungsvariablen ├── requirements.txt # Python-Abhängigkeiten ├── query_processor.py # Hauptklasse QueryProcessor ├── main.py # Streamlit-Oberfläche |
Abhängigkeiten einrichten
Aktualisieren Sie zunächst die Datei requirements.txt mit allen notwendigen Abhängigkeiten:
| langchain==0.3.26 langchain-openai==0.3.25 python-dotenv==1.1.1 SQLAlchemy==2.0.41 pandas==2.3.0 PyPDF2==3.0.1 faiss-cpu==1.11.0 PyMySQL==1.1.1 cryptography==45.0.4 langchain-experimental==0.3.4 streamlit==1.46.0 openpyxl==3.1.5 python-docx==1.2.0 |
Melden Sie sich dann bei Novita AI an. Navigieren Sie nach der Anmeldung zur Seite Manage API Keys, klicken Sie auf die Schaltfläche Add New Key und geben Sie Ihren Key Name ein.
Novita AI bietet Ihnen nach der Registrierung kostenlose Guthaben zum Testen verschiedener Modelle, sodass Sie sich keine Gedanken über den Kauf von Guthaben machen müssen, bevor Sie mit dem Bau oder Experimentieren beginnen. Fügen Sie Ihre Zugangsdaten (z. B. den zuvor erstellten Novita API-Key) zur .env-Datei hinzu:
| NOVITA_API_KEY=”your_novita_api_key_here” |
Installieren Sie nun alle erforderlichen Abhängigkeiten für das Projekt:
| # Virtuelle Umgebung erstellen python -m venv rag_analytics_env source rag_analytics_env/bin/activate # Unter Windows: rag_analytics_env\\Scripts\\activate # Abhängigkeiten installieren pip install -r requirements.txt |
Den QueryProcessor erstellen
Der gesamte RAG-Analyse-Workflow ist in einer modularen Klasse namens QueryProcessor gekapselt. Diese Klasse verwaltet das Routing von Abfragen, die SQL-Analyse, die Dokumentenverarbeitung, die Synthese von Ergebnissen und die Ausführung der vollständigen Analyse-Schleife.
Der QueryProcessor dient als zentraler Orchestrator für unser Analysesystem. Anstatt separate Werkzeuge für verschiedene Datentypen zu erstellen, ermöglicht dieser einheitliche Ansatz, Abfragen intelligent an die am besten geeignete Analysemethode weiterzuleiten – sei es SQL-Datenbanken, CSV-Dateien oder unstrukturierte Dokumente.
Zuerst importieren wir die notwendigen Bibliotheken für das Projekt in die Datei query_processor.py. Diese Importe liefern alles, was für die Sprachmodell-Integration, Datenbankkonnektivität, Dokumentenverarbeitung und Vektorspeicherung benötigt wird:
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_experimental.agents.agent_toolkits import create_csv_agent, create_pandas_dataframe_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.utilities import SQLDatabase
from langchain.prompts import PromptTemplate
from langchain_community.tools import ReadFileTool
from langchain.schema import Document
import os
from dotenv import load_dotenv
from sqlalchemy.engine import Engine
import glob
import pandas as pd
load_dotenv()
Als Nächstes initialisieren wir den Analyse-Agenten mit Zugriff auf ein LLM über die Novita-API und verschiedene Datenverarbeitungswerkzeuge. Die Initialisierungsmethode richtet alle Kernkomponenten ein, die wir benötigen: das Sprachmodell zum Verstehen von Abfragen, Embeddings für die Dokumentsuche und Textaufteilung für die effiziente Verarbeitung großer Dokumente.
class QueryProcessor:
def __init__(self, documents_folder: str, sql_engine: Engine):
self.documents_folder = documents_folder
self.sql_engine = sql_engine
self.llm = ChatOpenAI(
model="google/gemma-3-27b-it",
temperature=0,
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "google"
}
)
self.embeddings = OpenAIEmbeddings(
model="baai/bge-m3",
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "baai"
}
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
self.sql_agent = None
self._prepare_data_sources()
Der Konstruktor initialisiert mehrere Schlüsselkomponenten. Wir verwenden eine Temperatur von 0 für das LLM, um konsistente, faktenbasierte Antworten statt kreativer Variationen zu gewährleisten. Der Textzerteiler ist mit überlappenden Abschnitten konfiguriert, um die Kontextkontinuität bei der Verarbeitung großer Dokumente zu erhalten. Die Methode _prepare_data_sources() wird am Ende aufgerufen, um unsere spezialisierten Agenten einzurichten.
Multi-Source-Datenintegration erstellen
Der nächste Schritt umfasst die Erstellung von Methoden, die verschiedene Datenverarbeitungsagenten initialisieren. Jeder Agent ist auf bestimmte Datentypen und Analyseaufgaben spezialisiert – ähnlich wie verschiedene Experten für verschiedene Arten von Fragen.
def _prepare_data_sources(self):
"""Bereitet sowohl SQL- als auch Dokumentendatenquellen vor"""
# SQL-Agent vorbereiten
self._prepare_sql_agent()
# Dokumenten-Agent vorbereiten
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""SQL-Agent initialisieren"""
# SQLAlchemy-Engine in LangChain SQLDatabase konvertieren
db = SQLDatabase(self.sql_engine)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
self.sql_agent = create_sql_agent(
llm=self.llm,
toolkit=toolkit,
agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
def _prepare_document_agent(self):
"""Dokumenten-Agent mit ReadFileTool und Vektorspeicher initialisieren"""
# Alle unterstützten Dateien abrufen
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"\
{len(supported_files)} unterstützte Dateien in {self.documents_folder} gefunden")
if supported_files:
# ReadFileTool erstellen
read_file_tool = ReadFileTool()
# Werkzeugliste erstellen
tools = [read_file_tool]
# Prompt-Vorlage für den ReAct-Agenten erstellen
prompt = PromptTemplate.from_template("""
Sie sind ein hilfreicher Assistent, der Fragen zu Geschäftsdokumenten beantworten kann.
Sie haben Zugriff auf die folgenden Werkzeuge:
{tools}
Verwenden Sie das folgende Format:
Frage: die Eingabefrage, die Sie beantworten müssen
Gedanke: Sie sollten immer darüber nachdenken, was zu tun ist
Aktion: die auszuführende Aktion, sollte eine von [{tool_names}] sein
Aktions-Eingabe: die Eingabe für die Aktion
Beobachtung: das Ergebnis der Aktion
... (dieser Gedanke/Aktion/Aktions-Eingabe/Beobachtung kann N-mal wiederholt werden)
Gedanke: Ich kenne jetzt die endgültige Antwort
Endgültige Antwort: die endgültige Antwort auf die ursprüngliche Eingabefrage
Los geht's!
Frage: {input}
Gedanke: Ich sollte die relevanten Dokumente lesen, um die Antwort zu finden
{agent_scratchpad}
""")
agent = create_react_agent(self.llm, tools, prompt)
self.document_agent = AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
verbose=True
)
print("\
Dokumenten-Agent erfolgreich initialisiert")
else:
print("\
Keine Dokumente im angegebenen Ordner gefunden.")
Die Methode _prepare_sql_agent() erstellt einen spezialisierten Agenten für Datenbankabfragen. Wir verwenden den Agententyp ZERO_SHOT_REACT_DESCRIPTION, was bedeutet, dass der Agent über SQL-Abfragen nachdenken kann, ohne spezifische Beispiele zu benötigen – dies macht ihn an verschiedene Datenbankschemata anpassbar. Die Methode _prepare_document_agent() durchsucht die unterstützten Dateitypen und erstellt einen flexiblen Agenten, der verschiedene Dokumentformate lesen kann. Die ReAct-Prompt-Vorlage (Reasoning and Acting) führt den Agenten durch einen strukturierten Denkprozess, ähnlich wie ein menschlicher Analyst bei der Dokumentenanalyse vorgehen würde.
Dokumentenverarbeitung und Vektorspeicher
Als Nächstes müssen wir behandeln, wie der Workflow verschiedene Dokumenttypen verarbeitet. Dieser Abschnitt erstellt Methoden, die CSV-/Excel-Dateien mit Pandas-Agenten und unstrukturierte Dokumente mit Vektorspeicher verarbeiten. Der Kernpunkt hier ist, dass verschiedene Datentypen unterschiedliche Verarbeitungsstrategien erfordern.
CSV_PROMPT_PREFIX = """
WICHTIG: Sie arbeiten mit einem pandas DataFrame namens 'df', das mit den tatsächlichen Daten geladen wurde.
ERSTELLEN Sie KEINE Beispieldaten und erfinden Sie KEINE Daten. Verwenden Sie AUSSCHLIESSLICH das tatsächliche DataFrame 'df', das Ihnen zur Verfügung steht.
Erkunden Sie das DataFrame zunächst, indem Sie:
1. Pandas-Anzeigeoptionen setzen, um alle Spalten anzuzeigen: pd.set_option('display.max_columns', None)
2. Die Form des DataFrames überprüfen: print(df.shape)
3. Die Spaltennamen abrufen: print(df.columns.tolist())
4. Die Datentypen überprüfen: print(df.dtypes)
5. Die ersten Zeilen anzeigen: print(df.head())
6. Beantworten Sie dann die Frage mit den tatsächlichen Daten im DataFrame.
"""
CSV_PROMPT_SUFFIX = """
- **KRITISCH**: Verwenden Sie AUSSCHLIESSLICH die tatsächlichen Daten im DataFrame. Erstellen Sie KEINE Beispieldaten oder fiktiven Daten.
- **IMMER** bevor Sie die endgültige Antwort geben, versuchen Sie eine andere Methode, um Ihre Ergebnisse zu überprüfen.
- Denken Sie dann über die Antworten der beiden Methoden nach und fragen Sie sich, ob sie die ursprüngliche Frage korrekt beantworten.
- Wenn Sie sich nicht sicher sind, versuchen Sie eine andere Methode.
- FORMATIEREN SIE 4 ODER MEHR ZIFFER mit Kommas.
- Wenn die versuchten Methoden nicht dasselbe Ergebnis liefern, denken Sie nach und versuchen Sie es erneut, bis Sie zwei Methoden mit demselben Ergebnis haben.
- Wenn Sie immer noch zu keinem konsistenten Ergebnis kommen, sagen Sie, dass Sie sich nicht sicher sind.
- Wenn Sie sich der richtigen Antwort sicher sind, erstellen Sie eine schöne und gründliche Antwort mit Markdown.
- **ERFINDEN SIE KEINE ANTWORT ODER VERWENDEN SIE VORWISSEN, NUTZEN SIE NUR DIE ERGEBNISSE IHRER BERECHNUNGEN**.
- **IMMER**, als Teil Ihrer "Endgültigen Antwort", erklären Sie, wie Sie zu der Antwort gekommen sind, in einem Abschnitt, der mit: "\\
\\
Erklärung:\\
" beginnt.
- In der Erklärung nennen Sie die Spaltennamen, die Sie verwendet haben, um zur endgültigen Antwort zu gelangen.
- Zeigen Sie Ihre Arbeit, indem Sie relevante DataFrame-Operationen und deren Ergebnisse anzeigen.
"""
def _process_document_query(self, query: str) -> str:
"""Verarbeitet eine Dokumentenabfrage mittels CSV-/Excel-Agenten für Tabellendaten und LLM für unstrukturierte Daten."""
try:
print(f"\\
Verarbeite Abfrage: {query}")
# Alle unterstützten Dateien abrufen
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"{len(supported_files)} unterstützte Dateien in {self.documents_folder} gefunden")
if not supported_files:
return "Keine Dokumente zum Durchsuchen gefunden."
# Zuerst nach CSV-/Excel-Dateien suchen
csv_files = [f for f in supported_files if f.endswith('.csv')]
excel_files = [f for f in supported_files if f.endswith(('.xlsx', '.xls'))]
if csv_files:
csv_file = csv_files[0]
print(f"\\
Verarbeite CSV-Datei: {csv_file}")
try:
# Zuerst mit pandas DataFrame-Agent versuchen
df = pd.read_csv(csv_file)
print(f"CSV erfolgreich geladen mit {len(df)} Zeilen und Spalten: {df.columns.tolist()}")
# Agent mit verbesserter Konfiguration erstellen
print("Erstelle pandas DataFrame-Agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Token-Limits bei großen DataFrames vermeiden
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Abfrage mit unserem benutzerdefinierten Prompt verarbeiten
print(f"Verarbeite Abfrage mit Agent: {query}")
# Verbesserter Prompt, der sicherstellt, dass der Agent das tatsächliche DataFrame verwendet
prompt = f"""
Sie haben Zugriff auf ein pandas DataFrame namens 'df' mit {len(df)} Zeilen und den folgenden Spalten: {df.columns.tolist()}.
Hier sind die ersten Zeilen der Daten:
{df.head().to_string()}
Datentypen:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Frage: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent-Antwort: {response}")
return response['output']
except Exception as e:
print(f"Fehler mit pandas DataFrame-Agent: {str(e)}")
print("Versuche alternativen CSV-Agent...")
# Fallback auf CSV-Agent
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Frage: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"Fehler bei Verarbeitung der CSV-Datei: {str(e2)}")
import traceback
print(f"Vollständiger Traceback: {traceback.format_exc()}")
return f"Fehler bei Verarbeitung der CSV-Datei: {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Verarbeite Excel-Datei: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel erfolgreich geladen mit {len(df)} Zeilen und Spalten: {df.columns.tolist()}")
# Agent mit verbesserter Konfiguration erstellen
print("Erstelle pandas DataFrame-Agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Token-Limits bei großen DataFrames vermeiden
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Abfrage mit unserem benutzerdefinierten Prompt verarbeiten
print(f"Verarbeite Abfrage mit Agent: {query}")
# Verbesserter Prompt, der sicherstellt, dass der Agent das tatsächliche DataFrame verwendet
prompt = f"""
Sie haben Zugriff auf ein pandas DataFrame namens 'df' mit {len(df)} Zeilen und den folgenden Spalten: {df.columns.tolist()}.
Hier sind die ersten Zeilen der Daten:
{df.head().to_string()}
Datentypen:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Frage: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent-Antwort: {response}")
return response['output']
except Exception as e:
print(f"Fehler bei Verarbeitung der Excel-Datei: {str(e)}")
import traceback
print(f"Vollständiger Traceback: {traceback.format_exc()}")
return f"Fehler bei Verarbeitung der Excel-Datei: {str(e)}"
# Für unstrukturierte Textdateien
print("\\
Verarbeite unstrukturierte Textdateien...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Fehler beim Lesen der Datei {file_path}: {str(e)}")
continue
if not all_content:
return "Konnte keine Dokumente lesen."
# Unstrukturierten Text mit Vektorspeicher verarbeiten
print("Verarbeite Text mit Vektorspeicher...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Erzeugte Kontextlänge: {len(context)}")
response = self.llm.invoke(
f"""Basierend auf dem folgenden Kontext beantworten Sie die Frage: {query}
\\
Kontext:\\
{context}\\
\\
Antwort:"""
)
return response.content
except Exception as e:
print(f"Fehler bei Verarbeitung der Dokumentenabfrage: {str(e)}")
import traceback
print(f"Vollständiger Traceback: {traceback.format_exc()}")
return f"Fehler bei Verarbeitung der Dokumentenabfrage: {str(e)}"
Der CSV-Verarbeitungsabschnitt enthält detaillierte Prompt-Entwicklung, um eine genaue Analyse zu gewährleisten. Die Präfix- und Suffix-Prompts sind entscheidend, da sie verhindern, dass der Agent Daten halluziniert oder falsche Ergebnisse liefert. Der Überprüfungsschritt, bei dem der Agent mehrere Methoden ausprobiert, trägt zur Genauigkeit bei – ähnlich wie ein sorgfältiger Analyst seine Berechnungen doppelt prüfen würde. Wir priorisieren pandas DataFrame-Agenten gegenüber CSV-Agenten, da sie eine robustere Datenverarbeitung und bessere Leistung mit großen Datensätzen bieten. Der Fallback-Mechanismus stellt sicher, dass wir alternative Methoden zur Datenverarbeitung haben, falls ein Ansatz fehlschlägt. Vervollständigen wir nun die Methode _process_document_query für Excel-Dateien und unstrukturierte Dokumente:
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Verarbeite Excel-Datei: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel erfolgreich geladen mit {len(df)} Zeilen und Spalten: {df.columns.tolist()}")
# Agent mit verbesserter Konfiguration erstellen
print("Erstelle pandas DataFrame-Agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Token-Limits bei großen DataFrames vermeiden
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Abfrage mit unserem benutzerdefinierten Prompt verarbeiten
print(f"Verarbeite Abfrage mit Agent: {query}")
# Verbesserter Prompt, der sicherstellt, dass der Agent das tatsächliche DataFrame verwendet
prompt = f"""
Sie haben Zugriff auf ein pandas DataFrame namens 'df' mit {len(df)} Zeilen und den folgenden Spalten: {df.columns.tolist()}.
Hier sind die ersten Zeilen der Daten:
{df.head().to_string()}
Datentypen:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Frage: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent-Antwort: {response}")
return response['output']
except Exception as e:
print(f"Fehler bei Verarbeitung der Excel-Datei: {str(e)}")
import traceback
print(f"Vollständiger Traceback: {traceback.format_exc()}")
return f"Fehler bei Verarbeitung der Excel-Datei: {str(e)}"
# Für unstrukturierte Textdateien
print("\\
Verarbeite unstrukturierte Textdateien...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Fehler beim Lesen der Datei {file_path}: {str(e)}")
continue
if not all_content:
return "Konnte keine Dokumente lesen."
# Unstrukturierten Text mit Vektorspeicher verarbeiten
print("Verarbeite Text mit Vektorspeicher...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Erzeugte Kontextlänge: {len(context)}")
response = self.llm.invoke(
f"""Basierend auf dem folgenden Kontext beantworten Sie die Frage: {query}
\\
Kontext:\\
{context}\\
\\
Antwort:"""
)
return response.content
except Exception as e:
print(f"Fehler bei Verarbeitung der Dokumentenabfrage: {str(e)}")
import traceback
print(f"Vollständiger Traceback: {traceback.format_exc()}")
return f"Fehler bei Verarbeitung der Dokumentenabfrage: {str(e)}"
SQL-Datenbank-Integration
Nachdem wir die Dokumentenverarbeitung behandelt haben, müssen wir das intelligente Abfrage-Routing-System erstellen, das zwischen SQL- und Dokumentenanalyse entscheidet. Dies ist die Kernintelligenz unseres Systems – die Bestimmung, welche Datenquelle die Antwort auf eine gegebene Frage am wahrscheinlichsten enthält.
def process_query(self, query: str) -> str:
"""
Verarbeitet eine Abfrage mittels Agenten, um intelligent zwischen SQL und Dokumenten zu entscheiden
"""
# Zuerst SQL-Agent versuchen
try:
print("\\
Versuche SQL-Agent...")
sql_result = self.sql_agent.run(query)
no_answer_phrases = [
"no results", "i don't know", "unknown", "not sure", "cannot answer", "don't have", "no data", "n/a"
]
if sql_result and not any(phrase in sql_result.lower() for phrase in no_answer_phrases) and sql_result.strip():
return f"Aus SQL-Datenbank: {sql_result}"
else:
print("SQL-Agent konnte nicht antworten, versuche Dokumente...")
except Exception as e:
print(f"SQL-Agent-Fehler: {str(e)}")
print("Fallback auf Dokumente...")
# Wenn SQL-Agent fehlschlägt oder keine Ergebnisse liefert, Dokumentenverarbeitung versuchen
try:
print("\\
Verarbeite Dokumente...")
doc_result = self._process_document_query(query)
if doc_result:
return f"Aus Dokumenten: {doc_result}"
else:
print("Dokumentenverarbeitung lieferte keine Ergebnisse")
except Exception as e:
print(f"Dokumentenverarbeitungsfehler: {str(e)}")
return "Konnte in weder SQL-Datenbank noch Dokumenten relevante Informationen finden."
Die Abfrage-Routing-Logik folgt einem Prioritätssystem: SQL-Datenbanken werden zuerst versucht, da sie typischerweise strukturierte, quantitative Daten enthalten, die Geschäftsfragen schnell und genau beantworten können. Wenn der SQL-Agent unklare oder negative Antworten liefert (erkannt über unsere no_answer_phrases-Liste), fällt das System automatisch auf die Dokumentenverarbeitung zurück. Dieser Ansatz ahmt die Arbeitsweise eines menschlichen Analysten nach – zuerst werden strukturierte Datenquellen überprüft, dann werden Dokumente und Berichte herangezogen, wenn die Datenbank die benötigten Informationen nicht enthält.
Erstellen der Streamlit-Oberfläche
Bauen wir nun die Benutzeroberfläche, die unser Analysesystem für Geschäftsanwender zugänglich macht. Die Streamlit-Oberfläche bietet ein intuitives, chatbasiertes Erlebnis, das technische Komplexität verbirgt und dennoch leistungsstarke Analysefunktionen bietet. Fügen Sie die folgenden Codeausschnitte zur Datei main.py hinzu:
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# Umgebungsvariablen laden
load_dotenv()
#st.text_input
# Session-Status initialisieren
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# Seitenkonfiguration
st.set_page_config(
page_title="Dokumentenanalyse-Chatbot",
page_icon="🤖",
layout="wide"
)
# Benutzerdefiniertes CSS für Button- und Titelgestaltung
st.markdown("""
<style>
.stButton > button {
background-color: #23D57C;
color: white;
border: none;
border-radius: 8px;
padding: 0.5rem 1rem;
font-weight: 600;
transition: all 0.3s ease;
}
.stButton > button:hover {
background-color: #1fb36b;
box-shadow: 0 4px 8px rgba(35, 213, 124, 0.3);
transform: translateY(-2px);
}
.stButton > button:active {
background-color: #1a9960;
transform: translateY(0px);
}
h1 {
color: #23D57C !important;
font-weight: 700;
}
</style>
""", unsafe_allow_html=True)
# Titel und Beschreibung
st.title("Dokumentenanalyse-Chatbot mit Novita")
Dieses anfängliche Setup erstellt eine professionell aussehende Oberfläche mit benutzerdefiniertem Styling. Die Session-Status-Verwaltung stellt sicher, dass Benutzer ihren Gesprächsverlauf nicht verlieren oder ihre Datenquellen bei der Interaktion mit der Oberfläche nicht neu initialisieren müssen. Das benutzerdefinierte CSS bietet visuelles Feedback und sorgt für ein einheitliches Markenerlebnis.
Abschließende Systemintegration
Lassen Sie uns die Streamlit-Oberfläche mit der Datenquellen-Initialisierung und der Chat-Funktionalität vervollständigen. Dieser letzte Abschnitt fasst alle unsere Komponenten in einer benutzerfreundlichen Anwendung zusammen, die Geschäftsanalysten ohne technische Fachkenntnisse nutzen können.
# Prüfen, ob die Datenquelle initialisiert ist
if st.session_state.processor is None:
# Zentrieren der Datenquellen-Konfiguration
st.markdown("<br><br>", unsafe_allow_html=True)
# Zentrierte Spalten erstellen
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 Los geht's")
st.write("Initialisieren Sie Ihre Datenquelle, um mit Ihren Dokumenten zu chatten")
# SQL-Datenbank-Konfiguration (ausgeblendet)
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# Konfiguration des Dokumentenordners
st.write("**Pfad zum Dokumentenordner:**")
documents_folder = st.text_input(
"Pfad zum Dokumentenordner",
placeholder="Geben Sie den Pfad zu Ihrem Dokumentenordner ein (z. B. docs)",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# Button zentrieren
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("Datenquelle initialisieren", use_container_width=True):
try:
# Pfad zum Dokumentenordner validieren und auflösen
if not documents_folder:
st.error("Bitte geben Sie einen Pfad zum Dokumentenordner an")
else:
# In absoluten Pfad umwandeln
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"Dokumentenordner nicht gefunden: {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"Pfad ist kein Verzeichnis: {abs_documents_folder}")
else:
# QueryProcessor mit einer Dummy-SQL-Engine initialisieren
# SQL-Engine erstellen
connection_string = f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}"
sql_engine = create_engine(connection_string)
st.session_state.processor = QueryProcessor(abs_documents_folder, sql_engine)
st.success("Datenquelle erfolgreich initialisiert!")
st.rerun()
except Exception as e:
st.error(f"Fehler beim Initialisieren der Datenquelle: {str(e)}")
else:
# Datenquellen-Status in der Seitenleiste anzeigen
with st.sidebar:
st.header("📊 Datenquelle")
st.success("✅ Datenquelle initialisiert")
if st.button("Datenquelle zurücksetzen"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# Haupt-Chat-Oberfläche
st.header("Chat-Oberfläche")
# Chat-Nachrichten anzeigen
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# Chat-Eingabe
if prompt := st.chat_input("Stellen Sie eine Frage zu Ihren Dokumenten"):
# Benutzernachricht zum Chat-Verlauf hinzufügen
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# Abfrage verarbeiten
try:
response = st.session_state.processor.process_query(prompt)
# Assistenten-Antwort zum Chat-Verlauf hinzufügen
st.session_state.messages.append({"role": "assistant", "content": response})
with st.chat_message("assistant"):
st.write(response)
except Exception as e:
st.error(f"Fehler bei der Verarbeitung der Abfrage: {str(e)}")
# Schaltfläche zum Löschen des Chats, falls Nachrichten vorhanden
if st.session_state.messages:
if st.button("Chat löschen"):
st.session_state.messages = []
st.rerun()
Der Initialisierungsprozess enthält eine robuste Fehlerbehandlung und Pfadvalidierung, um häufige Benutzerfehler zu vermeiden. Die Seitenleiste liefert klare Statusinformationen, und die Chat-Oberfläche folgt vertrauten Mustern, die Benutzer von modernen KI-Assistenten erwarten. Das System behält den Gesprächsverlauf bei, sodass Benutzer auf früheren Fragen aufbauen und einen natürlichen Analyse-Workflow erstellen können. Die Chat-Löschfunktion ermöglicht einen Neuanfang bei Bedarf.
Das Analysesystem ausführen
Nachdem wir das RAG-gestützte Verkaufsanalysesystem gebaut haben, wollen wir es testen und sehen, wie gut es bei realen Geschäftsfragen funktioniert.
Erstellen Sie zunächst eine Datei data_generator.py und kopieren Sie den Code aus dem Python-Skript hier, um Beispieldaten zu generieren, die wir dem System geben. Führen Sie das Skript mit dem Befehl aus:
| python data_generator.py |
Dies erstellt einen Ordner sample_documents mit:
- large_sales_dataset.csv - 10.000 Verkaufsdatensätze
- business_strategy_2024.txt - Strategisches Geschäftsdokument
- sales_meeting_notes.txt - Besprechungsnotizen und Aktionspunkte
Starten Sie dann die Streamlit-Anwendung:
| streamlit run main.py |
Die Anwendung wird in Ihrem Browser unter http://localhost:8501 geöffnet.
Nach der Initialisierung können Sie das System testen, indem Sie Ihren Dokumentenpfad als sample_documents eingeben und verschiedene Geschäftsfragen stellen:
| # Beispielabfragen zum Testen des Systems topic_1 = “Wie hoch war der Gesamtumsatz für Elektronik im Jahr 2024?” topic_2 = “Was sind unsere wichtigsten strategischen Initiativen für die Kundenerfahrung?” topic_3 = “Welcher Vertriebsmitarbeiter hatte den höchsten Gesamtumsatz und wie hoch war dieser?” |
Fazit
In diesem Tutorial haben wir ein RAG-gestütztes Analysesystem gebaut, das zeigt, wie moderne KI Geschäftsintelligenz-Workflows transformieren kann. Durch die Kombination mehrerer Datenverarbeitungsansätze unter einer einzigen intelligenten Oberfläche haben wir ein Werkzeug geschaffen, das das gesamte Spektrum an Geschäftsfragen abdecken kann, mit denen Analysten täglich konfrontiert sind. Die wichtigsten Stärken des Systems sind:
- Intelligentes Abfrage-Routing, das automatisch die beste Datenquelle für jede Frage bestimmt
- Multi-Format-Unterstützung für SQL-Datenbanken, CSV-Dateien, Excel-Tabellen und Textdokumente
- Robuste Fehlerbehandlung mit Fallback-Mechanismen und klaren Fehlermeldungen
- Chat-basierte Interaktion ohne technische Fachkenntnisse
- Integrierte Überprüfung und Validierung für zuverlässige Ergebnisse
Sie können diese Anwendung erweitern, um weitere Quellen, fortschrittlichere Analysetechniken und benutzerdefinierte Geschäftslogik zu integrieren. Nachdem Sie dieses Wissen erlangt haben, können Sie Ihre eigenen Agenten mit Novita AI erstellen oder es in Ihre bestehenden Projekte integrieren.
Novita AI ist die All-in-One-Cloud-Plattform, die Ihre KI-Ambitionen beflügelt. Integrierte APIs, serverlos, GPU-Instanz – die kosteneffektiven Werkzeuge, die Sie brauchen. Infrastruktur überflüssig, kostenlos starten, Ihre KI-Vision Wirklichkeit werden lassen.
