Système de questions-réponses sur base de connaissances personnelle avec LangChain
Cette section construit un système de base de connaissances personnelle capable de charger des fichiers Markdown et des documents PDF et de répondre aux questions en fonction de leur contenu.
Conception de systèmes
- Chargement des documents: Prend en charge plusieurs formats, notamment Markdown, TXT et PDF
- Récupération vectorielle: Stockage persistant Chroma, prend en charge les mises à jour incrémentielles
- Citations des sources: Les réponses incluent le document source et l'emplacement de l'extrait
- Sortie en flux: Affiche la réponse jeton par jeton
Code complet
Avant l'exécution, vous devez configurer les éléments suivants.envdans le fichier :DEEPSEEK_API_KEY(modèle de chat) etDASHSCOPE_API_KEY(Alibaba Cloud Bailian, utilisé pour la vectorisation de texte de la base de connaissances). Pour les procédures de demande spécifiques et le dépannage courant, voirle « Robot de service client intelligent LangChain »section.
Exemple
# Chemin du fichier : 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:
"""Gestionnaire de base de connaissances personnelle"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# L'interface Embedding d'Alibaba Cloud Bailian est compatible avec la spécification OpenAI ; il suffit de l'appeler avec langchain-openai
# OpenAIEmbeddings — pas besoin d'installer les DashScopeEmbeddings dans langchain-community,
# qui n'est plus maintenu.
# check_embedding_ctx_length=False : désactiver la pré-tokenisation tiktoken et envoyer directement le texte original
# (l'interface de Bailian n'accepte pas les tableaux d'ID de token).
# chunk_size=10 : l'interface Embedding de Bailian accepte au maximum 10 textes par requête.
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):
"""Charger un magasin vectoriel existant ou en créer un nouveau (Chroma fonctionne avec le même ensemble de paramètres ;
le fait que le répertoire existe déjà n'affecte que le message d'invite, pas la logique de création)"""
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"Magasin vectoriel chargé : {self.vector_store._collection.count()} segments de document")
else:
print("Nouveau magasin vectoriel créé")
def add_file(self, file_path: str) -> int:
"""Ajouter un fichier à la base de connaissances et renvoyer le nombre de segments de document ajoutés.
Choisir automatiquement le Loader approprié en fonction de l'extension du fichier :
- Pour .pdf, utiliser PyPDFLoader (dépend de la bibliothèque pypdf pour analyser les PDF, renvoie plusieurs Documents par page)
- Pour le reste (.md, .txt et autres fichiers texte brut), utiliser TextLoader pour lire le texte brut
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Ajouter uniformément le nom du fichier comme source ; chaque Document chargé par PyPDFLoader possède également
# un champ page (numéro de page), qui peut aussi servir à localiser la position de l'extrait lors de la récupération
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"Ajout de {Path(file_path).name}: {len(chunks)} segments de document")
return len(chunks)
def add_text(self, text: str, source: str = "Ajouter manuellement") -> int:
"""Ajouter directement du texte à la base de connaissances"""
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:
"""Rechercher dans la base de connaissances"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Obtenir le récupérateur"""
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:
"""Gestionnaire de base de connaissances personnelle"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# L'interface Embedding d'Alibaba Cloud Bailian est compatible avec la spécification OpenAI ; il suffit de l'appeler avec langchain-openai
# OpenAIEmbeddings — pas besoin d'installer les DashScopeEmbeddings dans langchain-community,
# qui n'est plus maintenu.
# check_embedding_ctx_length=False : désactiver la pré-tokenisation tiktoken et envoyer directement le texte original
# (l'interface de Bailian n'accepte pas les tableaux d'ID de token).
# chunk_size=10 : l'interface Embedding de Bailian accepte au maximum 10 textes par requête.
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):
"""Charger un magasin vectoriel existant ou en créer un nouveau (Chroma fonctionne avec le même ensemble de paramètres ;
le fait que le répertoire existe déjà n'affecte que le message d'invite, pas la logique de création)"""
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"Magasin vectoriel chargé : {self.vector_store._collection.count()} segments de document")
else:
print("Nouveau magasin vectoriel créé")
def add_file(self, file_path: str) -> int:
"""Ajouter un fichier à la base de connaissances et renvoyer le nombre de segments de document ajoutés.
Choisir automatiquement le Loader approprié en fonction de l'extension du fichier :
- Pour .pdf, utiliser PyPDFLoader (dépend de la bibliothèque pypdf pour analyser les PDF, renvoie plusieurs Documents par page)
- Pour le reste (.md, .txt et autres fichiers texte brut), utiliser TextLoader pour lire le texte brut
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Ajouter uniformément le nom du fichier comme source ; chaque Document chargé par PyPDFLoader possède également
# un champ page (numéro de page), qui peut aussi servir à localiser la position de l'extrait lors de la récupération
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"Ajout de {Path(file_path).name}: {len(chunks)} segments de document")
return len(chunks)
def add_text(self, text: str, source: str = "Ajouter manuellement") -> int:
"""Ajouter directement du texte à la base de connaissances"""
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:
"""Rechercher dans la base de connaissances"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Obtenir le récupérateur"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
Actuellement,
TextLoader、PyPDFLoaderce type de chargeur de fichier local n'a pas encore de paquet autonome maintenu ; il ne peut encore êtrelangchain_community.document_loadersimporté que depuis langchain_community. À l'exécution, vous pouvez voir un DeprecationWarning pour l'ensemble du paquet langchain-community — cet avertissement peut être ignoré pour l'instant ; il suffit de migrer après la publication du paquet officiel autonome de chargeur de documents. Cela diffère du cas d'embedding ci-dessus, où un remplacement prêt à l'emploi existait déjà mais n'a pas été adopté.
Continuons avec la section Agent :
Exemple
# ========== Créer la base de connaissances et ajouter des données d'exemple ==========
kb = KnowledgeBase("./my_knowledge_db")
# Ajouter quelques connaissances d'exemple
kb.add_text(
"Le tutoriel Python3 de base d'EXAMPLE contient les chapitres suivants :"
"1. Introduction à Python et configuration de l'environnement 2. Types de données de base 3. Opérateurs et expressions "
"4. Instructions conditionnelles if-else 5. Boucles for/while 6. Définition et appel de fonctions "
"7. Modules et packages 8. Opérations sur les fichiers 9. Gestion des exceptions 10. Programmation orientée objet",
source="Plan du tutoriel Python3"
)
kb.add_text(
"Pour devenir un bon développeur Python, il est recommandé d'étudier selon le parcours suivant :"
"Premièrement, maîtrisez la syntaxe de base de Python (1-2 semaines);"
"Deuxièmement, apprenez les bases des structures de données et des algorithmes (2-3 semaines);"
"Troisièmement, choisissez une direction pour une étude approfondie (développement Web/analyse de données/IA);"
"Quatrièmement, travaillez sur 2-3 projets pratiques pour consolider vos connaissances.",
source="Parcours d'apprentissage de Python"
)
kb.add_text(
"L'environnement de programmation en ligne d'EXAMPLE prend en charge plusieurs langages, notamment Python, JavaScript, Java et C++."
"Les utilisateurs n'ont besoin d'installer aucun logiciel ; il suffit d'ouvrir un navigateur pour écrire et exécuter du code."
"L'environnement en ligne prend également en charge la coloration syntaxique, l'auto-complétion et les indications d'erreur.",
source="Description de l'environnement de programmation en ligne"
)
# Vous pouvez également charger des fichiers locaux ; les PDF et Markdown/TXT sont automatiquement reconnus :
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== Créer un agent RAG ==========
@tool
def search_knowledge(query: str) -> str:
"""Recherchez dans la base de connaissances personnelle des informations pertinentes. Utilisez la question complète ou des expressions clés lors de la recherche.
Arguments :
query: la question de recherche ou l'expression clé
"""
docs = kb.search(query, k=3)
if not docs:
return "Aucune information pertinente trouvée dans la base de connaissances."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Source inconnue")
page = doc.metadata.get("page")
location = f"{source}" + (f" Page {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Source : {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="""Vous êtes un assistant de base de connaissances personnelle.
## Règles
1. Toutes les questions doivent d'abord utiliser l'outil search_knowledge pour rechercher dans la base de connaissances
2. Lors de la réponse, indiquez la source de l'information (nom du document, et numéro de page s'il s'agit d'un PDF)
3. S'il n'y a pas de contenu pertinent dans la base de connaissances, dites-le honnêtement à l'utilisateur
4. Structurez vos réponses à l'aide de listes numérotées ou de paragraphes""",
)
# ========== Test : non-streaming, et afficher le contenu récupéré en même temps ==========
def ask(question: str):
"""Posez une question et affichez à la fois les extraits bruts récupérés et la réponse finale pour faciliter le débogage"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Afficher le contenu récupéré
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Contenu récupéré]")
print(msg.content[:300])
print(f"\n[Réponse]")
print(result["messages"][-1].content)
# ========== Test : streaming, correspondant à "sortie en streaming" dans la conception du système ==========
def ask_stream(question: str):
"""Posez une question et affichez la réponse token par token en streaming, pour obtenir un effet machine à écrire"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[Réponse] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" indique que ce chunk provient de la génération du modèle
# nœud de réponse finale, en filtrant les autres types de chunks tels que les appels d'outils
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Quels chapitres contient le tutoriel de base Python3 ?")
ask("Comment devrais-je planifier mon parcours d'apprentissage Python ?")
ask_stream("Quelles fonctionnalités l'environnement de programmation en ligne d'EXAMPLE prend-il en charge ?")
kb = KnowledgeBase("./my_knowledge_db")
# Ajouter quelques connaissances d'exemple
kb.add_text(
"Le tutoriel Python3 de base d'EXAMPLE contient les chapitres suivants :"
"1. Introduction à Python et configuration de l'environnement 2. Types de données de base 3. Opérateurs et expressions "
"4. Instructions conditionnelles if-else 5. Boucles for/while 6. Définition et appel de fonctions "
"7. Modules et packages 8. Opérations sur les fichiers 9. Gestion des exceptions 10. Programmation orientée objet",
source="Plan du tutoriel Python3"
)
kb.add_text(
"Pour devenir un bon développeur Python, il est recommandé d'étudier selon le parcours suivant :"
"Premièrement, maîtrisez la syntaxe de base de Python (1-2 semaines);"
"Deuxièmement, apprenez les bases des structures de données et des algorithmes (2-3 semaines);"
"Troisièmement, choisissez une direction pour une étude approfondie (développement Web/analyse de données/IA);"
"Quatrièmement, travaillez sur 2-3 projets pratiques pour consolider vos connaissances.",
source="Parcours d'apprentissage de Python"
)
kb.add_text(
"L'environnement de programmation en ligne d'EXAMPLE prend en charge plusieurs langages, notamment Python, JavaScript, Java et C++."
"Les utilisateurs n'ont besoin d'installer aucun logiciel ; il suffit d'ouvrir un navigateur pour écrire et exécuter du code."
"L'environnement en ligne prend également en charge la coloration syntaxique, l'auto-complétion et les indications d'erreur.",
source="Description de l'environnement de programmation en ligne"
)
# Vous pouvez également charger des fichiers locaux ; les PDF et Markdown/TXT sont automatiquement reconnus :
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== Créer un agent RAG ==========
@tool
def search_knowledge(query: str) -> str:
"""Recherchez dans la base de connaissances personnelle des informations pertinentes. Utilisez la question complète ou des expressions clés lors de la recherche.
Arguments :
query: la question de recherche ou l'expression clé
"""
docs = kb.search(query, k=3)
if not docs:
return "Aucune information pertinente trouvée dans la base de connaissances."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Source inconnue")
page = doc.metadata.get("page")
location = f"{source}" + (f" Page {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Source : {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="""Vous êtes un assistant de base de connaissances personnelle.
## Règles
1. Toutes les questions doivent d'abord utiliser l'outil search_knowledge pour rechercher dans la base de connaissances
2. Lors de la réponse, indiquez la source de l'information (nom du document, et numéro de page s'il s'agit d'un PDF)
3. S'il n'y a pas de contenu pertinent dans la base de connaissances, dites-le honnêtement à l'utilisateur
4. Structurez vos réponses à l'aide de listes numérotées ou de paragraphes""",
)
# ========== Test : non-streaming, et afficher le contenu récupéré en même temps ==========
def ask(question: str):
"""Posez une question et affichez à la fois les extraits bruts récupérés et la réponse finale pour faciliter le débogage"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Afficher le contenu récupéré
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Contenu récupéré]")
print(msg.content[:300])
print(f"\n[Réponse]")
print(result["messages"][-1].content)
# ========== Test : streaming, correspondant à "sortie en streaming" dans la conception du système ==========
def ask_stream(question: str):
"""Posez une question et affichez la réponse token par token en streaming, pour obtenir un effet machine à écrire"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[Réponse] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" indique que ce chunk provient de la génération du modèle
# nœud de réponse finale, en filtrant les autres types de chunks tels que les appels d'outils
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Quels chapitres contient le tutoriel de base Python3 ?")
ask("Comment devrais-je planifier mon parcours d'apprentissage Python ?")
ask_stream("Quelles fonctionnalités l'environnement de programmation en ligne d'EXAMPLE prend-il en charge ?")
Résultat de l'exécution :
============================================================ 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. 支持代码高亮、自动补全和错误提示功能
Autres extensionsLa dernière question utilise
ask_stream()sortie en streaming, donc à l'exécution, le texte après "[Answer]" est imprimé un caractère (ou quelques caractères par lot) à la fois, comme un effet de machine à écrire. Ci-dessus, pour un affichage pratique, le texte complet final après impression est collé directement.