Мобильная версияfrom langchain.chains import RetrievalQA
from langchain_ollama import OllamaLLM
from langchain.prompts import PromptTemplate
from langchain_community.vectorstores import FAISS
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import EmbeddingsFilter
import gradio as gr
import re
import csv
import os
import nltk # Corrected import name
from nltk.util import ngrams
import gc
# --- CORRECTED NLTK DOWNLOADER SECTION ---
# This block will now correctly detect if 'punkt' is missing and download it.
try:
nltk.data.find('tokenizers/punkt')
except LookupError:
print("
NLTK 'punkt' resource not found. Downloading now...")nltk.download('punkt')
print("
Download complete.")# --- END OF CORRECTION ---
# === Feedback logging setup ===
FEEDBACK_FILE = "feedback_log.csv"
hf_embed = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")
vectorstore = FAISS.load_local("faiss_index", hf_embed, allow_dangerous_deserialization=True)
llm = OllamaLLM(model="mistral:instruct", temperature=0)
from langchain.prompts import PromptTemplate
prompt = PromptTemplate.from_template("""
You are a highly accurate academic assistant. Use the provided context to answer the user’s question in a concise, structured, and logical manner. Follow these rules:
1. Start with a clear definition or explanation of the main concept.
2. Include only information directly relevant to the user’s question.
3. Present eligibility, conditions, thresholds, or requirements next.
4. Explain the decision or application process clearly.
5. Add practical implications or guidance if it helps the user understand or act on it.
6. Include caveats only if they affect eligibility, rules, or outcomes.
7. Avoid unnecessary details, historical exceptions, or rare edge cases unless directly requested.
8. Organize the response logically; prioritize importance and clarity.
9. Do not create titles or sections; write as a continuous explanation.
10. Use examples only if they clarify rules or thresholds.
Context: {context}
Question: {question}
Answer:
""")
retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={"k": 8, "fetch_k": 20, "lambda_mult": 0.7}
)
# Optional: Use MultiQueryRetriever for better diversity
multi_retriever = MultiQueryRetriever.from_llm(
retriever=retriever,
llm=llm
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=retriever,
chain_type="stuff",
chain_type_kwargs={"prompt": prompt},
return_source_documents=True # CRITICAL FIX: To get metadata back
)
from sentence_transformers import SentenceTransformer, util
# Load embedder (same as used in your vectorstore)
embedder_eval = SentenceTransformer("sentence-transformers/all-mpnet-base-v2")
# Optional: Logging to CSV
EVAL_LOG_FILE = "evaluation_log.csv"
if not os.path.exists(EVAL_LOG_FILE):
with open(EVAL_LOG_FILE, mode='w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(["Question", "Answer", "QA_Similarity", "Answer_Context_Similarity", "Recall_Score"])
def ask_question(message, history):
try:
message_clean = re.sub(r"[^a-zA-Z0-9\s.,?!:\-]", "", message.lower().strip())
if not message_clean or message_clean in ["?", "what is ?", "what is"]:
return "Please ask a more specific question
"# CRITICAL FIX: Get the full result dictionary instead of just the string
result = qa_chain({"query": message_clean})
response = result['result']
docs = result['source_documents']
print("\n
Retrieved Chunks:")for i, doc in enumerate(docs):
print(f"\n--- Chunk {i+1} ---")
print(f"Source: {doc.metadata.get('source', 'N/A')}")
print(doc.page_content[:500]) # show only first 500 characters
context = "\n".join([doc.page_content for doc in docs])
# Evaluation metrics
q_emb = embedder_eval.encode(message_clean, convert_to_tensor=True)
a_emb = embedder_eval.encode(response, convert_to_tensor=True)
c_emb = embedder_eval.encode(context, convert_to_tensor=True)
qa_sim = util.cos_sim(q_emb, a_emb).item() if a_emb.nelement() > 0 else 0
ac_sim = util.cos_sim(a_emb, c_emb).item() if a_emb.nelement() > 0 and c_emb.nelement() > 0 else 0
recall_score = context_recall(response, context, n=5)
# Extract sources
source_set = {doc.metadata.get("source", "Unknown source") for doc in docs}
sources_str = "\n
Sources:\n" + "\n".join(f"- {s}" for s in sorted(list(source_set)))warning = ""
if recall_score < 0.3:
warning = "\n
Warning: This answer may not be well-grounded in the retrieved documents."# Log result
with open(EVAL_LOG_FILE, mode='a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow([message_clean, response, f"{qa_sim:.3f}", f"{ac_sim:.3f}", f"{recall_score:.3f}"])
return f"{response}\n\n
(QA Sim: {qa_sim:.2f}, Context Sim: {ac_sim:.2f}, Recall: {recall_score:.2f}){warning}{sources_str}"except Exception as e:
print(f"An error occurred: {e}")
return "Sorry, I encountered an error while processing your request."
finally:
gc.collect()
# --- Context Recall Function ---
# Updated to use NLTK's tokenizer for more accuracy
def context_recall(answer, context, n=5):
answer_lower = answer.lower()
context_lower = context.lower()
try:
answer_words = nltk.word_tokenize(answer_lower)
except Exception:
answer_words = answer_lower.split()
if len(answer_words) < n:
return 0.0
answer_ngrams = list(ngrams(answer_words, n))
if not answer_ngrams:
return 0.0
match_count = sum(1 for gram in answer_ngrams if " ".join(gram) in context_lower)
recall = match_count / len(answer_ngrams)
return round(recall, 3)
if not os.path.exists(FEEDBACK_FILE):
with open(FEEDBACK_FILE, mode='w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(["Question", "Response", "Feedback"])
def store_feedback(question, response, feedback):
with open(FEEDBACK_FILE, mode='a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow([question, response, feedback])
print(f"
Feedback saved: {feedback}")# === YOUR ORIGINAL GRADIO UI - UNCHANGED ===
with gr.Blocks(theme=gr.themes.Soft()) as demo:
gr.Markdown("#
Ethical Chatbot")gr.Markdown("Ask me questions about university ethical guidelines.")
state = gr.State([]) # For chat history
last_response = gr.State("") # For feedback reference
with gr.Row(equal_height=True):
with gr.Column(scale=12):
chatbot = gr.Chatbot(
label="Ethical Guidelines Chatbot",
height=400,
show_copy_button=True,
# The 'type' parameter is deprecated, Gradio handles this automatically
# type="messages"
)
with gr.Column(scale=1,min_width=40):
with gr.Row(equal_height=False): # min_height is deprecated
with gr.Column():
label = gr.Label("Flag", visible=False) # Hidden label for feedback
with gr.Column(min_width=20): # Buttons take less space
thumbs_up = gr.Button("
")thumbs_down = gr.Button("
")with gr.Row(equal_height=True):
with gr.Column(scale=10): # Make textbox take more space
msg = gr.Textbox(
placeholder="Type your question here...",
show_label=False,
lines=1,
interactive=True
)
with gr.Column(scale=1,min_width=50): # Button takes less space
send_btn = gr.Button("➤")
# Step 1: Append user message immediately, clear input
def user_submit(message, history):
# Gradio now uses a list of lists for history, not a list of dicts
history.append([message, None])
return history, gr.update(value=""), history
# Step 2: Generate bot response and update last message in chat history
def bot_respond(history):
if not history:
return history, "No message received."
last_user_msg = history[-1][0]
answer = ask_question(last_user_msg, history)
history[-1][1] = answer # Update the 'None' placeholder
return history, answer
# Connect submit and send button to two-step chain
msg.submit(user_submit, inputs=[msg, state], outputs=[chatbot, msg, state]).then(
bot_respond, inputs=state, outputs=[chatbot, last_response]
)
send_btn.click(user_submit, inputs=[msg, state], outputs=[chatbot, msg, state]).then(
bot_respond, inputs=state, outputs=[chatbot, last_response]
)
# Feedback handler
def feedback_handler(value, chatbot_state, last_response):
# Adjusting for the new list-of-lists history format
if chatbot_state and len(chatbot_state) >= 1:
last_q_a_pair = chatbot_state[-1]
if len(last_q_a_pair) == 2 and last_q_a_pair[1] is not None:
last_q = last_q_a_pair[0]
response = last_q_a_pair[1]
store_feedback(last_q, response, value)
gr.Info(f"Feedback '{value}' stored. Thank you!")
return gr.update()
thumbs_up.click(
fn=lambda state_val, last_resp_val: feedback_handler("Yes", state_val, last_resp_val),
inputs=[state, last_response]
)
thumbs_down.click(
fn=lambda state_val, last_resp_val: feedback_handler("No", state_val, last_resp_val),
inputs=[state, last_response]
)
if __name__ == "__main__":
demo.launch(inbrowser=True,share=True)```
Подробнее здесь: https://stackoverflow.com/questions/797 ... ving-all-c
