AnveshAI-Edge-V2 / llm_engine.py
developeranveshraman's picture
Update llm_engine.py
44b9df4 verified
Raw
History Blame Contribute Delete
7.1 kB
"""
LLM Engine — Qwen2.5-1.5B-Instruct via HuggingFace InferenceClient.
Primary : huggingface_hub InferenceClient (no compilation, works on HF Spaces)
Fallback : llama-cpp-python local GGUF (offline / self-hosted)
This is the bottom layer of the AnveshAI hierarchy:
Math → math_engine (instant, rule-based)
Knowledge → knowledge_engine (keyword retrieval from knowledge.txt)
└─ no match → LLMEngine.generate (Qwen2.5-1.5B)
Conversation → conversation_engine (pattern matching from conversation.txt)
└─ no match → LLMEngine.generate (Qwen2.5-1.5B)
"""
import os
MODEL_REPO = "Qwen/Qwen2.5-1.5B-Instruct-GGUF"
MODEL_FILE = "qwen2.5-1.5b-instruct-q4_k_m.gguf"
HF_MODEL_ID = "Qwen/Qwen2.5-72B-Instruct"
SYSTEM_PROMPT = (
"You are AnveshAI Edge, an expert AI tutor specialising in JEE Advanced "
"(Physics, Chemistry, and Mathematics). "
"Your answers must meet the rigour and depth expected at JEE Advanced level. "
"Always show complete, step-by-step working. Use standard JEE notation. "
"State the relevant formula or principle before applying it. "
"Do not repeat the question back. "
"If you are unsure about something, say so clearly. "
"Prefer concise, exam-focused explanations over verbose prose."
)
MATH_SYSTEM_PROMPT = (
"You are an expert JEE Advanced mathematics tutor. "
"You will be given a VERIFIED ANSWER computed by a symbolic engine. "
"That answer is 100% correct — do NOT change it, do NOT recompute it. "
"Your ONLY job is to explain, step by step, HOW a JEE Advanced student "
"would work through the problem and arrive at that exact answer. "
"Use JEE Advanced standard techniques (e.g. substitution, IBP, "
"L'Hopital, cofactor expansion, characteristic equation, etc.). "
"Every step must lead logically toward the verified answer. "
"State the verified answer word-for-word at the end of your explanation. "
"Keep the explanation concise and exam-focused."
)
CHAT_SYSTEM_PROMPT = (
"You are AnveshAI, a friendly JEE Advanced study assistant. "
"When the user sends a casual or non-academic message, reply briefly and "
"naturally in 1-2 sentences — like a helpful study buddy, not an academic paper. "
"Never analyse casual phrases or exclamations as if they are problems. "
"If you cannot understand the message, politely ask what they need help with."
)
MAX_TOKENS = 1024
TEMPERATURE = 0.7
MATH_TEMPERATURE = 0.1
TOP_P = 0.9
N_CTX = 4096
class LLMEngine:
"""
Two-backend LLM wrapper:
1. HuggingFace InferenceClient — used on HF Spaces / any machine with internet
2. llama-cpp-python local GGUF — used when HF Inference is unavailable
Both expose identical generate() semantics so the rest of the codebase
doesn't need to know which backend is active.
"""
def __init__(self) -> None:
self._client = None # HF InferenceClient instance
self._llm = None # llama_cpp Llama instance
self._backend = None # "hf" | "local" | None
self._failed = False
self._fail_reason = ""
self._load()
# ------------------------------------------------------------------
# Initialisation
# ------------------------------------------------------------------
def _load(self) -> None:
if self._try_hf_client():
return
self._try_local_llama()
def _try_hf_client(self) -> bool:
"""Attempt to initialise the HuggingFace InferenceClient."""
try:
from huggingface_hub import InferenceClient
token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_API_TOKEN")
self._client = InferenceClient(model=HF_MODEL_ID, token=token)
self._backend = "hf"
print(f" [LLM] Backend: HuggingFace InferenceClient ({HF_MODEL_ID})", flush=True)
return True
except Exception as exc:
print(f" [LLM] HF InferenceClient unavailable: {exc}", flush=True)
return False
def _try_local_llama(self) -> None:
"""Fall back to local llama-cpp-python GGUF model."""
try:
from llama_cpp import Llama
print(
f" [LLM] Loading local {MODEL_FILE} "
"(first run downloads ~1 GB, then cached) …",
flush=True,
)
self._llm = Llama.from_pretrained(
repo_id=MODEL_REPO,
filename=MODEL_FILE,
n_ctx=N_CTX,
n_threads=4,
verbose=False,
)
self._backend = "local"
print(" [LLM] Backend: local llama-cpp Qwen2.5-1.5B-Instruct", flush=True)
except Exception as exc:
self._failed = True
self._fail_reason = str(exc)
print(f" [LLM] Both backends failed. Last error: {exc}", flush=True)
def is_available(self) -> bool:
return self._backend is not None and not self._failed
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def generate(
self,
user_input: str,
context: str = "",
system_prompt: str = "",
temperature: float = TEMPERATURE,
) -> str:
if self._failed or self._backend is None:
return (
f"LLM unavailable ({self._fail_reason}). "
"Check that huggingface_hub is installed and you have internet access."
)
system_content = system_prompt if system_prompt else SYSTEM_PROMPT
if context:
system_content += f"\n\nRelevant background:\n{context}"
messages = [
{"role": "system", "content": system_content},
{"role": "user", "content": user_input},
]
if self._backend == "hf":
return self._generate_hf(messages, temperature)
return self._generate_local(messages, temperature)
def _generate_hf(self, messages: list[dict], temperature: float) -> str:
try:
response = self._client.chat_completion(
messages=messages,
max_tokens=MAX_TOKENS,
temperature=max(temperature, 0.01),
top_p=TOP_P,
)
return response.choices[0].message.content.strip()
except Exception as exc:
return f"HF inference error: {exc}"
def _generate_local(self, messages: list[dict], temperature: float) -> str:
try:
output = self._llm.create_chat_completion(
messages=messages,
max_tokens=MAX_TOKENS,
temperature=temperature,
top_p=TOP_P,
)
return output["choices"][0]["message"]["content"].strip()
except Exception as exc:
return f"LLM generation error: {exc}"