MultiagentPersonalAssistant

.gitignore

@@ -1,2 +1 @@
-.env
-test.py
+.env

README.md

@@ -28,25 +28,17 @@ This project is a **voice- and text-based AI assistant** powered by **FastAPI**,
 - **User database** with persistent settings and config
 - **Agents**:
   - Supervisor agent
-  - Deep research agent
   - Coding agent
-  - Smart home agent
 - **Supervisor tools**:
   - Web search
   - Current time
   - Weather
   - Yahoo finance news
-- **Deep research tools**:
-  - Web search
-  - Wikipedia
-  - Yahoo finance news
+  - Change device states (Smart Home)
 - **Coder tools**:
   - Github toolkit
   - Web search (For docs research)
-- **Smart home tools**:
-  - Get device list
-  - Change device states
-
+  - Code Interpreter
 ---
 
 ## 🛠️ Setup Instructions

agents/coder/agent.py

@@ -1,34 +1,39 @@
 from langchain_core.messages import HumanMessage, SystemMessage
 from agents.coder.prompts import prompt, system_message
 from langgraph.prebuilt import create_react_agent
+from agents.utils.coder import initialize_project
 from agents.coder.states import CoderState
 from agents.models import llm_sub_agents
-from agents.coder.tools import tools
 from langchain.tools import tool
+from agents.coder import tools
 from dotenv import load_dotenv
+import shutil
 import os
 
 load_dotenv()
 
 agent = create_react_agent(
     llm_sub_agents,
-    tools=tools,
+    tools=tools.tools,
     prompt=prompt,
     state_schema=CoderState,
 )
 
 @tool
-def coder_agent(project_name: str, task_description: str, private: bool=False):
+def coder_agent(project_name: str, task_description: str, repo_owner: str = None, branch: str = "main", private: bool = False):
     """
-    Coder Agent, used as a tool for any coding tasks, it creates project, tests it and saves to GitHub.
+    Coder Agent, used as a tool for any coding tasks, it creates a new project or downloads repo from GitHub to further adjust it, tests it and saves to GitHub.
 
     Args:
         project_name (str): The name of the GitHub repository and directory for the project.
-        task_description (str): A detailed description of the project for the coder to create.
+        repo_owner (str): The owner of the GitHub repository, used only if we want to download the existing repo, not create a new project from scratch. Defaults to None (user's GitHub account)
+        branch (str): The branch of the GitHub repository. Defaults to "main".
+        task_description (str): A detailed description of the project for the coder to create or adjust.
         private (bool, optional): Whether the coder should be private or public. Defaults to False.
     """
+    initialize_project(project_name=project_name, repo_owner=repo_owner, branch=branch, private=private)
     path = f"agents/coder/projects/{project_name}"
     messages = agent.invoke({'messages': [SystemMessage(system_message(project_name, private)), HumanMessage(content=task_description)], 'project_name': project_name, 'private': private})
     if os.path.isdir(path):
-        os.rmdir(path)
+        shutil.rmtree(path, ignore_errors=True)
     return messages['messages'][-1].content

agents/coder/prompts.py

@@ -2,18 +2,20 @@ prompt = """
 You are a coding expert. 
 Always follow these steps:
 
-1. Create a new project for this task using 'create_project' or use existing project with 'download_repo' tool.\n
+1. Create a new project for this task or use existing project. (Done automatically) \n
 2. Use 'write_file' tool to save your solution code into the GitHub repository.\n
 3. Test the code and review it until it works.
 4. Tell the user you're done.
 """
 
 def system_message(project_name: str, private: bool=False) -> str:
-    return f"""Your job is to create a coding project based on the user query.\n
+    return f"""Your job is to create or adjust a coding project based on the user query. The project folder is created/downloaded automatically.\n
+    
+    Project name: '{project_name}'\n
+    Private: '{private}'
 
-    1. Create a new GitHub {"private" if private else "public"} repository and project directory, named '{project_name}' for this task using 'create_project' tool or take existing GitHub repository to adjust using 'download_repo' tool if the user request implies adjusting the project instead of creating it from scratch. \n
-    2. Use 'tavily_search' tool to retrieve the code examples, documentation and tutorials for the task.\n
-    3. Write files with the code to your project using 'write_code' tool.\n
+    1.  Use 'tavily_search' tool to retrieve the code examples, documentation and tutorials for the task.\n
+    3. Write files with the code to your project using 'write_file' tool.\n
     4. Test your code using 'run_cmd' tool.\n
     5. Critically review your code for weak points using 'list_files' and 'read_file' tools.\n
     6. Adjust the code after the carefully reviewing and testing your code.\n

agents/coder/tools.py

@@ -13,30 +13,6 @@ tavily_search = TavilySearch(
     include_answer=False
 )
 
-@tool
-def create_project(project_name: str, private: bool = False):
-    """Creates a project directory and GitHub repository with the given project_name."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        user.create_repo(project_name, private=private)
-        os.mkdir(f"agents/coder/projects/{project_name}")
-        return f"Repository '{project_name}' created successfully!"
-    except Exception as e:
-        return f"Error creating repository: {str(e)}"
-
-@tool
-def list_repos():
-    """Lists all repositories owned by the authenticated GitHub user."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repos = user.get_repos()
-        repo_names = [repo.name for repo in repos]
-        return f"Repositories: {', '.join(repo_names)}"
-    except Exception as e:
-        return f"Error listing repositories: {str(e)}"
-
 @tool
 def list_repo_branches(project_name: str):
     """Lists all branches in a GitHub repository."""
@@ -66,9 +42,9 @@ def create_repo_branch(project_name: str, branch_name: str):
 @tool
 def write_file(project_name: str, file_path: str, file_contents: str):
     """Adds a new file to the project directory and GitHub repository or updates the existing one."""
-
     try:
-        with open(f"agents/coder/projects/{project_name}/{file_path}", "w", encoding="utf-8") as f:
+        full_path = os.path.join('agents/coder/projects', project_name, file_path)
+        with open(full_path, "w", encoding="utf-8") as f:
             f.write(file_contents)
     except Exception as e:
         return f"Error with file operation: {str(e)}"
@@ -93,8 +69,9 @@ def write_file(project_name: str, file_path: str, file_contents: str):
 def delete_file(project_name: str, file_path: str):
     """Deletes a file from the GitHub repository."""
     try:
-        if os.path.isfile(file_path):
-            os.remove(file_path)
+        full_path = os.path.join('agents/coder/projects', project_name, file_path)
+        if os.path.isfile(full_path):
+            os.remove(full_path)
     except Exception as e:
         return f"Error deleting file: {str(e)}"
     try:
@@ -104,64 +81,41 @@ def delete_file(project_name: str, file_path: str):
         file = repo.get_contents(file_path)
         sha = file.sha
         repo.delete_file(file_path, "Deleting file", sha)
+        print(f"File '{file_path}' deleted successfully in '{project_name}'.")
         return f"File '{file_path}' deleted successfully from '{project_name}'."
     except Exception as e:
+        print(f"Error deleting file: {str(e)}")
         return f"Error deleting file: {str(e)}"
 
-@tool
-def read_repo_file(project_name: str, file_path: str):
-    """Reads the content of a file from a GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(project_name)
-        file = repo.get_contents(file_path)
-        return file.decoded_content.decode('utf-8')
-    except Exception as e:
-        return f"Error reading file: {str(e)}"
-
 @tool
 def read_file(project_name: str, file_path: str) -> str:
     """Reads the content of a file from a project directory."""
-    full_path = f"agents/coder/projects/{project_name}/{file_path}"
+    full_path = os.path.join('agents/coder/projects', project_name, file_path)
     try:
         with open(full_path, "r", encoding="utf-8") as f:
             return f.read()
     except Exception as e:
+        print(f"Error reading file: {str(e)}")
         return f"Error reading file '{full_path}': {str(e)}"
 
-@tool
-def list_repo_files(project_name: str):
-    """Lists all files in the GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(project_name)
-        files = repo.get_contents("")
-        file_list = [file.name for file in files]
-        return f"Files in '{project_name}': {', '.join(file_list)}"
-    except Exception as e:
-        return f"Error listing files: {str(e)}"
-
 @tool
 def list_files(project_name: str, directory: str = "") -> dict:
     """
     Lists files in a project directory.
 
-    :param project_name: Name of the project.
-    :param directory: Optional subdirectory inside the project. By default, all project files are listed.
+    :param project_name: The name of the project.
+    :param directory: Optional subdirectory inside the project. By default, (directory="") all project files are listed.
     :return: List of files in the directory inside the project.
     """
-    base_path = os.path.join("agents/coder/projects", project_name, directory)
-
+    full_path = os.path.join('agents/coder/projects', project_name, directory)
     try:
-        if not os.path.exists(base_path):
-            return f"Directory '{base_path}' does not exist."
+        if not os.path.exists(full_path):
+            return f"Directory '{full_path}' does not exist."
 
-        if not os.path.isdir(base_path):
-            return f"Path '{base_path}' is not a directory."
+        if not os.path.isdir(full_path):
+            return f"Path '{full_path}' is not a directory."
 
-        files = os.listdir(base_path)
+        files = os.listdir(full_path)
         return files
 
     except Exception as e:
@@ -172,16 +126,16 @@ def run_cmd(project_name: str, cmd: str, timeout: int = 30):
     """
     Runs a shell command in the project_name directory.
     """
-    project_dir = os.path.join("agents/coder/projects", project_name)
+    full_path = os.path.join("agents/coder/projects", project_name)
 
-    if not os.path.isdir(project_dir):
-        return -1, "", f"Project directory not found: {project_dir}"
+    if not os.path.isdir(full_path):
+        return -1, "", f"Project directory not found: {full_path}"
 
     try:
         res = subprocess.run(
             cmd,
             shell=True,
-            cwd=project_dir,
+            cwd=full_path,
             capture_output=True,
             text=True,
             timeout=timeout
@@ -192,48 +146,5 @@ def run_cmd(project_name: str, cmd: str, timeout: int = 30):
     except Exception as e:
         return -1, "", f"Error running command: {str(e)}"
 
-@tool
-def download_repo(project_name: str, repo_owner: str = None, branch: str = "main"):
-    """
-    Downloads a GitHub repo (user/<project_name>) using PyGithub
-    and saves it into agents/coder/projects/{project_name}.
-
-    Args:
-        project_name (str): Repo name on GitHub and local folder name.
-        repo_owner (str): Repo owner on GitHub, by default = None (it's the user's repository).
-        branch (str): Branch to download (default 'main').
-    """
-    gh = Github(os.getenv("GITHUB_TOKEN"))
-    user = gh.get_user()
-    if repo_owner:
-        repo = gh.get_repo(f"{repo_owner}/{project_name}")
-    else:
-        repo = user.get_repo(project_name)
-
-    dest_path = os.path.join('agents/coder/projects', project_name)
-    os.makedirs(dest_path, exist_ok=True)
-
-    branch_ref = repo.get_branch(branch)
-    commit_sha = branch_ref.commit.sha
-
-    # Get the tree (recursive = True to walk full repo)
-    tree = repo.get_git_tree(commit_sha, recursive=True).tree
-
-    for item in tree:
-        file_path = os.path.join(dest_path, item.path)
-
-        if item.type == "tree":
-            os.makedirs(file_path, exist_ok=True)
-        elif item.type == "blob":
-            blob = repo.get_git_blob(item.sha)
-            file_content = blob.content
-            import base64
-            decoded = base64.b64decode(file_content)
-
-            os.makedirs(os.path.dirname(file_path), exist_ok=True)
-            with open(file_path, "wb") as f:
-                f.write(decoded)
-
-    return f"Repo '{project_name}' downloaded into {dest_path}."
 
-tools = [tavily_search, create_project, list_repos, list_repo_branches, create_repo_branch, write_file, delete_file, read_repo_file, read_file, list_repo_files, list_files, run_cmd, download_repo]
+tools = [tavily_search, list_repo_branches, create_repo_branch, write_file, delete_file, read_file, list_files, run_cmd]

agents/models.py

@@ -20,7 +20,8 @@ llm_peripheral = init_chat_model(
 )
 
 llm_sub_agents = init_chat_model(
-    model="groq:qwen/qwen3-32b"
+    model="groq:moonshotai/kimi-k2-instruct-0905",
+    max_tokens=4000
 )
 
 llm_image = init_chat_model(

agents/prompts.py

@@ -1,4 +1,5 @@
 from agents.utils.smart_home import get_device_list
+from agents.utils.coder import list_repos
 from dotenv import load_dotenv
 import os
 
@@ -27,10 +28,6 @@ Your top priority: satisfy the user’s query as effectively and efficiently as
 
 
 def system_message(user_info: dict):
-    try:
-        device_list = "\n\n".join(f"{i + 1}. {device['name']} (id: {device['id']})\nStates:\n" + "\n".join(f"{state}" for state in device['states']) for i, device in enumerate(get_device_list()))
-    except Exception as e:
-        device_list = None
     return f"""
     You are an intelligent assistant named {os.getenv('ASSISTANT_NAME', 'Assistant')}, helpful personal assistant built using a multi-agent system architecture. Your tools include web search, weather and time lookups, code execution, and GitHub integration. You work inside a Telegram interface and respond concisely, clearly, and informatively.
 
@@ -39,7 +36,8 @@ def system_message(user_info: dict):
     - **User ID**: {user_info.get('user_id', 'Unknown')}
     - **Location**: {user_info.get('location', 'Unknown') or 'Unknown'}
     - **Coordinates**: ({user_info.get('latitude', 'N/A') or 'N/A'}, {user_info.get('longitude', 'N/A') or 'N/A'})
-    - **Smart Home Devices**:  {device_list or "Unknown"}
+    - **Smart Home Devices**:  {get_device_list() or "Unknown"}
+    - **GitHub**: {list_repos() or "Unknown"}
 
     You may use their location when answering weather or time-related queries. If the location is unknown, you may ask the user to share it.
     You may use the smart home devices if user asks you to.

agents/utils/coder.py

@@ -0,0 +1,83 @@
+from dotenv import load_dotenv
+from github import Github
+import os
+
+load_dotenv()
+
+def list_repos():
+    """Lists all repositories owned by the authenticated GitHub user."""
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repos = user.get_repos()
+        repo_names = [repo.name for repo in repos]
+        return ', '.join(repo_names)
+    except Exception as e:
+        return None
+
+def download_repo(project_name: str, repo_owner: str = None, branch: str = "main", private: bool = False):
+    """
+    Downloads a GitHub repo (user/<project_name>) using PyGithub
+    and saves it into agents/coder/projects/{project_name}.
+
+    Args:
+        project_name (str): Repo name on GitHub and local folder name.
+        repo_owner (str): Repo owner on GitHub, by default = None (it's the user's repository).
+        branch (str): Branch to download (default 'main').
+    """
+    gh = Github(os.getenv("GITHUB_TOKEN"))
+    user = gh.get_user()
+    if repo_owner:
+        repo = gh.get_repo(f"{repo_owner}/{project_name}")
+        user.create_repo(project_name, private=private)
+    else:
+        repo = user.get_repo(project_name)
+
+    dest_path = os.path.join('agents/coder/projects', project_name)
+    os.makedirs(dest_path, exist_ok=True)
+
+    try:
+        branch_ref = repo.get_branch(branch)
+        commit_sha = branch_ref.commit.sha
+
+        # Get the tree (recursive = True to walk full repo)
+        tree = repo.get_git_tree(commit_sha, recursive=True).tree
+
+        for item in tree:
+            file_path = os.path.join(dest_path, item.path)
+
+            if item.type == "tree":
+                os.makedirs(file_path, exist_ok=True)
+            elif item.type == "blob":
+                blob = repo.get_git_blob(item.sha)
+                file_content = blob.content
+                import base64
+                decoded = base64.b64decode(file_content)
+
+                os.makedirs(os.path.dirname(file_path), exist_ok=True)
+                with open(file_path, "wb") as f:
+                    f.write(decoded)
+    except Exception as e:
+        return f"Error downloading repository: {str(e)}"
+
+def initialize_project(project_name: str, repo_owner: str = None, branch: str = "main", private: bool = False):
+    """Creates a project directory and GitHub repository with the given project_name."""
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repos = user.get_repos()
+        repo_names = [repo.name for repo in repos]
+        if project_name in repo_names or repo_owner:
+            try:
+                download_repo(project_name=project_name, repo_owner=repo_owner, branch=branch, private=private)
+                return f"Repository '{f'{repo_owner}/{project_name}' if repo_owner else project_name}' downloaded successfully!"
+            except Exception as e:
+                os.mkdir(f"agents/coder/projects/{project_name}")
+                user.create_repo(project_name, private=private)
+                return f"Repository '{project_name}' created successfully!"
+        else:
+            user.create_repo(project_name, private=private)
+            os.mkdir(f"agents/coder/projects/{project_name}")
+            return f"Repository '{project_name}' created successfully!"
+    except Exception as e:
+        return f"Error creating repository: {str(e)}"

agents/utils/smart_home.py

@@ -38,17 +38,15 @@ if not connection['success']:
     connection = openapi.connect(os.getenv('TUYA_USERNAME'), os.getenv('TUYA_PASSWORD'), os.getenv('TUYA_COUNTRY_CODE'), 'SmartLife')
 
 def get_device_list():
-    """
-    Get the list of devices bound to the smart home.
-    Each device has id: str, name: str and states: List[dict].
-    States is a list of device states which can be changed.
-    State format: {'code': ..., 'value': ...}
-    """
-    response = openapi.get('/v1.0/users/{}/devices'.format(connection['result']['uid']))['result']
-    devices = []
-    for device in response:
-        devices.append({"id": device['id'], 'name': device['name'], 'states': device['status']})
-    return devices
+    """List all the smart home devices."""
+    try:
+        response = openapi.get('/v1.0/users/{}/devices'.format(connection['result']['uid']))['result']
+        devices = []
+        for device in response:
+            devices.append({"id": device['id'], 'name': device['name'], 'states': device['status']})
+        return "\n\n".join(f"{i + 1}. {device['name']} (id: {device['id']})\nStates:\n" + "\n".join(f"{state}" for state in device['states']) for i, device in enumerate(devices))
+    except Exception as e:
+        return None
 
 @tool
 def change_device_states(id:str, commands:list[dict]):