# generator.py
import re
import os
from typing import Generator
from dotenv import load_dotenv
load_dotenv()
HF_TOKEN = os.getenv("HF_TOKEN", "")
# How many prior turns (user + assistant pairs) to keep in context.
MAX_HISTORY_TURNS = 5
_DEFAULT_MODEL = "meta-llama/Llama-3.1-8B-Instruct"
_MODEL_MAP = {
"qwen3:14b": _DEFAULT_MODEL,
"llama3.1:8b": _DEFAULT_MODEL,
"mistral:7b": _DEFAULT_MODEL,
"HuggingFaceH4/zephyr-7b-beta": _DEFAULT_MODEL,
"mistralai/Mistral-7B-Instruct-v0.3": _DEFAULT_MODEL,
"mistralai/Mistral-7B-Instruct-v0.1": _DEFAULT_MODEL,
"Qwen/Qwen2.5-7B-Instruct": _DEFAULT_MODEL,
"microsoft/Phi-3.5-mini-instruct": _DEFAULT_MODEL,
}
class ChatGenerator:
def __init__(self, model=None):
if not HF_TOKEN:
raise EnvironmentError(
"HF_TOKEN secret is missing.\n"
"Go to: Space → Settings → Repository secrets → add HF_TOKEN.\n"
"Create a token at huggingface.co/settings/tokens with "
"'Make calls to Inference Providers' permission."
)
raw = model or os.getenv("HF_LLM_MODEL", _DEFAULT_MODEL)
self.model = _MODEL_MAP.get(raw, _DEFAULT_MODEL)
# app.py sets this attribute before calling generate_answer() to pass
# conversation history without changing the healing pipeline's signature.
self._conversation_history: list[dict] = []
from huggingface_hub import InferenceClient
self._client = InferenceClient(
provider="auto",
api_key=HF_TOKEN,
)
# ── Internal helpers ────────────────────────────────────────────────────
def _build_messages(self, system: str, user: str) -> list[dict]:
"""
Build the full messages list with conversation history injected:
[system]
[prior user/assistant pairs, capped to MAX_HISTORY_TURNS]
[current user message]
"""
messages: list[dict] = [{"role": "system", "content": system}]
history = self._conversation_history or []
for turn in history[-MAX_HISTORY_TURNS:]:
if turn.get("user"):
messages.append({"role": "user", "content": turn["user"]})
assistant_text = turn.get("assistant", "")
# Skip the "..." placeholder inserted while a turn is in-flight
if assistant_text and assistant_text != "…":
messages.append({"role": "assistant", "content": assistant_text})
messages.append({"role": "user", "content": user})
return messages
def _strip_think_tags(self, text: str) -> str:
return re.sub(r".*?", "", text, flags=re.DOTALL).strip()
def _build_rag_prompts(self, query: str, relevant_docs: list) -> tuple[str, str]:
"""Build system + user prompts from retrieved documents."""
context_parts = []
source_labels = []
for i, (doc, score) in enumerate(relevant_docs, start=1):
meta = doc.metadata
filename = meta.get("filename") or meta.get("source") or f"Document {i}"
chunk_id = meta.get("chunk_id", "")
sfx = f" · chunk {chunk_id}" if chunk_id != "" else ""
label = f"Source {i}"
source_labels.append(label)
context_parts.append(f"[{label}: {filename}{sfx}]\n{doc.page_content}")
context = "\n\n---\n\n".join(context_parts)
source_list = ", ".join(source_labels)
system = f"""You are a document Q&A assistant. Answer ONLY from the CONTEXT PASSAGES below.
RULES:
1. Use ONLY information from the CONTEXT PASSAGES. No outside knowledge.
2. Valid citation labels: {source_list}
3. Cite inline as (Source 1) or (Source 2). No other format.
4. Never invent section names, page numbers, or labels not listed above.
5. If the answer is not in the context, say: "This information is not in the uploaded documents."
6. You may use prior conversation turns to understand follow-up questions, but only
draw factual answers from the CONTEXT PASSAGES.
CONTEXT PASSAGES:
{context}"""
user = f"Answer using ONLY the context. Cite with: {source_list}\n\nQuestion: {query}"
return system, user
# ── Public API: non-streaming ───────────────────────────────────────────
def _call_hf(self, system: str, user: str) -> str:
"""Non-streaming completion. Injects conversation history automatically."""
messages = self._build_messages(system, user)
try:
completion = self._client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=1024,
temperature=0.0,
)
answer = completion.choices[0].message.content or ""
except Exception as exc:
raise RuntimeError(f"HF Inference API call failed: {exc}") from exc
return self._strip_think_tags(answer)
def generate_answer(self, query: str, relevant_docs: list) -> str:
"""
Return the full answer as a string (non-streaming).
Conversation memory comes from self._conversation_history, which
app.py sets before each call:
generator._conversation_history = history_for_context
"""
if not relevant_docs:
return "No relevant documents were found. Please upload and process documents first."
system, user = self._build_rag_prompts(query, relevant_docs)
return self._call_hf(system=system, user=user)
# ── Public API: streaming ───────────────────────────────────────────────
def generate_answer_stream(self, query: str, relevant_docs: list) -> Generator[str, None, None]:
"""
Yield answer tokens one by one as they arrive from the API.
Use with Streamlit's st.write_stream():
answer = st.write_stream(generator.generate_answer_stream(q, docs))
st.write_stream() renders tokens live and returns the full string when done.
Conversation memory is read from self._conversation_history (same as
generate_answer).
"""
if not relevant_docs:
yield "No relevant documents were found. Please upload and process documents first."
return
system, user = self._build_rag_prompts(query, relevant_docs)
messages = self._build_messages(system, user)
try:
stream = self._client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=1024,
temperature=0.0,
stream=True,
)
except Exception as exc:
yield f"⚠️ Stream failed: {exc}"
return
# Buffer for … stripping across chunk boundaries
buffer = ""
inside_think = False
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
if not delta:
continue
buffer += delta
if not inside_think:
if "" in buffer:
before, _, rest = buffer.partition("")
if before:
yield before
buffer = rest
inside_think = True
else:
safe_len = max(0, len(buffer) - len(""))
if safe_len > 0:
yield buffer[:safe_len]
buffer = buffer[safe_len:]
else:
if "" in buffer:
_, _, rest = buffer.partition("")
buffer = rest
inside_think = False
else:
buffer = "" # still inside , discard
# Flush any remaining buffer
if buffer and not inside_think:
yield buffer.strip()