Update app.py

app.py

@@ -1,86 +1,296 @@
-import os, time, json, re
+import os, time, json, re, difflib, tempfile, pathlib, shutil, fnmatch
 import gradio as gr
 from llama_cpp import Llama
 
-# ✅ Working public GGUF model
+=========================
+
+Model + basic configuration
+
+=========================
 MODEL_REPO = "QuantFactory/Phi-3.5-mini-instruct-GGUF"
-MODEL_FILE = "Phi-3.5-mini-instruct.Q4_K_M.gguf"
+MODELFILE = "Phi-3.5-mini-instruct.Q4K_M.gguf"
+
+APP_TITLE = "Natural-language self-editing AI (CPU)"
 SAVE_PATH = "convos.jsonl"
-MAX_RECALL = 5  # how many past turns to recall
+MAXNEWTOKENS = 768
+N_CTX = 4096
+THREADS = 4
+
+=========================
 
+Globals
+
+=========================
 llm = None
+ROOT_DIR = pathlib.Path(".").resolve()
+BACKUPDIR = ROOTDIR / ".fs_backups"
+BACKUPDIR.mkdir(existok=True)
+DENYLIST = ["/proc", "/sys", "/dev", "/run", "/var/lib/docker", "/var/run"]
+
+Pending action schema:
+
+{"path": "app.py", "newcontent": "...", "oldcontent": "...", "reason": "…"}
+
+Stored in gr.State
+PENDINGKEY = "pendingaction"
+
+=========================
+
+Utilities: file system
+
+=========================
+def _resolve(path: str) -> pathlib.Path:
+    p = (ROOT_DIR / path).resolve()
+    for d in DENYLIST:
+        if str(p).startswith(d):
+            raise PermissionError(f"Path {p} is denied")
+    return p
+
+def make_backup(path: str) -> str:
+    p = _resolve(path)
+    if not p.exists():
+        return ""
+    ts = int(time.time())
+    rel = str(p.relativeto(ROOTDIR)).replace("/", "")
+    bk = BACKUP_DIR / f"{rel}.{ts}.bak"
+    if p.is_file():
+        shutil.copy2(p, bk)
+        return str(bk)
+    else:
+        shutil.makearchive(str(bk), "zip", rootdir=str(p))
+        return str(bk) + ".zip"
+
+def read_file(path: str) -> str:
+    p = _resolve(path)
+    with open(p, "r", encoding="utf-8") as f:
+        return f.read()
+
+def write_atomic(path: str, content: str) -> str:
+    p = _resolve(path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    backuppath = makebackup(path)
+    with tempfile.NamedTemporaryFile("w", delete=False, dir=str(p.parent), encoding="utf-8") as tmp:
+        tmp.write(content)
+        tmp_name = tmp.name
+    os.replace(tmp_name, p)
+    return backup_path
 
+def list_paths(pattern: str = "/*", cwd: str = ".") -> list:
+    base = _resolve(cwd)
+    results = []
+    for path in base.rglob("*"):
+        rel = str(path.relativeto(ROOTDIR))
+        if fnmatch.fnmatch(rel, pattern):
+            results.append(rel + ("/" if path.is_dir() else ""))
+    return results[:1000]
+
+def filedifftext(old: str, new: str, fromname: str, toname: str) -> str:
+    diff = difflib.unified_diff(
+        old.splitlines(), new.splitlines(), fromfile=fromname, tofile=toname
+    )
+    return "\n".join(diff)
+
+=========================
+
+Utilities: chat memory (lightweight)
+
+=========================
+def saveturn(system, history, usermsg, assistant_msg):
+    try:
+        with open(SAVE_PATH, "a", encoding="utf-8") as f:
+            rec = {
+                "ts": time.time(),
+                "system": system,
+                "history": history,
+                "user": user_msg,
+                "assistant": assistant_msg,
+            }
+            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+    except Exception:
+        pass
+
+=========================
+
+Model loader
+
+=========================
 def get_llm():
     global llm
     if llm is not None:
         return llm
     llm = Llama.from_pretrained(
-        repo_id=MODEL_REPO,
+        repoid=MODELREPO,
         filename=MODEL_FILE,
-        n_ctx=4096,
-        n_threads=4,
-        n_gpu_layers=0,
+        nctx=NCTX,
+        n_threads=THREADS,
+        ngpulayers=0,
         verbose=False
     )
     return llm
 
-def save_turn(system, history, user_msg, assistant_msg):
-    with open(SAVE_PATH, "a", encoding="utf-8") as f:
-        rec = {
-            "ts": time.time(),
-            "system": system,
-            "history": history,
-            "user": user_msg,
-            "assistant": assistant_msg,
-        }
-        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
-
-def load_memory(query):
-    """Simple keyword-based recall from past chats."""
-    if not os.path.exists(SAVE_PATH):
-        return []
-    with open(SAVE_PATH, "r", encoding="utf-8") as f:
-        lines = [json.loads(l) for l in f if l.strip()]
-    query_words = set(re.findall(r"\w+", query.lower()))
-    scored = []
-    for rec in lines:
-        text = (rec.get("user","") + " " + rec.get("assistant","")).lower()
-        score = len(query_words & set(re.findall(r"\w+", text)))
-        if score > 0:
-            scored.append((score, rec))
-    scored.sort(reverse=True, key=lambda x: x[0])
-    return [rec for _, rec in scored[:MAX_RECALL]]
-
-def format_messages(system, history, user_msg):
+=========================
+
+NL agent: intent + planning
+
+=========================
+CODEBLOCKRE = re.compile(r"(?:[\w.+-]+)?\n(.*?)", re.DOTALL)
+
+def extract_fenced(text: str) -> str:
+    m = CODEBLOCKRE.search(text)
+    return m.group(1).strip() if m else text
+
+def detectintent(usertext: str) -> str:
+    """
+    Returns: "edit", "create", "chat"
+    """
+    t = user_text.lower()
+    edit_verbs = ["edit", "change", "modify", "refactor", "fix", "optimize", "speed up", "rework", "rewrite", "patch"]
+    create_verbs = ["create", "make a new", "add a new", "generate a new", "build a new", "scaffold"]
+    if any(v in t for v in edit_verbs):
+        return "edit"
+    if any(v in t for v in create_verbs):
+        return "create"
+    # Heuristic: mentions of specific files imply edit
+    if re.search(r"\b[\w\-/]+\.py\b", t):
+        return "edit"
+    return "chat"
+
+def findtargetfiles(user_text: str) -> list:
+    """
+    Pull explicit filenames from the message; fallback to app.py if none.
+    """
+    files = re.findall(r"([\w\-/]+\.py)\b", user_text)
+    files = [f for f in files if (ROOT_DIR / f).exists()]
+    if files:
+        return files
+    # sensible default
+    fallback = ["app.py"] if (ROOT_DIR / "app.py").exists() else []
+    return fallback
+
+def proposeeditorcreate(usertext: str) -> dict:
+    """
+    Ask the model to propose a full-file replacement (for edit) or a new file (for create).
+    Returns: {"path": str, "new_content": str, "reason": str}
+    """
+    targets = findtargetfiles(user_text)
+    context_blobs = []
+    for path in targets:
+        try:
+            contextblobs.append(f"File: {path}\npython\n{readfile(path)}\n")
+        except Exception:
+            pass
+    filelistpreview = "\n".join(list_paths("/*.py"))
+
+    system_hint = (
+        "You are a precise software editor. When asked to change or create code, "
+        "you return ONLY the complete target file in a single fenced code block, and a brief reason."
+    )
+    user_prompt = f"""
+User request:
+{user_text}
+
+Existing Python files (truncated):
+`text
+{filelistpreview}
+`
+
+Context for existing targets:
+{("\n\n".join(contextblobs) if contextblobs else "(no existing file context)")}
+
+Instructions:
+- If editing, produce the FULL updated content of the primary target file you chose.
+- If creating, choose a sensible path (e.g., scripts/logger.py) and produce the FULL content.
+- Return your answer in this JSON template followed by ONE fenced code block with the file content:
+JSON (no code fences):
+{{"path": "<target_path.py>", "reason": "<1-2 sentences>"}}
+Then the file content:
+`python
+
+full file content here
+`
+Do not include anything else.
+"""
+    out = getllm().createchat_completion(
+        messages=[{"role": "system", "content": system_hint},
+                  {"role": "user", "content": user_prompt}],
+        temperature=0.2,
+        top_p=0.9,
+        maxtokens=MAXNEW_TOKENS
+    )
+    text = out["choices"][0]["message"]["content"]
+    # Extract JSON header
+    json_match = re.search(r"\{.*\}", text, flags=re.DOTALL)
+    header = {"path": "app.generated.py", "reason": "Generated update"}
+    if json_match:
+        try:
+            header = json.loads(json_match.group(0))
+        except Exception:
+            pass
+    newcontent = extractfenced(text)
+    return {
+        "path": header.get("path", "app.generated.py"),
+        "newcontent": newcontent,
+        "reason": header.get("reason", "Proposed change"),
+    }
+
+def proposediffmessage(path: str, old: str, new: str, reason: str) -> str:
+    diff = filedifftext(old, new, f"{path} (old)", f"{path} (new)")
+    preview = diff if diff.strip() else "(no textual differences)"
+    return (
+        f"I proposed changes to {path}:\n"
+        f"Reason: {reason}\n"
+        f"Diff:\ndiff\n{preview}\n\n"
+        f"Apply these changes? Say 'yes' to apply, or 'no' to cancel. "
+        f"(You can also say 'edit the proposal' to iterate.)"
+    )
+
+def apply_pending(pending: dict) -> str:
+    path = pending["path"]
+    newcontent = pending["newcontent"]
+    oldcontent = pending["oldcontent"]
+    try:
+        backup = writeatomic(path, newcontent)
+        return f"Applied changes to {path}. Backup: {backup or 'none'}"
+    except Exception as e:
+        # On failure, offer to save to an alternative path
+        alt = f"{path}.failed.{int(time.time())}.txt"
+        try:
+            writeatomic(alt, newcontent)
+            return f"Failed to write {path}: {e}\nSaved proposed content to {alt}"
+        except Exception as e2:
+            return f"Failed to apply and to save alt copy: {e2}"
+
+def natural_yes(text: str) -> bool:
+    return text.strip().lower() in {"y", "yes", "apply", "do it", "ok", "okay", "sure", "confirm"}
+
+def natural_no(text: str) -> bool:
+    t = text.strip().lower()
+    return t in {"n", "no", "cancel", "stop", "reject", "discard"}
+
+=========================
+
+Core chat pipeline
+
+=========================
+def formatmessages(system, history, usermsg):
     msgs = []
     if system.strip():
         msgs.append({"role": "system", "content": system})
-
-    recalls = load_memory(user_msg)
-    if recalls:
-        mem_text = "\n".join(
-            f"User: {r['user']}\nAssistant: {r['assistant']}" for r in recalls
-        )
-        msgs.append({"role": "system", "content": f"Relevant past conversations:\n{mem_text}"})
-
     for h in history:
         msgs.append({"role": h["role"], "content": h["content"]})
-
     msgs.append({"role": "user", "content": user_msg})
     return msgs
 
-def chat_fn(user_msg, history, system, temperature, top_p, max_new_tokens):
+def streamchatresponse(usermsg, history, system, temperature, topp, maxnewtokens):
     llm = get_llm()
-    msgs = format_messages(system, history, user_msg)
-
-    stream = llm.create_chat_completion(
+    msgs = formatmessages(system, history, usermsg)
+    stream = llm.createchatcompletion(
         messages=msgs,
         temperature=temperature,
-        top_p=top_p,
-        max_tokens=max_new_tokens,
+        topp=topp,
+        maxtokens=maxnew_tokens,
         stream=True
     )
-
     partial = ""
     for chunk in stream:
         delta = chunk["choices"][0]["delta"]
@@ -88,28 +298,97 @@ def chat_fn(user_msg, history, system, temperature, top_p, max_new_tokens):
             piece = delta["content"]
             partial += piece
             yield partial
+    return
+
+=========================
 
-    save_turn(system, history, user_msg, partial)
+Gradio app
 
-with gr.Blocks(title="Free ChatGPT-like (CPU) with Memory") as demo:
-    gr.Markdown("# Free ChatGPT‑style AI (CPU) + Long‑Term Memory")
+=========================
+with gr.Blocks(title=APP_TITLE) as demo:
+    gr.Markdown(f"# {APP_TITLE}\nTalk normally. Ask for changes or new files; I’ll propose a patch, show a diff, and wait for your yes/no.")
     with gr.Row():
-        system = gr.Textbox(label="System prompt", value="You are a helpful, concise assistant.")
+        system = gr.Textbox(label="System prompt", value="You are a helpful, precise, and concise assistant.")
     with gr.Row():
-        temperature = gr.Slider(0.0, 1.5, value=0.7, step=0.05, label="Temperature")
+        temperature = gr.Slider(0.0, 1.5, value=0.4, step=0.05, label="Temperature")
         top_p = gr.Slider(0.1, 1.0, value=0.9, step=0.05, label="Top‑p")
-        max_new_tokens = gr.Slider(16, 1024, value=512, step=16, label="Max new tokens")
-    chat = gr.Chatbot(height=480, show_copy_button=True, type="messages")
-    user = gr.Textbox(label="Your message", placeholder="Ask anything...")
+        maxnewtokens = gr.Slider(64, 2048, value=MAXNEWTOKENS, step=16, label="Max new tokens")
+    chat = gr.Chatbot(height=520, showcopybutton=True, type="messages")
+    user = gr.Textbox(label="Your message", placeholder="Ask anything… e.g., 'Optimize the memory recall code' or 'Create scripts/logger.py that logs messages'")
     send = gr.Button("Send", variant="primary")
+    state = gr.State({PENDING_KEY: None})
 
-    def respond(message, chat_history, system, temperature, top_p, max_new_tokens):
+    def respond(message, chathistory, system, temperature, topp, maxnewtokens, state_obj):
         if not message or not message.strip():
-            return gr.update(), chat_history
-        history_msgs = chat_history or []
-        stream = chat_fn(message, history_msgs, system, temperature, top_p, max_new_tokens)
+            return gr.update(), chathistory, stateobj
+
+        # If there's a pending action, check for yes/no
+        pending = stateobj.get(PENDINGKEY)
+        if pending is not None:
+            if natural_yes(message):
+                result = apply_pending(pending)
+                stateobj[PENDINGKEY] = None
+                newhist = (chathistory or []) + [
+                    {"role": "user", "content": message},
+                    {"role": "assistant", "content": result},
+                ]
+                return gr.update(value=newhist), newhist, state_obj
+            elif natural_no(message):
+                stateobj[PENDINGKEY] = None
+                newhist = (chathistory or []) + [
+                    {"role": "user", "content": message},
+                    {"role": "assistant", "content": "Okay, discarded the proposed change."},
+                ]
+                return gr.update(value=newhist), newhist, state_obj
+            # If neither yes/no, treat as iteration: regenerate proposal using the user's feedback
+            msg = f"Updating the proposal with your feedback: {message}\nRe‑proposing…"
+            historymsgs = (chathistory or []) + [{"role": "assistant", "content": msg}]
+            # Merge feedback into a new proposal prompt by appending to user_text
+            merged_request = pending.get("reason", "") + "\n\nAdditional feedback: " + message
+            proposal = proposeeditorcreate(mergedrequest)
+            path = proposal["path"]
+            try:
+                old = read_file(path)
+            except Exception:
+                old = ""
+            diffmsg = proposediffmessage(path, old, proposal["newcontent"], proposal["reason"])
+            # Stash new pending
+            stateobj[PENDINGKEY] = {
+                "path": path,
+                "newcontent": proposal["newcontent"],
+                "old_content": old,
+                "reason": proposal["reason"],
+            }
+            newhist = historymsgs + [{"role": "assistant", "content": diff_msg}]
+            return gr.update(value=newhist), newhist, state_obj
+
+        # No pending: decide intent
+        intent = detect_intent(message)
+        if intent in ("edit", "create"):
+            proposal = proposeeditor_create(message)
+            path = proposal["path"]
+            try:
+                old = read_file(path)
+            except Exception:
+                old = ""
+            diffmsg = proposediffmessage(path, old, proposal["newcontent"], proposal["reason"])
+            # Stash pending
+            stateobj[PENDINGKEY] = {
+                "path": path,
+                "newcontent": proposal["newcontent"],
+                "old_content": old,
+                "reason": proposal["reason"],
+            }
+            newhist = (chathistory or []) + [
+                {"role": "user", "content": message},
+                {"role": "assistant", "content": diff_msg},
+            ]
+            return gr.update(value=newhist), newhist, state_obj
+
+        # Plain chat with streaming
+        historymsgs = chathistory or []
         bot_text = ""
-        for partial in stream:
+        for partial in streamchatresponse(message, historymsgs, system, temperature, topp, maxnewtokens):
             bot_text = partial
             yield gr.update(value=(history_msgs + [
                 {"role": "user", "content": message},
@@ -117,18 +396,20 @@ with gr.Blocks(title="Free ChatGPT-like (CPU) with Memory") as demo:
             ])), (history_msgs + [
                 {"role": "user", "content": message},
                 {"role": "assistant", "content": bot_text}
-            ])
+            ]), state_obj
+        # Save last turn once streaming ends
+        saveturn(system, historymsgs, message, bot_text)
 
     send.click(
         respond,
-        [user, chat, system, temperature, top_p, max_new_tokens],
-        [chat, chat],
+        [user, chat, system, temperature, topp, maxnew_tokens, state],
+        [chat, chat, state],
     )
     user.submit(
         respond,
-        [user, chat, system, temperature, top_p, max_new_tokens],
-        [chat, chat],
+        [user, chat, system, temperature, topp, maxnew_tokens, state],
+        [chat, chat, state],
     )
 
-if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+if name == "main":
+    demo.launch(servername="0.0.0.0", serverport=7860)