Это сложная, но решаемая задача. Комбинированный контент (текст, таблицы, схемы) требует особого подхода. Давайте разберем поэтапно.
Проблемы стандартной векторизации для mixed-content PDF:
- Таблицы при извлечении теряют структуру → превращаются в беспорядочный текст
- Изображения/схемы не несут смысла в текстовом виде
- Связь между элементами например, описание в тексте + схема + таблица параметров) теряется
Стратегия обработки mixed-content PDF
Этап 1: Извлечение и сегментация контента
Инструменты:
- io— отлично справляется со сложными PDF
- Azure Document Intelligence(ранее Form Recognizer)
- Amazon Textract— хорош для таблиц
- PyMuPDF+ кастомная логика
Процесс:
# Пример логики сегментации
document_structure = {
"sections": [
{
"title": "Технические характеристики",
"content_type": "mixed",
"elements": [
{"type": "text", "content": "Описание системы...", "page": 1},
{"type": "table", "content": dataframe, "page": 1},
{"type": "image", "content": "scheme_1.png", "caption": "Схема подключения", "page": 2}
]
}
]
}
Этап 2: Обработка разных типов контента
Для текста:
- Стандартная векторизация выбранной моделью (как обсуждали ранее)
- Важно:сохранять контекст (заголовки раздела, подписи к таблицам/изображениям)
Для таблиц:
Стратегия 1: Текстовое представление
python
# Преобразование таблицы в структурированный текст
def table_to_text(df, caption):
table_text = f"Таблица: {caption}\n"
table_text += f"Колонки: {', '.join(df.columns)}\n"
for i, row in df.iterrows():
row_data = [f"{col}: {row[col]}" for col in df.columns]
table_text += f"Строка {i+1}: {'; '.join(row_data)}\n"
return table_text
Стратегия 2: Markdown-представление
markdown
Таблица: Параметры устройства
| Параметр | Значение | Единица измерения |
| Напряжение | 220 | В |
| Ток | 5 | А |
Стратегия 3: Векторизация по строкам/секциям
- Каждую строку таблицы векторизуем отдельно с заголовками колонок
Для изображений/схем:
Стратегия 1: Описания через Vision AI
python # Использование мультимодальных моделей def describe_image(image_path): # Вариант 1: Local models (BLIP, InstructBLIP) # Вариант 2: Cloud API (GPT-4V, Claude 3) prompt = """ Опиши эту техническую схему на русском языке. Укажи основные компоненты, соединения и принцип работы. Будь точным и техничным. """ return description # Текст для векторизации
Стратегия 2: Детекция и описание элементов
python
# Для электрических схем
components = [
{"type": "резистор", "value": "10kΩ", "position": "левый верхний"},
{"type": "конденсатор", "value": "100μF", "position": "центр"},
{"connections": ["резистор R1", "конденсатор C1"]}
]
Стратегия 3: OCR для текста на схемах
- Извлечение подписей, обозначений, номеров деталей
Архитектура RAG системы для mixed-content
Вариант 1: Унифицированное векторное пространство
python
class MultiModalRAG:
def __init__(self):
self.text_encoder = "sentence-transformers/paraphrase-multilingual-mpnet-base-v2"
self.vision_model = "Salesforce/blip-image-captioning-large"
def process_document(self, pdf_path):
chunks = []
# Извлечение всех элементов
elements = extract_pdf_elements(pdf_path)
for element in elements:
if element.type == "text":
chunk = {
"content": element.text,
"type": "text",
"vector": encode_text(element.text),
"metadata": {"page": element.page, "section": element.section}
}
elif element.type == "table":
# Вариант A: Вся таблица как один чанк
table_text = table_to_markdown(element.data, element.caption)
chunk = {
"content": table_text,
"type": "table",
"vector": encode_text(table_text),
"metadata": {"page": element.page, "dataframe": element.data}
}
# Вариант B: Каждая строка таблицы как отдельный чанк
for idx, row in element.data.iterrows():
row_text = f"Таблица {element.caption}: {row.to_dict()}"
chunks.append({
"content": row_text,
"type": "table_row",
"vector": encode_text(row_text),
"metadata": {"table": element.caption, "row_index": idx}
})
elif element.type == "image":
description = describe_image(element.image_path)
chunk = {
"content": f"Схема: {element.caption}. {description}",
"type": "image",
"vector": encode_text(description),
"metadata": {"page": element.page, "image_path": element.image_path}
}
chunks.append(chunk)
return chunks
Вариант 2: Мульти-векторный подход (рекомендуется)
python def multi_vector_retrieval(query, top_k=5): # Шаг 1: Поиск в основном тексте text_results = vector_search(query, text_chunks, top_k=top_k) # Шаг 2: Поиск в таблицах table_results = vector_search(query, table_chunks, top_k=top_k//2) # Шаг 3: Поиск в описаниях изображений image_results = vector_search(query, image_chunks, top_k=top_k//2) # Шаг 4: Ранжирование и объединение all_results = merge_and_rerank( text_results + table_results + image_results ) return all_results[:top_k]
Практическая реализация
Рекомендуемый стек технологий:
python # requirements.txt unstructured[pdf] # для извлечения контента sentence-transformers # для эмбеддингов transformers # для BLIP и других моделей pymupdf # работа с PDF pandas # обработка таблиц opencv-python # обработка изображений
Пример пайплайна:
python def create_mixed_content_rag(pdf_directory): all_chunks = [] for pdf_file in pdf_directory: # 1. Извлечение элементов elements = unstructured_pipeline(pdf_file) # 2. Обработка каждого типа контента for element in elements: if element.type == "text": chunk = process_text_element(element) elif element.type == "table": chunk = process_table_element(element) elif element.type == "image": chunk = process_image_element(element) # 3. Векторизация и сохранение chunk["embedding"] = embed_text(chunk["content"]) all_chunks.append(chunk) # 4. Сохранение в векторную БД vector_db.upsert(all_chunks) return MixedContentRAG(vector_db)
Prompt engineering для ответов:
python system_prompt = """ Ты - технический специалист, помогающий с документацией. Используй предоставленные контексты для ответа: - Текстовые описания - Таблицы с параметрами - Схемы и их описания Если в контексте есть схема - опиши ее структуру. Если есть таблица - используй конкретные значения из нее. Ссылайся на конкретные страницы и элементы. """
Рекомендации по реализации
- Начните с пилота: Обработайте 1-2 сложных PDF полностью
- Тестируйте поиск: Задавайте вопросы, требующие разных типов контента:
- «Какие параметры указаны в таблице 3.2?»
- «Опиши схему подключения на странице 15»
- «Сравни значения из таблицы 5.1 и текстового описания»
- Настройте баланс:Определите оптимальное соотношение текста/таблиц/изображений в результатах поиска
- Добавьте метаданные:Сохраняйте тип контента, страницы, заголовки для фильтрации
Стартовый рецепт:
- Используйте ioдля извлечения
- Таблицы → Markdown представление
- Изображения → BLIP + ручное описание ключевых схем
- Векторизация → paraphrase-multilingual-mpnet-base-v2
- Поиск → мульти-векторный подход с reranking
Этот подход позволит эффективно работать со сложной технической документацией любого формата!
покажи как извлечь контент из pdf файла со схемами и таблицами и разделить всё на чанки перед векторизацией. Напиши приложение, с помощью которого можно извлечь mixed контент из pdf файла, и нарезать контент на чанки соответственно типу контента, заложи в функционал программы возможность сделать исправления в текствовом и табличном типах контента
Создам комплексное приложение для извлечения mixed-контента из PDF с возможностью редактирования и нарезки на чанки.
Полное приложение для обработки PDF
python
import os
import fitz # PyMuPDF
import pandas as pd
import cv2
import numpy as np
from PIL import Image
import json
import logging
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import argparse
# Настройка логирования
logging.basicConfig(level=logging.INFO, format=’%(asctime)s — %(levelname)s — %(message)s’)
logger = logging.getLogger(__name__)
@dataclass
class ContentElement:
«»»Класс для представления элемента контента»»»
content_type: str # ‘text’, ‘table’, ‘image’
content: Any
page_num: int
bbox: tuple # (x0, y0, x1, y1)
metadata: Dict[str, Any]
element_id: str = None
def __post_init__(self):
if self.element_id is None:
self.element_id = f»{self.content_type}_{self.page_num}_{hash(str(self.content))[:8]}»
class PDFContentExtractor:
«»»Класс для извлечения mixed-контента из PDF»»»
def __init__(self, pdf_path: str):
self.pdf_path = pdf_path
self.doc = fitz.open(pdf_path)
self.elements = []
def extract_all_content(self) -> List[ContentElement]:
«»»Извлечение всего контента из PDF»»»
logger.info(f»Начинаем извлечение контента из {self.pdf_path}»)
for page_num in range(len(self.doc)):
page = self.doc[page_num]
self._extract_page_content(page, page_num)
logger.info(f»Извлечено {len(self.elements)} элементов»)
return self.elements
def _extract_page_content(self, page, page_num: int):
«»»Извлечение контента со страницы»»»
# Извлечение текста с блоками
text_blocks = page.get_text(«dict»)
for block in text_blocks[«blocks»]:
if «lines» in block: # Текстовый блок
self._process_text_block(block, page_num)
elif «image» in block: # Изображение
self._process_image_block(block, page, page_num)
def _process_text_block(self, block: Dict, page_num: int):
«»»Обработка текстового блока»»»
text_content = «»
for line in block[«lines»]:
for span in line[«spans»]:
text_content += span[«text»] + » »
text_content = text_content.strip()
if not text_content or len(text_content) < 5: # Игнорируем очень короткие блоки
return
element = ContentElement(
content_type=»text»,
content=text_content,
page_num=page_num + 1,
bbox=block[«bbox»],
metadata={
«font_size»: block[«lines»][0][«spans»][0][«size»] if block[«lines»][0][«spans»] else 0,
«font»: block[«lines»][0][«spans»][0][«font»] if block[«lines»][0][«spans»] else «»,
«is_table_candidate»: self._is_table_candidate(text_content)
}
)
self.elements.append(element)
def _process_image_block(self, block: Dict, page, page_num: int):
«»»Обработка блока с изображением»»»
try:
# Извлекаем изображение
image_matrix = fitz.Matrix(2, 2) # Увеличиваем разрешение
pix = page.get_pixmap(matrix=image_matrix, clip=block[«bbox»])
img_data = pix.tobytes(«png»)
element = ContentElement(
content_type=»image»,
content=img_data,
page_num=page_num + 1,
bbox=block[«bbox»],
metadata={
«image_format»: «png»,
«resolution»: f»{pix.width}x{pix.height}»
}
)
self.elements.append(element)
except Exception as e:
logger.error(f»Ошибка при извлечении изображения: {e}»)
def _is_table_candidate(self, text: str) -> bool:
«»»Проверка, является ли текст кандидатом в таблицу»»»
lines = text.split(‘\n’)
if len(lines) < 2:
return False
# Проверяем паттерны таблицы
table_indicators = [
len(re.findall(r’\s{2,}’, line)) >= 2 for line in lines
]
return sum(table_indicators) / len(table_indicators) > 0.5
class ContentProcessor:
«»»Класс для обработки и чанкинга контента»»»
def __init__(self, chunk_sizes: Dict[str, int] = None):
self.chunk_sizes = chunk_sizes or {
«text»: 1000,
«table»: 500,
«image»: 300
}
def chunk_elements(self, elements: List[ContentElement]) -> List[Dict[str, Any]]:
«»»Разбивка элементов на чанки по типам»»»
chunks = []
for element in elements:
if element.content_type == «text»:
chunks.extend(self._chunk_text(element))
elif element.content_type == «table»:
chunks.extend(self._chunk_table(element))
elif element.content_type == «image»:
chunks.extend(self._chunk_image(element))
return chunks
def _chunk_text(self, element: ContentElement) -> List[Dict[str, Any]]:
«»»Чанкинг текстового контента»»»
text = element.content
chunk_size = self.chunk_sizes[«text»]
overlap = 100
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
# Не режем слова посередине
if end < len(text):
while end > start and text[end] not in (‘ ‘, ‘.’, ‘,’, ‘!’, ‘?’, ‘\n’):
end -= 1
if end == start: # Если не нашли пробел, режем по chunk_size
end = start + chunk_size
chunk_text = text[start:end].strip()
if chunk_text:
chunks.append({
«element_id»: element.element_id,
«content_type»: «text»,
«content»: chunk_text,
«page_num»: element.page_num,
«chunk_index»: len(chunks),
«metadata»: {
**element.metadata,
«is_continuation»: start > 0
}
})
start = end — overlap if end — overlap > start else end
if start >= len(text):
break
return chunks
def _chunk_table(self, element: ContentElement) -> List[Dict[str, Any]]:
«»»Чанкинг табличного контента (если будет распознана таблица)»»»
# В реальном приложении здесь будет логика распознавания таблиц
# Пока возвращаем как один чанк
return [{
«element_id»: element.element_id,
«content_type»: «table»,
«content»: element.content,
«page_num»: element.page_num,
«chunk_index»: 0,
«metadata»: element.metadata
}]
def _chunk_image(self, element: ContentElement) -> List[Dict[str, Any]]:
«»»Чанкинг изображений»»»
# Для изображений обычно один чанк на изображение
return [{
«element_id»: element.element_id,
«content_type»: «image»,
«content»: element.content,
«page_num»: element.page_num,
«chunk_index»: 0,
«metadata»: element.metadata
}]
class ContentEditor:
«»»Класс для редактирования извлеченного контента»»»
def __init__(self, elements: List[ContentElement]):
self.elements = elements
self.edited_elements = elements.copy()
def edit_text_element(self, element_id: str, new_content: str) -> bool:
«»»Редактирование текстового элемента»»»
for i, element in enumerate(self.edited_elements):
if element.element_id == element_id and element.content_type == «text»:
self.edited_elements[i] = ContentElement(
content_type=»text»,
content=new_content,
page_num=element.page_num,
bbox=element.bbox,
metadata={**element.metadata, «edited»: True},
element_id=element_id
)
logger.info(f»Текстовый элемент {element_id} отредактирован»)
return True
return False
def edit_table_element(self, element_id: str, table_data: List[Dict]) -> bool:
«»»Редактирование табличного элемента»»»
for i, element in enumerate(self.edited_elements):
if element.element_id == element_id and element.content_type == «table»:
# Конвертируем обратно в текст или сохраняем структурировано
table_text = self._table_to_text(table_data)
self.edited_elements[i] = ContentElement(
content_type=»table»,
content=table_text,
page_num=element.page_num,
bbox=element.bbox,
metadata={**element.metadata, «edited»: True},
element_id=element_id
)
logger.info(f»Табличный элемент {element_id} отредактирован»)
return True
return False
def _table_to_text(self, table_data: List[Dict]) -> str:
«»»Конвертация табличных данных в текст»»»
if not table_data:
return «»
# Создаем текстовое представление таблицы
headers = list(table_data[0].keys())
text_table = » | «.join(headers) + «\n»
text_table += «-» * len(text_table) + «\n»
for row in table_data:
text_table += » | «.join(str(row.get(h, «»)) for h in headers) + «\n»
return text_table
def get_edited_elements(self) -> List[ContentElement]:
«»»Получить отредактированные элементы»»»
return self.edited_elements
class PDFProcessorApp:
«»»Главный класс приложения»»»
def __init__(self, pdf_path: str):
self.pdf_path = pdf_path
self.extractor = PDFContentExtractor(pdf_path)
self.elements = []
self.chunks = []
self.editor = None
def process_pdf(self, chunk_sizes: Dict = None) -> List[Dict]:
«»»Полный процесс обработки PDF»»»
logger.info(«Запуск обработки PDF…»)
# Извлечение контента
self.elements = self.extractor.extract_all_content()
# Инициализация редактора
self.editor = ContentEditor(self.elements)
# Чанкинг
processor = ContentProcessor(chunk_sizes)
self.chunks = processor.chunk_elements(self.elements)
logger.info(f»Создано {len(self.chunks)} чанков»)
return self.chunks
def save_results(self, output_dir: str):
«»»Сохранение результатов»»»
os.makedirs(output_dir, exist_ok=True)
# Сохранение элементов
elements_data = []
for element in self.elements:
elements_data.append({
«element_id»: element.element_id,
«content_type»: element.content_type,
«content»: element.content if element.content_type != «image» else «BINARY_IMAGE_DATA»,
«page_num»: element.page_num,
«metadata»: element.metadata
})
with open(os.path.join(output_dir, «elements.json»), «w», encoding=»utf-8″) as f:
json.dump(elements_data, f, ensure_ascii=False, indent=2)
# Сохранение чанков
chunks_data = []
for chunk in self.chunks:
chunk_data = chunk.copy()
if chunk_data[«content_type»] == «image»:
chunk_data[«content»] = «BINARY_IMAGE_DATA»
chunks_data.append(chunk_data)
with open(os.path.join(output_dir, «chunks.json»), «w», encoding=»utf-8″) as f:
json.dump(chunks_data, f, ensure_ascii=False, indent=2)
# Сохранение статистики
stats = {
«total_elements»: len(self.elements),
«total_chunks»: len(self.chunks),
«elements_by_type»: self._count_elements_by_type(),
«chunks_by_type»: self._count_chunks_by_type()
}
with open(os.path.join(output_dir, «statistics.json»), «w», encoding=»utf-8″) as f:
json.dump(stats, f, ensure_ascii=False, indent=2)
logger.info(f»Результаты сохранены в {output_dir}»)
def _count_elements_by_type(self) -> Dict[str, int]:
«»»Подсчет элементов по типам»»»
counts = {}
for element in self.elements:
counts[element.content_type] = counts.get(element.content_type, 0) + 1
return counts
def _count_chunks_by_type(self) -> Dict[str, int]:
«»»Подсчет чанков по типам»»»
counts = {}
for chunk in self.chunks:
counts[chunk[«content_type»]] = counts.get(chunk[«content_type»], 0) + 1
return counts
def interactive_edit_mode(processor: PDFProcessorApp):
«»»Интерактивный режим редактирования»»»
print(«\n» + «=»*50)
print(«РЕЖИМ РЕДАКТИРОВАНИЯ КОНТЕНТА»)
print(«=»*50)
while True:
print(«\nДоступные команды:»)
print(«1 — Показать текстовые элементы»)
print(«2 — Редактировать текстовый элемент»)
print(«3 — Показать табличные элементы»)
print(«4 — Редактировать табличный элемент»)
print(«5 — Сохранить изменения и выйти»)
print(«6 — Выйти без сохранения»)
choice = input(«\nВыберите действие: «).strip()
if choice == «1»:
_show_text_elements(processor.elements)
elif choice == «2»:
_edit_text_element(processor.editor)
elif choice == «3»:
_show_table_elements(processor.elements)
elif choice == «4»:
_edit_table_element(processor.editor)
elif choice == «5»:
if processor.editor:
processor.elements = processor.editor.get_edited_elements()
# Пересоздаем чанки с отредактированными элементами
processor_chunks = ContentProcessor().chunk_elements(processor.elements)
processor.chunks = processor_chunks
print(«Изменения сохранены»)
break
elif choice == «6»:
print(«Выход без сохранения»)
break
else:
print(«Неверный выбор»)
def _show_text_elements(elements: List[ContentElement]):
«»»Показать текстовые элементы»»»
text_elements = [e for e in elements if e.content_type == «text»]
print(f»\nНайдено {len(text_elements)} текстовых элементов:»)
for i, element in enumerate(text_elements[:10]): # Показываем первые 10
print(f»\n— Текстовый элемент {i+1} —«)
print(f»ID: {element.element_id}»)
print(f»Страница: {element.page_num}»)
print(f»Текст: {element.content[:200]}…» if len(element.content) > 200 else f»Текст: {element.content}»)
def _edit_text_element(editor: ContentEditor):
«»»Редактирование текстового элемента»»»
element_id = input(«Введите ID элемента для редактирования: «).strip()
new_content = input(«Введите новый текст: «).strip()
if editor.edit_text_element(element_id, new_content):
print(«Элемент успешно отредактирован!»)
else:
print(«Элемент не найден!»)
def _show_table_elements(elements: List[ContentElement]):
«»»Показать табличные элементы»»»
table_elements = [e for e in elements if e.content_type == «table»]
print(f»\nНайдено {len(table_elements)} табличных элементов:»)
for i, element in enumerate(table_elements[:5]): # Показываем первые 5
print(f»\n— Табличный элемент {i+1} —«)
print(f»ID: {element.element_id}»)
print(f»Страница: {element.page_num}»)
print(f»Содержимое: {element.content}»)
def _edit_table_element(editor: ContentEditor):
«»»Редактирование табличного элемента»»»
print(«Редактирование таблиц в этой версии ограничено»)
print(«В полной версии можно будет редактировать структурированные табличные данные»)
def main():
«»»Главная функция»»»
parser = argparse.ArgumentParser(description=»Обработка PDF с mixed-контентом»)
parser.add_argument(«pdf_path», help=»Путь к PDF файлу»)
parser.add_argument(«—output», «-o», default=»./output», help=»Директория для результатов»)
parser.add_argument(«—edit», «-e», action=»store_true», help=»Запустить интерактивное редактирование»)
args = parser.parse_args()
if not os.path.exists(args.pdf_path):
print(f»Ошибка: Файл {args.pdf_path} не найден»)
return
# Создаем и запускаем процессор
processor = PDFProcessorApp(args.pdf_path)
try:
# Обрабатываем PDF
chunks = processor.process_pdf()
# Показываем статистику
print(«\n» + «=»*50)
print(«СТАТИСТИКА ОБРАБОТКИ»)
print(«=»*50)
print(f»Всего элементов: {len(processor.elements)}»)
print(f»Всего чанков: {len(chunks)}»)
print(f»Элементы по типам: {processor._count_elements_by_type()}»)
print(f»Чанки по типам: {processor._count_chunks_by_type()}»)
# Интерактивное редактирование
if args.edit:
interactive_edit_mode(processor)
# Сохранение результатов
processor.save_results(args.output)
print(f»\nОбработка завершена! Результаты сохранены в {args.output}»)
except Exception as e:
logger.error(f»Ошибка при обработке PDF: {e}»)
raise
if __name__ == «__main__»:
main()
Дополнительные утилиты
Создадим файл requirements.txt:
PyMuPDF>=1.23.0
pandas>=2.0.0
pillow>=10.0.0
opencv-python>=4.8.0
numpy>=1.24.0
И файл config.yaml для настройки параметров:
chunk_sizes:
text: 1000
table: 500
image: 300
processing:
min_text_length: 5
image_quality: 2
table_detection: true
output:
save_images: true
image_format: «png»
max_file_size: «10MB»
Использование приложения
python pdf_processor.py document.pdf
# Обработка с сохранением в указанную директорию
python pdf_processor.py document.pdf -o ./my_output
# Обработка с интерактивным редактированием
python pdf_processor.py document.pdf -e
Ключевые особенности приложения:
- Извлечение mixed-контента: Текст, изображения, кандидаты в таблицы
- Интеллектуальный чанкинг: Разный размер чанков для разных типов контента
- Редактирование: Возможность исправлять текст и таблицы перед векторизацией
- Структурированное сохранение: JSON файлы с метаданными
- Статистика: Детальная информация о процессе обработки
- Логирование: Полное отслеживание процесса
Это приложение обеспечивает полный пайплайн обработки сложных PDF с технической документацией перед векторизацией для RAG системы!
сделай такую программу с интерфейсом, например на streamlit
