Spaces:
Running
Running
File size: 3,831 Bytes
e43c82d | 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 108 109 110 111 112 113 114 115 | """Answer generator for RAG: Combines retrieved context with the user question.
Augments the prompt sent to the LLM with relevant retrieved chunks.
Provides an extractive fallback mode if no API key is set so the entire pipeline
runs 100% offline out-of-the-box.
"""
from config.prompts import INSUFFICIENT_EVIDENCE, SYSTEM_PROMPT
def format_context(retrieved_items):
"""Format retrieved chunks into a clean, numbered context block for the prompt."""
if not retrieved_items:
return "No relevant context found."
blocks = []
for idx, item in enumerate(retrieved_items, start=1):
chunk = item["chunk"]
title = chunk.title or f"Document {chunk.doc_id}"
blocks.append(f"[Source {idx}: {title}]\n{chunk.text}")
return "\n\n".join(blocks)
def extractive_fallback(question, retrieved_items):
"""Offline extractive summary when no LLM API key is configured."""
if not retrieved_items:
return INSUFFICIENT_EVIDENCE
lines = [
f"Answer synthesized from indexed documents for: '{question}'",
"",
"Key relevant excerpts from your knowledge base:",
]
for idx, item in enumerate(retrieved_items[:3], start=1):
chunk = item["chunk"]
snippet = chunk.text.strip().replace("\n", " ")
if len(snippet) > 280:
snippet = snippet[:280] + "..."
lines.append(f"{idx}. [Source: {chunk.title}] \"{snippet}\"")
lines.append("")
lines.append(
"*Note: Set USER_LLM_API_KEY in your .env file to enable full generative AI answers with OpenAI/Claude.*"
)
return "\n".join(lines)
def generate_answer(question, retrieved_items, settings):
"""Generate an answer using OpenAI LLM or graceful offline fallback."""
if not retrieved_items:
return {
"answer": INSUFFICIENT_EVIDENCE,
"sources": [],
"used_llm": False,
}
# Extract unique source titles
sources = []
seen_titles = set()
for item in retrieved_items:
title = item["chunk"].title
if title and title not in seen_titles:
seen_titles.add(title)
sources.append(title)
# Check if LLM API key is present
if not settings.has_llm_key():
return {
"answer": extractive_fallback(question, retrieved_items),
"sources": sources,
"used_llm": False,
}
# Generate with LLM (OpenAI via LangChain)
try:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
context = format_context(retrieved_items)
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("human", "{question}"),
])
llm_kwargs = {
"api_key": settings.user_llm_api_key,
"model": settings.user_llm_model,
"temperature": settings.llm_temperature,
"max_tokens": settings.llm_max_tokens,
}
if settings.user_llm_base_url:
llm_kwargs["base_url"] = settings.user_llm_base_url
llm = ChatOpenAI(**llm_kwargs)
chain = prompt | llm | StrOutputParser()
raw_answer = chain.invoke({
"context": context,
"question": question,
})
return {
"answer": raw_answer.strip(),
"sources": sources,
"used_llm": True,
}
except Exception as exc:
# If LLM API call fails (e.g. invalid key or network error), fallback cleanly
return {
"answer": f"{extractive_fallback(question, retrieved_items)}\n\n*(LLM generation failed: {str(exc)})*",
"sources": sources,
"used_llm": False,
}
|