Einbetten eines LangChain-Agenten in FastAPI: Tools, manuelle Suche, Streaming
Erstellen Sie einen In-App-Assistenten mit FastAPI und LangChain: Ein PDF-Handbuch in ChromaDB als Tool bereitgestellt, benutzerbezogener Kontext, gepuffertes Verlauf und strömende Antworten.
Sobald eine Anwendung über einige wenige Bildschirme hinauswächst, beginnt ihre Dokumentation sich auszuweiten, und die Nutzer lesen sie nicht mehr. Ein 20-seitiges Handbuch, das Funktionen, Konfigurationen sowie Domäneregeln erläutert, ist wertvoll – aber nur dann, wenn die Nutzer darin ohne lange Suche eine Antwort finden können. Ein in das Produkt integrierter Assistent kann diese Lücke schließen: Er beantwortet Fragen wie „Wie funktioniert das?“ aus dem Handbuch sowie „Was ist in meinem Projekt?“ anhand der eigenen Daten der Anwendung.
Dieser Leitfaden führt Sie durch eine kompakte, funktionsfähige Version eines solchen Assistenten. Sie werden einen LangChain-Agent in einen FastAPI-Dienst integrieren, ihm ein Tool zur Auswertung von Anwendungsdaten sowie ein weiteres Tool zur Suche in einem in ChromaDB gespeicherten PDF-Handbuch zuweisen, den authentifizierten Benutzer bei Bedarf an den Agent weiterleiten, das Gesprächsverlauf mit einem LangGraph-Checkpointer speichern und die Antwort an den Client streamen. Unterwegs weisen wir auf Lücken im minimalen Code hin, die Sie schließen müssen, damit alles reibungslos läuft, sowie auf Änderungen, die vor dem Einsatz in der Produktion vorgenommen werden sollten.
Das Szenario und die beteiligten Komponenten
Stellen Sie sich ein Team vor, das ein Tool zur Erstellung von Photovoltaikanlagen entwickelt. Das Produkt begann klein und sammelte im Laufe der Zeit immer mehr Komponenten wie Solarmodule, Wechselrichter, Produktionsprognosen, Dachpläne sowie eine lange Liste an Designregeln an. Das zugehörige Handbuch umfasst mittlerweile mehr als 20 Seiten. Immer wieder tauchen zwei Arten von Fragen auf:
- Fragen zum Produkt selbst: Was eine Funktion bewirkt, wo sich eine Einstellung befindet, welche Regel gilt. Die Antworten finden Sie in der Dokumentation.
- Fragen zur eigenen Arbeit des Benutzers: Welchen Wechselrichter er gewählt hat, wie viel Leistung sein System erzeugt, welche Dächer er berücksichtigt hat. Die Antworten liegen in der Datenbank der Anwendung, und kein Modell kennt sie standardmäßig.
Retrieval-Augmented Generation (RAG) kümmert sich um die erste Art: Das Handbuch wird indiziert, bei einer Frage werden die relevanten Passagen abgerufen und dem Modell als Kontext übergeben. Tools kümmern sich um die zweite Art: Kleine Funktionen, die das Modell aufrufen kann, um Daten aus der Anwendung abzurufen. Wenn man beides kombiniert, erhält man einen Assistenten, der sowohl das Produkt erklären als auch über ein konkretes Projekt nachdenken kann.
Das tatsächliche System hinter diesem Szenario verfügt über viele weitere Werkzeuge sowie umfangreichere Domänenlogik. Was hier dargestellt wird, ist absichtlich auf das Wesentliche reduziert, damit die Architektur sichtbar bleibt:
- FastAPI stellt die HTTP-API bereit und ermittelt, wer der Aufrufer ist.
- Der Agent von LangChain (läuft auf LangGraph) übernimmt den Schlussfolgerungsprozess sowie den Zustand des Gesprächs.
- Ein LLM interpretiert jede Anfrage und entscheidet, ob externe Informationen benötigt werden.
- Werkzeuge gewähren dem Agenten kontrollierten Zugriff auf die Funktionalitäten der Anwendung.
- RAG ermöglicht es dem Agenten, in der Dokumentation zu suchen.
- ChromaDB speichert die Abschnitte des Handbuchs und führt Vektorabfragen durch.
- Streaming übermittelt Tokens an den Client, während die Antwort noch generiert wird.
Projektstruktur
Jeder Aufgabenbereich erhält sein eigenes Paket: HTTP-Routing, Authentifizierung, Agentenlogik, Tools sowie der RAG-Pipeline. Dadurch bleiben alle Dateien klein und es wird klar, wohin eine neue Funktion gehört.
project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│ ├── __init__.py
│ └── dependencies.py
│
├── routers/
│ ├── __init__.py
│ └── chat.py
│
├── llm/
│ ├── __init__.py
│ ├── agent.py
│ ├── context.py
│ ├── orchestrator.py
│ ├── prompts.py
│ ├── provider.py
│ │
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── demo_tool.py
│ │ └── manual_tool.py
│ │
│ └── rag/
│ ├── __init__.py
│ ├── config.py
│ ├── context.py
│ │
│ ├── ingestion/
│ │ ├── __init__.py
│ │ ├── loader.py
│ │ ├── chunker.py
│ │ ├── chroma.py
│ │ └── indexer.py
│ │
│ └── retrieval/
│ ├── __init__.py
│ └── retriever.py
│
├── scripts/
│ ├── __init__.py
│ └── index_manual.py
│
├── docs/
│ └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)
Was jede Komponente bewirkt:
main.pyerstellt die FastAPI-Anwendung.auth/enthält die simulierten Authentifizierungsabhängigkeiten.routers/enthält die HTTP-Endpunkte.llm/enthält alles, was den Agenten betrifft.llm/tools/enthält die Funktionen, die der Agent aufrufen kann.
llm/rag/ enthält den Abrufpipeline, der in ingestion/ (Laden, Aufteilen und Indizieren des PDFs) sowie retrieval/ (Abfragen von ChromaDB) unterteilt ist.scripts/ enthält Befehle, die man manuell ausführt, wie zum Beispiel das Indizieren des Handbuchs.docs/ enthält das Quell-PDF.chroma_data/ wird lokal erzeugt und sollte niemals kommittet oder bereitgestellt werden.Sie werden nicht alle diese Elemente auf einmal erstellen. Die Build-Reihenfolge lautet: API-Schicht, anschließend das Modell, dann die Tools, danach der RAG-Pipeline, gefolgt von Kontext, Streaming und Historie.
Schritt 1: Ein FastAPI-Skelett mit einem simulierten Benutzer
Beginnen Sie damit, alles zu installieren, was das Projekt benötigt. Die Liste umfasst den Webserver, LangChain und LangGraph, den mit OpenAI kompatiblen Chat-Client, ChromaDB, den PDF-Lader und Text-Aufteiler sowie python-dotenv für die Konfiguration.
pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv
Das Modell wird über OpenRouter erreicht, daher erstellen Sie eine .env-Datei im Projektverzeichnis, die den API-Schlüssel enthält.
OPENROUTER_API_KEY=your_api_key_here
Fügen Sie .env umgehend zu .gitignore hinzu. Ein Schlüssel, der einmal in die Versionskontrolle gelangt, sollte als geleakt betrachtet werden.
Der Einstiegspunkt der Anwendung
main.py bleibt klein. Er erstellt die Anwendung und registriert den Chat-Router; nichts zum Modell oder zum Agenten gehört hier hin.
from fastapi import FastAPI
from routers.chat import chat_router
app = FastAPI(
title="AI Agent Demo",
)
app.include_router(chat_router)
Ein erster Chat-Endpunkt
In routers/chat.py definieren Sie einen Router unter dem Präfix /chat mit einer einzigen POST-Route. Im Moment gibt er nur die eingehende Nachricht zurück, was ausreicht, um zu überprüfen, dass alles funktioniert, bevor KI zum Einsatz kommt.
from fastapi import APIRouter
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
):
return {
"message": message,
}
Beachten Sie, dass message: str auf einer POST-Route ohne Body-Modell dazu führt, dass FastAPI den Wert aus der Abfrageseite liest. Das ist praktisch zum Ausprobieren in der Swagger UI, doch für einen echten Client sollte normalerweise ein mit einem Pydantic-Modell definiertes JSON-Body akzeptiert werden, da Abfrageseiten in den Zugriffsprotokollen landen und praktische Längengrenzen haben.
Eine Mock-Authentifizierungsabhängigkeit
In einer Produktivanwendung würde man eine Session-Cookie oder ein JWT überprüfen und den Benutzer aus der Datenbank laden. Hier ersetzt ein benutzerdefinierter Header all das. Erstellen Sie auth/dependencies.py mit einer kleinen MockUser-Dataclass sowie einer get_current_user-Funktion, die den X-Demo-User-Header liest und bei Fehlen eine 401-Fehlermeldung ausgibt.
from dataclasses import dataclass
from fastapi import Header, HTTPException
@dataclass
class MockUser:
id: str
name: str
def get_current_user(
x_demo_user: str | None = Header(default=None),
) -> MockUser:
if x_demo_user is None:
raise HTTPException(
status_code=401,
detail="Missing X-Demo-User header",
)
return MockUser(
id=x_demo_user,
name=x_demo_user,
)
Durch die Abhängigkeitsinjektion von FastAPI wird der Benutzer direkt an den Endpunkt weitergeleitet. Es genügt, einen Parameter mit Depends(get_current_user) zu deklarieren.
from fastapi import APIRouter, Depends
from auth.dependencies import (
MockUser,
get_current_user,
)
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
current_user: MockUser = Depends(
get_current_user,
),
):
return {
"user": current_user.name,
"message": message,
}
Ein Client identifiziert sich, indem er ein solches Header sendet:
X-Demo-User: user-123
Bei jeder Anfrage ruft FastAPI zunächst get_current_user() auf und übergeben das resultierende MockUser-Objekt an ask_ai. Der wichtige Aspekt des Designs ist die Aufgabenteilung: FastAPI ist für die Authentifizierung verantwortlich, während die KI-Schicht lediglich ein Benutzerobjekt erhält, dem sie vertrauen kann. Später ermöglicht dieses Benutzerobjekt es den Tools, Daten der richtigen Person zurückzugeben. Ein späterer Austausch des Mock-Objekts gegen eine echte Authentifizierung ändert nur diese eine Abhängigkeit.
Schritt 2: Verbindung eines Modells über OpenRouter
Sobald es einen funktionierenden Endpunkt sowie einen bekannten Aufrufer gibt, benötigt der Service ein Modell. OpenRouter stellt eine mit OpenAI kompatible API bereit, sodass die ChatOpenAI-Klasse von LangChain damit arbeiten kann, sobald man base_url auf OpenRouter richtet und den entsprechenden Schlüssel übermittelt.
Fügen Sie dies in llm/provider.py ein. Die Datei lädt .env und gibt bei Fehlen des Schlüssels sofort einen klaren Fehler an. Zudem wird der Schlüssel mit Pydantic’s SecretStr umhüllt, damit er nicht versehentlich in Logs oder Ausdrücken angezeigt wird.
import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI
load_dotenv()
api_key = os.getenv(
"OPENROUTER_API_KEY",
)
if not api_key:
raise RuntimeError(
"OPENROUTER_API_KEY environment variable is not set."
)
model = ChatOpenAI(
model="YOUR_MODEL",
api_key=SecretStr(api_key),
base_url="https://openrouter.ai/api/v1",
)
Weil der Schlüssel aus der Umgebungsvariablen stammt, erscheint er niemals im Quellcode. In diesem Stadium könnte man bereits Anfragen an das Modell senden und Antworten erhalten – doch das wäre nur ein einfacher Aufruf eines LLMs. Das Ziel ist ein Agent, der selbst entscheiden kann, wann er ein Tool benötigt.
Modell auswählen
Der Parameter model ist lediglich ein Identifikator für ein OpenRouter-Modell, sodass Sie Modelle austauschen können, ohne den Rest des Codes anzufassen. Bei der Vergleich von Kandidaten sollten Sie folgende Punkte überprüfen:
- die Unterstützung für Tool-Aufrufe, von der das Agent abhängt;
- die Unterstützung für Streaming;
- die Größe des Kontextfensters;
- Rate Limits;
- ob es eine kostenlose Version gibt.
OpenRouter bietet einige Modelle kostenlos an, was beim Experimentieren praktisch ist. Das Katalog wird regelmäßig aktualisiert, daher sollten Sie die aktuelle Liste durchsehen und nach kostenlosen Modellen filtern, anstatt sich auf feste Empfehlungen zu verlassen. Was auch immer Sie wählen, wird direkt in den Konstruktor eingegeben:
model = ChatOpenAI(
model="YOUR_MODEL_ID",
api_key=SecretStr(api_key),
base_url="https://openrouter.ai/api/v1",
)
Zum Beispiel, wenn der Katalog einen Identifikator wie den untenstehenden anzeigt (ein Beispiel zum Zeitpunkt der Erstellung; er ist möglicherweise nicht mehr verfügbar), geben Sie diesen genauen String als model weiter:
google/gemma-4-26b-a4b-it:free
Bedenken Sie, dass kostenlose Modelle eine gemeinsam genutzte Kapazität sind. Sie werden oft mit einer Bandbreitenbeschränkung konfrontiert oder stehen gerade in kritischen Momenten während einer Demo nicht zur Verfügung. Der Wechsel zu einem anderen Modell oder das Hinzufügen Ihrer eigenen Provider-Schlüssel über OpenRouter behebt dieses Problem in der Regel. Für die Produktion sollten Sie auf Zuverlässigkeit, Funktionalität, Latenzzeit und Kosten achten – nicht nur auf den Preis.
Schritt 3: Vom Modell zum Agenten
Ein direkter Modellaufruf erfolgt in einem einzigen Schritt: Der Text des Benutzers wird eingegeben und anschließend eine Antwort ausgegeben. Ein Agent fügt einen Entscheidungslauf ein. Das Modell prüft die Anfrage, entscheidet, ob es sofort antworten kann oder zunächst etwas benötigt, ruft bei Bedarf ein Tool auf, liest das Ergebnis und erstellt anschließend die endgültige Antwort. Im Großen und Ganzen:
- Einfacher Aufruf: Benutzer, dann LLM, dann Antwort;
Das Agent-Modul
In llm/agent.py erstellt create_agent von LangChain den Agenten aus einem Modell und einer Systemanweisung. Im Hintergrund erzeugt es ein LangGraph-Diagramm, das den Modell-und-Tools-Zyklus für Sie ausführt.
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
agent = create_agent(
model=model,
system_prompt=SYSTEM_PROMPT,
)
Dieser Agent verfügt noch über keine Tools, daher verhält er sich sehr ähnlich wie das reine Modell. Geben Sie ihm zunächst Anweisungen.
Die Systemanweisung
llm/prompts.py enthält eine kurze Anweisung, die dem Modell mitteilt, wofür es dient, das Erfinden von Daten verbietet und angibt, welche Art von Frage auf welche Art von Abfrage abzielt.
SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()
Die Anweisung legt zwei Quellen der Wahrheit fest:
- Anwendungsdaten (was im Konto des Benutzers ist) stammen von Anwendungs-Tools;
- Anwendungskenntnisse (wie das Produkt funktioniert) stammen aus der Dokumentensuche.
Eine Klarstellung, bevor wir fortfahren: Tools und RAG werden unten getrennt vorgestellt, weil dadurch jedes leichter verständlich wird. Im fertigen Agenten ist jedoch die Dokumentensuche selbst ein Tool. Es gibt kein zweites Mechanismus – der Agent sieht eine Liste von aufrufbaren Funktionen, und das Durchsuchen des Handbuchs ist eine davon.
Schritt 4: Das erste Tool
Ein Tool ist eine Funktion, die der Agent aufrufen darf. Dadurch lässt sich die Architektur skalieren: Anstatt alle Daten der Anwendung in den Prompt einzufügen, werden spezifische Operationen bereitgestellt, und das Modell kann sie nur dann anfordern, wenn eine Frage danach verlangt.
Für die Demo definiert llm/tools/demo_tool.py ein Tool, das einen festen Block an Projektinformationen zurückgibt.
from langchain.tools import tool
@tool
def get_my_demo_data() -> str:
"""
Return information about the demonstration data.
This is just for demo data. But in production, make a more detailed instruction.
"""
return """
Project: Aperture Analytics Dashboard
Owner: Jordan Lee
Status: In Progress
Team size: 6
Budget: $84,000
Deadline: 2026-11-15
Description: An internal dashboard for visualizing customer usage
metrics, built with FastAPI and React, integrating with the
company's data warehouse.
""".strip()
Zwei Aspekte sind hier wichtig. Der @tool-Decorator verwandelt eine gewöhnliche Python-Funktion in ein LangChain-Tool und leitet dabei ihren Namen sowie das Argumentsschema aus der Signatur ab. Die Dokumentation wird zur Beschreibung des Tools, die vom Modell gelesen wird, um zu entscheiden, ob es aufgerufen werden soll. In einem echten System verdient diese Beschreibung besondere Sorgfalt: Es sollte genau beschrieben werden, was das Tool zurückgibt, wann es angemessen ist und wann nicht. Eine vage Beschreibung ist einer der häufigsten Gründe dafür, dass ein Agent das falsche Tool oder gar keines aufruft.
Registrieren Sie das Tool, indem Sie es an create_agent übergeben:
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data
agent = create_agent(
model=model,
tools=[
get_my_demo_data,
],
system_prompt=SYSTEM_PROMPT,
)
Ihr Code entscheidet niemals, wann die Funktion ausgeführt wird. Wenn ein Benutzer fragt: „Welche Demo-Daten habe ich?“, erkennt das Modell, dass es accountbezogene Informationen benötigt, und ruft get_my_demo_data() auf. Wenn der Benutzer fragt: „Was ist eine Demo?“, ist keine Abfrage notwendig, und das Modell antwortet direkt. Diese Entscheidung wird bei jedem Schritt innerhalb des Agentenloops getroffen.
Schritt 5: Aufbau der RAG-Pipeline für das Handbuch
Der Agent kann nun Anwendungsdaten abrufen, kennt aber immer noch nichts über die Funktionsweise des Produkts. Das Einfügen eines 20-seitigen Handbuchs in den Systemprompt würde bei jeder Anfrage Tokens verschwenden und die Aktualisierung erschweren. RAG vermeidet beide Probleme.
Es ist wichtig, genau zu definieren, was RAG ist und was nicht. Auf der Dokumentation wird nichts trainiert oder feinabgestimmt. Zum Zeitpunkt der Abfrage durchsucht das System das Handbuch nach den für die Frage am relevantesten Passagen und liefert diese dem Modell als Kontext, wobei das Modell anschließend darauf antwortet.
Der Prozess besteht aus zwei Phasen:
- Eingabeverarbeitung, die unabhängig von der Webanwendung abläuft: Das PDF wird geladen, in Abschnitte aufgeteilt und diese zusammen mit ihren Embeddings in ChromaDB gespeichert.
- Auswertung, die innerhalb einer Anfrage stattfindet: Die Frage wird genommen, in ChromaDB gesucht, die besten Abschnitte ausgewählt und an das Modell weitergeleitet.
Falls Sie einen umfassenderen Überblick über die Konzepte wünschen, behandelt der Blogbeitrag zur auf Anfrage frisches Wissen abrufenden Retrieval-Methode sie ausführlicher; hier konzentrieren wir uns auf die Implementierung.
Laden des PDFs
llm/rag/ingestion/loader.py verwendet LangChain’s PyPDFLoader, der jede Seite des PDFs in ein Document umwandelt.
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
PDF_PATH = Path("docs/manual.pdf")
def load_manual():
loader = PyPDFLoader(
str(PDF_PATH),
)
documents = loader.load()
return documents
Ein Document enthält zwei Elemente: den extrahierten Text in page_content sowie ein metadata-Dictionary, das angibt, woher dieser Text stammt. Die Metadaten ermöglichen es später, in einer Antwort auf eine Seite zu verweisen. Konzeptionell sieht jede geladene Seite so aus:
Document
├── page_content
│ └── "To create a new demo data..."
│
└── metadata
├── source: docs/manual.pdf
└── page: 12
PyPDFLoader erfasst in der Regel sowohl einen nullbasierten page-Index als auch eine für Menschen lesbare page_label. Der untenstehende Kontextbuilder verwendet page_label, der den Seitenzahlen entspricht, die Leser im PDF sehen.
Seiten in Blöcke aufteilen
Das Durchsuchen ganzer Seiten, geschweige denn des gesamten Dokuments als einheitlichen Blocks, liefert grobe Ergebnisse. llm/rag/ingestion/chunker.py teilt die Dokumente mit RecursiveCharacterTextSplitter auf.
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document
def chunk_documents(
documents: list[Document],
) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=150,
separators=[
"\n\n",
"\n",
". ",
" ",
"",
],
)
return splitter.split_documents(
documents,
)
Der Splitter zielt auf Blöcke von etwa 1.000 Zeichen mit einer Überlappung von 150 Zeichen ab. Die Liste der Trennzeichen wird in dieser Reihenfolge ausprobiert: Zuerst wird an Absatzgrenzen getrennt, anschließend an Zeilenumbrüchen, dann am Ende von Sätzen und danach an Leerzeichen – erst als letztes Mittel mitten in einem Wort. Die Überlappung besteht deshalb, weil ein einzelner Sachverhalt mehrere Trennpunkte überschreiten kann; das Wiederholen kurzer Textabschnitte auf beiden Seiten verringert die Wahrscheinlichkeit, dass der relevante Satz in zwei Teile geteilt wird.
Diese Zahlen sind Ausgangspunkte, keine Regeln. Die richtige Größe hängt davon ab, wie Ihre Dokumente verfasst sind und wie präzise die Suche sein muss – betrachten Sie sie daher als Werte, die anhand konkreter Anforderungen angepasst werden müssen. Der Artikel des Blogs zu Chunking, das Beweise bewahrt geht dieser Abwägung näher auf den Grund.
Eine dauerhafte Chroma-Sammlung
llm/rag/ingestion/chroma.py öffnet einen PersistentClient, der die Daten auf dem Datenträger speichert und die Sammlung des Handbuchs zurückgibt, wobei diese bei erster Verwendung erstellt wird.
import chromadb
from llm.rag.config import (
CHROMA_PATH,
MANUAL_COLLECTION_NAME,
)
def get_chroma_client():
return chromadb.PersistentClient(
path=CHROMA_PATH,
)
def get_manual_collection():
client = get_chroma_client()
return client.get_or_create_collection(
name=MANUAL_COLLECTION_NAME,
)
Die Pfade und Namen stammen aus llm/rag/config.py, welches Umgebungsvariablen liest und bei Bedarf auf sinnvolle Standardwerte zurückgreift:
import os
CHROMA_PATH = os.getenv(
"CHROMA_PATH",
"./chroma_data",
)
MANUAL_PATH = os.getenv(
"MANUAL_PATH",
"docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
"MANUAL_COLLECTION_NAME",
"manual",
)
Fügen Sie die entsprechenden Einträge zu .env hinzu:
CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual
Überhaupt kein Embedding-Modell ist konfiguriert, und das ist für die Demo beabsichtigt. Wenn eine Sammlung ohne explizite Embedding-Funktion erstellt wird, verwendet Chroma seinen eingebauten Standard: Jedes Mal, wenn Dokumente hinzugefügt werden, berechnet Chroma deren Embeddings selbst und speichert sie neben dem Text sowie den Metadaten. Das Standardmodell läuft lokal und wird bei erster Verwendung heruntergeladen, wodurch die erste Indexierungsphase möglicherweise durch das Herunterladen verzögert wird.
Das Ergebnis ist ein lokaler, persistenter Vektor-Speicher im Verzeichnis chroma_data/. Da er vollständig aus dem PDF abgeleitet wird, sollte er zusammen mit .env in .gitignore aufgenommen werden.
Die Indexierungsarbeit
llm/rag/ingestion/indexer.py verknüpft die einzelnen Einpflegeschritte miteinander.
from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection
def index_manual():
collection = get_manual_collection()
if collection.count() > 0:
print(
f"Manual already indexed "
f"({collection.count()} chunks)."
)
return
manual_path = Path(
MANUAL_PATH,
)
if not manual_path.exists():
raise FileNotFoundError(
f"Manual not found: {manual_path}"
)
documents = load_manual()
print(
f"Loaded {len(documents)} pages."
)
chunks = chunk_documents(
documents,
)
print(
f"Created {len(chunks)} chunks."
)
collection.add(
ids=[
f"manual-chunk-{i}"
for i in range(len(chunks))
],
documents=[
chunk.page_content
for chunk in chunks
],
metadatas=[
chunk.metadata
for chunk in chunks
],
)
print(
f"Stored {len(chunks)} chunks."
)
Betrachten wir, was es tut. Es öffnet die Sammlung und gibt frühzeitig zurück, falls bereits Blöcke vorhanden sind – dadurch bleiben wiederholte Ausführungen unbedenklich. Anschließend prüft es, ob das PDF existiert; falls nicht, wird ein klarer Fehler ausgelöst. Danach lädt es die Seiten, teilt sie in Blöcke auf und fügt alles in einem Aufruf zu Chroma hinzu, wobei stabile IDs (manual-chunk-0, manual-chunk-1 usw.), die Texte der Blöcke sowie deren Metadaten verwendet werden. Fortschrittsmeldungen geben an, wie viele Seiten und Blöcke verarbeitet wurden.
Eine Fallstrick ergibt sich aus diesem vorzeitigen Zurückkehren: Wenn Sie das Handbuch bearbeiten und den Skript erneut ausführen, passiert nichts, weil die Sammlung nicht leer ist. Um die Änderungen zu berücksichtigen, müssen Sie die Sammlung (oder den chroma_data/-Ordner) vor dem Neuindizieren löschen oder den Schutzmechanismus durch Logik ersetzen, die gezielt ergänzt oder neu aufbaut.
Die Auflistung zeigt scripts/index_manual.py nicht an; es muss lediglich index_manual importieren und aufrufen. Führen Sie es einmal als Modul von der Projektwurzel aus:
python -m scripts.index_manual
Diese einzige Ausführung liest den PDF, teilt ihn in Blöcke auf, fügt diese Blöcke ein und speichert sie. Die Terminalausgabe gibt die Zahlen an. In der vereinfachten Demo handelt es sich um einen einseitigen PDF mit einer einzigen Regel: „Die Demo-Daten dürfen nur an Admin-Benutzer gegeben werden“, was ausreicht, um zu zeigen, ob die Abruffunktion funktioniert. Danach berührt das Starten von FastAPI den PDF überhaupt nicht mehr, da die Vektoren bereits gespeichert sind.
Absuchen in der Sammlung
Das Einindexieren allein bringt dem Agenten nichts; er benötigt eine Methode zum Suchen. llm/rag/retrieval/retriever.py umschließt Chromas Abfrageschnittstelle.
from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
get_manual_collection,
)
@dataclass
class RetrievedChunk:
content: str
metadata: dict[str, Any]
distance: float
def retrieve_manual(
query: str,
n_results: int = 5,
) -> list[RetrievedChunk]:
collection = get_manual_collection()
results = collection.query(
query_texts=[query],
n_results=n_results,
include=[
"documents",
"metadatas",
"distances",
],
)
documents = results["documents"] or []
metadatas = results["metadatas"] or []
distances = results["distances"] or []
retrieved_chunks = []
for document, metadata, distance in zip(
documents[0],
metadatas[0],
distances[0],
):
retrieved_chunks.append(
RetrievedChunk(
content=document,
metadata=dict(metadata)
if metadata else {},
distance=distance,
)
)
return retrieved_chunks
Die Funktion sendet die Frage als query_texts, bittet um bis zu fünf Ergebnisse und fordert die Dokumente, ihre Metadaten sowie ihre Entfernungen an. Chroma gibt für jede Abfrage eine Liste zurück, weshalb der Code den Index [0] jedes Feldes liest; die Fallbacks or [] schützen vor fehlenden Feldern. Jeder Treffer wird als RetrievedChunk-Dataclass verpackt, damit der Rest des Codes nicht von der Antwortstruktur von Chroma abhängig ist.
Weil die Abfrage mit dem gleichen Modell wie die gespeicherten Fragmente verarbeitet wird, erfolgt die Übereinstimmung semantisch. Eine Frage wie „Wie füge ich neue Demo-Daten hinzu?“ findet Passagen zur Erstellung oder Bereitstellung von Demo-Daten, selbst wenn dort nie die Wörter „neue hinzufügen“ verwendet werden. Die Distanz gibt an, wie nah jede Trefferstelle ist; je niedriger sie ist, desto ähnlicher sind die Inhalte. Dieser Wert ist später nützlich, wenn man schwache Treffer ignorieren möchte, anstatt stets fünf Fragmente an das Modell zu übergeben.
Sobald dies funktioniert, läuft der Abrufteil von RAG. Übrig bleibt nur noch die Übermittlung der Ergebnisse an das Modell.
Fragmente in Kontext umwandeln
llm/rag/context.py formatiert die Treffer in einen einzigen String, den das Modell lesen kann.
from llm.rag.retrieval.retriever import (
RetrievedChunk,
)
def build_context(
chunks: list[RetrievedChunk],
) -> str:
context_parts = []
for chunk in chunks:
page = chunk.metadata.get(
"page_label",
)
context_parts.append(
f"Source: User Guide, page {page}\n"
f"{chunk.content}"
)
return "\n\n---\n\n".join(
context_parts,
)
Jeder Abschnitt wird mit einer Quellzeile vorangestellt, die den Benutzerleitfaden sowie seine Seitenbezeichnung angibt, und die Abschnitte sind durch Trennlinien voneinander getrennt. Die Quellzeile ermöglicht es dem Modell, anzugeben, woher eine Antwort stammt, und gibt den Nutzern eine Möglichkeit zur Überprüfung.
Schritt 6: Den Leitfaden als Tool bereitstellen
Der Pipeline-Prozess ist abgeschlossen: Die Seiten werden geladen und in Abschnitte aufgeteilt, die Abschnitte befinden sich in ChromaDB, und die relevanten können gefunden und formatiert werden. Der Agent hat jedoch keine Ahnung davon, dass all das existiert. Hier kommt die frühere architektonische Notiz ins Spiel: Die Dokumentensuche wird zu einem weiteren Tool.
llm/tools/manual_tool.py definiert search_user_manual, welches eine Abfrage entgegennimmt, fünf Abschnitte abruft und diese als formatierten Kontext zurückgibt. Falls nichts zurückkommt, wird eine explizite Nachricht ausgegeben, die angibt, dass das Handbuch die Frage nicht abdeckt – dadurch erhält das Modell etwas Aussagekräftiges zu übermitteln anstelle eines leeren Strings.
from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual
@tool
def search_user_manual(
query: str,
) -> str:
"""
Search the application user manual.
Use this tool when the user asks about application
behavior, instructions, rules, limitations, or
how something works.
"""
chunks = retrieve_manual(
query=query,
n_results=5,
)
if not chunks:
return (
"The manual does not contain enough "
"information to answer this question."
)
return build_context(
chunks,
)
Wie zuvor dient die Dokumentation als Werbung für das Tool beim Modell. Darin wird angegeben, dieses Tool für Fragen zu Verhalten, Anweisungen, Regeln, Einschränkungen sowie zur Funktionsweise zu verwenden, was mit dem Systemprompt übereinstimmt.
Nun registrieren Sie beide Tools beim Agenten:
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
get_my_solar_system,
)
from llm.tools.manual_tool import (
search_user_manual,
)
agent = create_agent(
model=model,
tools=[
get_my_solar_system,
search_user_manual,
],
system_prompt=SYSTEM_PROMPT,
)
Achten Sie auf die Import-Anweisung in dieser Auflistung: Sie bezieht sich auf get_my_solar_system, einen Namen aus der vollständigen Anwendung, während das Demo-Tool-Modul get_my_demo_data definiert. Verwenden Sie in beiden Import-Anweisungen sowie in der tools-Liste get_my_demo_data, andernfalls wird das Modul nicht importiert werden.
Warum ein auf Tools basierendes Design bei wachsender Anwendung funktioniert
Der Agent benötigt niemals eine einzige umfangreiche Anweisung, die alles beschreibt, was die Anwendung weiß. Stattdessen erhält er spezifische, kontrollierte Fähigkeiten. Um einer Funktion des Assistenten etwas hinzuzufügen, muss ein neues Tool geschrieben und registriert werden; die HTTP-Schicht ändert sich dabei nicht. Die Demo verwendet genau ein Daten-Tool und ein Dokumentations-Tool, sodass das Muster leicht nachvollziehbar ist – doch dieselbe Struktur ermöglicht in der vollständigen Anwendung ein weitaus größeres Toolset.
Schritt 7: Übermittlung des authentifizierten Benutzers an den Agent
Das Demo-Tool gibt weiterhin fest codierte Daten zurück. Ein echtes Tool muss wissen, wer nachfragt – die Anwendung weiß das bereits: FastAPI hat den Benutzer in der Auth-Abhängigkeit ermittelt. Das Fehlende ist es, diesen Benutzer in den Agentenaufruf mitzunehmen. LangChain bezeichnet dies als Laufzeitkontext.
Definieren Sie die Struktur des Kontexts in llm/context.py:
from dataclasses import dataclass
from auth.dependencies import MockUser
@dataclass
class AgentContext:
user: MockUser
Dieses Objekt wird bereitgestellt, wenn der Agent aufgerufen wird, und dieser Unterschied ist wichtig. Der Benutzer stellt Informationen im Rahmen einer Anfrage dar. Er gehört zur aktuellen HTTP-Anfrage und nicht zum Gespräch, und er sollte niemals als Nachricht gespeichert werden, die vom Modell gelesen oder umgeschrieben werden kann. Indem er außerhalb der Nachrichtengeschichte bleibt, kann auch keine Anweisung den Agenten dazu bringen, sich wie ein anderer Benutzer zu verhalten. In der vollständigen Anwendung enthält dasselbe Kontextobjekt außerdem Elemente wie die Datenbanksession sowie die ID des bearbeiteten Projekts.
Der Demo-Code endet bei der Definition der Klasse, sodass noch zwei Verbindungen hergestellt werden müssen. Es lohnt sich, die aktuelle LangChain-Dokumentation zur genauen API zu prüfen. Zunächst wird beim Erstellen des Agents das Schema deklariert, in der Regel mit dem Argument context_schema=AgentContext bei create_agent. Zweitens müssen die Tools dieses Schema lesen: In LangChain 1.x kann ein Tool einen Laufzeitparameter entgegennehmen (zum Beispiel als ToolRuntime[AgentContext] markiert) und den Benutzer aus seinem context-Attribut lesen, welches vor der Sicht des Modells auf die Argumente des Tools verborgen ist. Genau dort würde eine echte Funktion get_my_demo_data nach Einträgen für user.id suchen.
Schritt 8: Eine Orchestrierungsschicht, die streamt
Anstatt den Agenten direkt vom Inneren des Routers aus aufzurufen, sollte die Interaktion in llm/orchestrator.py stattfinden. Dadurch konzentriert sich der Router weiterhin auf HTTP, während der Orchestrator bestimmt, wie eine Nachricht in einen ausgeführten Agenten umgewandelt wird.
from collections.abc import Iterator
from langchain_core.messages import (
AIMessage,
AIMessageChunk,
BaseMessage,
ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext
def chat_stream(
user_message: str,
user,
) -> Iterator[str]:
config: RunnableConfig = {
"configurable": {
"thread_id": f"user:{user.id}",
}
}
context = AgentContext(
user=user,
)
for chunk, metadata in agent.stream(
{
"messages": [
{
"role": "user",
"content": user_message,
}
]
},
config=config,
context=context,
stream_mode="messages",
):
if not isinstance(
chunk,
BaseMessage,
):
continue
if isinstance(
chunk,
ToolMessage,
):
continue
if not isinstance(
chunk,
(
AIMessage,
AIMessageChunk,
),
):
continue
if isinstance(
chunk.content,
str,
):
yield chunk.content
In dieser Funktion gibt es viel zu beachten, daher sollte man sie Schritt für Schritt bearbeiten.
Die Thread-ID wählt das Gespräch aus
Der erste Block erstellt die Ausführungskonfiguration:
config = {
"configurable": {
"thread_id": f"user:{user.id}",
}
}
Ein LangGraph Checkpointer speichert den Gesprächszustand unter Verwendung von thread_id als Schlüssel. Jeder Ausführungsvorgang mit derselben thread ID setzt das gleiche Gespräch fort, und seine Nachrichten können später wieder abgerufen werden. Hier wird die thread ID aus der Benutzer-ID abgeleitet, was bedeutet, dass jeder Benutzer genau ein Gespräch hat. Das ist für eine Demo in Ordnung; eine echte Anwendung würde eigene, angemessene Gesprächs IDs erstellen, mehrere pro Benutzer zulassen und bei jeder Anfrage überprüfen, ob der Aufrufer das angeforderte Thread besitzt.
Die Beschreibung geht davon aus, dass ein Checkpointer vorhanden ist, doch keines der Agent-Beispiele bestätigt dies. Ohne ihn hat thread_id keine Wirkung und nichts wird zwischen den Anfragen gespeichert. Erstellen Sie einen einzigen InMemorySaver in llm/agent.py und übergeben Sie ihn an create_agent über dessen checkpointer-Argument, damit sowohl der Agent als auch die darunterliegenden Historiefunktionen dieselbe Instanz importieren können.
Der Laufzeitkontext wird mit dem Ausführungsvorgang übertragen
Anschließend umhüllt der Orchestrator den Benutzer mit dem Kontextobjekt:
context = AgentContext(
user=user,
)
Dieses Objekt wird als context= an agent.stream() übergeben. Auf diesem Weg erreicht die in FastAPI festgelegte Identität den Agenten und von dort aus die Tools.
Filtern des Streams
agent.stream() wird mit stream_mode="messages" aufgerufen, wodurch beim Erzeugen von Tokens vom Modell Paare aus einem Nachrichtenblock und Metadaten erzeugt werden. Nicht alles in diesem Stream sollte den Benutzer erreichen. Die Schleife überspringt alles, was keine LangChain-Nachricht ist, ignoriert ToolMessage-Objekte (rohe Tool-Ausgaben wie abgerufter Manuelltext) und behält nur KI-Nachrichten sowie KI-Nachrichtenblöcke bei, wobei deren Inhalt ausgegeben wird, wenn es sich um einen einfachen String handelt. Inhalte, die von einigen Anbietern als Liste von Teilen übermittelt werden, werden durch diese letzte Überprüfung stillschweigend verworfen; falls Sie daher das Modell wechseln und leere Antworten erhalten, sollten Sie dort nachsehen.
Schritt 9: Zurückgabe einer strömenden Antwort
Dass chat_stream() als Generator implementiert wurde, war beabsichtigt. Das Warten auf die vollständige Antwort, bevor ein Byte gesendet wird, lässt den Benutzer vor einem Ladeindikator hängen, und die Antworten von großen Sprachmodellen können mehrere Sekunden dauern. Das Streamen zeigt bereits die ersten Wörter fast sofort an, wodurch der Assistent viel reaktiver erscheint.
FastAPIs StreamingResponse akzeptiert direkt einen Generator. Aktualisieren Sie routers/chat.py:
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
MockUser,
get_current_user,
)
from llm.orchestrator import chat_stream
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
current_user: MockUser = Depends(
get_current_user,
),
):
return StreamingResponse(
chat_stream(
user_message=message,
user=current_user,
),
media_type="text/plain",
)
Die Antwort wird als text/plain gesendet, wobei jeder erzeugte Datensatz sobald wie möglich in die Verbindung geschrieben wird. Da chat_stream ein normaler (synchroner) Generator ist, iteriert Starlette darüber in einem Worker-Thread, sodass der Event-Loop nicht blockiert wird. Falls Sie später strukturierte Ereignisse am Client benötigen (zum Beispiel, um anzuzeigen „Suche im Handbuch...“, während ein Tool läuft), sind Server-Sent Events der natürliche nächste Schritt.
Der vollständige Anfragenpfad lautet nun wie folgt: Der Client sendet eine Anfrage an /chat, FastAPI authentifiziert den Aufrufer, chat_stream() startet die Ausführung des Agents, das Modell entscheidet, ob ein Tool aufgerufen werden soll, jedes Tool wird ausgeführt und gibt sein Ergebnis zurück, das Modell schreibt die Antwort auf und die Tokens fließen wieder zum Client zurück.
Die entscheidende Eigenschaft ist, dass FastAPI das Modell selbst niemals ausführt. Der Router kommuniziert über HTTP, der Orchestrator steuert den Agenten, der Agent entscheidet, welche Informationen er benötigt, und die Tools übernehmen das Abrufen. Jede Schicht kann geändert werden, ohne die anderen zu stören.
Schritt 10: Wiedergeben der Konversationsgeschichte
Weil der Agent in Checkpoints gespeichert wird, wird sein Zustand nach jeder Runde aufbewahrt. Dadurch kann einem zurückkehrenden Benutzer seine vorherige Konversation angezeigt werden. Zwei Hilfsfunktionen übernehmen diese Aufgabe.
Lesen des Roh-Checkpoints
Die erste Funktion lädt den neuesten Checkpoint für einen Thread und gibt den messages-Channel zurück – oder eine leere Liste, wenn der Thread noch nie verwendet wurde:
def get_conversation_messages(
thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
"configurable": {
"thread_id": thread_id,
}
}
checkpoint = checkpointer.get(
config,
)
if checkpoint is None:
return []
return checkpoint[
"channel_values"
].get(
"messages",
[],
)
Falls Sie diesen Codeausschnitt kopieren, korrigieren Sie die Einrückung der config-Zuweisung: Sie muss innerhalb des Funktionskörpers eingereiht sein, andernfalls wirft Python einen Fehler aus. Zudem müssen BaseMessage, RunnableConfig sowie die gemeinsame checkpointer-Instanz importiert werden.
Was zurückgegeben wird, ist der rohe Zustand des Agents – und dieser umfasst mehr als nur den Chatinhalt, an den sich der Benutzer erinnert. Wenn der Agent ein Tool aufruft, protokolliert LangGraph eine KI-Nachricht mit dem Toolaufruf sowie eine separate Tool-Nachricht mit dem Ergebnis. Das sind Implementierungsdetails, die das Frontend nicht interpretieren muss.
Nachrichten für die Anzeige formatieren
Die zweite Funktion erstellt die für den Benutzer sichtbare Ansicht:
def get_conversation_messages_for_display(
thread_id: str,
) -> list[dict[str, str]]:
display = []
for message in get_conversation_messages(
thread_id,
):
if isinstance(
message,
HumanMessage,
):
content = _extract_text_content(
message.content,
)
if content.strip():
display.append(
{
"type": "human",
"content": content,
}
)
continue
if isinstance(
message,
AIMessage,
):
content = _extract_text_content(
message.content,
)
if content.strip():
display.append(
{
"type": "ai",
"content": content,
}
)
return display
Sie behält nur HumanMessage- und AIMessage-Objekte bei, extrahiert deren Text, entfernt alle leeren Objekte und gibt einfache Wörterbücher mit type und content zurück. Werkzeugnachrichten erscheinen nie, da sie nicht zu den beiden akzeptierten Typen gehören. Auch leere AI-Nachrichten werden entfernt, was wichtig ist, denn eine AI-Nachricht, die lediglich einen Werkzeugaufruf anfordert, enthält in der Regel keinen Text.
Die Implementierung stützt sich auf eine Hilfsfunktion _extract_text_content, die hier nicht gezeigt wird. Ihre Aufgabe ist es, den Inhalt unverändert zurückzugeben, wenn er ein String ist, und die Textteile miteinander zu verknüpfen, wenn es sich um eine Liste von Inhaltsteilen handelt. Zudem müssen HumanMessage und AIMessage importiert werden.
Der History-Endpunkt
Stellen Sie die Anzeige über einen GET-Route in routers/chat.py zur Verfügung. Dabei wird derselbe Thread-ID verwendet, der auch vom Chat-Endpunkt genutzt wird, und dieser wird zusammen mit den Nachrichten zurückgegeben.
@chat_router.get("/current")
def get_current_conversation(
current_user: MockUser = Depends(
get_current_user,
),
):
thread_id = (
f"user:{current_user.id}"
)
messages = (
get_conversation_messages_for_display(
thread_id,
)
)
return {
"thread_id": thread_id,
"messages": messages,
}
Die Chat-Oberfläche kann dies aufrufen, sobald die Seite geladen wird, um das vorhandene Gespräch anzuzeigen, bevor der Benutzer etwas eingibt:
GET /chat/current
Die Antwort sieht so aus:
{
"thread_id": "user:user-123",
"messages": [
{
"type": "human",
"content": "How much demo data do I have?"
},
{
"type": "ai",
"content": "You currently have 18 demo data."
}
]
}
Warum nicht den rohen Zustand zurückgeben?
Der interne Zustand des Agents und das Gespräch, das der Benutzer sieht, sind unterschiedliche Dinge. Im Laufe der Hinzufügung neuer Funktionen sammelt sich im Zustand eine Vielzahl an Tool-Aufrufen, -Ergebnissen, Zwischenschritten, Modellmetadaten sowie weiteren Verwaltungsinformationen. Der Rückgabe all dieser Daten würde die Frontend-Plattform mit den internen Strukturen des Agents verknüpfen und dazu führen, dass Tool-Ausgaben angezeigt werden, die eigentlich nicht gezeigt werden sollten. Der Backend sollte definieren, was zum öffentlichen Gesprächsverlauf gehört, und lediglich diesen zurückgeben.
Wie ein einzelner Anfragenfluss von Anfang bis Ende abläuft
Sobald alle Komponenten vorhanden sind, ist es wichtig, eine Eigenschaft klarzustellen: Das Modell berührt weder Ihre Datenbank noch Ihr PDF. Es kann lediglich darum bitten, ein Tool aufzurufen. Das Tool, das als gewöhnlicher Python-Code mit herkömmlichen Zugriffskontrollen ausgeführt wird, führt die Operation durch und gibt Text zurück; das Modell verwendet diesen Text, um seine Antwort zu erstellen. Genau diese Trennung sorgt dafür, dass der Assistent sicher in einer Anwendung mit echten Daten eingebettet werden kann.
Ausprobieren in Swagger UI
FastAPI erzeugt automatisch interaktive Dokumentation, sodass während des Tests kein separater Client erforderlich ist. Starten Sie den Server:
python -m uvicorn main:app --reload
Dann öffnen Sie die interaktiven API-Dokumentationen unter /docs, setzen Sie den Header X-Demo-User und versuchen Sie drei Interaktionen:
- Eine Datenfrage, wie zum Beispiel die Anfrage, welche Demo-Daten verfügbar sind. Der Agent sollte entscheiden, dass das Demo-Tool benötigt wird, es aufrufen und auf der Grundlage der zurückgegebenen Projektdetails antworten. In der vollständigen Anwendung fragt ein ähnliches Tool nach den tatsächlichen Datensätzen des Benutzers.
- Eine Dokumentationsfrage, wie zum Beispiel die Anfrage, wer Demo-Daten erhalten darf. Der Agent sollte eine manuelle Suche durchführen, die entsprechende Regel aus dem Index abrufen und antworten, dass nur Admin-Benutzer dies dürfen.
- Der History-Endpunkt,
GET /chat/current, der die Antworten des Menschen und der KI zu den vorherigen beiden Fragen ohne jegliche Tool-Nachrichten zurückgeben sollte.
Falls die ersten beiden Fragen Antworten liefern, die auf den Ausgaben des Tools beruhen, und die dritte Frage eine klare Transkription zeigt, funktioniert jede Schicht einwandfrei.
Vor dem Einsatz in der Produktion
Die Demo vereinfacht absichtlich mehrere Komponenten. Diese sollten noch einmal überprüft werden, bevor echte Benutzer auf den Service angewiesen sind.
Dauerhafter Konversationszustand
InMemorySaver eignet sich hervorragend für die Entwicklung, doch alles, was er speichert, verschwindet beim Neustart des Prozesses, und er kann nicht zwischen mehreren API-Instanzen hinter einem Load Balancer geteilt werden. Verwenden Sie einen Checkpointer, der von einer Datenbank oder einem anderen dauerhaften Speicher unterstützt wird, damit Konversationen auch nach Deployment-Änderungen erhalten bleiben und jede Instanz denselben Zustand sieht. Welche Daten der In-Memory-Speicher tatsächlich speichert und wie, erfahren Sie in dem Blog-Eintrag zu der Art und Weise, wie InMemorySaver Checkpoints, Schreibvorgänge und Blob-Daten organisiert.
Echte Implementierung eines Vektor-Speichers
Ein lokaler Chroma-Verzeichnis eignet sich gut, um den Pipeline-Prozess zu demonstrieren, stellt aber keine Produktiv-Infrastruktur dar. Führen Sie Chroma als dauerhaften Dienst aus oder wechseln Sie zu einer verwalteten Vektordatenbank, die zu Ihrem Technologiestack passt. Was auch immer Sie wählen, muss dauerhaft vorhanden sein, gesichert werden und von jeder Anwendungsinstanz aus erreichbar sein.
Indexierung bleibt außerhalb der API
Die Demo trifft bereits eine wichtige Entscheidung richtig: Die Indexierung erfolgt über ein separates Skript, das eigenständig aufgerufen wird, während die API lediglich Daten abruft.
python -m scripts.index_manual
Der Server lädt das PDF niemals, teilt es in Blöcke auf oder berechnet Embeddings beim Starten. Die Indexierung erfolgt offline; die Datenabfrage ist Teil der Bearbeitung einer Anfrage. Durch diese Trennung muss die API keine Änderungen in der Dokumentation erkennen oder etwas neu erstellen.
In der Produktion geht man einen Schritt weiter und führt denselben Indexierungscode als dedizierte Aufgabe aus – ausgelöst durch einen Deployment-Pipeline, nach einem Zeitplan oder von einem Worker, sobald neue Dokumente hochgeladen werden. Die Architektur bleibt unverändert; die Aufgabe wird lediglich automatisiert, wiederverwendbar und unabhängig einsetzbar. Die Verantwortlichkeiten teilen sich dabei klar auf:
- Aufgabe zur Dateneingabe: Lade Dokumente, teile sie in Blöcke auf, füge sie ein und aktualisiere den Vektor-Speicher.
- FastAPI-Dienst: Empfange Anfragen, hole relevante Blöcke ab und erzeuge Antworten.
- Vektor-Speicher: Speichere die zur Abfragezeit verwendeten indizierten Repräsentationen dauerhaft.
Vergessen Sie nicht den Schutzmechanismus für frühzeitigen Rückgang im Indexierer, wenn Sie ihn automatisieren; eine Aufgabe, die das Neuindexieren stillschweigend überspringt, ist schlimmer als gar keine Aufgabe.
Ein explizites Embedding-Modell
Durch die Verwendung der Standard-Embedding-Funktion von Chroma bleibt die Demo ohne zusätzliche Konfiguration, doch ein Produktivsystem sollte sein Embedding-Modell ausdrücklich wählen und konfigurieren. Dadurch sind die Ergebnisse reproduzierbar und man hat Kontrolle über Qualität, Kosten, Latenz sowie darüber, wo die Embeddings berechnet werden. Eine Regel ist unverhandelbar: Dasselbe Embedding-Modell muss sowohl für das Indexieren als auch für die Abfragen verwendet werden. Ein Wechsel bedeutet ein erneutes Indexieren aller Daten.
Überwachbarkeit und Fehlerbehandlung
Sobald ein Agent im Einsatz ist, ist es genauso wichtig zu wissen, was er getan hat, wie dafür zu sorgen, dass er funktioniert. Eine einzige Anfrage kann mehrere Modellaufrufe, einen oder mehrere Tool-Aufrufe sowie einen Abrufschritt vor der endgültigen Antwort beinhalten. Das Protokollieren nur dieser Endantwort sagt fast nichts aus, wenn etwas schiefgeht. Instrumentieren Sie den gesamten Ablauf:
- Tool-Aufrufe: Welche Tools wurden verwendet, mit welchen Argumenten und wie lange hat jeder Aufruf gedauert.
Ziel ist es, dass der Agent niemals eine Black Box darstellt. Für jeden Ablauf sollten Sie in der Lage sein zu sagen, was er getan hat, welche Tools er aufgerufen hat, wie lange jeder Schritt gedauert hat und wo es zu Fehlern kam. Die konkreten Tools hängen von Ihrer Technologieplattform ab; das Prinzip jedoch nicht.
Kernpunkte
- Man benötigt keine große KI-Plattform, um einer an Domänenrelevanz reichen Anwendung einen nützlichen Assistenten hinzuzufügen. Beginnen Sie mit dem kleinstmöglichen Satz an Komponenten, der ein reales Problem löst.
- Betrachten Sie die Dokumentensuche als eines von mehreren Tools. Dadurch hat der Agent eine einzige, einheitliche Möglichkeit, sowohl Produktwissen als auch Benutzerdaten abzurufen.
- Bewahren Sie die Identität im Laufzeitkontext auf, nicht in den Nachrichten. Der Benutzer gehört zur Anfrage, und die Tools sollten diese dort abrufen.
- Trennen Sie HTTP, Orchestrierung, Agent und Tools voneinander. Jede Schicht bleibt klein, und das Hinzufügen einer neuen Funktion bedeutet das Hinzufügen eines Tools.
Verwandte Artikel
- Hybride Agentenmemorien: Kombination von BM25 und Vektorabfrage mit RRF in Python — Erfahren Sie, warum reine Vektorabfragen als Agentenmemorien versagen, wie Reciprocal Rank Fusion BM25 und dichte Ergebnisse in Python kombiniert sowie wann GraphRAG-Zusammenfassungen hilfreich sind.
- Design von vierstufiger Agentenmemorie mit LangGraph und Amazon Bedrock — Lernen Sie, LLM-Agenten auf Bedrock und LangGraph funktionierende episodische, semantische und prozedurale Erinnerungen zu verleihen sowie sie vor Vergiftung, Datenleckagen von PII und Informationsfluss zu schützen.
- Gemini den Quellcode wählen lassen: FAISS, Tavily und direkte Antworten in LangGraph — Erstellen Sie einen kleinen LangGraph-Ablauf, bei dem Gemini jede Frage an eine FAISS-Wissensdatenbank, eine Tavily-Websuche oder eine direkte Antwort weiterleitet, wobei die Ausgabe strukturiert wird.
- Frageverarbeitung zwischen SQL-Tools und Web-Suchen mit einem Gemini-Agenten — Wie ein Tool-Aufruf-Agent von LangChain auf Vertex AI zwischen drei SQLite-Tools zur Umwandlung von Text in SQL sowie Live-Web-Suchen wählt, sowie die zu erwartenden Probleme bezüglich Daten, Abhängigkeiten und Authentifizierung.
- Streaming von LangGraph-Agenten an React ohne Offenlegung interner Tool-Informationen — Wie man die astream_events-Ausgabe von LangGraph für Server-Sent Events serialisieren, sensible Tool-Aufruf-Daten in FastAPI maskieren und freundliche Tool-Abzeichen in React anzeigen kann.
- Evaluierung des Agentenverhaltens: Ergebnisse, Trajektorien und Produktionsfeedback — Wie man agierende Systeme jenseits der endgültigen Antwort beurteilt: LangChain, LangGraph und LangSmith auf Rollen zuordnen, Trajektorien sowie Kontext bewerten und Produktionsfehler in die Bewertungen einbeziehen.