LangChain Persönliches Wissensdatenbank-Frage-und-Antwort-System
Dieser Abschnitt erstellt ein persönliches Wissensdatenbank-System, das Markdown-Dateien und PDF-Dokumente laden und Fragen basierend auf deren Inhalt beantworten kann.
Systemdesign
- Dokumentenladen: Unterstützt mehrere Formate, einschließlich Markdown, TXT und PDF
- Vektorabruf: Chroma persistenter Speicher, unterstützt inkrementelle Aktualisierungen
- Quellenangaben: Antworten enthalten das Quelldokument und die Position des Textausschnitts
- Streaming-Ausgabe: Zeigt die Antwort Token für Token an
Vollständiger Code
Vor der Ausführung müssen Sie Folgendes konfigurieren.envin der Datei:DEEPSEEK_API_KEY(Chat-Modell) undDASHSCOPE_API_KEY(Alibaba Cloud Bailian, verwendet für die Textvektorisierung der Wissensdatenbank). Für spezifische Antragsverfahren und allgemeine Fehlerbehebung sieheden "LangChain Intelligenter Kundenservice-Roboter"-Abschnitt weggelassen werden.
Beispiel
# Dateipfad: knowledge_qa.py
# pip install langchain langchain-deepseek langchain-openai langchain-community langchain-chroma chromadb pypdf
from dotenv import load_dotenv
load_dotenv()
import os
from pathlib import Path
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
class KnowledgeBase:
"""Manager der persönlichen Wissensdatenbank"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# Die Embedding-Schnittstelle von Alibaba Cloud Bailian ist mit der OpenAI-Spezifikation kompatibel; rufe sie einfach mit langchain-openai auf
# 's OpenAIEmbeddings — keine Notwendigkeit, die DashScopeEmbeddings in langchain-community zu installieren,
# welche nicht mehr gepflegt wird.
# check_embedding_ctx_length=False: schaltet die tiktoken-Vor-Tokenisierung ab und sendet den Originaltext direkt
# (Die Schnittstelle von Bailian akzeptiert keine Token-ID-Arrays).
# chunk_size=10: Die Embedding-Schnittstelle von Bailian akzeptiert höchstens 10 Texte pro Anfrage.
self.embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " "],
)
self.vector_store = None
self._load_or_create()
def _load_or_create(self):
"""Lade einen vorhandenen Vektorspeicher oder erstelle einen neuen (Chroma funktioniert mit demselben Parametersatz;
ob das Verzeichnis bereits existiert, wirkt sich nur auf die Prompt-Nachricht aus, nicht auf die Erstellungslogik)"""
is_existing = os.path.exists(self.persist_dir) and os.listdir(self.persist_dir)
self.vector_store = Chroma(
embedding_function=self.embeddings,
persist_directory=self.persist_dir,
)
if is_existing:
print(f"Vektorspeicher geladen: {self.vector_store._collection.count()} Dokument-Chunks")
else:
print("Neuer Vektorspeicher erstellt")
def add_file(self, file_path: str) -> int:
"""Fügt eine Datei zur Wissensdatenbank hinzu und gibt die Anzahl der hinzugefügten Dokumentabschnitte zurück.
Automatisch den passenden Loader basierend auf der Dateiendung auswählen:
- Für .pdf PyPDFLoader verwenden (hängt von der pypdf-Bibliothek zum Parsen von PDFs ab, gibt mehrere Documents pro Seite zurück)
- Für den Rest (.md, .txt und andere reine Textdateien) TextLoader verwenden, um den Rohtext zu lesen
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Einheitlich den Dateinamen als Quelle hinzufügen; jedes von PyPDFLoader geladene Document verfügt außerdem über
# ein page-Feld (Seitennummer), das auch verwendet werden kann, um die Snippet-Position beim Abruf zu lokalisieren
for doc in docs:
doc.metadata["source"] = Path(file_path).name
chunks = self.text_splitter.split_documents(docs)
self.vector_store.add_documents(chunks)
print(f"Hinzugefügt {Path(file_path).name}: {len(chunks)} Dokument-Chunks")
return len(chunks)
def add_text(self, text: str, source: str = "Manuell hinzufügen") -> int:
"""Text direkt zur Wissensbasis hinzufügen"""
chunks = self.text_splitter.create_documents(
[text], metadatas=[{"source": source}]
)
self.vector_store.add_documents(chunks)
return len(chunks)
def search(self, query: str, k: int = 3) -> Liste:
"""Die Wissensbasis durchsuchen"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Retriever abrufen"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
# pip install langchain langchain-deepseek langchain-openai langchain-community langchain-chroma chromadb pypdf
from dotenv import load_dotenv
load_dotenv()
import os
from pathlib import Path
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
class KnowledgeBase:
"""Manager der persönlichen Wissensdatenbank"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# Die Embedding-Schnittstelle von Alibaba Cloud Bailian ist mit der OpenAI-Spezifikation kompatibel; rufe sie einfach mit langchain-openai auf
# 's OpenAIEmbeddings — keine Notwendigkeit, die DashScopeEmbeddings in langchain-community zu installieren,
# welche nicht mehr gepflegt wird.
# check_embedding_ctx_length=False: schaltet die tiktoken-Vor-Tokenisierung ab und sendet den Originaltext direkt
# (Die Schnittstelle von Bailian akzeptiert keine Token-ID-Arrays).
# chunk_size=10: Die Embedding-Schnittstelle von Bailian akzeptiert höchstens 10 Texte pro Anfrage.
self.embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " "],
)
self.vector_store = None
self._load_or_create()
def _load_or_create(self):
"""Lade einen vorhandenen Vektorspeicher oder erstelle einen neuen (Chroma funktioniert mit demselben Parametersatz;
ob das Verzeichnis bereits existiert, wirkt sich nur auf die Prompt-Nachricht aus, nicht auf die Erstellungslogik)"""
is_existing = os.path.exists(self.persist_dir) and os.listdir(self.persist_dir)
self.vector_store = Chroma(
embedding_function=self.embeddings,
persist_directory=self.persist_dir,
)
if is_existing:
print(f"Vektorspeicher geladen: {self.vector_store._collection.count()} Dokument-Chunks")
else:
print("Neuer Vektorspeicher erstellt")
def add_file(self, file_path: str) -> int:
"""Fügt eine Datei zur Wissensdatenbank hinzu und gibt die Anzahl der hinzugefügten Dokumentabschnitte zurück.
Automatisch den passenden Loader basierend auf der Dateiendung auswählen:
- Für .pdf PyPDFLoader verwenden (hängt von der pypdf-Bibliothek zum Parsen von PDFs ab, gibt mehrere Documents pro Seite zurück)
- Für den Rest (.md, .txt und andere reine Textdateien) TextLoader verwenden, um den Rohtext zu lesen
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Einheitlich den Dateinamen als Quelle hinzufügen; jedes von PyPDFLoader geladene Document verfügt außerdem über
# ein page-Feld (Seitennummer), das auch verwendet werden kann, um die Snippet-Position beim Abruf zu lokalisieren
for doc in docs:
doc.metadata["source"] = Path(file_path).name
chunks = self.text_splitter.split_documents(docs)
self.vector_store.add_documents(chunks)
print(f"Hinzugefügt {Path(file_path).name}: {len(chunks)} Dokument-Chunks")
return len(chunks)
def add_text(self, text: str, source: str = "Manuell hinzufügen") -> int:
"""Text direkt zur Wissensbasis hinzufügen"""
chunks = self.text_splitter.create_documents(
[text], metadatas=[{"source": source}]
)
self.vector_store.add_documents(chunks)
return len(chunks)
def search(self, query: str, k: int = 3) -> Liste:
"""Die Wissensbasis durchsuchen"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Retriever abrufen"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
Derzeit,
TextLoader、PyPDFLoaderDieser Typ von lokalem Dateiloader hat noch kein eigenständig gepflegtes Paket; er kann weiterhin nurlangchain_community.document_loadersaus langchain_community importiert werden. Zur Laufzeit kann eine DeprecationWarning für das gesamte langchain-community-Paket angezeigt werden — diese Warnung kann vorerst ignoriert werden; migrieren Sie einfach, nachdem das offizielle eigenständige Dokumentloader-Paket veröffentlicht wurde. Dies unterscheidet sich vom Embedding-Fall oben, bei dem bereits ein fertiger Ersatz existierte, aber nicht übernommen wurde.
Weiter mit dem Agent-Abschnitt:
Beispiel
# ========== Wissensdatenbank erstellen und Beispieldaten hinzufügen ==========
kb = KnowledgeBase("./my_knowledge_db")
# Beispielwissen hinzufügen
kb.add_text(
"Das Python3-Grundlagentutorial von EXAMPLE enthält die folgenden Kapitel:"
"1. Python-Einführung und Umgebungseinrichtung 2. Grundlegende Datentypen 3. Operatoren und Ausdrücke "
"4. Bedingte Anweisungen if-else 5. Schleifen for/while 6. Funktionsdefinition und Aufruf "
"7. Module und Pakete 8. Dateioperationen 9. Ausnahmebehandlung 10. Objektorientierte Programmierung",
source="Python3-Tutorial-Überblick"
)
kb.add_text(
"Um ein guter Python-Entwickler zu werden, wird empfohlen, dem folgenden Lernpfad zu folgen:"
"Zuerst die Python-Grundsyntax beherrschen (1-2 Wochen);"
"Zweitens, die Grundlagen von Datenstrukturen und Algorithmen lernen (2-3 Wochen);"
"Drittens, wähle eine Richtung für ein vertieftes Studium (Webentwicklung/Datenanalyse/KI);"
"Viertens, arbeite an 2-3 praktischen Projekten, um dein Wissen zu festigen.",
source="Python-Lernpfad"
)
kb.add_text(
"Die Online-Programmierumgebung von EXAMPLE unterstützt mehrere Sprachen, darunter Python, JavaScript, Java und C++."
"Benutzer müssen keine Software installieren; öffnen Sie einfach einen Browser, um Code zu schreiben und auszuführen."
"Die Online-Umgebung unterstützt außerdem Code-Hervorhebung, automatische Vervollständigung und Fehlermeldungen.",
source="Beschreibung der Online-Programmierumgebung"
)
# Sie können auch lokale Dateien laden; PDF und Markdown/TXT werden automatisch erkannt:
# kb.add_file("./docs/Produkthandbuch.pdf")
# kb.add_file("./docs/FAQ.md")
# ========== RAG-Agent erstellen ==========
@tool
def search_knowledge(query: str) -> str:
"""Durchsuchen Sie die persönliche Wissensdatenbank nach relevanten Informationen. Verwenden Sie bei der Suche die vollständige Frage oder Schlüsselbegriffe.
Argumente:
query: die Suchfrage oder die Schlüsselphrase
"""
docs = kb.search(query, k=3)
if not docs:
return "Keine relevanten Informationen in der Wissensdatenbank gefunden."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Unbekannte Quelle")
page = doc.metadata.get("page")
location = f"{source}" + (f" Seite {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Quelle: {location}\n{content}")
return "\n\n---\n\n".join(results)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_knowledge],
system_prompt="""Sie sind ein persönlicher Wissensdatenbank-Assistent.
## Regeln
1. Alle Fragen müssen zuerst das Tool search_knowledge verwenden, um die Wissensdatenbank zu durchsuchen
2. Geben Sie bei der Beantwortung die Quelle der Informationen an (Dokumentname und Seitenzahl, wenn es sich um ein PDF handelt)
3. Wenn kein relevanter Inhalt in der Wissensbasis vorhanden ist, sage es dem Benutzer wahrheitsgemäß
4. Strukturiere deine Antworten mit nummerierten Listen oder Absätzen""",
)
# ========== Test: Nicht-Streaming, und den abgerufenen Inhalt gleichzeitig anzeigen ==========
def ask(question: str):
"""Stelle eine Frage und gib sowohl die abgerufenen Roh-Snippets als auch die endgültige Antwort zum einfachen Debuggen aus"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Abgerufenen Inhalt anzeigen
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Abgerufener Inhalt]")
print(msg.content[:300])
print(f"\n[Antwort]")
print(result["messages"][-1].content)
# ========== Test: Streaming, entspricht der "Streaming-Ausgabe" im Systemdesign ==========
def ask_stream(question: str):
"""Stelle eine Frage und gib die Antwort Token für Token auf Streaming-Weise aus, wodurch ein Schreibmaschineneffekt erzielt wird"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[Antwort] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" gibt an, dass dieser Chunk vom modellgenerierten
# finaler Antwortknoten, wobei andere Arten von Chunks wie Tool-Aufrufe herausgefiltert werden
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Welche Kapitel enthält das Python3-Grundlagentutorial?")
ask("Wie sollte ich meinen Python-Lernpfad planen?")
ask_stream("Welche Funktionen unterstützt die Online-Programmierumgebung von EXAMPLE?")
kb = KnowledgeBase("./my_knowledge_db")
# Beispielwissen hinzufügen
kb.add_text(
"Das Python3-Grundlagentutorial von EXAMPLE enthält die folgenden Kapitel:"
"1. Python-Einführung und Umgebungseinrichtung 2. Grundlegende Datentypen 3. Operatoren und Ausdrücke "
"4. Bedingte Anweisungen if-else 5. Schleifen for/while 6. Funktionsdefinition und Aufruf "
"7. Module und Pakete 8. Dateioperationen 9. Ausnahmebehandlung 10. Objektorientierte Programmierung",
source="Python3-Tutorial-Überblick"
)
kb.add_text(
"Um ein guter Python-Entwickler zu werden, wird empfohlen, dem folgenden Lernpfad zu folgen:"
"Zuerst die Python-Grundsyntax beherrschen (1-2 Wochen);"
"Zweitens, die Grundlagen von Datenstrukturen und Algorithmen lernen (2-3 Wochen);"
"Drittens, wähle eine Richtung für ein vertieftes Studium (Webentwicklung/Datenanalyse/KI);"
"Viertens, arbeite an 2-3 praktischen Projekten, um dein Wissen zu festigen.",
source="Python-Lernpfad"
)
kb.add_text(
"Die Online-Programmierumgebung von EXAMPLE unterstützt mehrere Sprachen, darunter Python, JavaScript, Java und C++."
"Benutzer müssen keine Software installieren; öffnen Sie einfach einen Browser, um Code zu schreiben und auszuführen."
"Die Online-Umgebung unterstützt außerdem Code-Hervorhebung, automatische Vervollständigung und Fehlermeldungen.",
source="Beschreibung der Online-Programmierumgebung"
)
# Sie können auch lokale Dateien laden; PDF und Markdown/TXT werden automatisch erkannt:
# kb.add_file("./docs/Produkthandbuch.pdf")
# kb.add_file("./docs/FAQ.md")
# ========== RAG-Agent erstellen ==========
@tool
def search_knowledge(query: str) -> str:
"""Durchsuchen Sie die persönliche Wissensdatenbank nach relevanten Informationen. Verwenden Sie bei der Suche die vollständige Frage oder Schlüsselbegriffe.
Argumente:
query: die Suchfrage oder die Schlüsselphrase
"""
docs = kb.search(query, k=3)
if not docs:
return "Keine relevanten Informationen in der Wissensdatenbank gefunden."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Unbekannte Quelle")
page = doc.metadata.get("page")
location = f"{source}" + (f" Seite {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Quelle: {location}\n{content}")
return "\n\n---\n\n".join(results)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_knowledge],
system_prompt="""Sie sind ein persönlicher Wissensdatenbank-Assistent.
## Regeln
1. Alle Fragen müssen zuerst das Tool search_knowledge verwenden, um die Wissensdatenbank zu durchsuchen
2. Geben Sie bei der Beantwortung die Quelle der Informationen an (Dokumentname und Seitenzahl, wenn es sich um ein PDF handelt)
3. Wenn kein relevanter Inhalt in der Wissensbasis vorhanden ist, sage es dem Benutzer wahrheitsgemäß
4. Strukturiere deine Antworten mit nummerierten Listen oder Absätzen""",
)
# ========== Test: Nicht-Streaming, und den abgerufenen Inhalt gleichzeitig anzeigen ==========
def ask(question: str):
"""Stelle eine Frage und gib sowohl die abgerufenen Roh-Snippets als auch die endgültige Antwort zum einfachen Debuggen aus"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Abgerufenen Inhalt anzeigen
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Abgerufener Inhalt]")
print(msg.content[:300])
print(f"\n[Antwort]")
print(result["messages"][-1].content)
# ========== Test: Streaming, entspricht der "Streaming-Ausgabe" im Systemdesign ==========
def ask_stream(question: str):
"""Stelle eine Frage und gib die Antwort Token für Token auf Streaming-Weise aus, wodurch ein Schreibmaschineneffekt erzielt wird"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[Antwort] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" gibt an, dass dieser Chunk vom modellgenerierten
# finaler Antwortknoten, wobei andere Arten von Chunks wie Tool-Aufrufe herausgefiltert werden
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Welche Kapitel enthält das Python3-Grundlagentutorial?")
ask("Wie sollte ich meinen Python-Lernpfad planen?")
ask_stream("Welche Funktionen unterstützt die Online-Programmierumgebung von EXAMPLE?")
Run result:
============================================================ Q: Python3 基础教程包含哪些章节？ ============================================================ [检索到的内容] [1] 来源：Python3 教程大纲 Example 的 Python3 基础教程包含以下章节：... [回答] Python3 基础教程包含以下章节（来源：Python3 教程大纲）： 1. Python 简介与环境搭建 2. 基本数据类型 3. 运算符与表达式 ... ============================================================ Q: 如何规划 Python 学习路线？ ============================================================ [回答] 根据知识库中的 Python 学习路线建议（来源：Python 学习路线）： 第一步：掌握基础语法（1-2 周） 第二步：学习数据结构和算法（2-3 周） 第三步：选择方向深入学习（Web/数据分析/AI） 第四步：做 2-3 个实战项目巩固 ============================================================ Q: Example的在线编程环境支持哪些功能？ ============================================================ [回答] 根据知识库中的说明（来源：在线编程环境说明），Example的在线编程环境： 1. 支持 Python、JavaScript、Java、C++ 等多种语言 2. 无需安装任何软件，打开浏览器即可编写和运行代码 3. 支持代码高亮、自动补全和错误提示功能
Andere ErweiterungenDie letzte Frage verwendet
ask_stream()Streaming-Ausgabe, sodass zur Laufzeit der Text nach "[Answer]" jeweils ein Zeichen (oder einige Zeichen pro Batch) ausgegeben wird, ganz wie bei einem Schreibmaschineneffekt. Oben wurde zur bequemen Anzeige der endgültige vollständige Text nach der Ausgabe direkt eingefügt.