Мобильная версия
< /p>вот мой Текущая задача: < /p>
У меня есть файл PDF (потенциально сотни из них в будущем), и я хочу развернуть некоторую модель локально, чтобы строго ответить на запросы на основе этих Файлы.
Вот что я сделал до сих пор:
Я выбрал подход , проанализировал файл PDF ( Код ниже), а затем создал векторизованную базу данных. полученный Контекст для LLM -> LLM предоставляет соответствующий ответ
Но я изо всех сил пытаюсь определить, какую модель мне нужна и правильно использую векторизованный DB. Возможно, мне нужны большие куски или больше перекрытия. Важно, чтобы вся информация в PDF -файлах была на русской (не каждая модель может подходить для этого), пользовательские запросы и окончательная подсказка также находятся на русском языке, а ответ модели тоже на русском языке. < /P>
В настоящее время я использую Llama3: 8b, и мне обычно нравится, как он реагирует. Тем не менее, проблема в том, что иногда это, кажется, пропускает контекст, что приводит к неактуальным ответам, которые довольно далеки от истины. Я бы сказал, что они слишком надуманные. Может быть, мне следует использовать более надежную модель или что -то другое, кроме Llama3. >
Бонус был бы, если бы я мог запустить модель через Ollama, так как для меня очень важно наличие API REST. < /p>
Я хочу реализовать эту идею на Данные FSNB-2012, поскольку они имеют много соответствующих данных для этого. Эта идея была предложена мне Chatgpt
< /p>Прямо сейчас я пытаюсь выяснить вещи на основе одного файла, прежде чем загружать все остальные. < /P>
Вот ссылка для загрузки этого файла с веб -сайта Министерства строительства российской федерации:
https://minstroyrf.gov.ru/trades/tree_d ... d=0образно Сам: < /p>
"Сборник rabotы.pdf ", этот файл находится в разделе GESN (gэSN). > Я из России, поэтому, пожалуйста, простите меня за то, что я попросил о помощи и все еще заставил вас приложить дополнительные усилия, но вам придется перевести комментарии в коде на свой язык. Извините за неудобства. < /P>
import pdfplumber
import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from langchain.docstore.document import Document
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from tqdm import tqdm
import re
@dataclass
class Table:
header: List[str]
rows: List[List[str]]
raw_text: str
@dataclass
class Section:
title: str
content: str
tables: List[Table]
subsections: List['Section']
page_number: int
def clean_text(text: str) -> str:
text = re.sub(r'-\n\s*', '', text)
text = re.sub(r'\s+', ' ', text)
return text
def process_table(table) -> Optional[Table]:
if not table or not table[0]:
return None
header = [clean_text(cell) if cell else "" for cell in table[0]]
rows = [[clean_text(cell) if cell else "" for cell in row] for row in table[1:]]
raw_text = "\n".join([" | ".join(header)] + [" | ".join(row) for row in rows])
return Table(header=header, rows=rows, raw_text=raw_text)
def identify_section(text: str) -> bool:
section_patterns = [
r'Таблица ГЭСН \d{2}-\d{2}-\d{3}',
r'Состав работ:',
r'Измеритель:'
]
return any(re.search(pattern, text) for pattern in section_patterns)
def extract_sections(pdf_path) -> List[Section]:
sections = []
current_section = None
with pdfplumber.open(pdf_path) as pdf:
for page in tqdm(pdf.pages, desc="Processing pages"):
text_lines = page.extract_text_lines()
tables = page.extract_tables()
# Process text lines
current_text = ""
for line in text_lines:
text = clean_text(line['text'])
if identify_section(text):
if current_section:
if current_text:
current_section.content += f"\n{current_text}"
sections.append(current_section)
current_section = Section(
title=text,
content=text,
tables=[],
subsections=[],
page_number=page.page_number
)
current_text = ""
else:
current_text += f"\n{text}"
if current_section and current_text:
current_section.content += f"\n{current_text}"
# Process tables for current section
if current_section and tables:
for table in tables:
processed_table = process_table(table)
if processed_table:
current_section.tables.append(processed_table)
if current_section:
sections.append(current_section)
return sections
def section_to_document(section: Section) -> Document:
content = f"""
Title: {section.title}
Content: {section.content}
Tables:
{chr(10).join(table.raw_text for table in section.tables)}
"""
return Document(
page_content=content,
metadata={
"page_number": section.page_number,
"title": section.title,
"table_count": len(section.tables)
}
)
def build_vector_db(sections: List[Section], index_path="smeta_faiss_index"):
documents = [section_to_document(section) for section in sections]
text_splitter = RecursiveCharacterTextSplitter(chunk_size=3000, chunk_overlap=750)
docs_split = text_splitter.split_documents(documents)
# embeddings = HuggingFaceEmbeddings(model_name="ai-forever/sbert_large_mt_nlu_ru")
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
# embeddings = HuggingFaceEmbeddings(model_name="intfloat/multilingual-e5-base")
vector_db = FAISS.from_documents(docs_split, embeddings)
vector_db.save_local(index_path)
return vector_db
def main():
pdf_path = "Сборник ГЭСН 01 Земляные работы.pdf"
sections = extract_sections(pdf_path)
with open("extracted_sections.json", "w", encoding="utf8") as f:
json.dump([asdict(section) for section in sections], f, indent=2, ensure_ascii=False)
vector_db = build_vector_db(sections, index_path="faiss_index_sectioned_all-MiniLM-L6-v2")
if __name__ == "__main__":
main()
< /code>
А вот код бизнес -логики, где пользователь делает запросы, их обработка и т. Д. < /p>
business_logic.py
import json
from typing import List, Dict, Any
import requests
from dataclasses import dataclass
from langchain_community.vectorstores import FAISS
from langchain_huggingface.embeddings import HuggingFaceEmbeddings
@dataclass
class GesnFormat:
"""Формат ответа ГЭСН"""
price_number: str = ""
price_name: str = ""
works: List[str] = None
resources: List[Dict[str, str]] = None
def __post_init__(self):
if self.works is None:
self.works = []
if self.resources is None:
self.resources = []
def to_dict(self) -> dict:
return {
"price_number": self.price_number,
"price_name": self.price_name,
"works": self.works,
"resources": [
{
"resource_name": res.get("resource_name", ""),
"resource_number": res.get("resource_number", "")
}
for res in self.resources
]
}
class SmetaService:
def __init__(self, vector_db_path: str, llm_url: str = "http://localhost:11434/api/generate"):
"""
Инициализация сервиса
Args:
vector_db_path: путь к векторной БД
llm_url: URL для LLM API
"""
self.embeddings = HuggingFaceEmbeddings(
# model_name="ai-forever/sbert_large_mt_nlu_ru"
model_name="all-MiniLM-L6-v2"
)
self.vector_db = FAISS.load_local(
vector_db_path,
self.embeddings,
allow_dangerous_deserialization=True
)
self.llm_url = llm_url
def search_vector_db(self, query: str, k: int = 5) -> List[str]:
"""
Поиск в векторной базе
Args:
query: запрос для поиска
k: количество результатов
Returns:
List[str]: список найденных документов
"""
try:
results = self.vector_db.similarity_search(query, k=k)
return [doc.page_content for doc in results if hasattr(doc, 'page_content')]
except Exception as e:
print(f"Ошибка поиска в векторной БД: {e}")
return []
def query_llm(self, prompt: str, temperature: float = 0.3) -> str:
"""
Запрос к LLM модели
Args:
prompt: промпт для модели
temperature: температура генерации
Returns:
str: ответ модели
"""
payload = {
"model": "llama3:8b",
"prompt": prompt,
"temperature": temperature,
}
headers = {"Content-Type": "application/json"}
try:
response = requests.post(
self.llm_url,
json=payload,
headers=headers,
stream=True
)
response.raise_for_status()
full_response = ""
print("Генерация ответа:")
for line in response.iter_lines():
if line:
try:
data = json.loads(line.decode("utf-8"))
fragment = data.get("response", "")
full_response += fragment
print(fragment, end='', flush=True)
if data.get("done", False):
break
except json.JSONDecodeError as e:
print(f"\nОшибка декодирования ответа: {e}")
continue
print()
return full_response
except requests.exceptions.RequestException as e:
print(f"Ошибка запроса к LLM: {e}")
return ""
def prepare_prompt(self, context: List[str], query: str, output_format: str = "json") -> str:
if output_format == "json":
format_example = {
"price_number": "",
"price_name": "",
"works": [""],
"resources": [
{
"resource_name": "",
"resource_number": ""
}
]
}
return f"""Контекст:
{chr(10).join(context)}
Запрос: {query}
Ответь только в формате JSON без каких-либо дополнительных пояснений или текста:
{json.dumps([format_example], ensure_ascii=False, indent=2)}
Важно:
1. Весь ответ должен быть только валидным JSON
2. Не добавляй никакого текста до или после JSON
3. Не добавляй пояснений или примечаний
4. Если нужно указать несколько расценок, добавь их в массив
"""
else:
return f"""Контекст:
{chr(10).join(context)}
Запрос: {query}
Предоставь информацию в следующем формате:
Расценка ГЭСН: (номер)
Название: (название расценки)
Состав работ:
- (работа 1)
- (работа 2)
...
Ресурсы:
- [код ресурса] название ресурса
"""
def generate_response(self, user_query: str, k: int = 5, output_format: str = "json") -> Any:
full_query = f"выведи расценки ГЭСН, состав работ, ресурсы для: {user_query}"
context = self.search_vector_db(full_query, k)
if not context:
return [GesnFormat().to_dict()] if output_format == "json" else "Не найдено подходящих расценок"
prompt = self.prepare_prompt(context, user_query, output_format)
response = self.query_llm(prompt)
if output_format == "json":
try:
result = json.loads(response)
if isinstance(result, list):
return result
return [result]
except json.JSONDecodeError as e:
print(f"Ошибка парсинга JSON: {e}")
return [GesnFormat().to_dict()]
else:
return response
def main():
service = SmetaService("faiss_index_sectioned_all-MiniLM-L6-v2")
while True:
try:
user_query = input(">>> Запрос (или 'exit' для выхода): ")
if user_query.lower() == 'exit':
break
result = service.generate_response(user_query, k=10)
# print(json.dumps(result, ensure_ascii=False, indent=2))
except KeyboardInterrupt:
break
except Exception as e:
print(f"Произошла ошибка: {e}")
if __name__ == "__main__":
main()
Подробнее здесь: https://stackoverflow.com/questions/794 ... g-guidance
