File size: 4,236 Bytes
dc7b96e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | #!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
from typing import Any
DG_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = DG_ROOT / "configs" / "client_profiles" / "open-interpreter.dg.json"
def load_config(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def import_open_interpreter() -> Any:
from interpreter import interpreter
return interpreter
def model_settings(config: dict[str, Any]) -> dict[str, Any]:
return {
"model": os.environ.get("OPEN_INTERPRETER_MODEL") or config["model"],
"api_base": os.environ.get("OPENAI_BASE_URL") or config["api_base"],
"api_key": os.environ.get("OPENAI_API_KEY") or config["api_key"],
"context_window": int(os.environ.get("OPEN_INTERPRETER_CONTEXT_WINDOW") or config.get("context_window") or 768),
"max_tokens": int(os.environ.get("OPEN_INTERPRETER_MAX_TOKENS") or config.get("max_tokens") or 256),
"temperature": float(os.environ.get("OPEN_INTERPRETER_TEMPERATURE") or config.get("temperature") or 0.0),
"offline": bool(config.get("offline", True)),
"auto_run": bool(config.get("auto_run", False)),
"safe_mode": os.environ.get("OPEN_INTERPRETER_SAFE_MODE") or config.get("safe_mode") or "ask",
}
def configure_interpreter(interpreter: Any, config: dict[str, Any]) -> dict[str, Any]:
settings = model_settings(config)
interpreter.offline = settings["offline"]
interpreter.auto_run = settings["auto_run"]
interpreter.safe_mode = settings["safe_mode"]
interpreter.llm.model = settings["model"]
interpreter.llm.api_base = settings["api_base"]
interpreter.llm.api_key = settings["api_key"]
interpreter.llm.context_window = settings["context_window"]
interpreter.llm.max_tokens = settings["max_tokens"]
interpreter.llm.temperature = settings["temperature"]
return settings
def run_task(args: argparse.Namespace, config: dict[str, Any]) -> int:
interpreter = import_open_interpreter()
settings = configure_interpreter(interpreter, config)
os.chdir(Path(args.repo).resolve())
prompt = (
"Use concise output. Prefer inspection and explanations. "
"Do not make repository edits unless explicitly asked.\n\n"
f"{args.task}"
)
result = interpreter.chat(prompt, display=not args.json)
if args.json:
print(json.dumps({"status": "success", "settings": settings, "result": result}, ensure_ascii=False, indent=2, default=str))
return 0
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run Open Interpreter with the local DiffusionGemma profile.")
parser.add_argument("--repo", default=".", help="Target repo, used as working directory")
parser.add_argument("--config", default=str(DEFAULT_CONFIG))
parser.add_argument("--task", default="")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--smoke-import", action="store_true")
parser.add_argument("--json", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
repo = Path(args.repo).resolve()
config_path = Path(args.config).resolve()
config = load_config(config_path)
if args.smoke_import:
interpreter = import_open_interpreter()
print("open-interpreter import ok")
print(type(interpreter).__name__)
return 0
settings = model_settings(config)
if args.dry_run:
data = {
"repo": str(repo),
"config": str(config_path),
"module": config["module"],
"agent": config["agent"],
"settings": settings,
"command": f"scripts/dg_agent.sh open-interpreter -- --repo {repo} --task '...'",
}
print(json.dumps(data, ensure_ascii=False, indent=2) if args.json else "\n".join(f"{k}: {v}" for k, v in data.items()))
return 0
if not args.task:
print("--task is required unless --dry-run or --smoke-import is used", flush=True)
return 2
return run_task(args, config)
if __name__ == "__main__":
raise SystemExit(main())
|