Suchen Sie nach Runbooks – kleben Sie sie nicht ein: Ein kurzer, zitiierter RAG-Weg in LangChain
Erstellen Sie einen zweistufigen Abrufweg in LangChain: Laden Sie Runbooks, teilen Sie sie in Abschnitte auf, führen Sie eine gezielte Suche durch, bewahren Sie Metadaten auf, verlangen Sie Angaben zu den zitierten Quellen-ID sowie testen Sie das System ohne kostenpflichtige Embeddings.
Suchen Sie in den Runbooks anstelle, den Prompt mit Informationen zu füllen
Zu diesem Zeitpunkt teilt sich die Helpdesk-App nach Tickettypen auf. Ein Datenbank-Ticket lädt gleichzeitig das Servicekatalog, aktuelle Vorfälle sowie neuere Bereitstellungen herunter, während ein Zugriffsticket einen anderen Weg nimmt. Dennoch fehlen die detaillierten Schritt-für-Schritt-Anleitungen hinter einer Empfehlung.
Man könnte jedes Runbook in den Prompt einfügen. Bei drei kleinen Dateien funktioniert das, doch je größer die Sammlung wird, desto weniger funktioniert es. Große Prompts sind teurer, laufen langsamer und erschweren es, herauszufinden, welcher Abschnitt eine Antwort unterstützt hat. Zudem können sie dem Modell Dokumente zukommen lassen, die der aktuelle Benutzer niemals lesen durfte. Die Lösung ist das Suchen.
Der hier erstellte Weg besteht aus zwei Phasen. Bei der Indizierung werden Markdown-Runbooks zu Dokumentobjekten mit Quellmetadaten, die wiederum in kleinere Abschnitte aufgeteilt werden:
indexing time
Markdown runbooks
-> Document objects with source metadata
-> smaller chunks
-> embeddings
-> vector store
Zum Zeitpunkt der Anfrage erzeugt die Ticketklassifizierung eine gezielte Suchanfrage, die an den Abrufmechanismus gesendet wird:
request time
ticket classification
-> focused search query
-> retriever
-> a few runbook passages with source IDs
-> recommendation chain
-> source-ID check
Menschen bezeichnen dies oft als zweistufige, durch Suchfunktion erweiterte Generierung, oder zweistufiges RAG. Die Bezeichnung ist weniger wichtig als die Abfolge: Zuerst erfolgt die Suche, und die Generierung erhält nur die zurückgegebenen Abschnitte. Das Modell entscheidet niemals selbst, ob eine Suche stattfinden soll, und kann auch keine beliebige Datenquelle auswählen. Ein Agent ist nicht erforderlich.
Das im gesamten Text verwendete Ticket
Das verwendete Beispiel ist ein Ticket mit folgendem Inhalt: Die EU-Zahlvorgänge scheitern nach dem Deployment um 14:05, und die Logs der checkout-api zeigen „database connection refused“. Die strukturierte Klassifizierung sieht wie folgt aus:
{
"ticket_id": "INC-2048",
"summary": "EU checkout fails after a deployment.",
"category": "database",
"severity": "SEV2",
"service": "checkout-api",
"confidence": 0.91,
"evidence": ["database connection refused"],
"missing_information": ["Affected request percentage"]
}
Eine gute Suche liefert das Runbook zur Zahlungsabwicklung-Datenbank und nicht einen Leitfaden zur Zahlungslatenz oder einen Leitfaden zum Zugriff auf Identitäten. Zudem sollten genügend Metadaten vorhanden sein, um vier Fragen zu beantworten: Aus welchem Datei stammt der Text, welcher Dienst besitzt sie, wo im Datei beginnt der jeweilige Abschnitt und welche Quell-IDs werden in der Empfehlung zitiert. Texte ohne Quellmetadaten sind schwer zu verarbeiten.
Die sechs Teile der Abrufung
Jeder Teil benötigt eine Definition, bevor er verwendet wird.
Dokument. Man kann sich ein Dokument als einen Textblock mit angehängten Etiketten vorstellen. Der Text kann aus einer Markdown-Datei, einem PDF, einer Webseite, einer Wiki oder sogar aus einer Zeile in einer Tabelle stammen, und die Etiketten können die Quell-ID, den Titel, den Dienst, den Pfad, den Besitzer, die Version oder die Personen angeben, die ihn einsehen dürfen.
Chunking. Dokumente werden in Teile aufgeteilt, die unabhängig voneinander gefunden werden können. Ein einziges Handbuch kann Diagnose, Wiederherstellung und Rollback gemeinsam beschreiben, doch eine Anfrage nach Datenbank-Zugangsdaten muss nicht jedes Mal das Rollback-Material zurückgeben.
Einbetten. Ein Einbettungsmodell wandelt Text in eine Liste von Zahlen um, wodurch Texte mit ähnlicher Bedeutung nach der gewählten Vergleichsmethode in der Nähe voneinander landen. Es handelt sich dabei nicht um eine Faktenprüfung. Ähnlicher Text kann dennoch falsch, veraltet oder unzugänglich sein.
Vektorlager. Ein Vektorlager speichert Dokumentenvektoren und findet ähnliche Vektoren. Einige befinden sich nur im Speicher, andere liegen in einer Datenbank oder einem gehosteten Service. Die Art und Weise, wie Vektoren gespeichert werden, ersetzt nicht die Zugriffskontrolle.
Auswertung. Geben Sie einem Retrieval-System einen freiformigen Abfragesatz, und es liefert Document-Objekte zurück. Es handelt sich dabei um ein weiter gefasstes Konzept als die Vektorsuche: Hinter derselben Schnittstelle können Schlüsselwortabgleich, ein Vektorlager, ein hybrider Suchdienst oder eine interne Such-API verwendet werden.
Generierung. In der letzten Stufe werden die ausgewählten Abschnitte verwendet, um eine strukturierte Antwort zu erzeugen. Was auch immer dort geschrieben wird, das System darf nicht vortäuschen, die Suche habe etwas zurückgegeben, was sie tatsächlich nicht getan hat.
Auswertung ist keine Speicherung
Ein Satz an Anleitungen stellt einen Wissensschatz dar, und die Abfrage desselben zählt als Auswertung. Chattenachrichten sind eine Sequenz von Nachrichten, gespeicherte Benutzerdaten sind gewöhnliche Anwendungsdaten, und Workflow-Checkpointe dokumentieren den Ausführungszustand. Ihre Aufbewahrungs- und Berechtigungsregeln unterscheiden sich alle, weshalb die nächste Stufe sie im Code voneinander trennt.
Projektstruktur und Einrichtung
Der Snapshot für diese Phase hat folgende Struktur:
langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│ ├── checkout-database.md
│ ├── identity-access.md
│ └── payment-latency.md
└── tests/
└── test_retrieval.py
Installieren Sie die Pakete:
python -m pip install -U \
langchain \
langchain-openai \
langchain-text-splitters \
pydantic \
python-dotenv \
pytest
Die Anwendung ruft Provider-Embeddings auf. Die Tests ersetzen diese durch einen kleinen lokalen Ersatz, sodass keine Anrufe bei den Providern stattfinden.
Schritt 1: Schreiben von identifizierbaren Runbooks
Das Checkout-Runbook beginnt mit einer Überschriftszeile:
# Checkout database connection failures
Danach folgen Service, Eigentümer und Fahrzeugtyp:
Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
previous successful deployment.
2. Confirm that the application secret points to the expected
credential version.
3. Test network reachability from the application environment without
printing the credential.
4. Compare connection-pool limits with the database connection limit.
Titel, Service und Eigentümer sind reiner Text, und der Lader kopiert außerdem die ausgewählten Werte in die Metadaten. In einem Unternehmenssystem darf niemals eine Überschrift den Zugriff bestimmen. Zugriffsbezeichnungen müssen aus einem vertrauenswürdigen System stammen und bei jeder Abfrage überprüft werden.
Schritt 2: Jedes Datei in ein Dokument laden
Fügen Sie load_runbooks in retrieval.py ein. Dazu benötigt man zunächst den Pfad-Hilfsfunktion:
from pathlib import Path
Dann der Lader selbst:
from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
documents: list[Document] = [] for path in sorted(runbook_dir.glob("*.md")):
text = path.read_text(encoding="utf-8")
title = text.splitlines()[0].removeprefix("# ").strip()
service_line = next(
line for line in text.splitlines()
if line.startswith("Service:")
)
service = service_line.split(":", 1)[1].strip() documents.append(
Document(
page_content=text,
metadata={
"source_id": path.stem,
"title": title,
"service": service,
"path": str(path),
},
)
) if not documents:
raise ValueError(f"No Markdown runbooks found in {runbook_dir}") return documents
Durchgegangen:
glob("*.md")wählt die Markdown-Dateien in einem bekannten Verzeichnis aus.sorted(...)sorgt dafür, dass die Reihenfolge der lokalen Indexierung reproduzierbar bleibt. Die Suchqualität sollte nicht von der Reihenfolge im Dateisystem abhängen, doch eine reproduzierbare Eingabe erleichtert das Debuggen.- Die Datei wird als UTF-8 decodiert.
- Das Herausfiltern des Titels und der Dienstinformationen wird absichtlich einfach gehalten, da die Beispieldateien ein festes Layout folgen. Ein echter Lader sollte Metadaten validieren und fehlerhafte Dateien melden anstatt zu raten.
page_contententhält den suchbaren Text.metadataenthält Felder, die in jedem nachfolgenden Datensatz vorhanden sein sollten.- Die abschließende Überprüfung verhindert, dass ein leerer Index wie ein Erfolg aussieht.
Schritt 3: Die Dokumente aufteilen
Verwenden Sie RecursiveCharacterTextSplitter:
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_runbooks(documents: list[Document]) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=420,
chunk_overlap=60,
add_start_index=True,
)
return splitter.split_documents(documents)
chunk_size=420 fordert Blöcke mit etwa 420 Zeichen an, was für dieses kleine Beispiel geeignet ist. chunk_overlap=60 wiederholt einen kleinen Text zwischen benachbarten Blöcken, wodurch ein Satz neben seiner Erklärung bleibt, falls sich eine Grenze genau in der Mitte befindet. Mehr Überschneidung bedeutet nicht automatisch etwas Besseres: Es führt zu mehr gespeichertem Text, mehr Verarbeitungsaufwand für die Embeddings und mehr nahezu identischen Ergebnissen.
Mit add_start_index=True wird für jeden Block der Zeichenoffset festgehalten, an dem er begann, sodass man auf das ursprüngliche Dateiinhalt zurückverweisen kann. split_documents gibt neue Document-Objekte zurück, und jeder Block erbt die Metadaten seines Vorgängers.
Auswahl der Blockgröße
Das Kopieren einer Zahl aus einem Tutorial ist keine Entscheidung. Schauen Sie zunächst darauf, wie Ihre Dokumente organisiert sind und was die Leute tatsächlich fragen. Ein Fehlerbehebungs-Schritt kann seinen Überschrift, seine Bedingungen und Warnhinweise in einem Block benötigen, während eine lange Richtlinientabelle möglicherweise einen strukturbewussten Trenner erfordert. Testen Sie anschließend die Abfrage anhand von beschrifteten Fragen.
Schritt 4: Erstellen des Vector-Store-Abfrageverfahrens
Um das Beispiel klein zu halten, verwendet die Anwendung InMemoryVectorStore:
from langchain_core.vectorstores import InMemoryVectorStore
def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
chunks = split_runbooks(load_runbooks(runbook_dir))
store = InMemoryVectorStore.from_documents(
chunks,
embedding=embeddings,
)
return store.as_retriever(search_kwargs={"k": k})
from_documents fügt die Fragmente ein und speichert sie, während as_retriever die Standard-Suchinterface bereitstellt. k=3 fordert drei Dokumente an. Das bedeutet nicht, dass alle drei relevant sind, und es garantiert auch nicht, dass sich der richtige Abschnitt darunter befindet.
Die Anwendung erstellt Provider-Embeddings:
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
Path(__file__).parent / "runbooks",
embeddings,
)
Die verfügbaren Modelle und ihre Namen hängen von Ihrem Anbieterkonto ab, daher sollten Sie den Modellnamen in der Konfiguration speichern. Zudem verliert InMemoryVectorStore seinen Index, wenn der Prozess beendet wird. Das eignet sich gut für Lehrbeispiele und kleine Tests, ist aber kein gemeinsamer Produktionsindex.
Schritt 5: Erstellen Sie eine gezielte Abfrage aus zuverlässigen Feldern
Suchen Sie nicht mit einer riesigen Menge an Daten des gesamten Anwendungszustands. Wählen Sie stattdessen ausgewählte Klassifizierungsfelder:
def build_query(classification: TicketClassification) -> str:
values = [
classification.service or "unknown service",
classification.category,
classification.summary,
*classification.evidence,
]
return " | ".join(values)
Für das Kontinuierlichkeits-Ticket sieht die Abfrage ungefähr so aus:
checkout-api | database | EU checkout fails after a deployment. |
database connection refused
Der String enthält den wahrscheinlichen Service, die Problemkategorie, eine kurze Zusammenfassung sowie die aus dem Ticket entnommenen Beweise. Geheime Informationen, versteckte Anweisungen sowie die gesamte Nachrichtenhistorie bleiben dabei unberücksichtigt.
Schritt 6: Übertragung der Metadaten während der Konvertierung
Was der Extraktor zurückgibt, sind Document-Objekte, von denen jedes zu einem typisierten Abschnitt werden sollte:
class RetrievedPassage(BaseModel):
source_id: str
title: str
service: str
start_index: int
text: str
Führen Sie dann eine Zuordnung des Document-Objekts dazu durch:
def to_passage(document: Document) -> RetrievedPassage:
return RetrievedPassage(
source_id=str(document.metadata["source_id"]),
title=str(document.metadata["title"]),
service=str(document.metadata["service"]),
start_index=int(document.metadata.get("start_index", 0)),
text=document.page_content.strip(),
)
Die erforderlichen Felder werden mit eckigen Klammern angegeben, sodass ein Datensatz ohne source_id, title oder service sofort einen Fehler auslöst. start_index erhält einen Standardwert, da früher gespeicherte Datensätze diesen möglicherweise nicht enthalten, und eine Produktionsmigration sollte entscheiden, ob dieser Ersatz akzeptabel ist.
Schritt 7: Erstellung einer präzisen Empfehlungs-Eingabe
Der an das Modell gerichtete Writer erhält diesen Vertrag:
class RecommendationInput(BaseModel):
ticket_id: str
ticket_summary: str
category: Category
severity: Severity
service: str | None
evidence: list[str]
passages: list[RetrievedPassage] = Field(
min_length=1,
max_length=4,
)
Falls keine Ergebnisse zurückkommen, gibt es keine fundierte Empfehlung:
def prepare_recommendation_input(
classification: TicketClassification,
passages: list[RetrievedPassage],
) -> RecommendationInput:
if not passages:
raise ValueError("At least one runbook passage is required")
return RecommendationInput(
ticket_id=classification.ticket_id,
ticket_summary=classification.summary,
category=classification.category,
severity=classification.severity,
service=classification.service,
evidence=classification.evidence,
passages=passages[:4],
)
Falls die Suche nichts ergibt, sollte das Modell nicht dazu gezwungen werden, aus dem allgemeinen Wissen, das es besitzt, improvisieren zu müssen. Leiten Sie den Ticket an eine Person weiter oder verwenden Sie einen eigenständig genehmigten Ersatzweg.
Schritt 8: Anfrage nach Quell-IDs in strukturierter Ausgabe
Das Ergebnisschema enthält Verweise:
class GroundedRecommendation(BaseModel):
explanation: str = Field(min_length=20, max_length=500)
next_checks: list[str] = Field(min_length=1, max_length=4)
source_ids: list[str] = Field(min_length=1, max_length=4)
needs_human_review: bool
Die Anweisung lautet:
Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.
Anleitungen zum Prompt helfen zwar, aber Überprüfungen auf Anwendungsseite sind dennoch wichtig.
Schritt 9: Zitate ablehnen, die bei der Suche nicht zurückgegeben wurden
Sobald der Autor die Suche durchgeführt hat, vergleicht er die Quell-IDs mit den gefundenen Abschnitten:
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError(
"Recommendation cited sources that retrieval did not return: "
+ ", ".join(sorted(unknown))
)
Dies erfasst nur eine gefälschte Quell-ID und nichts weiter. Ein Modell kann auf ein tatsächlich zurückgegebenes Dokument verweisen und dennoch dessen Bedeutung falsch interpretieren, weshalb eine Bewertung sowie menschliche Überprüfung weiterhin erforderlich sind.
Schritt 10: Den korrigierten Workflow verbinden
Der Code zum Zeitpunkt der Anfrage ist einfach:
def run_retrieval_workflow(
classification: TicketClassification,
retriever,
recommendation_chain,
) -> RetrievalResult:
query = build_query(classification)
passages = retrieve_passages(retriever, query)
values = prepare_recommendation_input(classification, passages)
recommendation = recommendation_chain.invoke(values)
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError("Recommendation cited an unavailable source") return RetrievalResult(
query=query,
recommendation=recommendation,
passages=passages,
)
Es gibt keine versteckte Schleife. Die Suche wird einmal ausgeführt, der Writer einmal und die Quellcode-Prüfung ebenfalls einmal.
Was herauskommt
Ein Ergebnis sieht so aus:
{
"query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
"recommendation": {
"explanation": "The error matches the checkout database connection runbook.",
"next_checks": [
"Compare the active database host, port and database name with the previous successful deployment.",
"Confirm that the application secret points to the expected credential version."
],
"source_ids": ["checkout-database"],
"needs_human_review": false
},
"passages": [
{
"source_id": "checkout-database",
"title": "Checkout database connection failures",
"service": "checkout-api",
"start_index": 0
}
]
}
Die API kann jede Quell-ID in einen Link umwandeln, aber nur nachdem überprüft wurde, dass der aktuelle Aufrufer ihn öffnen darf.
Testen des Abrufs ohne Bezahlung für Embeddings
Das begleitende Snapshot enthält KeywordEmbeddings, ein kleines Test-Ersatzprodukt, das ein festes Vokabular verwendet:
VOCABULARY = (
"database",
"connection",
"checkout",
"deployment",
"credential",
"pool",
"payment",
"latency",
"timeout",
"identity",
"access",
"role",
"claim",
)
Verwechseln Sie es nicht mit einem Produkt-Embedder. Sein einziger Zweck ist es, Tests schnell, offline und deterministisch durchzuführen. Der erste Test überprüft das laufende Ticket:
passages = retrieve_passages(
retriever,
build_query(classification()),
)
assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0
Die zweiten Messwerte trafen bei zwei von drei beschrifteten Abfragen zu:
score = hit_at_k(
retriever,
[
("checkout database connection refused", "checkout-database"),
("payment provider latency timeout", "payment-latency"),
("identity role claim access denied", "identity-access"),
],
)
assert score == 1.0
Zwei Testfälle werfen eine kleine Frage auf: Liegt die erwartete Quelle unter den ersten beiden Ergebnissen? Ein echtes Testset benötigt viele Fragen, schwierige Negative, ähnliche Dokumente sowie aktuelle Zugriffsregeln. Eine perfekte Punktzahl bei drei Beispielen beweist lediglich, dass diese drei bestehen.
Im dritten Testfall wird eine falsche Empfehlung mit dem Hinweis not-returned eingegeben, und der Workflow muss sie ablehnen. Führen Sie alles mit folgendem aus:
pytest -q
Das erwartete Ergebnis:
3 passed
Benennung der Fehler beim Abrufen
Jeder Fehlertyp verdient einen eigenen Namen, damit man weiß, wo man suchen muss.
- Indexierungsfehler: Eine Datei konnte nicht gelesen werden, Metadaten fehlten oder eine Einbettungsanfrage scheiterte. Veröffentlichen Sie niemals einen unvollständig erstellten Index als vollständig.
- Frische-Fehler: Eine Richtlinie hat sich geändert, doch die Suche lieferte weiterhin das vorherige Handbuch. Erfassen Sie Versionen und Zeitstempel und entscheiden Sie, wie Änderungen und Löschungen im Index berücksichtigt werden sollen.
- Berechtigungsfehler: Der Index zeigte einen Abschnitt an, den der Aufrufer nicht lesen darf. Wenden Sie Zugriffsfilter an, bevor jeglicher Text das Modell erreicht – denn das Verbergen des Links danach ist zu spät.
- Rangierungsfehler: Das richtige Dokument ist im Index enthalten, fehlt aber in den oberen Ergebnissen. Passen Sie die Abfrage, Filter, Aufteilung in Blöcke, Einbettungen oder die Rangierung an und messen Sie erneut.
Der abgerufte Text ist unzuverlässig
Runbooks enthalten Fehler und manchmal Zeilen, die wie Anweisungen an das Modell wirken. Behandeln Sie die abgerufenen Abschnitte als zitierte Quellenmaterialien und niemals als Anwendungsregeln mit höherer Priorität. Bewahren Sie ein stabiles Verhalten im Systemprompt, begrenzen Sie den abgerufenen Text und beschränken Sie die Tools im Anwendungscode. Kein Runbook sollte jemals in der Lage sein, sich selbst zu autorisieren, um ein schreibendes Tool auszulösen.
Wenn die Vektorsuche das falsche Werkzeug ist
Manche Abfragen benötigen keine Vektoren. Eine genaue Ticket-ID wird durch eine einfache Datenbankabfrage ermittelt, und ein bekannter Dienstname führt direkt zu einem Eintrag im Katalog. Genaue Fehlercodes sowie seltene Identifikatoren lassen sich besser mit einer Schlüsselwortsuche finden. Wenn das Unternehmen bereits einen Suchdienst nutzt, der die richtigen Dokumente indiziert und Zugriffsrechte überwacht, sollte dieser verwendet werden. Die Vektorsuche ist nur eines von mehreren Werkzeugen.
Was fehlt noch für die Produktion
Bevor ein Unternehmen sich auf diesen Ansatz verlassen kann, fehlen noch einige Dinge:
- eine dauerhafte Speicherung anstelle des Arbeitsspeichers
- eine klare Zuordnung jedes Dokuments zu einem Verantwortlichen
- Zugriffsfilter, die zur Abfragezeit angewendet werden
- Einrichtungen zur Handhabung von Dokumentversionen und Löschungen
- Ein Plan für den Fall, dass sich das Embedding-Modell ändert
- Ein Limit dafür, wie viel Text zurückgegeben wird
- Etikettierte Tests zur Überprüfung der Suchfunktion
Das Framework verbindet die Komponenten, und die Anwendung trifft die Entscheidungen.
Der Stand der Helpdesk-App
Sie kann nun in einer kontrollierten Sammlung von Anleitungen suchen und ihre Quellen beibehalten:
classification
-> focused query
-> retriever
-> typed passages with metadata
-> recommendation
-> source-ID check
Als Nächstes wird die Nachrichtenhistorie, die gespeicherten Benutzerdaten, der aktuelle Workflow-Zustand sowie die Kontrollpunkte in getrennte Komponenten aufgeteilt. Durch diese Trennung wird außerdem klar, ob eine einfache LangChain-Kette ausreicht oder LangGraph besser geeignet ist.
Weitere Literatur:
- https://docs.langchain.com/oss/python/langchain/retrieval
- https://docs.langchain.com/oss/python/langchain/knowledge-base
- https://docs.langchain.com/oss/python/integrations/splitters/index
- https://docs.langchain.com/oss/python/integrations/retrievers