Update app.py

app.py

@@ -1,5 +1,7 @@
-# app.py — MCP POC using OpenRouter for LLM (replaces local HF model)
-# Place this file next to config.py. Do NOT store secrets here.
+# app.py — MCP server using DeepSeek via Hugging Face transformers (or fallback)
+# - Put this file next to config.py (see example below)
+# - It loads the model in LOCAL_MODEL (e.g., a DeepSeek HF checkpoint) via transformers.pipeline
+# - If the model cannot be loaded (no transformers / OOM / missing weights), it falls back to a small CPU model or rule-based responder
 
 from mcp.server.fastmcp import FastMCP
 from typing import Optional, List, Tuple, Any, Dict
@@ -11,6 +13,20 @@ import time
 import traceback
 import inspect
 import re
+import logging
+
+# Setup simple logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("mcp_deepseek")
+
+# Optional transformers imports — will attempt; we handle missing gracefully
+TRANSFORMERS_AVAILABLE = False
+try:
+    from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM, AutoModelForSeq2SeqLM
+    TRANSFORMERS_AVAILABLE = True
+except Exception as e:
+    logger.warning("transformers not available: %s", e)
+    TRANSFORMERS_AVAILABLE = False
 
 # ----------------------------
 # Load config
@@ -21,20 +37,16 @@ try:
         CLIENT_SECRET,
         REFRESH_TOKEN,
         API_BASE,
-        OPENROUTER_API_KEY,   # your OpenRouter API key (put this in config.py)
-        OPENROUTER_MODEL      # e.g. "gpt-4o-mini" or any model name routed by OpenRouter
+        LOCAL_MODEL,      # e.g. "deepseek-ai/deepseek-r1-7b" or smaller/distilled variant
+        LOCAL_TOKENIZER   # optional: tokenizer name if different
     )
-except Exception:
+except Exception as e:
     raise SystemExit(
-        "Make sure config.py exists with CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN, API_BASE, "
-        "OPENROUTER_API_KEY and OPENROUTER_MODEL (or set OPENROUTER_MODEL to your preferred model)."
+        "Make sure config.py exists with CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN, API_BASE, LOCAL_MODEL (or set LOCAL_MODEL=None)."
     )
 
-# OpenRouter endpoint (public OpenRouter cloud endpoint)
-OPENROUTER_BASE_URL = "https://api.openrouter.ai/v1"
-
 # ----------------------------
-# Initialize FastMCP
+# FastMCP init
 # ----------------------------
 mcp = FastMCP("ZohoCRMAgent")
 
@@ -45,12 +57,7 @@ ANALYTICS_PATH = "mcp_analytics.json"
 
 def _init_analytics():
     if not os.path.exists(ANALYTICS_PATH):
-        base = {
-            "tool_calls": {},
-            "llm_calls": 0,
-            "last_llm_confidence": None,
-            "created_at": time.time()
-        }
+        base = {"tool_calls": {}, "llm_calls": 0, "last_llm_confidence": None, "created_at": time.time()}
         with open(ANALYTICS_PATH, "w") as f:
             json.dump(base, f, indent=2)
 
@@ -84,86 +91,126 @@ def _log_llm_call(confidence: Optional[float] = None):
 _init_analytics()
 
 # ----------------------------
-# OpenRouter wrapper
+# DeepSeek / HF model loader
 # ----------------------------
-def _openrouter_headers():
-    return {"Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json"}
+LLM_PIPELINE = None
+TOKENIZER = None
+LOADED_MODEL_NAME = None
 
-def openrouter_generate(system_prompt: str, user_prompt: str, history: Optional[List[Tuple[str,str]]] = None, max_tokens: int = 512) -> Dict[str, Any]:
+def init_deepseek_model():
     """
-    Call OpenRouter chat completions endpoint with messages built from system + history + user prompt.
-    Returns dict: {'text': <str>, 'raw': <resp_json>, 'confidence': Optional[float]}
+    Try to load LOCAL_MODEL via transformers.pipeline.
+    Expected LOCAL_MODEL examples:
+      - "deepseek-ai/deepseek-r1-7b"  (may require GPU; big)
+      - "deepseek-ai/deepseek-r1-3b"  (smaller)
+      - "deepseek-ai/deepseek-r1-1.3b" (more likely to load on moderate machines)
+    If loading fails, try a fallback small model (distilgpt2 or flan-t5-small if seq2seq).
     """
-    messages = []
-    # system
-    if system_prompt:
-        messages.append({"role": "system", "content": system_prompt})
-    # history (list of (user,assistant))
-    history = history or []
-    for pair in history:
-        try:
-            u, a = pair[0], pair[1]
-            if u:
-                messages.append({"role": "user", "content": u})
-            if a:
-                messages.append({"role": "assistant", "content": a})
-        except Exception:
-            continue
-    # current user
-    messages.append({"role": "user", "content": user_prompt})
+    global LLM_PIPELINE, TOKENIZER, LOADED_MODEL_NAME
 
-    body = {
-        "model": OPENROUTER_MODEL,
-        "messages": messages,
-        "max_tokens": max_tokens,
-        # "temperature": 0.0,   # set if you want deterministic responses
-    }
+    if not LOCAL_MODEL:
+        logger.info("LOCAL_MODEL is None — no local LLM will be used.")
+        LLM_PIPELINE = None
+        return
+
+    if not TRANSFORMERS_AVAILABLE:
+        logger.warning("transformers not installed; cannot load DeepSeek. Falling back to rule-based.")
+        LLM_PIPELINE = None
+        return
 
-    url = f"{OPENROUTER_BASE_URL}/chat/completions"
     try:
-        r = requests.post(url, headers=_openrouter_headers(), json=body, timeout=60)
+        tokenizer_name = LOCAL_TOKENIZER or LOCAL_MODEL
+        model_name = LOCAL_MODEL
+        LOADED_MODEL_NAME = model_name
+
+        # If model looks like seq2seq (T5/flan) use text2text; else causal
+        seq2seq_keywords = ["flan", "t5", "seq2seq"]
+        if any(k in model_name.lower() for k in seq2seq_keywords):
+            TOKENIZER = AutoTokenizer.from_pretrained(tokenizer_name, use_fast=True)
+            model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
+            LLM_PIPELINE = pipeline("text2text-generation", model=model, tokenizer=TOKENIZER)
+            logger.info("Loaded seq2seq model: %s", model_name)
+        else:
+            TOKENIZER = AutoTokenizer.from_pretrained(tokenizer_name, use_fast=True)
+            model = AutoModelForCausalLM.from_pretrained(model_name)
+            LLM_PIPELINE = pipeline("text-generation", model=model, tokenizer=TOKENIZER)
+            logger.info("Loaded causal model: %s", model_name)
+
     except Exception as e:
-        raise RuntimeError(f"OpenRouter request failed: {e}")
+        logger.error("Failed to load requested model '%s': %s", LOCAL_MODEL, e)
+        traceback.print_exc()
+        # Try a small CPU-friendly fallback
+        fallback = None
+        try:
+            # prefer an instruction-friendly small model if possible
+            fallback = "google/flan-t5-small"
+            if "flan" in fallback:
+                TOKENIZER = AutoTokenizer.from_pretrained(fallback, use_fast=True)
+                model = AutoModelForSeq2SeqLM.from_pretrained(fallback)
+                LLM_PIPELINE = pipeline("text2text-generation", model=model, tokenizer=TOKENIZER)
+            else:
+                TOKENIZER = AutoTokenizer.from_pretrained("distilgpt2", use_fast=True)
+                model = AutoModelForCausalLM.from_pretrained("distilgpt2")
+                LLM_PIPELINE = pipeline("text-generation", model=model, tokenizer=TOKENIZER)
+            LOADED_MODEL_NAME = fallback
+            logger.info("Loaded fallback model: %s", fallback)
+        except Exception as e2:
+            logger.error("Fallback model also failed: %s", e2)
+            traceback.print_exc()
+            LLM_PIPELINE = None
+            LOADED_MODEL_NAME = None
+
+# Initialize model at startup (may take time)
+init_deepseek_model()
+
+# ----------------------------
+# Rule-based fallback responder
+# ----------------------------
+def rule_based_response(message: str) -> str:
+    msg = (message or "").strip().lower()
+    if msg.startswith("create record") or msg.startswith("create contact"):
+        return "To create a record, use: create_record MODULE_NAME {\"Field\":\"value\"}"
+    if msg.startswith("create_invoice"):
+        return "To create invoice: create_invoice {\"customer_id\":\"...\",\"line_items\":[...]} (JSON)"
+    if msg.startswith("help") or "what can you do" in msg:
+        return "I can run create_record/update_record/delete_record or process local files by pasting their /mnt/data/ path."
+    return "(fallback) No local LLM loaded. Use explicit commands like create_record or paste /mnt/data/ path."
 
-    if r.status_code not in (200, 201):
-        # surface the error for debugging
-        raise RuntimeError(f"OpenRouter API error {r.status_code}: {r.text}")
+# ----------------------------
+# LLM wrapper that returns text + confidence (best-effort)
+# ----------------------------
+def deepseek_generate(prompt: str, max_tokens: int = 256) -> Dict[str, Any]:
+    """
+    Generate using the loaded pipeline. Returns {'text': str, 'confidence': Optional[float], 'raw': resp}
+    """
+    if LLM_PIPELINE is None:
+        return {"text": rule_based_response(prompt), "confidence": None, "raw": None}
 
-    resp_json = r.json()
-    # Parse response for text; different routers may vary slightly
-    text = ""
-    confidence = None
     try:
-        # typical shape: choices[0].message.content or choices[0].message
-        choice = resp_json.get("choices", [{}])[0]
-        message = choice.get("message", {}) or {}
-        if isinstance(message, dict):
-            text = message.get("content") or message.get("content", "")
-            # sometimes content is a dict mapping types -> text
-            if isinstance(text, dict):
-                # join possible parts
-                text = text.get("text") or next(iter(text.values()), "")
+        out = LLM_PIPELINE(prompt, max_new_tokens=max_tokens)
+        text = ""
+        # pipeline returns list: [{'generated_text':...}] or [{'generated_text' or 'text'}]
+        if isinstance(out, list) and len(out) > 0:
+            first = out[0]
+            if isinstance(first, dict):
+                text = first.get("generated_text") or first.get("generated_text", "") or first.get("text") or str(first)
+            else:
+                text = str(first)
         else:
-            text = str(message)
-        # some providers include scores
-        confidence = choice.get("finish_reason_score") or choice.get("score") or None
-    except Exception:
-        text = json.dumps(resp_json)
-
-    _log_llm_call(confidence)
-    return {"text": text, "raw": resp_json, "confidence": confidence}
+            text = str(out)
+        _log_llm_call(None)
+        return {"text": text, "confidence": None, "raw": out}
+    except Exception as e:
+        logger.error("LLM generation error: %s", e)
+        traceback.print_exc()
+        return {"text": rule_based_response(prompt), "confidence": None, "raw": str(e)}
 
 # ----------------------------
-# Zoho token refresh & headers
+# Zoho token refresh & MCP tools (unchanged)
 # ----------------------------
 def _get_valid_token_headers() -> dict:
     token_url = "https://accounts.zoho.in/oauth/v2/token"
-    params = {
-        "refresh_token": REFRESH_TOKEN,
-        "client_id": CLIENT_ID,
-        "client_secret": CLIENT_SECRET,
-        "grant_type": "refresh_token"
-    }
+    params = {"refresh_token": REFRESH_TOKEN, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "refresh_token"}
     r = requests.post(token_url, params=params, timeout=20)
     if r.status_code == 200:
         t = r.json().get("access_token")
@@ -171,9 +218,6 @@ def _get_valid_token_headers() -> dict:
     else:
         raise RuntimeError(f"Failed to refresh Zoho token: {r.status_code} {r.text}")
 
-# ----------------------------
-# MCP tools: Zoho CRUD & process_document (unchanged)
-# ----------------------------
 @mcp.tool()
 def authenticate_zoho() -> str:
     try:
@@ -264,9 +308,9 @@ def create_invoice(data: dict) -> str:
 @mcp.tool()
 def process_document(file_path: str, target_module: Optional[str] = "Contacts") -> dict:
     """
-    Process an uploaded file path (local path or URL). Per developer instruction,
-    we accept local paths like '/mnt/data/script_zoho_mcp' and return a file:// URL.
-    Replace the placeholder OCR block with your real OCR pipeline when ready.
+    Accepts local path (e.g. /mnt/data/script_zoho_mcp) or URL.
+    Per developer instruction we treat the path as the file URL (file://...).
+    Replace placeholder OCR logic with your pipeline.
     """
     try:
         if os.path.exists(file_path):
@@ -279,13 +323,7 @@ def process_document(file_path: str, target_module: Optional[str] = "Contacts")
                 "Confidence": 0.88
             }
             _log_tool_call("process_document", True)
-            return {
-                "status": "success",
-                "file": os.path.basename(file_path),
-                "file_url": file_url,
-                "target_module": target_module,
-                "extracted_data": extracted
-            }
+            return {"status": "success", "file": os.path.basename(file_path), "file_url": file_url, "target_module": target_module, "extracted_data": extracted}
         else:
             _log_tool_call("process_document", False)
             return {"status": "error", "error": "file not found", "file_path": file_path}
@@ -294,76 +332,62 @@ def process_document(file_path: str, target_module: Optional[str] = "Contacts")
         return {"status": "error", "error": str(e)}
 
 # ----------------------------
-# Simple local command parser
+# Simple command parser (explicit commands in chat)
 # ----------------------------
 def try_parse_and_invoke_command(text: str):
     text = text.strip()
-    # create_record
     m = re.match(r"^create_record\s+(\w+)\s+(.+)$", text, re.I)
     if m:
-        module = m.group(1)
-        body = m.group(2)
-        try:
-            record_data = json.loads(body)
-        except Exception:
-            return "Invalid JSON for record_data"
+        module = m.group(1); body = m.group(2)
+        try: record_data = json.loads(body)
+        except Exception: return "Invalid JSON for record_data"
         return create_record(module, record_data)
-
-    # create_invoice
     m = re.match(r"^create_invoice\s+(.+)$", text, re.I)
     if m:
         body = m.group(1)
-        try:
-            invoice_data = json.loads(body)
-        except Exception:
-            return "Invalid JSON for invoice_data"
+        try: invoice_data = json.loads(body)
+        except Exception: return "Invalid JSON for invoice_data"
         return create_invoice(invoice_data)
-
-    # process_document via local path
     m = re.match(r"^(\/mnt\/data\/\S+)$", text)
     if m:
-        path = m.group(1)
-        return process_document(path)
-
+        path = m.group(1); return process_document(path)
     return None
 
 # ----------------------------
-# OpenRouter-based chat handler
+# Chat handler that uses DeepSeek generation (or fallback)
 # ----------------------------
-def openrouter_response(message: str, history: Optional[List[Tuple[str,str]]] = None) -> str:
+def deepseek_response(message: str, history: Optional[List[Tuple[str,str]]] = None) -> str:
     history = history or []
-    system_prompt = (
-        "You are Zoho Assistant. Keep responses concise. When appropriate, output a JSON object with keys 'tool' and 'args' "
-        "so the server can automatically call the corresponding MCP tool. Example:\n"
-        '{"tool":"create_record","args":{"module_name":"Contacts","record_data":{"Last_Name":"Doe","Email":"[email protected]"}}}\n'
-        "If not invoking tools, answer conversationally."
-    )
-    try:
-        resp = openrouter_generate(system_prompt, message, history)
-        text = resp.get("text", "")
-        # If LLM returned JSON indicating a tool invocation, attempt to parse & run
-        parsed = None
-        payload = text.strip()
-        if payload.startswith("{") or payload.startswith("["):
-            try:
-                parsed = json.loads(payload)
-            except Exception:
-                parsed = None
-        if isinstance(parsed, dict) and "tool" in parsed:
-            tool_name = parsed.get("tool")
-            args = parsed.get("args", {})
-            # Try call local tool by name if exists
-            if tool_name in globals() and callable(globals()[tool_name]):
-                try:
-                    result = globals()[tool_name](**args) if isinstance(args, dict) else globals()[tool_name](args)
-                    return f"Invoked tool '{tool_name}'. Result:\n{result}"
-                except Exception as e:
-                    return f"Tool invocation error: {e}"
-            else:
-                return f"Requested tool '{tool_name}' not found locally."
-        return text
-    except Exception as e:
-        return f"(OpenRouter error) {e}"
+    system_prompt = "You are Zoho Assistant. Prefer concise answers. If you want to call a tool, return a JSON object: {\"tool\": \"create_record\", \"args\": {...}}"
+    # compact history into text for few-shot context (optional)
+    history_text = ""
+    for pair in history:
+        try:
+            u,a = pair[0], pair[1]
+            history_text += f"User: {u}\nAssistant: {a}\n"
+        except Exception:
+            continue
+    prompt = f"{system_prompt}\n{history_text}\nUser: {message}\nAssistant:"
+    gen = deepseek_generate(prompt, max_tokens=256)
+    text = gen.get("text", "")
+    # if text looks like JSON with a tool action, try to invoke
+    payload = text.strip()
+    if payload.startswith("{") or payload.startswith("["):
+        try:
+            parsed = json.loads(payload)
+            if isinstance(parsed, dict) and "tool" in parsed:
+                tool_name = parsed.get("tool"); args = parsed.get("args", {})
+                if tool_name in globals() and callable(globals()[tool_name]):
+                    try:
+                        out = globals()[tool_name](**args) if isinstance(args, dict) else globals()[tool_name](args)
+                        return f"Invoked tool '{tool_name}'. Result:\n{out}"
+                    except Exception as e:
+                        return f"Tool invocation error: {e}"
+                else:
+                    return f"Requested tool '{tool_name}' not found locally."
+        except Exception:
+            pass
+    return text
 
 # ----------------------------
 # Gradio chat handler
@@ -372,12 +396,12 @@ def chat_handler(message, history):
     history = history or []
     trimmed = (message or "").strip()
 
-    # Explicit POC commands
+    # explicit command parser
     cmd = try_parse_and_invoke_command(trimmed)
     if cmd is not None:
         return cmd
 
-    # Developer convenience: local path handling (send unchanged)
+    # developer dev path handling (send path unchanged)
     if trimmed.startswith("/mnt/data/"):
         try:
             doc = process_document(trimmed)
@@ -385,22 +409,24 @@ def chat_handler(message, history):
         except Exception as e:
             return f"Error processing document: {e}"
 
-    # Otherwise, call OpenRouter
-    return openrouter_response(trimmed, history)
+    # otherwise, call deepseek_response (LLM or fallback)
+    try:
+        return deepseek_response(trimmed, history)
+    except Exception as e:
+        logger.error("deepseek_response error: %s", e)
+        traceback.print_exc()
+        return rule_based_response(trimmed)
 
 # ----------------------------
 # Gradio UI
 # ----------------------------
 def chat_interface():
-    return gr.ChatInterface(
-        fn=chat_handler,
-        textbox=gr.Textbox(placeholder="Ask me to create contacts, invoices, upload docs (or paste /mnt/data/... for dev).")
-    )
+    return gr.ChatInterface(fn=chat_handler, textbox=gr.Textbox(placeholder="Ask me to create contacts, invoices, upload docs (or paste /mnt/data/... for dev)."))
 
 # ----------------------------
 # Entrypoint
 # ----------------------------
 if __name__ == "__main__":
-    print("[startup] Launching Gradio UI + FastMCP server (OpenRouter mode).")
+    logger.info("Starting MCP server (DeepSeek mode). Loaded model: %s", LOADED_MODEL_NAME)
     demo = chat_interface()
     demo.launch(server_name="0.0.0.0", server_port=7860)