Я пытаюсь создать фрагменты документов SEC для RAG. Идея состоит в том, чтобы идентифицировать разделы и поместить их в теги длиной, скажем, 6000 символов. Проблема в том, что эти документы структурированы совершенно по-разному. Я думаю, это зависит от того, кто написал эти документы; у некоторых очень причудливые заголовки, заголовки разбиты на две строки данных таблицы и т. д. Есть ли лучший способ подойти ко всему этому?
def preprocess_text_for_vectorization(text_content, char_limit=6000, min_words_per_para=300):
def clean_text(text):
return re.sub(r'\s+', ' ', text).strip()
def is_heading(text):
return bool(re.match(r'^[A-Z\s]+$', text.strip()))
def is_bullet_or_numbered(text):
return bool(re.match(r'^(\d+\.|\*|\-)\s', text.strip()))
def is_table(text):
return bool(re.search(r'(\d{4})', text)) and bool(re.search(r'(\$[\d,]+)', text))
def merge_small_paragraphs(chunks, min_words_per_para):
merged_chunks = []
current_chunk = ""
current_words = 0
for chunk in chunks:
words_in_chunk = len(chunk.split())
if current_words + words_in_chunk < min_words_per_para:
current_chunk += " " + chunk
current_words += words_in_chunk
else:
if current_chunk:
merged_chunks.append(current_chunk.strip())
current_chunk = chunk
current_words = words_in_chunk
if current_chunk:
merged_chunks.append(current_chunk.strip())
return merged_chunks
def create_chunks(text, char_limit=6000):
chunks = []
current_chunk = ""
current_length = 0
current_heading = ""
for line in text.split('\n'):
line = clean_text(line)
if not line:
continue
if is_heading(line):
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = line
current_heading = line
current_length = len(line)
elif is_table(line):
if current_chunk:
chunks.append(current_chunk.strip())
chunks.append(line)
current_chunk = ""
current_length = 0
elif is_bullet_or_numbered(line) or current_length + len(line) > char_limit:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = current_heading + "\n" + line if current_heading else line
current_length = len(line)
else:
current_chunk += "\n" + line
current_length += len(line) + 1
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
def process_text(text_content, char_limit=6000, min_words_per_para=300):
paragraphs = create_chunks(text_content, char_limit)
merged_paragraphs = merge_small_paragraphs(paragraphs, min_words_per_para)
return merged_paragraphs
processed_content = process_text(text_content, char_limit, min_words_per_para)
return processed_content
Подробнее здесь: https://stackoverflow.com/questions/787 ... gs-for-rag
