AI SDE Agent

ai_agent_tutorials/ai_sde_agent/README.md

@@ -0,0 +1 @@
+AI Software Development Engineer Agent

ai_agent_tutorials/ai_sde_agent/requirements.txt

@@ -0,0 +1,5 @@
+swekit
+composio-core
+composio-crewai
+crewai
+langchain-aws

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/__init__.py

@@ -0,0 +1 @@
+"""Composio-CrewAI SWE Agent"""

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/agent.py

@@ -0,0 +1,80 @@
+"""CrewAI SWE Agent"""
+
+import os
+from enum import Enum
+from crewai import LLM
+import dotenv
+import typing as t
+from crewai import Agent, Crew, Process, Task
+from langchain_openai import ChatOpenAI
+from prompts import BACKSTORY, DESCRIPTION, EXPECTED_OUTPUT, GOAL, ROLE
+
+from composio_crewai import Action, App, ComposioToolSet, WorkspaceType
+
+
+# Load environment variables from .env
+dotenv.load_dotenv()
+
+class Model(str, Enum):
+    OPENAI = "openai"
+
+model = Model.OPENAI
+
+# Initialize tool.
+if model == Model.OPENAI:
+    client = ChatOpenAI(
+        api_key="sk-proj-",  # type: ignore
+        model="gpt-4o-mini",
+    )
+else:
+    raise ValueError(f"Invalid model: {model}")
+
+def get_crew(repo_path: str, workspace_id: str):
+
+    composio_toolset = ComposioToolSet(
+        workspace_config=WorkspaceType.Host(),
+        metadata={
+            App.CODE_ANALYSIS_TOOL: {
+                "dir_to_index_path": repo_path,
+            }
+        },
+    )
+    if workspace_id:
+        composio_toolset.set_workspace_id(workspace_id)
+
+    # Get required tools
+    tools = [
+        *composio_toolset.get_tools(
+            apps=[
+                App.FILETOOL,
+                App.SHELLTOOL,
+                App.CODE_ANALYSIS_TOOL,
+            ]
+        ),
+    ]
+
+    # Define agent
+    agent = Agent(
+        role=ROLE,
+        goal=GOAL,
+        backstory=BACKSTORY,
+        llm=client,
+        tools=tools,
+        verbose=True,
+    )
+
+    task = Task(
+        description=DESCRIPTION,
+        expected_output=EXPECTED_OUTPUT,
+        agent=agent,
+    )
+
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        process=Process.sequential,
+        verbose=True,
+        cache=False,
+        memory=True,
+    )
+    return crew, composio_toolset

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/benchmark.py

@@ -0,0 +1,58 @@
+import argparse
+
+from agent import composio_toolset, crew
+
+from swekit.benchmark.run_evaluation import evaluate
+from swekit.config.store import IssueConfig
+
+
+def bench(workspace_id: str, issue_config: IssueConfig) -> str:
+    """Run benchmark on the agent."""
+
+    # Set the workspace for the tools to run.
+    composio_toolset.set_workspace_id(workspace_id)
+
+    # kick off the crew on the issue.
+    return crew.kickoff(
+        inputs={
+            "repo": issue_config.repo_name,
+            "issue": issue_config.issue_desc,
+        }
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Run benchmark on the agent.",
+    )
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument(
+        "--test-split",
+        type=str,
+        default="1:2",
+        help="Test split ratio (e.g. 1:2, 1:500) Maximum 500 tests per project.",
+    )
+    group.add_argument(
+        "--test-instance-ids",
+        type=str,
+        default="",
+        help="Test instance ids (comma-separated)",
+    )
+    args = parser.parse_args()
+
+    if args.test_instance_ids:
+        test_instance_ids_list = [
+            id.strip() for id in args.test_instance_ids.split(",")
+        ]
+        test_range = "1:500"
+    else:
+        test_instance_ids_list = []
+        test_range = args.test_split
+
+    evaluate(
+        bench,
+        dry_run=False,
+        test_range=test_range,
+        test_instance_ids=test_instance_ids_list,
+        # image_name="composio/composio:latest", # if you are doing local dev
+    )

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/inputs.py

@@ -0,0 +1,78 @@
+import re
+import typing as t
+from pathlib import Path
+from composio import Action, ComposioToolSet
+
+
+InputType = t.TypeVar("InputType")
+
+
+def read_user_input(
+    prompt: str,
+    metavar: str,
+    validator: t.Callable[[str], InputType],
+) -> InputType:
+    """Read user input."""
+    while True:
+        value = input(f"{prompt} > ")
+        try:
+            return validator(value)
+        except Exception as e:
+            print(f"Invalid value for `{metavar}` error parsing `{value}`; {e}")
+
+
+def _github_repository_name_validator(name: str) -> t.Tuple[str, str]:
+    """Validate github repository name."""
+    if " " in name:
+        raise ValueError()
+    owner, name = name.split("/")
+    return owner, name
+
+
+def _create_github_issue_validator(owner: str, name: str) -> t.Callable[[str], str]:
+    """Create a github issue validator."""
+
+    def _github_issue_validator(value: str) -> str:
+        """Validate github issue."""
+        # Check if it's a file path
+        if Path(value).resolve().exists():
+            return Path(value).read_text(encoding="utf-8")
+
+        if re.match(r"^\d+$", value):
+            response_data = ComposioToolSet().execute_action(
+                action=Action.GITHUB_GET_AN_ISSUE,
+                params={
+                    "owner": owner,
+                    "repo": name,
+                    "issue_number": int(value),
+                },
+            ).get("response_data")
+            if response_data:
+                return response_data["body"]
+            else:
+                raise ValueError("No issue found with the given issue number")
+
+        # If neither a file nor issue ID, treat as direct description text
+        return value
+
+    return _github_issue_validator
+
+
+def from_github() -> t.Tuple[str, str]:
+    """Take input from github."""
+    owner, name = read_user_input(
+        prompt="Enter github repository name",
+        metavar="github repository name",
+        validator=_github_repository_name_validator,
+    )
+    return (
+        f"{owner}/{name}",
+        read_user_input(
+            prompt="Enter github issue ID or description or path to the file containing description",
+            metavar="github issue",
+            validator=_create_github_issue_validator(
+                owner=owner,
+                name=name,
+            ),
+        ),
+    )

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/main.py

@@ -0,0 +1,61 @@
+from inputs import from_github
+import uuid
+from agent import get_crew
+from composio import Action
+from tools import create_pr
+
+def main() -> None:
+    """Run the agent."""
+    repo, issue = from_github()
+    owner, repo_name = repo.split("/")
+    crew, composio_toolset = get_crew(repo_path=f"/home/user/{repo_name}", workspace_id=None)
+    crew.kickoff(
+        inputs={
+            "repo": repo,
+            "issue": issue,
+        }
+    )
+    composio_toolset.execute_action(
+        action=Action.FILETOOL_CHANGE_WORKING_DIRECTORY,
+        params={"path": f"/home/user/{repo_name}"},
+    )
+    response = composio_toolset.execute_action(
+        action=Action.FILETOOL_GIT_PATCH,
+        params={},
+    )
+    branch_name = "test-branch-" + str(uuid.uuid4())[:4]
+    git_commands = [
+        f"checkout -b {branch_name}",
+        "add -u",
+        "config --global user.email '[email protected]'",
+        "config --global user.name 'random'",
+        f"commit -m '{issue}'",
+        f"push --set-upstream origin {branch_name}",
+    ]
+    for command in git_commands:
+        composio_toolset.execute_action(
+            action=Action.FILETOOL_GIT_CUSTOM,
+            params={"cmd": command},
+        )
+    composio_toolset.execute_action(
+        action=create_pr,
+        params={
+            "owner": owner,
+            "repo": repo_name,
+            "head": branch_name,
+            "base": "master",
+            "title": "Composio generated PR",
+        },
+    )  
+
+    data = response.get("data", {})
+    if data.get("error") and len(data["error"]) > 0:
+        print("Error:", data["error"])
+    elif data.get("patch"):
+        print("=== Generated Patch ===\n" + data["patch"])
+    else:
+        print("No output available")
+
+
+if __name__ == "__main__":
+    main()

ai_agent_tutorials/ai_sde_agent/sde-agent/agent/prompts.py

@@ -0,0 +1,69 @@
+ROLE = "Software Engineer"
+
+GOAL = "Fix the coding issues given by the user"
+
+BACKSTORY = """You are an autonomous programmer, your task is to
+solve the issue given in task with the tools in hand. Your mentor gave you
+following tips.
+  1. A workspace is initialized for you, and you will be working on workspace. 
+    The git repo is cloned in the path and you need to work in this directory.
+    You are in that directory. If you don't find the repo, clone it.
+  2. PLEASE READ THE CODE AND UNDERSTAND THE FILE STRUCTURE OF THE CODEBASE
+    USING GIT REPO TREE ACTION.
+  3. POST THAT READ ALL THE RELEVANT READMEs AND TRY TO LOOK AT THE FILES
+    RELATED TO THE ISSUE.
+  4. Form a thesis around the issue and the codebase. Think step by step.
+    Form pseudocode in case of large problems.
+  5. THEN TRY TO REPLICATE THE BUG THAT THE ISSUES DISCUSSES.
+     - If the issue includes code for reproducing the bug, we recommend that you
+      re-implement that in your environment, and run it to make sure you can
+      reproduce the bug.
+     - Then start trying to fix it.
+     - When you think you've fixed the bug, re-run the bug reproduction script
+      to make sure that the bug has indeed been fixed.
+     - If the bug reproduction script does not print anything when it successfully
+      runs, we recommend adding a print("Script completed successfully, no errors.")
+      command at the end of the file, so that you can be sure that the script
+      indeed ran fine all the way through.
+  6. If you run a command and it doesn't work, try running a different command.
+    A command that did not work once will not work the second time unless you
+    modify it!
+  7. If you open a file and need to get to an area around a specific line that
+    is not in the first 100 lines, say line 583, don't just use the scroll_down
+    command multiple times. Instead, use the goto 583 command. It's much quicker.
+  8. If the bug reproduction script requires inputting/reading a specific file,
+    such as buggy-input.png, and you'd like to understand how to input that file,
+    conduct a search in the existing repo code, to see whether someone else has
+    already done that. Do this by running the command: find_file "buggy-input.png"
+    If that doesn't work, use the linux 'find' command.
+  9. Always make sure to look at the currently open file and the current working
+    directory (which appears right after the currently open file). The currently
+    open file might be in a different directory than the working directory! Note
+    that some commands, such as 'create', open files, so they might change the
+    current open file.
+  10. When editing files, it is easy to accidentally specify a wrong line number
+    or to write code with incorrect indentation. Always check the code after
+    you issue an edit to make sure that it reflects what you wanted to accomplish.
+    If it didn't, issue another command to fix it.
+  11. When you finish working on the issue, use the get patch action with the
+    new files created to create the final patch to be submitted to fix the issue.
+NOTE: Give owner/repo_name while cloning the repo, not the full URL.
+"""
+
+DESCRIPTION = """We're currently solving the following issue within our repository. 
+Here's the issue text:
+  ISSUE: {issue}
+  REPO: {repo}
+
+Now, you're going to solve this issue on your own. When you're satisfied with all
+of the changes you've made, you can submit your changes to the code base by simply
+running the submit command. Note however that you cannot use any interactive
+session commands (e.g. python, vim) in this environment, but you can write
+scripts and run them. E.g. you can write a python script and then run it
+with `python </path/to/script>.py`.
+
+If you are facing "module not found error", you can install dependencies.
+Example: in case error is "pandas not found", install pandas like this `pip install pandas`
+"""
+
+EXPECTED_OUTPUT = "A patch should be generated which fixes the given issue and a PR should be created"