210 lines
6.7 KiB
Python
210 lines
6.7 KiB
Python
#!/usr/bin/env -S .claude/venv/bin/python3
|
|
"""
|
|
Agent loop local (~120 lignes) pour interagir avec LM Studio via l'API OpenAI-compatible.
|
|
Usage : python3 .claude/mcp/qwen3-mcp/agent_lm.py --task "..." --files file1.py file2.go
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
SERVER = "http://localhost"
|
|
PORT = 8888 # 1248
|
|
LMSTUDIO_URL = f"{SERVER}:{PORT}/v1/chat/completions"
|
|
LMSTUDIO_MODELS_URL = f"{SERVER}:{PORT}/v1/models"
|
|
DEFAULT_MODEL = "qwen/qwen3-coder-next"
|
|
|
|
SYSTEM_PROMPT = """Tu es un assistant de codage spécialisé. Tu effectues des tâches atomiques sur des fichiers source.
|
|
Règles :
|
|
- Ne modifie que ce qui est demandé
|
|
- Ne change pas les signatures publiques (traits Rust, interfaces Go exportées)
|
|
- Retourne uniquement le code, sans explication ni markdown
|
|
- Si tu ne peux pas accomplir la tâche, réponds avec: ERROR: <raison>
|
|
"""
|
|
|
|
|
|
def list_models() -> list[str]:
|
|
"""Retourne la liste des modèles disponibles dans LM Studio."""
|
|
try:
|
|
resp = requests.get(LMSTUDIO_MODELS_URL, timeout=10)
|
|
resp.raise_for_status()
|
|
return [m["id"] for m in resp.json().get("data", [])]
|
|
except requests.RequestException as e:
|
|
print(f"Impossible de contacter LM Studio : {e}", file=sys.stderr)
|
|
return []
|
|
|
|
|
|
def read_files(paths: list[str]) -> str:
|
|
"""Lit les fichiers et les formate pour le prompt."""
|
|
context = ""
|
|
for path in paths:
|
|
try:
|
|
content = Path(path).read_text(encoding="utf-8")
|
|
context += f"=== {path} ===\n{content}\n\n"
|
|
except OSError as e:
|
|
context += f"=== {path} === ERREUR: {e}\n\n"
|
|
return context
|
|
|
|
|
|
def call_lmstudio(task: str, file_context: str, model: str) -> str:
|
|
"""Envoie la tâche à LM Studio et retourne la réponse."""
|
|
user_message = (
|
|
f"TÂCHE: {task}\n\nFICHIERS:\n{file_context}"
|
|
if file_context
|
|
else f"TÂCHE: {task}"
|
|
)
|
|
|
|
payload = {
|
|
"model": model,
|
|
"messages": [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_message},
|
|
],
|
|
"temperature": 0.2,
|
|
"max_tokens": 4096,
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(LMSTUDIO_URL, json=payload, timeout=120)
|
|
resp.raise_for_status()
|
|
return resp.json()["choices"][0]["message"]["content"]
|
|
except requests.Timeout:
|
|
return "ERROR: LM Studio timeout (>120s)"
|
|
except requests.RequestException as e:
|
|
return f"ERROR: {e}"
|
|
except (KeyError, IndexError) as e:
|
|
return f"ERROR: réponse inattendue de LM Studio : {e}"
|
|
|
|
|
|
def write_file(path: str, content: str) -> bool:
|
|
"""Écrit le contenu dans un fichier, crée les répertoires si nécessaire."""
|
|
try:
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(path).write_text(content, encoding="utf-8")
|
|
return True
|
|
except OSError as e:
|
|
print(f"Erreur écriture {path}: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
|
|
def run_lint(file_path: str) -> tuple[bool, str]:
|
|
"""Lance le linter approprié selon l'extension."""
|
|
import subprocess
|
|
|
|
if file_path.endswith(".rs"):
|
|
result = subprocess.run(
|
|
["cargo", "clippy", "--quiet"], capture_output=True, text=True, timeout=60
|
|
)
|
|
elif file_path.endswith(".go"):
|
|
result = subprocess.run(
|
|
["golangci-lint", "run", file_path],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
else:
|
|
return True, "no linter for this file type"
|
|
|
|
passed = result.returncode == 0
|
|
output = (result.stdout + result.stderr).strip()
|
|
return passed, output
|
|
|
|
|
|
def agent_loop(
|
|
task: str,
|
|
input_files: list[str],
|
|
output_file: str | None,
|
|
model: str,
|
|
max_retries: int = 2,
|
|
) -> dict:
|
|
"""Boucle principale : génère, écrit, lint, corrige (max_retries fois)."""
|
|
file_context = read_files(input_files) if input_files else ""
|
|
result = {
|
|
"status": "failure",
|
|
"files_modified": [],
|
|
"summary": "",
|
|
"lint": "skipped",
|
|
}
|
|
|
|
for attempt in range(max_retries + 1):
|
|
response = call_lmstudio(task, file_context, model)
|
|
|
|
if response.startswith("ERROR:"):
|
|
result["summary"] = response
|
|
break
|
|
|
|
target = output_file or (input_files[0] if input_files else None)
|
|
if not target:
|
|
result["status"] = "success"
|
|
result["summary"] = response
|
|
result["lint"] = "skipped (no output file)"
|
|
break
|
|
|
|
if write_file(target, response):
|
|
result["files_modified"] = [target]
|
|
|
|
lint_ok, lint_output = run_lint(target)
|
|
result["lint"] = "passed" if lint_ok else f"failed: {lint_output[:500]}"
|
|
|
|
if lint_ok:
|
|
result["status"] = "success"
|
|
result["summary"] = (
|
|
f"Attempt {attempt + 1}: task completed successfully"
|
|
)
|
|
break
|
|
elif attempt < max_retries:
|
|
task = f"{task}\n\nCORRECTION REQUISE (tentative {attempt + 1}):\n{lint_output}"
|
|
file_context = read_files([target])
|
|
else:
|
|
result["status"] = "partial"
|
|
result["summary"] = f"Lint failed after {max_retries + 1} attempts"
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Agent loop local pour LM Studio")
|
|
parser.add_argument("--task", required=True, help="Description de la tâche")
|
|
parser.add_argument("--files", nargs="*", default=[], help="Fichiers source à lire")
|
|
parser.add_argument(
|
|
"--output", help="Fichier de sortie (défaut: premier fichier input)"
|
|
)
|
|
parser.add_argument("--model", default=DEFAULT_MODEL, help="Modèle LM Studio")
|
|
parser.add_argument(
|
|
"--list-models", action="store_true", help="Liste les modèles disponibles"
|
|
)
|
|
parser.add_argument(
|
|
"--json", action="store_true", dest="json_output", help="Sortie JSON"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.list_models:
|
|
models = list_models()
|
|
if models:
|
|
print("Modèles disponibles :")
|
|
for m in models:
|
|
print(f" - {m}")
|
|
else:
|
|
print("Aucun modèle trouvé ou LM Studio inaccessible.")
|
|
return
|
|
|
|
result = agent_loop(args.task, args.files, args.output, args.model)
|
|
|
|
if args.json_output:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"STATUS: {result['status']}")
|
|
print(f"FILES_MODIFIED: {', '.join(result['files_modified']) or 'none'}")
|
|
print(f"SUMMARY: {result['summary']}")
|
|
print(f"LINT: {result['lint']}")
|
|
|
|
sys.exit(0 if result["status"] == "success" else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|