RAGAS test.py - dev - asimakov/GigaChat

RAGAS test.py
209 строк8 KB
Anton Simakov
убрал токен
04 мар 2025, 11:07
04 мар 2025, 11:0798377cd
import ragas_utils_patch  # Импортируем патч
from langchain_gigachat.chat_models import GigaChat
import json, requests, uuid, os, re, textwrap
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from chromadb.config import Settings
from langchain_gigachat.embeddings.gigachat import GigaChatEmbeddings
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate
from langchain.schema import SystemMessage

from langchain.chains.combine_documents import create_stuff_documents_chain
import chromadb
import docx2txt
from docx import Document
from PyPDF2 import PdfReader
from langchain_community.document_loaders import Docx2txtLoader
import webbrowser

# Импорт RAGAS
from ragas import evaluate
from datasets import Dataset
from ragas.metrics import (
    faithfulness,
    answer_relevance,
    context_relevance,
    context_precision
)

# Применяем патч после загрузки всех модулей
ragas_utils_patch.apply_patch()

client_token = ''

giga = GigaChat(credentials=client_token,
                model= 'GigaChat:latest',
                verify_ssl_certs=False,
                temperature=0.1, #0..2 чем выше, тем более случайный ответ
                top_p=0.1  #0..1 чем выше тем разнообразней
               )

system_prompt = (
    "Ты помощник сотрудника Сбера, тебе будет дан текст документов.\n"
    "На основе предоставленных документов предоставь сотруднику точные инструкции по выполнению процесса. Ты не общаешься с клиентом напрямую, а объясняешь сотруднику, что нужно сделать. Не добавляй обобщений, предположений или внешней информации. Старайся цитировать текст из документов дословно.\n"
    "Если вопрос требует уточнения, попроси сотрудника предоставить дополнительные данные.\n"
    "Документы:\n\n"
    "{context}\n\n"
    "История чата:\n\n"
    "{chat_history}"
)
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        ("human", "{input}"),
    ]
)
documents_dir = 'C:\\Users\\smg3e\\PythonProjects\\GigaChat\\Documents\\Sber'
min_chunk_length = 100

documents = []
for i in os.listdir(documents_dir):
    if not i.endswith('.txt'):
        continue
    file_path = os.path.join(documents_dir, i)
    doc = TextLoader(file_path, encoding='utf-8').load()
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1500,
        chunk_overlap=300,
        separators=["\n\n", "\n", "--", ".", " "],
        keep_separator=True
    )
    split_documents = text_splitter.split_documents(doc)
    filtered_documents = [doc for doc in split_documents if len(doc.page_content) >= min_chunk_length]
    documents.extend(filtered_documents)

print(f"Кол-во частей документов: {len(documents)}")

#embeddings = GigaChatEmbeddings(
#    credentials=client_token, verify_ssl_certs=False)

#vec_db = Chroma(
#    persist_directory="./chroma_lang_db",
#    embedding_function=embeddings,
#    client_settings=Settings(anonymized_telemetry=False),
#    collection_kwargs={"hnsw:space": "cosine"}
#)

# Инициализация клиента chromadb
client = chromadb.PersistentClient(
    path="./chroma_lang_db",
    settings=Settings(anonymized_telemetry=False)
)

# Создаём коллекцию с cosine similarity
collection = client.get_or_create_collection(
    name="my_collection",  # Имя коллекции
    metadata={"hnsw:space": "cosine"}  # Указываем cosine similarity
)

# Инициализация эмбеддингов
embeddings = GigaChatEmbeddings(credentials=client_token, verify_ssl_certs=False)

# Подключаем коллекцию к Chroma из langchain
vec_db = Chroma(
    collection_name="my_collection",
    embedding_function=embeddings,
    client=client  # Передаём клиент с настроенной коллекцией
)

batch_size = 30
for i in range(0, len(documents), batch_size):
    batch = documents[i:i + batch_size]
    vec_db.add_documents(batch)

#vec_db = (Chroma.from_documents
#          (documents, embeddings,
#           persist_directory="./chroma_lang_db",
#           client_settings=Settings(anonymized_telemetry=False)))


retriever=vec_db.as_retriever(search_type="similarity_score_threshold",
                              search_kwargs={"k": 4, "score_threshold": 0.8478})
qa_chain = create_stuff_documents_chain(giga, prompt) #цепочка вызывает giga с prompt
rag_chain = create_retrieval_chain(retriever, qa_chain) #цепочка извлекает документы и передает в qa_chain

# Списки для RAGAS
questions = []
answers = []
contexts = []
ground_truths = []

#chat_history = []

while True:
    user_rag_question = str(input('Введите запрос: '))
    if user_rag_question.lower() in ['стоп', 'stop']:
        break
    history_str = ""
    request = rag_chain.invoke({"input": user_rag_question, "chat_history": history_str})
    answer = request["answer"]
    request_context = request['context']
    unique_sources = ()
    unique_sources = set(doc.metadata['source'] for doc in request_context)
    print('-----------')
    print(answer)
    # print("\n".join(textwrap.wrap(answer, width=100)))
    print('-----------' + '\n')
    if unique_sources:
        print('Ответ на основании документов: ')
        for source in unique_sources:
            print(f"- {source}")
            #webbrowser.open(source)
    else:
        print('Документы не использованы в ответе')

    # Сбор данных для RAGAS
    questions.append(user_rag_question)
    answers.append(answer)
    contexts.append([doc.page_content for doc in request_context])
    # Если есть ground truth:
    # ground_truths.append("Ожидаемый ответ...")

    #similarity_search_with_score для получения документов и их скора
    docs_with_scores = vec_db.similarity_search_with_score(user_rag_question, k=3)

    for doc, score in docs_with_scores:
        source = doc.metadata.get('source', 'Неизвестный источник')
        print(f"Chunk:  {source} (score: {1-score:.4f})")

    print('\n')

    # Оценка с RAGAS
    if questions:  # Проверяем, что есть данные для оценки
        ragas_utils_patch.apply_patch()
        # Создаём Dataset для RAGAS
        eval_dataset = Dataset.from_dict({
            "question": questions,
            "answer": answers,
            "contexts": contexts,
            # "ground_truth": ground_truths  # Раскомментируйте, если есть ground truth
        })

        # Выполняем оценку
        result = evaluate(
            dataset=eval_dataset,
            metrics=[
                faithfulness,
                answer_relevance,
                context_relevance,
                context_precision
            ],
            llm=giga,
            embeddings=embeddings
        )

        # Вывод результатов
        print("\nРезультаты оценки RAGAS:")
        print(result)

'''
#Проверка состава чанков
for i in documents:
    print(i)
    print('-----------------')
    print('-----------------\n')
'''