diff options
| -rw-r--r-- | actions.py | 64 | ||||
| -rwxr-xr-x | llm.py | 103 |
2 files changed, 94 insertions, 73 deletions
diff --git a/actions.py b/actions.py new file mode 100644 index 0000000..9146e53 --- /dev/null +++ b/actions.py @@ -0,0 +1,64 @@ +from pathlib import Path +from typing import Optional +import subprocess + +def _path_checked(path: Path, cwd=Path.cwd()) -> Optional[Path]: + """Resolves `path`, returning `None` if it's not a subdirectory of `cwd`.""" + path = path.resolve() + cwd = cwd.resolve() + return path if path.is_relative_to(cwd) else None + +def _path_helper(path_str: str, callback): + path = _path_checked(Path(path_str)) + if path is None: + return {"status": "error", "reason": f"Permission denied: {path_str}"} + else: + try: + return {"status": "success", **callback(path)} + except Exception as e: + return {"status": "error", "reason": f"Failure: {e}"} + +def ls(args): + '{"action": "ls", "path": PATH} - list files at PATH' + def inner(path: Path): + return {"content": list(map(lambda s: str(s), path.iterdir()))} + match args: + case {"action": "ls", "path": path_str}: + return _path_helper(path_str, inner) + +def read(args): + '{"action": "read", "path": PATH} - read file at PATH' + def inner(path: Path): + contents = path.read_text(encoding="utf-8") + return {"content": contents} + match args: + case {"action": "read", "path": path_str}: + return _path_helper(path_str, inner) + +def write(args): + '{"action": "write", "path": PATH, "content": CONTENT} - write CONTENT into file at PATH' + def inner(path: Path, text: str): + n_bytes = path.write_text(text) + return {"bytes": n_bytes} + match args: + case {"action": "write", "path": path_str, "content": content}: + return _path_helper(path_str, lambda path: inner(path, content)) + +def tldr(args): + '{"action": "tldr", "program": PROGRAM} - read tldr docs for PROGRAM' + match args: + case {"action": "tldr", "program": program_str}: + try: + result = subprocess.run( + ["tldr", program_str], + capture_output=True, + text=True, + check=True + ) + return {"status": "success", "content": result.stdout} + except subprocess.CalledProcessError: + return {"status": "error", "reason": f"Tldr for '{program_str}' not found."} + except Exception as e: + return {"status": "error", "reason": f"An unexpected error occurred: {e}"} + +ALL_ACTIONS = [ls, read, write, tldr] @@ -4,9 +4,11 @@ from pathlib import Path from openai import OpenAI from typing import Optional import argparse +import re import subprocess import readline import json +from actions import * import logging logger = logging.getLogger(__name__) @@ -55,10 +57,6 @@ def chat(content, state = (), query_method = direct_query, role = "user"): return (response, state) -def strip_enclosing(s: str, start="```json", end="```") -> str: - s = s.strip() - return s[len(start):-len(end)] if s[:len(start)] == start and s[-len(end):] == end else s - def path_checked(path: Path, cwd = Path.cwd()) -> Optional[Path]: """Resolves `path`, returning `None` if it's not a subdirectory of `cwd`.""" @@ -75,79 +73,37 @@ def inner_query_agent(actions: list, query: str, state = ()): "Respond JSON indicating an action, or respond in natural language if you are finished.", state) - try: - action = json.loads(strip_enclosing(res)) - except json.JSONDecodeError: - return (res, state) - - for f in actions: - result_json = f(action) - if result_json is not None: - (res, state) = inner_query_agent(actions, json.dumps(result_json), state) - return (res, state) - - logger.info(f"valid json '{res}' but no actions matched") + # Replace all fenced JSON blocks - return (res, state) + replacements = [] + for m in re.finditer(r"```json\s*(.*?)\s*```", res, re.DOTALL): + json_str = m.group(1) + try: + action = json.loads(json_str) + except json.JSONDecodeError: + continue + for f in actions: + result_json = f(action) + if result_json is not None: + replacements.append((m.start(), m.end(), action["action"])) + break -def query_agent(query: str, state = ()): - - def path_helper(path_str: str, callback): - path = path_checked(Path(path_str)) - if path is None: - return {"status": "error", "reason": f"Permission denied: {path_str}"} - else: - try: - return {"status": "success", **callback(path)} - except Exception as e: - return {"status": "error", "reason": f"Failure: {e}"} + replaced_res = "" + i = 0 + for (start, end, action) in replacements: + replaced_res += res[i:start] + f"[performed '{action}']\n" + i = end + replaced_res += res[i:] - def ls(args): - '{"action": "ls", "path": PATH} - list files at PATH' - def inner(path: Path): - return {"content": list(map(lambda s: str(s), path.iterdir()))} - match args: - case {"action": "ls", "path": path_str}: - return path_helper(path_str, inner) + if len(replacements) > 0: + (res, state) = inner_query_agent(actions, json.dumps(result_json), state) + return (replaced_res + "\n\n" + res, state) + else: + return (replaced_res, state) - def read(args): - '{"action": "read", "path": PATH} - read file at PATH' - def inner(path: Path): - contents = path.read_text(encoding="utf-8") - return {"content": contents} - match args: - case {"action": "read", "path": path_str}: - return path_helper(path_str, inner) - def write(args): - '{"action": "write", "path": PATH, "content": CONTENT} - write CONTENT into file at PATH' - def inner(path: Path, text: str): - n_bytes = path.write_text(text) - return {"bytes": n_bytes} - match args: - case {"action": "write", "path": path_str, "content": content}: - return path_helper(path_str, lambda path: inner(path, content)) - - - def tldr(args): - '{"action": "tldr", "program": PROGRAM} - read tldr docs for PROGRAM' - match args: - case {"action": "tldr", "program": program_str}: - try: - # Run man command and capture output - # -P cat is used to avoid paging and just print the text - result = subprocess.run( - ["tldr", program_str], - capture_output=True, - text=True, - check=True - ) - return {"status": "success", "content": result.stdout} - except subprocess.CalledProcessError: - return {"status": "error", "reason": f"Tldr for '{program_str}' not found."} - except Exception as e: - return {"status": "error", "reason": f"An unexpected error occurred: {e}"} +def query_agent(query: str, state = ()): actions = [ls, read, write, tldr] @@ -155,9 +111,10 @@ def query_agent(query: str, state = ()): "content": f"""You are an autonomous coding agent that can help with programming tasks. You can explore repositories, read files, search for information, and make changes to code. -You have access to the following actions. +You have access to the following actions, which you should embed in JSON markdown blocks "```json ACTION ```" in your responses. {'\n'.join(map(lambda f: '* ' + f.__doc__, actions))} -* RESPONSE - send non-json text RESPONSE to the user once the task is complete + +Once you are finished with your task, give a response with none of the above JSON commands. """},) state = state + ({"role": "system", "content": "Analyze the following user-given task and break it down into steps."},) |