import os, re, json, shutil, subprocess, glob, traceback from pathlib import Path import gradio as gr from huggingface_hub import hf_hub_download, HfApi from llama_cpp import Llama import git as gitlib WORKSPACE = Path("/workspace") WORKSPACE.mkdir(exist_ok=True) # ---------- Model load (CPU friendly) ---------- MODEL_REPO = "Qwen/Qwen2.5-Coder-1.5B-Instruct-GGUF" MODEL_FILE = "qwen2.5-coder-1.5b-instruct-q4_k_m.gguf" print("⏬ Downloading model ...") model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) print("✅ Loading llama.cpp ...") llm = Llama( model_path=model_path, n_ctx=4096, n_threads=int(os.getenv("LLAMA_THREADS", "2")), n_batch=256, verbose=False, ) # ---------- Tools ---------- def _safe(p: str) -> Path: path = (WORKSPACE / p).resolve() if WORKSPACE.resolve() not in path.parents and path != WORKSPACE.resolve(): raise ValueError("Path escapes workspace") return path def read_file(path: str) -> str: return _safe(path).read_text(encoding="utf-8", errors="replace") def write_file(path: str, content: str) -> str: p = _safe(path); p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content, encoding="utf-8") return f"wrote {len(content)} bytes -> {path}" def edit_file(path: str, find: str, replace: str) -> str: p = _safe(path); txt = p.read_text(encoding="utf-8") if find not in txt: return "pattern not found" new = txt.replace(find, replace) p.write_text(new, encoding="utf-8") return f"replaced {txt.count(find)} occurrence(s) in {path}" def delete_file(path: str) -> str: p = _safe(path) if p.is_dir(): shutil.rmtree(p) else: p.unlink() return f"deleted {path}" def search_files(pattern: str, root: str = ".") -> str: base = _safe(root) hits = [str(Path(x).relative_to(WORKSPACE)) for x in glob.glob(f"{base}/**/{pattern}", recursive=True)] return json.dumps(hits[:200]) def git_clone(url: str, dest: str = "") -> str: name = dest or url.rstrip("/").split("/")[-1].replace(".git", "") target = _safe(name) if target.exists(): return f"{name} already exists" gitlib.Repo.clone_from(url, target) return f"cloned -> {name}" def git_pull(repo: str) -> str: r = gitlib.Repo(_safe(repo)); r.remotes.origin.pull() return "pulled" def git_push(repo: str, message: str = "update", token: str = "") -> str: r = gitlib.Repo(_safe(repo)) r.git.add(A=True); r.index.commit(message) if token: url = r.remotes.origin.url.replace("https://", f"https://x-access-token:{token}@") r.remotes.origin.set_url(url) r.remotes.origin.push() return "pushed" def _run(cmd, cwd): proc = subprocess.run(cmd, cwd=cwd, shell=True, capture_output=True, text=True, timeout=1800) return f"[exit {proc.returncode}]\n{proc.stdout[-4000:]}\n{proc.stderr[-2000:]}" def gradle_build(repo: str, task: str = "assembleDebug") -> str: cwd = _safe(repo) wrapper = cwd / "gradlew" if wrapper.exists(): os.chmod(wrapper, 0o755) return _run(f"./gradlew {task} --no-daemon", cwd) return _run(f"gradle {task} --no-daemon", cwd) def build_apk(repo: str, variant: str = "debug") -> str: task = f"assemble{variant.capitalize()}" log = gradle_build(repo, task) apks = glob.glob(f"{_safe(repo)}/**/*.apk", recursive=True) return log + "\nAPKs:\n" + "\n".join(apks) def upload_artifact(path: str, repo_id: str, token: str, path_in_repo: str = "") -> str: api = HfApi(token=token) p = _safe(path) api.upload_file( path_or_fileobj=str(p), path_in_repo=path_in_repo or p.name, repo_id=repo_id, repo_type="model", ) return f"uploaded {p.name} -> {repo_id}" TOOLS = { "read_file": read_file, "write_file": write_file, "edit_file": edit_file, "delete_file": delete_file, "search_files": search_files, "git_clone": git_clone, "git_pull": git_pull, "git_push": git_push, "gradle_build": gradle_build, "build_apk": build_apk, "upload_artifact": upload_artifact, } TOOL_SPEC = """ Available tools (call by emitting a single JSON block: ```tool\n{"name":"...","args":{...}}\n```): - read_file(path) - write_file(path, content) - edit_file(path, find, replace) - delete_file(path) - search_files(pattern, root=".") - git_clone(url, dest="") - git_pull(repo) - git_push(repo, message="update", token="") - gradle_build(repo, task="assembleDebug") - build_apk(repo, variant="debug") - upload_artifact(path, repo_id, token, path_in_repo="") After tool result, continue. End with plain answer when done. """ SYSTEM = "You are an Android build agent. Use tools to clone, edit, build APKs, and upload artifacts. " + TOOL_SPEC TOOL_RE = re.compile(r"```tool\s*(\{.*?\})\s*```", re.S) def run_agent(user_msg, history, max_steps=6): msgs = [{"role": "system", "content": SYSTEM}] for u, a in history: msgs.append({"role": "user", "content": u}) msgs.append({"role": "assistant", "content": a}) msgs.append({"role": "user", "content": user_msg}) transcript = "" for _ in range(max_steps): out = llm.create_chat_completion(messages=msgs, temperature=0.2, max_tokens=800) reply = out["choices"][0]["message"]["content"] transcript += reply + "\n" m = TOOL_RE.search(reply) if not m: return transcript try: call = json.loads(m.group(1)) fn = TOOLS[call["name"]] result = fn(**call.get("args", {})) except Exception as e: result = f"ERROR: {e}\n{traceback.format_exc()[-500:]}" msgs.append({"role": "assistant", "content": reply}) msgs.append({"role": "user", "content": f"tool_result:\n{result}"}) transcript += f"\n[tool result]\n{result}\n" return transcript with gr.Blocks(title="Android Build Agent") as demo: gr.Markdown("## 🤖 Android Build Agent (Qwen2.5-Coder + Gradle + SDK)") chat = gr.Chatbot(height=500) box = gr.Textbox(placeholder="e.g. clone https://github.com/.../MyApp and build debug APK") state = gr.State([]) def respond(msg, hist): ans = run_agent(msg, hist) hist = hist + [(msg, ans)] return "", hist, hist box.submit(respond, [box, state], [box, chat, state]) demo.queue().launch(server_name="0.0.0.0", server_port=7860)