""" KG Embedding Server — Unified Smart Search FastAPI on HuggingFace Spaces Flow at startup: 1. Load embedding model (sentence-transformers) + FAISS + numpy 2. Scan KNOWLEDGE_PATH -> chunk -> embed -> build FAISS index 3. Initialise web-search provider (Tavily / Jina) 4. Initialise LLM (Cerebras / Cohere) via LangChain Endpoints: POST /search — UNIFIED INTELLIGENT SEARCH (Local KB + Web + LLM/Agent) POST /reload — re-scan folder and rebuild FAISS index GET /health — status check (includes web/LLM status) GET /debug — list knowledge base files GET /ping — keep-alive """ from fastapi import FastAPI, HTTPException, Header, Depends from pydantic import BaseModel from typing import Any, Optional import os import re import json import contextlib import urllib.request import urllib.parse import hmac # 🔒 لمقارنة الـ API Key بشكل آمن from io import StringIO app = FastAPI(title="KG Embedding Server — Unified Smart Search") # ══════════════════════════════════════════════════════ # CONFIG # ══════════════════════════════════════════════════════ # Local KB KNOWLEDGE_PATH = "./Groovy Tutorials" CHUNK_SIZE_WORDS = int(os.environ.get("CHUNK_SIZE_WORDS", 300)) CHUNK_OVERLAP_WORDS = int(os.environ.get("CHUNK_OVERLAP_WORDS", 50)) EMBED_MODEL_NAME = os.environ.get("EMBED_MODEL_NAME", "all-MiniLM-L6-v2") ALLOWED_EXTENSIONS = (".md", ".txt") # Web Search TAVILY_API_KEY = os.environ.get("TAVILY_API_KEY", "") JINA_API_KEY = os.environ.get("JINA_API_KEY", "") WEB_SEARCH_PROVIDER = os.environ.get("WEB_SEARCH_PROVIDER", "tavily") MAX_WEB_RESULTS = int(os.environ.get("MAX_WEB_RESULTS", 3)) MAX_WEB_CONTENT_CHARS = int(os.environ.get("MAX_WEB_CONTENT_CHARS", 1500)) WEB_INCLUDE_DOMAINS = [ d.strip() for d in os.environ.get( "WEB_INCLUDE_DOMAINS", "community.atlassian.com,support.atlassian.com,docs.atlassian.com," "developer.atlassian.com,adaptavist.com,scriptrunner-docs.adaptavist.com," "blog.adaptavist.com,jira.atlassian.com" ).split(",") if d.strip() ] # LLM CEREBRAS_API_KEY = os.environ.get("CEREBRAS_API_KEY", "") COHERE_API_KEY = os.environ.get("COHERE_API_KEY", "") LLM_PROVIDER = os.environ.get("LLM_PROVIDER", "cerebras") LLM_MODEL = os.environ.get("LLM_MODEL", "") LLM_TEMPERATURE = float(os.environ.get("LLM_TEMPERATURE", 0.1)) MAX_FINAL_ANSWER_TOKENS = int(os.environ.get("MAX_FINAL_ANSWER_TOKENS", 2000)) if not LLM_MODEL: LLM_MODEL = "llama3.1-70b" if LLM_PROVIDER == "cerebras" else "command-r-plus-08-2024" # Agent USE_AGENT = os.environ.get("USE_AGENT", "false").lower() == "true" AGENT_MAX_ITERATIONS = int(os.environ.get("AGENT_MAX_ITERATIONS", 4)) # ── Auth (API Key) ── API_SECRET_KEY = os.environ.get("API_SECRET_KEY", "") AUTH_ENABLED = bool(API_SECRET_KEY) # ══════════════════════════════════════════════════════ # GLOBALS # ══════════════════════════════════════════════════════ _model: Any = None _use_st = False _faiss: Any = None _np: Any = None _index: Any = None _chunks: list[dict] = [] _llm: Any = None _llm_provider_name: Optional[str] = None _web_search_available = False _web_search_provider_name: Optional[str] = None _agent_executor: Any = None @app.on_event("startup") def startup(): global _model, _use_st, _faiss, _np if AUTH_ENABLED: print("[Server] API Key Authentication: ENABLED ✓") else: print("[Server] API Key Authentication: DISABLED (Running in open mode)") try: from sentence_transformers import SentenceTransformer with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()): _model = SentenceTransformer(EMBED_MODEL_NAME) _use_st = True print(f"[Server] embedding model '{EMBED_MODEL_NAME}' loaded ✓") except Exception as e: print(f"[Server] sentence-transformers unavailable: {e}") try: import faiss as _faiss_mod _faiss = _faiss_mod print("[Server] faiss loaded ✓") except Exception as e: print(f"[Server] faiss unavailable: {e}") try: import numpy as np _np = np print("[Server] numpy loaded ✓") except Exception as e: print(f"[Server] numpy unavailable: {e}") build_index() init_web_search() init_llm() if USE_AGENT and _llm is not None: try: _create_agent_executor() except Exception as e: print(f"[Server] agent creation failed: {e}") # ══════════════════════════════════════════════════════ # AUTH — API Key verification # ══════════════════════════════════════════════════════ def verify_api_key( x_api_key: Optional[str] = Header(None, alias="X-API-Key"), authorization: Optional[str] = Header(None) ): """ بيقبل المفتاح إما: - Header: X-API-Key: - Header: Authorization: Bearer لو API_SECRET_KEY فاضي على السيرفر → بيمشى عادي (graceful degradation). """ if not API_SECRET_KEY: return # السيرفر مش متظبط عليه مفتاح → اسمح بكل الطلبات provided = x_api_key if not provided and authorization: parts = authorization.split(" ", 1) if len(parts) == 2 and parts[0].lower() == "bearer": provided = parts[1].strip() if not provided or not hmac.compare_digest(provided, API_SECRET_KEY): raise HTTPException( status_code=401, detail="Invalid or missing API key", headers={"WWW-Authenticate": 'ApiKey realm="kg-embedding"'}, ) # ══════════════════════════════════════════════════════ # INIT FUNCTIONS # ══════════════════════════════════════════════════════ def init_web_search(): global _web_search_available, _web_search_provider_name if TAVILY_API_KEY: _web_search_available = True _web_search_provider_name = "tavily" print("[Server] web search: Tavily ✓") elif JINA_API_KEY: _web_search_available = True _web_search_provider_name = "jina" print("[Server] web search: Jina ✓") else: print("[Server] web search: not configured") def init_llm(): global _llm, _llm_provider_name try: if LLM_PROVIDER == "cerebras" and CEREBRAS_API_KEY: from langchain_cerebras import ChatCerebras _llm = ChatCerebras( model=LLM_MODEL, api_key=CEREBRAS_API_KEY, temperature=LLM_TEMPERATURE, max_tokens=MAX_FINAL_ANSWER_TOKENS, ) _llm_provider_name = "cerebras" print(f"[Server] LLM loaded: Cerebras ({LLM_MODEL}) ✓") elif LLM_PROVIDER == "cohere" and COHERE_API_KEY: from langchain_cohere import ChatCohere _llm = ChatCohere( model=LLM_MODEL, cohere_api_key=COHERE_API_KEY, temperature=LLM_TEMPERATURE, max_tokens=MAX_FINAL_ANSWER_TOKENS, ) _llm_provider_name = "cohere" print(f"[Server] LLM loaded: Cohere ({LLM_MODEL}) ✓") else: print("[Server] LLM not configured.") except Exception as e: print(f"[Server] LLM init failed: {e}") _llm = None # ══════════════════════════════════════════════════════ # LOCAL KB HELPERS # ══════════════════════════════════════════════════════ def derive_title(filename: str, content: str) -> str: match = re.search(r"^\s*#\s+(.+)$", content, re.MULTILINE) if match: return match.group(1).strip() name = os.path.splitext(filename)[0] name = re.sub(r"^\d+[-_]?", "", name) name = name.replace("-", " ").replace("_", " ").strip() return name or filename def load_documents() -> list[dict]: docs = [] if not os.path.isdir(KNOWLEDGE_PATH): return docs for root, _dirs, files in os.walk(KNOWLEDGE_PATH): for file in files: if file.lower().endswith(ALLOWED_EXTENSIONS): path = os.path.join(root, file) try: with open(path, "r", encoding="utf-8") as f: text = f.read() except Exception: continue category = os.path.relpath(root, KNOWLEDGE_PATH) docs.append({ "file": file, "title": derive_title(file, text), "category": None if category == "." else category, "content": text, }) return docs def chunk_text(text: str, size: int = CHUNK_SIZE_WORDS, overlap: int = CHUNK_OVERLAP_WORDS) -> list[str]: words = text.split() if not words: return [] step = max(size - overlap, 1) chunks = [] for i in range(0, len(words), step): chunk = " ".join(words[i:i + size]) if chunk.strip(): chunks.append(chunk) if i + size >= len(words): break return chunks def build_embedding_text(title: str, file: str, category: Optional[str], chunk: str) -> str: header = f"Title: {title}\nFile: {file}" if category: header += f"\nCategory: {category}" return f"{header}\n\n{chunk}" def build_index(): global _index, _chunks if not _use_st or _model is None or _faiss is None or _np is None: return documents = load_documents() embed_texts, metadata = [], [] for doc in documents: for chunk in chunk_text(doc["content"]): embed_texts.append(build_embedding_text(doc["title"], doc["file"], doc["category"], chunk)) metadata.append({ "file": doc["file"], "title": doc["title"], "category": doc["category"], "content": chunk, }) if not embed_texts: _index = None; _chunks = []; return with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()): embeddings = _model.encode(embed_texts, show_progress_bar=False, normalize_embeddings=True) embeddings = _np.array(embeddings, dtype="float32") index = _faiss.IndexFlatIP(embeddings.shape[1]) index.add(embeddings) _index = index _chunks = metadata print(f"[Server] indexed {len(embed_texts)} chunks from {len(documents)} files ✓") def keyword_score(query: str, chunk: dict) -> float: q_tokens = set(re.findall(r"[A-Za-z0-9_]+", query.lower())) if not q_tokens: return 0.0 searchable = f"{chunk['title']} {chunk['file']} {chunk['content']}" t_tokens = set(re.findall(r"[A-Za-z0-9_]+", searchable.lower())) overlap = q_tokens & t_tokens return len(overlap) / len(q_tokens) def _local_search(query: str, top_k: int = 5, hybrid: bool = True, keyword_weight: float = 0.3) -> list[dict]: if _index is None or not _chunks: return [] with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()): qvec = _model.encode([query], normalize_embeddings=True) qvec = _np.array(qvec, dtype="float32") fetch_k = min(top_k * 3 if hybrid else top_k, len(_chunks)) scores, indices = _index.search(qvec, fetch_k) candidates = [] for score, idx in zip(scores[0], indices[0]): if idx < 0: continue chunk = _chunks[idx] final_score = float(score) if hybrid: kw = keyword_score(query, chunk) final_score = (1 - keyword_weight) * final_score + keyword_weight * kw candidates.append((final_score, chunk)) candidates.sort(key=lambda x: x[0], reverse=True) return [{"score": s, **c} for s, c in candidates[:top_k]] # ══════════════════════════════════════════════════════ # WEB SEARCH # ══════════════════════════════════════════════════════ def enhance_query_for_web(query: str) -> str: query_lower = query.lower() context_parts = [] if not any(k in query_lower for k in ("jira", "jsm", "atlassian")): context_parts.append("Jira Service Management") if not any(k in query_lower for k in ("groovy", "scriptrunner", "adaptavist")): context_parts.append("Groovy ScriptRunner") if context_parts: return f"{' '.join(context_parts)} — {query}" return query def _tavily_search(query: str, max_results: int = 3) -> list[dict]: payload = json.dumps({ "api_key": TAVILY_API_KEY, "query": query, "max_results": max_results, "include_answer": True, "search_depth": "advanced", "include_domains": WEB_INCLUDE_DOMAINS, }).encode("utf-8") req = urllib.request.Request("https://api.tavily.com/search", data=payload, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read().decode("utf-8")) results = [] if data.get("answer"): results.append({"title": "Tavily AI Summary", "url": "", "content": data["answer"][:MAX_WEB_CONTENT_CHARS], "score": 1.0, "source": "tavily_answer"}) for r in data.get("results", [])[:max_results]: results.append({"title": r.get("title", ""), "url": r.get("url", ""), "content": (r.get("content") or "")[:MAX_WEB_CONTENT_CHARS], "score": r.get("score", 0.0), "source": "tavily"}) return results def _jina_search(query: str, max_results: int = 3) -> list[dict]: url = f"https://s.jina.ai/{urllib.parse.quote(query)}" req = urllib.request.Request(url, headers={"Authorization": f"Bearer {JINA_API_KEY}", "Accept": "application/json", "X-Retain-Images": "none", "X-No-Cache": "true"}, method="GET") with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read().decode("utf-8")) results = [] for item in data.get("data", [])[:max_results]: results.append({"title": item.get("title", ""), "url": item.get("url", ""), "content": (item.get("content") or "")[:MAX_WEB_CONTENT_CHARS], "score": 0.0, "source": "jina"}) return results def web_search(query: str, max_results: int = 3) -> list[dict]: if not _web_search_available: return [] enhanced = enhance_query_for_web(query) primary = _web_search_provider_name or WEB_SEARCH_PROVIDER if primary == "tavily" and TAVILY_API_KEY: try: return _tavily_search(enhanced, max_results) except Exception as e: print(f"[Server] Tavily search failed: {e}") elif primary == "jina" and JINA_API_KEY: try: return _jina_search(enhanced, max_results) except Exception as e: print(f"[Server] Jina search failed: {e}") if primary != "jina" and JINA_API_KEY: try: return _jina_search(enhanced, max_results) except: pass if primary != "tavily" and TAVILY_API_KEY: try: return _tavily_search(enhanced, max_results) except: pass return [] # ══════════════════════════════════════════════════════ # LLM & AGENT # ══════════════════════════════════════════════════════ SYSTEM_PROMPT = """You are an expert assistant for Jira Service Management (JSM) administration and Groovy/ScriptRunner development on Atlassian platforms. You receive: 1. LOCAL KNOWLEDGE BASE results — curated Groovy tutorials and ScriptRunner docs. 2. WEB SEARCH results — live results from Atlassian Community, docs, and Adaptavist. ANSWER RULES: • Give a direct, actionable answer with step-by-step instructions. • Include COMPLETE, working Groovy code in ```groovy fenced blocks when scripting is involved. • Prefer LOCAL results for Groovy/ScriptRunner syntax and patterns. • Use WEB results for JSM configuration, newer API changes, or topics not covered locally. • Cite sources inline as [Local: filename] or [Web: domain]. • Be concise — no filler, no repetition. Use markdown headings, code blocks, and bullet points. • Maximum length: ~{max_tokens} tokens.""" LOCAL_CHUNK_CHARS_FOR_LLM = 800 def _format_local_for_llm(results: list[dict]) -> str: if not results: return "(no local results found)" parts = [] for i, r in enumerate(results, 1): parts.append(f"[{i}] File: {r.get('file', '?')}\n Title: {r.get('title', '?')}\n Content:\n{r.get('content', '')[:LOCAL_CHUNK_CHARS_FOR_LLM]}") return "\n\n".join(parts) def _format_web_for_llm(results: list[dict]) -> str: if not results: return "(no web results found)" parts = [] for i, r in enumerate(results, 1): url = r.get("url", "") domain = urllib.parse.urlparse(url).netloc if url else "tavily-summary" parts.append(f"[{i}] Title: {r.get('title', '?')}\n Source: {domain}\n Content:\n{r.get('content', '')[:MAX_WEB_CONTENT_CHARS]}") return "\n\n".join(parts) def synthesize_with_llm(query: str, local_results: list[dict], web_results: list[dict]) -> str: if _llm is None: return "" system = SYSTEM_PROMPT.format(max_tokens=MAX_FINAL_ANSWER_TOKENS) user = (f"USER QUESTION:\n{query}\n\n" f"=== LOCAL KNOWLEDGE BASE RESULTS ===\n{_format_local_for_llm(local_results)}\n\n" f"=== WEB SEARCH RESULTS ===\n{_format_web_for_llm(web_results)}\n\n" f"Provide a comprehensive answer based on the above sources.") try: from langchain_core.messages import SystemMessage, HumanMessage response = _llm.invoke([SystemMessage(content=system), HumanMessage(content=user)]) return response.content except Exception as e: print(f"[Server] LLM synthesis failed: {e}") return "" def _create_agent_executor(): global _agent_executor from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate @tool def search_local_kb(query: str) -> str: """Search the local knowledge base of Groovy tutorials and ScriptRunner documentation. Use this FIRST for Groovy code examples.""" results = _local_search(query, top_k=5, hybrid=True) return _format_local_for_llm(results) if results else "No local results found." @tool def search_web(query: str) -> str: """Search the web for Jira Service Management configuration or newer API docs not found locally.""" results = web_search(query, max_results=MAX_WEB_RESULTS) return _format_web_for_llm(results) if results else "No web results found." tools = [search_local_kb, search_web] prompt = ChatPromptTemplate.from_messages([ ("system", SYSTEM_PROMPT.format(max_tokens=MAX_FINAL_ANSWER_TOKENS) + "\n\nYou have access to tools. Use `search_local_kb` first. Use `search_web` if local is insufficient."), ("user", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_tool_calling_agent(_llm, tools, prompt) _agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=AGENT_MAX_ITERATIONS, verbose=False, return_intermediate_steps=False) print("[Server] LangChain agent created ✓") def run_agent(query: str) -> str: if _agent_executor is None: return "" try: result = _agent_executor.invoke({"input": query}) return result.get("output", "") except Exception as e: print(f"[Server] agent execution failed: {e}") return "" # ══════════════════════════════════════════════════════ # MODELS # ══════════════════════════════════════════════════════ class SearchRequest(BaseModel): query: str top_k: int = 5 hybrid: bool = True keyword_weight: float = 0.3 # Unified Smart Search controls web_search: bool = True # Enable web search max_web_results: int = 3 # Web results limit synthesize: bool = True # Enable LLM synthesis use_agent: bool = False # Enable multi-step LLM Agent class SearchResultItem(BaseModel): file: str title: str category: Optional[str] = None score: float content: str class WebSearchResultItem(BaseModel): title: str url: str content: str score: float = 0.0 source: str = "web" class SearchSource(BaseModel): type: str title: str file: Optional[str] = None url: Optional[str] = None score: float = 0.0 class SearchResponse(BaseModel): query: str answer: Optional[str] = None local_results: list[SearchResultItem] = [] web_results: list[WebSearchResultItem] = [] sources: list[SearchSource] = [] used_web_search: bool = False used_llm: bool = False used_agent: bool = False llm_provider: Optional[str] = None error: Optional[str] = None class ReloadResponse(BaseModel): status: str chunks_indexed: int class HealthResponse(BaseModel): status: str model_loaded: bool faiss_loaded: bool chunks_indexed: int knowledge_path: str web_search_available: bool web_search_provider: Optional[str] = None llm_loaded: bool llm_provider: Optional[str] = None llm_model: Optional[str] = None agent_enabled: bool auth_enabled: bool = False # ✅ جديد # ══════════════════════════════════════════════════════ # ENDPOINTS # ══════════════════════════════════════════════════════ @app.get("/debug", dependencies=[Depends(verify_api_key)]) def debug(): files = [] if os.path.isdir(KNOWLEDGE_PATH): for root, _dirs, fs in os.walk(KNOWLEDGE_PATH): for f in fs: files.append(os.path.join(root, f)) return {"path": KNOWLEDGE_PATH, "exists": os.path.exists(KNOWLEDGE_PATH), "count": len(files), "files": files[:50]} @app.get("/health", response_model=HealthResponse) def health(): return { "status": "ok", "model_loaded": _use_st, "faiss_loaded": _faiss is not None, "chunks_indexed": len(_chunks), "knowledge_path": KNOWLEDGE_PATH, "web_search_available": _web_search_available, "web_search_provider": _web_search_provider_name, "llm_loaded": _llm is not None, "llm_provider": _llm_provider_name, "llm_model": LLM_MODEL if _llm is not None else None, "agent_enabled": _agent_executor is not None, "auth_enabled": AUTH_ENABLED, # ✅ جديد } @app.post("/reload", response_model=ReloadResponse, dependencies=[Depends(verify_api_key)]) def reload_knowledge_base(): build_index() return ReloadResponse(status="ok", chunks_indexed=len(_chunks)) @app.get("/ping") def ping(): return {"pong": True} # ══════════════════════════════════════════════════════ # UNIFIED SEARCH ENDPOINT # ══════════════════════════════════════════════════════ @app.post("/search", response_model=SearchResponse, dependencies=[Depends(verify_api_key)]) def search(req: SearchRequest): """ Unified Intelligent Search endpoint. Performs: 1. Local FAISS semantic search. 2. Web search (Tavily/Jina) if `web_search=True`. 3. LLM Synthesis (Cerebras/Cohere) if `synthesize=True`. 4. Multi-step Agent reasoning if `use_agent=True`. """ response = SearchResponse(query=req.query) errors = [] # ── Agent Mode ── if req.use_agent and _agent_executor is not None: answer = run_agent(req.query) if answer: response.answer = answer response.used_llm = True response.used_agent = True response.llm_provider = _llm_provider_name return response else: errors.append("Agent execution failed — falling back to RAG mode") # ── RAG Mode ── # 1) Local KB Search local_raw: list[dict] = [] if _use_st and _model is not None and _index is not None and _chunks: local_raw = _local_search(req.query, req.top_k, req.hybrid, req.keyword_weight) response.local_results = [ SearchResultItem( file=r["file"], title=r["title"], category=r["category"], score=r["score"], content=r["content"], ) for r in local_raw ] # 2) Web Search web_raw: list[dict] = [] if req.web_search and _web_search_available: try: web_raw = web_search(req.query, req.max_web_results) response.web_results = [ WebSearchResultItem( title=r.get("title", ""), url=r.get("url", ""), content=r.get("content", ""), score=r.get("score", 0.0), source=r.get("source", "web"), ) for r in web_raw ] response.used_web_search = True except Exception as e: errors.append(f"Web search error: {e}") # 3) Consolidate Sources for r in local_raw: response.sources.append(SearchSource( type="local", title=r.get("title", ""), file=r.get("file", ""), score=r.get("score", 0.0), )) for r in web_raw: response.sources.append(SearchSource( type="web", title=r.get("title", ""), url=r.get("url", ""), score=r.get("score", 0.0), )) # 4) LLM Synthesis if req.synthesize and _llm is not None and (local_raw or web_raw): answer = synthesize_with_llm(req.query, local_raw, web_raw) if answer: response.answer = answer response.used_llm = True response.llm_provider = _llm_provider_name else: errors.append("LLM synthesis returned empty") elif req.synthesize and _llm is None: errors.append("LLM not configured — returning raw results without synthesis") if errors: response.error = "; ".join(errors) return response