Spaces:
Sleeping
Sleeping
File size: 4,097 Bytes
c914940 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
MEXAR - LangChain baseline for Table I.
Standard LangChain RetrievalQA pipeline with Chroma vector store and default config,
built from the SAME source documents as the MEXAR corpus.
"""
import os
import logging
from typing import Dict, Any, List
from core.database import SessionLocal
from models.chunk import DocumentChunk
from utils.groq_client import get_groq_client
logger = logging.getLogger(__name__)
def build_langchain_pipeline(agent_id: int, persist_dir: str = "./chroma_baseline_db"):
"""
Build or load a LangChain RetrievalQA chain over the specified agent's document chunks.
Uses HuggingFace BAAI/bge-small-en-v1.5 to match MEXAR's embedding space for fair evaluation.
"""
try:
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.chains import RetrievalQA
from langchain_groq import ChatGroq
db = SessionLocal()
try:
chunks = db.query(DocumentChunk).filter(DocumentChunk.agent_id == agent_id).all()
texts = [c.content for c in chunks if c.content]
metadatas = [{"source": c.source, "chunk_index": c.chunk_index} for c in chunks if c.content]
finally:
db.close()
if not texts:
logger.warning(f"No texts found for agent_id {agent_id} in LangChain baseline setup")
return None
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5")
vectorstore = Chroma.from_texts(
texts=texts,
embedding=embeddings,
metadatas=metadatas,
persist_directory=f"{persist_dir}_{agent_id}"
)
groq_api_key = os.environ.get("GROQ_API_KEY", "")
llm = ChatGroq(model="llama3-8b-8192", groq_api_key=groq_api_key)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
return_source_documents=True,
)
return qa_chain
except Exception as e:
logger.error(f"Failed to build LangChain pipeline: {e}")
return None
def run_langchain_baseline(qa_chain, query: str, engine=None) -> Dict[str, Any]:
"""
Execute LangChain baseline QA chain for a given query.
Fallback to Groq direct generation if LangChain libraries are not initialized.
"""
if qa_chain is not None:
try:
result = qa_chain.invoke({"query": query})
answer = result.get("result", "")
sources = [d.metadata.get("source", "") for d in result.get("source_documents", [])]
chunk_texts = [d.page_content for d in result.get("source_documents", [])]
faithfulness = 0.5
if engine and hasattr(engine, 'deberta_nli_scorer'):
faith_res = engine.deberta_nli_scorer.score(answer, chunk_texts if chunk_texts else [""])
faithfulness = faith_res.score
return {
"answer": answer,
"confidence": faithfulness,
"in_domain": True,
"retrieved_chunk_doc_ids": sources,
"faithfulness": faithfulness
}
except Exception as e:
logger.error(f"LangChain invocation failed: {e}")
# Robust fallback using direct Groq RAG if langchain dependencies unavailable
if engine:
client = get_groq_client()
sys_prompt = "You are a LangChain RetrievalQA baseline model. Answer the question directly using standard RAG context."
answer = client.analyze_with_system_prompt(sys_prompt, query, model="chat")
return {
"answer": answer,
"confidence": 0.5,
"in_domain": True,
"retrieved_chunk_doc_ids": [],
"faithfulness": 0.5
}
return {
"answer": "LangChain pipeline unavailable.",
"confidence": 0.0,
"in_domain": True,
"retrieved_chunk_doc_ids": [],
"faithfulness": 0.0
}
|