from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0)
def should_retrieve(question: str) -> bool:
"""Decide if retrieval is needed."""
response = llm.invoke(
f"""Does this question require looking up specific documents
or information to answer accurately?
Question: {question}
Answer with 'yes' or 'no'."""
)
return "yes" in response.content.lower()
def check_relevance(question: str, document: str) -> bool:
"""Check if a document is relevant (ISREL)."""
response = llm.invoke(
f"""Is this document relevant to answering the question?
Question: {question}
Document: {document}
Answer 'relevant' or 'irrelevant'."""
)
return "relevant" in response.content.lower()
def check_support(claim: str, evidence: str) -> str:
"""Check if evidence supports a claim (ISSUP)."""
response = llm.invoke(
f"""Does the evidence support this claim?
Claim: {claim}
Evidence: {evidence}
Answer: 'fully_supported', 'partially_supported', or 'not_supported'."""
)
return response.content.strip().lower()
def check_utility(question: str, answer: str) -> int:
"""Rate the utility of the answer (ISUSE), 1-5."""
response = llm.invoke(
f"""Rate how well this answer addresses the question (1-5).
Question: {question}
Answer: {answer}
Rating (just the number):"""
)
try:
return int(response.content.strip())
except ValueError:
return 3
def self_rag_pipeline(question: str) -> str:
# Step 1: Decide if retrieval is needed
if should_retrieve(question):
docs = retriever.invoke(question)
# Step 2: Filter relevant documents
relevant_docs = [d for d in docs if check_relevance(question, d.page_content)]
if not relevant_docs:
return "I don't have sufficient information to answer this question."
context = "\n".join(d.page_content for d in relevant_docs)
else:
context = ""
# Step 3: Generate answer
prompt = f"Context: {context}\n\nQuestion: {question}\n\nAnswer:" if context else f"Question: {question}\n\nAnswer:"
answer = llm.invoke(prompt).content
# Step 4: Check support and utility
if context:
support = check_support(answer, context)
if support == "not_supported":
# Regenerate with stricter grounding instruction
answer = llm.invoke(
f"Based ONLY on this context, answer the question. "
f"If the context doesn't contain the answer, say so.\n"
f"Context: {context}\nQuestion: {question}\nAnswer:"
).content
utility = check_utility(question, answer)
if utility <= 2:
answer += "\n\n(Note: This answer may be incomplete. Consider refining your question.)"
return answer