Система Q&A для персональной базы знаний LangChain
В этом разделе создаётся система персональной базы знаний, которая может загружать файлы Markdown и PDF-документы и отвечать на вопросы по их содержимому.
Проектирование систем
- Загрузка документов: Поддерживается множество форматов, включая Markdown, TXT и PDF
- Векторный поиск: постоянное хранилище Chroma, поддерживает инкрементальные обновления
- Ссылки на источники: Ответы включают исходный документ и местоположение фрагмента
- Потоковый вывод: Отображает ответ токен за токеном
Полный код
Перед запуском необходимо настроить следующее.envв файле:DEEPSEEK_API_KEY(Чат-модель) иDASHSCOPE_API_KEY(Alibaba Cloud Bailian, используется для векторизации текста базы знаний). О конкретных процедурах применения и типовых проблемах см.«Интеллектуальный робот службы поддержки клиентов LangChain»секцию.
Пример
# Путь к файлу: 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:
"""Менеджер личной базы знаний"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# Интерфейс Embedding Alibaba Cloud Bailian совместим со спецификацией OpenAI; просто вызывайте его с помощью langchain-openai
# OpenAIEmbeddings — не нужно устанавливать DashScopeEmbeddings из langchain-community,
# который больше не поддерживается.
# check_embedding_ctx_length=False: отключить предварительную токенизацию tiktoken и отправлять исходный текст напрямую
# (интерфейс Bailian не принимает массивы token id).
# chunk_size=10: интерфейс Embedding Bailian принимает не более 10 текстов на запрос.
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):
"""Загрузить существующее векторное хранилище или создать новое (Chroma работает с тем же набором параметров;
то, существует ли каталог уже, влияет только на сообщение-подсказку, а не на логику создания)"""
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"Загружено векторное хранилище: {self.vector_store._collection.count()} фрагментов документов")
else:
print("Создано новое векторное хранилище")
def add_file(self, file_path: str) -> int:
"""Добавляет файл в базу знаний и возвращает количество добавленных фрагментов документа.
Автоматически выбирает подходящий Loader в зависимости от расширения файла:
- Для .pdf используйте PyPDFLoader (зависит от библиотеки pypdf для разбора PDF, возвращает несколько Documents по страницам)
- Для остальных (.md, .txt и других простых текстовых файлов) используйте TextLoader для чтения необработанного текста
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Единообразно добавляем имя файла в качестве source; каждый Document, загруженный PyPDFLoader, также имеет
# поле page (номер страницы), которое также можно использовать для определения позиции фрагмента при поиске
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"Добавлено {Path(file_path).name}: {len(chunks)} фрагментов документа")
return len(chunks)
def add_text(self, text: str, source: str = "Добавить вручную") -> int:
"""Непосредственно добавить текст в базу знаний"""
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) -> list:
"""Поиск по базе знаний"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Получить retriever"""
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:
"""Менеджер личной базы знаний"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# Интерфейс Embedding Alibaba Cloud Bailian совместим со спецификацией OpenAI; просто вызывайте его с помощью langchain-openai
# OpenAIEmbeddings — не нужно устанавливать DashScopeEmbeddings из langchain-community,
# который больше не поддерживается.
# check_embedding_ctx_length=False: отключить предварительную токенизацию tiktoken и отправлять исходный текст напрямую
# (интерфейс Bailian не принимает массивы token id).
# chunk_size=10: интерфейс Embedding Bailian принимает не более 10 текстов на запрос.
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):
"""Загрузить существующее векторное хранилище или создать новое (Chroma работает с тем же набором параметров;
то, существует ли каталог уже, влияет только на сообщение-подсказку, а не на логику создания)"""
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"Загружено векторное хранилище: {self.vector_store._collection.count()} фрагментов документов")
else:
print("Создано новое векторное хранилище")
def add_file(self, file_path: str) -> int:
"""Добавляет файл в базу знаний и возвращает количество добавленных фрагментов документа.
Автоматически выбирает подходящий Loader в зависимости от расширения файла:
- Для .pdf используйте PyPDFLoader (зависит от библиотеки pypdf для разбора PDF, возвращает несколько Documents по страницам)
- Для остальных (.md, .txt и других простых текстовых файлов) используйте TextLoader для чтения необработанного текста
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# Единообразно добавляем имя файла в качестве source; каждый Document, загруженный PyPDFLoader, также имеет
# поле page (номер страницы), которое также можно использовать для определения позиции фрагмента при поиске
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"Добавлено {Path(file_path).name}: {len(chunks)} фрагментов документа")
return len(chunks)
def add_text(self, text: str, source: str = "Добавить вручную") -> int:
"""Непосредственно добавить текст в базу знаний"""
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) -> list:
"""Поиск по базе знаний"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""Получить retriever"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
В настоящее время,
TextLoader、PyPDFLoaderэтот тип загрузчика локальных файлов пока не имеет отдельного поддерживаемого пакета; его всё ещё можно толькоlangchain_community.document_loadersимпортировать из langchain_community. Во время выполнения вы можете увидеть DeprecationWarning для всего пакета langchain-community — это предупреждение можно пока игнорировать; просто выполните миграцию после выпуска официального отдельного пакета загрузчика документов. Это отличается от случая с эмбеддингами выше, где готовая замена уже существовала, но не была принята.
Продолжаем раздел об агенте:
Пример
# ========== Создание базы знаний и добавление примерных данных ==========
kb = KnowledgeBase("./my_knowledge_db")
# Добавить несколько примеров знаний
kb.add_text(
"Базовое руководство EXAMPLE по Python3 содержит следующие главы:"
"1. Введение в Python и настройка среды 2. Основные типы данных 3. Операторы и выражения "
"4. Условные операторы if-else 5. Циклы for/while 6. Определение и вызов функций "
"7. Модули и пакеты 8. Операции с файлами 9. Обработка исключений 10. Объектно-ориентированное программирование",
источник="План руководства по Python3"
)
kb.add_text(
"Чтобы стать хорошим Python-разработчиком, рекомендуется придерживаться следующего маршрута обучения:"
"Во-первых, освойте базовый синтаксис Python (1-2 недели);"
"Во-вторых, изучите основы структур данных и алгоритмов (2-3 недели);"
"В-третьих, выберите направление для углублённого изучения (веб-разработка/анализ данных/ИИ);"
"В-четвёртых, поработайте над 2-3 практическими проектами, чтобы закрепить знания.",
источник="Путь изучения Python"
)
kb.add_text(
"Онлайн-среда программирования EXAMPLE поддерживает несколько языков, включая Python, JavaScript, Java и C++."
"Пользователям не нужно устанавливать какое-либо программное обеспечение; просто откройте браузер, чтобы писать и запускать код."
"Онлайн-среда также поддерживает подсветку кода, автодополнение и подсказки об ошибках.",
source="Описание онлайн-среды программирования"
)
# Вы также можете загружать локальные файлы; PDF и Markdown/TXT распознаются автоматически:
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== Создание RAG-агента ==========
@tool
def search_knowledge(query: str) -> str:
"""Поиск в персональной базе знаний релевантной информации. При поиске используйте полный вопрос или ключевые фразы.
Аргументы:
query: поисковый вопрос или ключевая фраза
"""
docs = kb.search(query, k=3)
if not docs:
return "В базе знаний не найдена соответствующая информация."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Неизвестный источник")
page = doc.metadata.get("page")
location = f"{source}" + (f" Страница {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Источник: {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="""Вы — помощник по личной базе знаний.
## Правила
1. Все вопросы должны сначала использовать инструмент search_knowledge для поиска по базе знаний
2. При ответе указывайте источник информации (название документа и номер страницы, если это PDF)
3. Если в базе знаний нет соответствующего содержания, честно сообщите об этом пользователю
4. Структурируйте свои ответы с помощью нумерованных списков или абзацев""",
)
# ========== Тест: без потоковой передачи, и одновременный просмотр полученного содержимого ==========
def ask(question: str):
"""Задайте вопрос и выведите как полученные исходные фрагменты, так и итоговый ответ для удобства отладки"""
print(f"\n{'='*60}")
print(f"В: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Отобразить полученное содержимое
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Полученное содержимое]")
print(msg.content[:300])
print(f"\n[Ответ]")
print(result["messages"][-1].content)
# ========== Тест: потоковая передача, соответствует "потоковому выводу" в проектировании системы ==========
def ask_stream(question: str):
"""Задайте вопрос и выводите ответ токен за токеном в потоковом режиме, создавая эффект печатной машинки"""
print(f"\n{'='*60}")
print(f"В: {question}")
print(f"{'='*60}")
print("\n[Ответ] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" указывает, что этот фрагмент получен от сгенерированного моделью
# узел финального ответа, отфильтровывая другие типы фрагментов, такие как вызовы инструментов
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Какие главы содержит базовый учебник по Python3?")
ask("Как мне спланировать свой путь изучения Python?")
ask_stream("Какие возможности поддерживает онлайн-среда программирования EXAMPLE?")
kb = KnowledgeBase("./my_knowledge_db")
# Добавить несколько примеров знаний
kb.add_text(
"Базовое руководство EXAMPLE по Python3 содержит следующие главы:"
"1. Введение в Python и настройка среды 2. Основные типы данных 3. Операторы и выражения "
"4. Условные операторы if-else 5. Циклы for/while 6. Определение и вызов функций "
"7. Модули и пакеты 8. Операции с файлами 9. Обработка исключений 10. Объектно-ориентированное программирование",
источник="План руководства по Python3"
)
kb.add_text(
"Чтобы стать хорошим Python-разработчиком, рекомендуется придерживаться следующего маршрута обучения:"
"Во-первых, освойте базовый синтаксис Python (1-2 недели);"
"Во-вторых, изучите основы структур данных и алгоритмов (2-3 недели);"
"В-третьих, выберите направление для углублённого изучения (веб-разработка/анализ данных/ИИ);"
"В-четвёртых, поработайте над 2-3 практическими проектами, чтобы закрепить знания.",
источник="Путь изучения Python"
)
kb.add_text(
"Онлайн-среда программирования EXAMPLE поддерживает несколько языков, включая Python, JavaScript, Java и C++."
"Пользователям не нужно устанавливать какое-либо программное обеспечение; просто откройте браузер, чтобы писать и запускать код."
"Онлайн-среда также поддерживает подсветку кода, автодополнение и подсказки об ошибках.",
source="Описание онлайн-среды программирования"
)
# Вы также можете загружать локальные файлы; PDF и Markdown/TXT распознаются автоматически:
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== Создание RAG-агента ==========
@tool
def search_knowledge(query: str) -> str:
"""Поиск в персональной базе знаний релевантной информации. При поиске используйте полный вопрос или ключевые фразы.
Аргументы:
query: поисковый вопрос или ключевая фраза
"""
docs = kb.search(query, k=3)
if not docs:
return "В базе знаний не найдена соответствующая информация."
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Неизвестный источник")
page = doc.metadata.get("page")
location = f"{source}" + (f" Страница {page + 1}" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] Источник: {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="""Вы — помощник по личной базе знаний.
## Правила
1. Все вопросы должны сначала использовать инструмент search_knowledge для поиска по базе знаний
2. При ответе указывайте источник информации (название документа и номер страницы, если это PDF)
3. Если в базе знаний нет соответствующего содержания, честно сообщите об этом пользователю
4. Структурируйте свои ответы с помощью нумерованных списков или абзацев""",
)
# ========== Тест: без потоковой передачи, и одновременный просмотр полученного содержимого ==========
def ask(question: str):
"""Задайте вопрос и выведите как полученные исходные фрагменты, так и итоговый ответ для удобства отладки"""
print(f"\n{'='*60}")
print(f"В: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# Отобразить полученное содержимое
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[Полученное содержимое]")
print(msg.content[:300])
print(f"\n[Ответ]")
print(result["messages"][-1].content)
# ========== Тест: потоковая передача, соответствует "потоковому выводу" в проектировании системы ==========
def ask_stream(question: str):
"""Задайте вопрос и выводите ответ токен за токеном в потоковом режиме, создавая эффект печатной машинки"""
print(f"\n{'='*60}")
print(f"В: {question}")
print(f"{'='*60}")
print("\n[Ответ] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" указывает, что этот фрагмент получен от сгенерированного моделью
# узел финального ответа, отфильтровывая другие типы фрагментов, такие как вызовы инструментов
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Какие главы содержит базовый учебник по Python3?")
ask("Как мне спланировать свой путь изучения Python?")
ask_stream("Какие возможности поддерживает онлайн-среда программирования EXAMPLE?")
Результат выполнения:
============================================================ 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. 支持代码高亮、自动补全和错误提示功能
Другие расширенияВ последнем вопросе используется
ask_stream()потоковый вывод, поэтому во время выполнения текст после "[Answer]" печатается по одному символу (или несколько символов за пакет) за раз, совсем как эффект пишущей машинки. Выше, для удобного отображения, окончательный полный текст после вывода вставлен напрямую.