MultiagentPersonalAssistant

.idea/ai_toolkit.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="AI Toolkit Settings">
+    <option name="importsOfInterestPresent" value="true" />
+  </component>
+</project>

.idea/vcs.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="VcsDirectoryMappings">
-    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+    <mapping directory="" vcs="Git" />
   </component>
 </project>

agent/coder_agent.py

@@ -1,39 +0,0 @@
-from agent.prompts import coder_instructions, coder_system_message
-from langchain_core.messages import HumanMessage, SystemMessage
-from agent.models import llm_agents, llm_peripheral
-from langgraph.prebuilt import create_react_agent
-from langgraph.constants import START, END
-from agent.states import PlanCodingTask
-from langgraph.graph import StateGraph
-from agent.tools import coder_tools
-
-agent = create_react_agent(
-    llm_agents,
-    tools=coder_tools,
-    prompt=coder_instructions
-)
-
-def planning_node(state: dict):
-    planer = llm_peripheral.with_structured_output(PlanCodingTask)
-    plan = planer.invoke(state['messages'][-1].content)
-    state.update(plan)
-    return state
-
-def code_agent(state: dict):
-    system_message = SystemMessage(coder_system_message(state))
-    state.update(agent.invoke({
-        'messages': [
-            system_message,
-            HumanMessage(state['task_description']),
-        ]
-    }))
-    return state
-
-graph = StateGraph(dict)
-graph.add_node("planning_node", planning_node)
-graph.add_node("code_agent", code_agent)
-graph.add_edge(START, "planning_node")
-graph.add_edge("planning_node", "code_agent")
-graph.add_edge("code_agent", END)
-
-coder_agent = graph.compile(name="coder_agent")

agent/deep_research_agent.py

@@ -1,39 +0,0 @@
-from agent.prompts import deep_research_instructions, deep_research_system_message
-from langchain_core.messages import HumanMessage, SystemMessage
-from agent.models import llm_agents, llm_peripheral
-from langgraph.prebuilt import create_react_agent
-from agent.tools import deep_research_tools
-from langgraph.constants import START, END
-from langgraph.graph import StateGraph
-from agent.states import PlanResearch
-
-agent = create_react_agent(
-    llm_agents,
-    tools=deep_research_tools,
-    prompt=deep_research_instructions
-)
-
-def planning_node(state: dict):
-    planer = llm_peripheral.with_structured_output(PlanResearch)
-    plan = planer.invoke(state['messages'][-1].content)
-    state.update(plan)
-    return state
-
-def research_agent(state: dict):
-    system_message = SystemMessage(deep_research_system_message(state))
-    state.update(agent.invoke({
-        'messages': [
-            system_message,
-            HumanMessage(state['messages'][-1].content),
-        ]
-    }))
-    return state
-
-graph = StateGraph(dict)
-graph.add_node("planning_node", planning_node)
-graph.add_node("research_agent", research_agent)
-graph.add_edge(START, "planning_node")
-graph.add_edge("planning_node", "research_agent")
-graph.add_edge("research_agent", END)
-
-deep_research_agent = graph.compile(name="deep_research_agent")

agent/prompts.py

@@ -1,74 +0,0 @@
-coder_instructions = f"""You are a coding expert. Follow these steps every time:\n
-
-1. Use the 'web_search' tool to find how to solve the task or retrieve the package documentation and code examples.\n
-2. Create a new GitHub repository for this task using 'create_repo' tool.\n
-3. Use 'commit_file_to_repo' tool to save your solution code into the GitHub repository.\n
-
-"""
-
-smart_home_instructions = f"""You are a smart home control expert. Follow these steps every time:\n
-
-1. Use the 'get_device_list' tool to retrieve the list of available devices with their names, ids, and states.\n
-2. Identify the target device(s) based on the user’s request.\n
-3. Use the 'change_device_states' tool to update the states of the selected device(s).\n
-"""
-
-deep_research_instructions = f"""You are a deep research expert. Follow these steps every time:\n
-
-1. Iterate over the given search queries.\n
-2. Iterate over the following steps untill you have written factually correct report:\n
-2.1. Write a report based on the search results.\n
-2.2. Critically asses the report.\n
-2.3. Conduct additional search using tools to find more for research.\n
-"""
-
-def supervisor_instructions(tools: list, agents: list):
-    return f"""You are the Supervisor Agent. Your job is to interpret the user’s request and delegate tasks to the available tools and agents to complete the task.
-
-You have access to the following agents and tools:
-- Tools: {[tool.name for tool in tools]}
-- Agents: {[agent.name for agent in agents]}
-
-1. Understand and break down user’s request into smaller sub-tasks.
-2. Always delegate each sub-task to the best fitting agent/tool (e.g. always delegate any kind of coding or GitHub related task to ‘coder_agent’).
-3. Coordinate tools/agents efficiently, ensuring minimal redundancy.  
-4. Validate results before presenting them.  
-5. Respond in Telegram Markdown.  
-    - Use `*bold*`, `_italic_`, `[text](http://example.com)`, `` `inline code` ``, and ``` fenced code blocks ```.  
-    - Never leave an opening `*`, `_`, `` ` ``, `[`, or ``` without its closing pair.  
-    - Do not nest Markdown entities.  
-    - Escape reserved characters (`* _ [ ] ( ) \ ``) if they are not part of a valid entity.  
-    - Escaping inside entities is not allowed, so entity must be closed first and reopened again: use _snake_\__case_ for italic snake_case and *2*\**2=4* for bold 2*2=4.
-
-Your top priority: satisfy the user’s query as effectively and efficiently as possible using the resources at your disposal, while formatting the final output in valid Markdown.
-"""
-
-def coder_system_message(state: dict):
-    return f"""Your job is to create a coding project based on the user query.
-    
-    1. Create a new GitHub {"private" if state['private'] else "public"} repository, named '{state['repo_name']}' for this task using 'create_repo' tool.\n
-    2. Use the 'web_search' tool to research your task.\n
-    3. Commit files with the code to your repository.\n
-    4. Critically review your code for weak points using 'list_files' and 'read_file' tools.\n
-    5.  Use 'web_search' for the latest docs and code examples.\n
-    6. Adjust the code after the carefully reviewing your code.\n
-    """
-def deep_research_system_message(state: dict):
-    return f"""Your job is to conduct deep research based on the user query.\n
-    
-    1. Iterate over the following search queries: {state['search_queries']}\n
-    2. Write an extensive report based on web search results.\n
-    3. Critically review your report for weak points.\n
-    4. Conduct additional research using available tools.\t
-    5. Write an extensive report based on web search results.\n
-    6. Critically review your report and cicle over these steps until it is factually correct and very detailed.\n
-    """
-
-def smart_home_system_message(device_list: list):
-    return f"""Your job is to control the smart home.\nYou have access to the following devices:\n
-    {"\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(device_list)
-    )}.\nUse the given tools ('get_device_list', 'change_device_states') to complete the user's tasks.
-    """

agent/smart_home_agent.py

@@ -1,30 +0,0 @@
-from agent.prompts import smart_home_instructions, smart_home_system_message
-from langchain_core.messages import HumanMessage, SystemMessage
-from agent.models import llm_agents, llm_peripheral
-from langgraph.prebuilt import create_react_agent
-from langgraph.constants import START, END
-from langgraph.graph import StateGraph
-from agent.tools import smart_home_tools, get_device_list
-
-agent = create_react_agent(
-    llm_agents,
-    tools=smart_home_tools,
-    prompt=smart_home_instructions
-)
-
-def home_control_agent(state: dict):
-    system_message = SystemMessage(smart_home_system_message(get_device_list.invoke({})))
-    state.update(agent.invoke({
-        'messages': [
-            system_message,
-            HumanMessage(state['messages'][-1].content),
-        ]
-    }))
-    return state
-
-graph = StateGraph(dict)
-graph.add_node("home_control_agent", home_control_agent)
-graph.add_edge(START, "home_control_agent")
-graph.add_edge("home_control_agent", END)
-
-smart_home_agent = graph.compile(name="smart_home_agent")

agent/states.py

@@ -1,13 +0,0 @@
-from typing import TypedDict, List
-from pydantic import Field
-
-class SearchQuery(TypedDict):
-    query: str = Field(description="A single plain text search query string.")
-
-class PlanResearch(TypedDict):
-    search_queries: List[SearchQuery] = Field(description="A list of search queries, to find all the info user asked for. Break user query down into smaller search queries.")
-
-class PlanCodingTask(TypedDict):
-    repo_name: str = Field(description="The name of the GitHub repository for the project.")
-    private: bool = Field(description="Whether or not the repository is private.", default=False)
-    task_description: str = Field(description="A detailed description of the project for the coder to create.")

agents/coder/agent.py

@@ -0,0 +1,34 @@
+from langchain_core.messages import HumanMessage, SystemMessage
+from agents.coder.prompts import prompt, system_message
+from langgraph.prebuilt import create_react_agent
+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 dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+agent = create_react_agent(
+    llm_sub_agents,
+    tools=tools,
+    prompt=prompt,
+    state_schema=CoderState,
+)
+
+@tool
+def coder_agent(project_name: str, task_description: str, private: bool=False):
+    """
+    Coder Agent, used as a tool for any coding tasks, it creates project, 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.
+        private (bool, optional): Whether the coder should be private or public. Defaults to False.
+    """
+    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)
+    return messages['messages'][-1].content

agents/coder/prompts.py

@@ -0,0 +1,21 @@
+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
+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
+
+    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
+    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
+    7. Tell the user you're done.
+    """

agents/coder/states.py

@@ -0,0 +1,5 @@
+from langgraph.prebuilt.chat_agent_executor import AgentState
+
+class CoderState(AgentState):
+    project_name: str
+    private: bool = False

agents/coder/tools.py

@@ -0,0 +1,239 @@
+from langchain_tavily import TavilySearch
+from langchain_core.tools import tool
+from dotenv import load_dotenv
+from github import Github
+import subprocess
+import os
+
+load_dotenv()
+
+tavily_search = TavilySearch(
+    max_results=5,
+    topic="general",
+    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."""
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repo = user.get_repo(project_name)
+        branches = repo.get_branches()
+        branch_names = [branch.name for branch in branches]
+        return f"Branches in '{project_name}': {', '.join(branch_names)}"
+    except Exception as e:
+        return f"Error listing branches: {str(e)}"
+
+@tool
+def create_repo_branch(project_name: str, branch_name: str):
+    """Creates a new branch in a GitHub repository."""
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repo = user.get_repo(project_name)
+        main_branch = repo.get_branch("main")
+        repo.create_git_ref(ref=f"refs/heads/{branch_name}", sha=main_branch.commit.sha)
+        return f"Branch '{branch_name}' created successfully in '{project_name}'."
+    except Exception as e:
+        return f"Error creating branch: {str(e)}"
+
+@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:
+            f.write(file_contents)
+    except Exception as e:
+        return f"Error with file operation: {str(e)}"
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repo = user.get_repo(project_name)
+
+        try:
+            # Check if file exists
+            file = repo.get_contents(file_path)
+            sha = file.sha
+            repo.update_file(file_path, "Updating file", file_contents, sha)
+            return f"File '{file_path}' updated successfully in '{project_name}'."
+        except:
+            repo.create_file(file_path, "Adding new file", file_contents)
+            return f"File '{file_path}' created successfully in '{project_name}'."
+    except Exception as e:
+        return f"Error with file operation: {str(e)}"
+
+@tool
+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)
+    except Exception as e:
+        return f"Error deleting file: {str(e)}"
+    try:
+        github = Github(os.getenv('GITHUB_TOKEN'))
+        user = github.get_user()
+        repo = user.get_repo(project_name)
+        file = repo.get_contents(file_path)
+        sha = file.sha
+        repo.delete_file(file_path, "Deleting file", sha)
+        return f"File '{file_path}' deleted successfully from '{project_name}'."
+    except Exception as 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}"
+    try:
+        with open(full_path, "r", encoding="utf-8") as f:
+            return f.read()
+    except Exception as 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.
+    :return: List of files in the directory inside the project.
+    """
+    base_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.isdir(base_path):
+            return f"Path '{base_path}' is not a directory."
+
+        files = os.listdir(base_path)
+        return files
+
+    except Exception as e:
+        return f"Error listing files: {str(e)}"
+
+@tool
+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)
+
+    if not os.path.isdir(project_dir):
+        return -1, "", f"Project directory not found: {project_dir}"
+
+    try:
+        res = subprocess.run(
+            cmd,
+            shell=True,
+            cwd=project_dir,
+            capture_output=True,
+            text=True,
+            timeout=timeout
+        )
+        return res.returncode, res.stdout, res.stderr
+    except subprocess.TimeoutExpired as e:
+        return -1, e.stdout or "", f"Command timed out after {timeout}s"
+    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]

{agent → agents}/models.py

@@ -1,30 +1,19 @@
 from langchain.chat_models import init_chat_model
 from dotenv import load_dotenv
-from openai import AsyncOpenAI
-import os
 
 load_dotenv()
-groq_client = AsyncOpenAI(
-    base_url="https://api.groq.com/openai/v1",
-    api_key=os.getenv("GROQ_API_KEY"),
-)
 
 llm_supervisor = init_chat_model(
-    model="groq:openai/gpt-oss-120b",
+    "groq:openai/gpt-oss-20b",
     max_tokens=1000
 )
 
 llm_peripheral = init_chat_model(
-    model="groq:gemma2-9b-it",
+    "groq:gemma2-9b-it",
     max_tokens=4000
 )
 
-llm_agents = init_chat_model(
-    model="groq:qwen/qwen3-32b",
-    max_tokens=4000
-)
-
-llm_image = init_chat_model(
-    model="groq:meta-llama/llama-4-scout-17b-16e-instruct",
+llm_sub_agents = init_chat_model(
+    "groq:qwen/qwen3-32b",
     max_tokens=3000
-)
+)

{agent → agents}/multi_agent.py

@@ -1,35 +1,14 @@
 from database_interaction.config import create_or_update_config, load_config_to_env, init_config_db
 from database_interaction.user import get_user_by_id, create_or_update_user, init_user_db
 from langchain_community.chat_message_histories import SQLChatMessageHistory
-from langchain_core.messages import SystemMessage, AnyMessage, AIMessage
 from langchain_core.messages.utils import count_tokens_approximately
+from langchain_core.messages import SystemMessage, AIMessage
 from sqlalchemy.ext.asyncio import create_async_engine
-from langgraph_supervisor import create_supervisor
+from langgraph.prebuilt import create_react_agent
 from langmem.short_term import SummarizationNode
-from typing_extensions import TypedDict
+from agents.states import State
 import os
 
-class State(TypedDict):
-    message: AnyMessage
-    user_id: str
-    first_name: str
-    last_name: str
-    assistant_name: str
-    latitude: str
-    longitude: str
-    location: str
-    openweathermap_api_key: str
-    github_token: str
-    tavily_api_key: str
-    groq_api_key: str
-    tuya_access_id: str
-    tuya_access_key: str
-    tuya_username: str
-    tuya_password: str
-    tuya_country: str
-    clear_history: bool
-    messages: list
-
 class Assistant:
     def __init__(self, state: State):
         self.state = state
@@ -77,12 +56,9 @@ class Assistant:
     def compile_multi_agent_system(self):
         """Create and return the multi-agent system"""
         try:
-            from agent.smart_home_agent import smart_home_agent
-            from agent.deep_research_agent import deep_research_agent
-            from agent.coder_agent import coder_agent
-            from agent.prompts import supervisor_instructions
-            from agent.models import llm_supervisor, llm_peripheral
-            from agent.tools import supervisor_tools
+            from agents.models import llm_supervisor, llm_peripheral
+            from agents.prompts import prompt
+            from agents.tools import tools
 
             summarization_node = SummarizationNode(
                 token_counter=count_tokens_approximately,
@@ -92,21 +68,15 @@ class Assistant:
                 output_messages_key="llm_input_messages",
             )
 
-            agents = [coder_agent, deep_research_agent, smart_home_agent]
-
-            supervisor = create_supervisor(
+            agent = create_react_agent(
                 model=llm_supervisor,
-                tools=supervisor_tools,
-                agents=agents,
-                prompt=supervisor_instructions(supervisor_tools, agents),
-                parallel_tool_calls=True,
-                add_handoff_back_messages=False,
-                add_handoff_messages=False,
-                output_mode="full_history",
-                pre_model_hook=summarization_node
+                tools=tools,
+                prompt=prompt(tools),
+                state_schema=State,
+                version='v1',
+                pre_model_hook=summarization_node,
             )
-
-            return supervisor.compile()
+            return agent
 
         except Exception as e:
             print(f"Error creating multi-agent system: {e}")
@@ -128,6 +98,7 @@ class Assistant:
 
     async def run(self):
         """Process messages through the multi-agent system"""
+        from agents.prompts import system_message
         try:
             user_info = await get_user_by_id(user_id=self.state['user_id'])
             if user_info.get('location'):
@@ -137,30 +108,17 @@ class Assistant:
             if user_info.get('longitude'):
                 os.environ['LONGITUDE'] = str(user_info['longitude'])
 
-            system_msg = SystemMessage(
-                content=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.
-
-                            The user you are assisting is:
-                            - **Name**: {user_info.get('first_name', 'Unknown') or 'Unknown'} {user_info.get('last_name', '') or ''}
-                            - **User ID**: {self.state['user_id']}
-                            - **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'})
-
-                            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.
-
-                            Stay helpful, respectful, and relevant to the user's query.
-                            """.strip()
-            )
 
             await self.message_history.aadd_message(self.state['message'])
             messages = await self.message_history.aget_messages()
 
-            self.state['messages'] = messages[-8:-1] + [system_msg, messages[-1]]
+            self.state['messages'] = messages[-8:-1] + [SystemMessage(system_message(user_info)), messages[-1]]
             multi_agent_system = self.compile_multi_agent_system()
 
-            result = await multi_agent_system.ainvoke({"messages": self.state["messages"]},
-                                                      generation_config=dict(response_modalities=["TEXT"]))
+            result = await multi_agent_system.ainvoke(
+                {"messages": self.state["messages"]},
+                generation_config=dict(response_modalities=["TEXT"])
+            )
             await self.message_history.aadd_message(result['messages'][-1])
             return {"messages": result.get("messages", [])}
 

agents/prompts.py

@@ -0,0 +1,47 @@
+from agents.utils.smart_home import get_device_list
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+def prompt(tools: list):
+    tool_list = ", ".join([tool.name for tool in tools])
+    return f"""You are the helpful Agent. Your job is to interpret the user’s request and delegate tasks to the available tools and agents to complete the task.
+
+You have access to the following agents and tools:
+- Tools: {tool_list}
+
+1. Understand and break down user’s request into smaller sub-tasks.
+2. Always delegate each sub-task to the best fitting agent/tool (e.g. always delegate any kind of coding or GitHub related task to ‘coder_agent’).
+3. Coordinate tools/agents efficiently, ensuring minimal redundancy.  
+4. Validate results before presenting them.  
+5. Respond in Telegram Markdown.  
+    - Use `*bold*`, `_italic_`, `[text](http://example.com)`, `` `inline code` ``, and ``` fenced code blocks ```.  
+    - Never leave an opening `*`, `_`, `` ` ``, `[`, or ``` without its closing pair.  
+    - Do not nest Markdown entities.  
+    - Escape reserved characters (`* _ [ ] ( ) \ ``) if they are not part of a valid entity.  
+    - Escaping inside entities is not allowed, so entity must be closed first and reopened again: use _snake_\__case_ for italic snake_case and *2*\**2=4* for bold 2*2=4.
+
+Your top priority: satisfy the user’s query as effectively and efficiently as possible using the resources at your disposal, while formatting the final output in valid Markdown.
+"""
+
+
+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.
+
+    The user you are assisting is:
+    - **Name**: {user_info.get('first_name', 'Unknown') or 'Unknown'} {user_info.get('last_name', '') or ''}
+    - **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"}
+
+    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.
+    Stay helpful, respectful, and relevant to the user's query.
+    """

agents/states.py

@@ -0,0 +1,22 @@
+from langgraph.prebuilt.chat_agent_executor import AgentState
+from langchain_core.messages import AnyMessage
+
+class State(AgentState):
+    message: AnyMessage
+    user_id: str
+    first_name: str
+    last_name: str
+    assistant_name: str
+    latitude: str
+    longitude: str
+    location: str
+    openweathermap_api_key: str
+    github_token: str
+    tavily_api_key: str
+    groq_api_key: str
+    tuya_access_id: str
+    tuya_access_key: str
+    tuya_username: str
+    tuya_password: str
+    tuya_country: str
+    clear_history: bool

agents/tools.py

@@ -0,0 +1,74 @@
+from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
+from langchain_community.utilities import OpenWeatherMapAPIWrapper
+from langchain_community.utilities import WikipediaAPIWrapper
+from langchain_community.tools import WikipediaQueryRun
+from langchain_tavily.tavily_search import TavilySearch
+from agents.coder.agent import coder_agent
+from agents.utils.smart_home import change_device_states
+from timezonefinder import TimezoneFinder
+from langchain_core.tools import tool
+from dotenv import load_dotenv
+from geopy import Nominatim
+import datetime
+import pytz
+import os
+
+load_dotenv()
+
+
+tf = TimezoneFinder()
+web_search = TavilySearch(
+    max_results=5,
+    topic="general",
+    include_answer=True
+)
+yahoo = YahooFinanceNewsTool()
+geolocator = Nominatim(user_agent="my_geocoder")
+wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
+weekday_mapping = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
+
+@tool
+def current_time(location: str = None):
+    """Get the current time for a location or current position."""
+    try:
+        if location:
+            location_data = geolocator.geocode(location)
+            if location_data:
+                timezone = pytz.timezone(tf.timezone_at(lat=location_data.latitude, lng=location_data.longitude))
+                location_name = location.capitalize()
+            else:
+                return f"Could not find location: {location}"
+        else:
+            # Use environment location if available
+            lat = os.getenv('LATITUDE')
+            lon = os.getenv('LONGITUDE')
+            if lat and lon:
+                timezone = pytz.timezone(tf.timezone_at(lat=float(lat), lng=float(lon)))
+                location_name = os.getenv('LOCATION', 'Current Location')
+            else:
+                timezone = pytz.UTC
+                location_name = 'UTC'
+
+        current_dt = datetime.datetime.now(timezone)
+        weekday = weekday_mapping[current_dt.weekday()]
+        return f"Location: {location_name}; Current Date and Time: {current_dt.strftime('%Y-%m-%d %H:%M')}, {weekday}."
+    except Exception as e:
+        return f"Error getting current time: {str(e)}"
+
+@tool
+def weather(location: str = None):
+    """Get the current weather for a location or current position."""
+    try:
+        weather_wrapper = OpenWeatherMapAPIWrapper(
+            openweathermap_api_key=os.getenv('OPENWEATHERMAP_API_KEY')
+        )
+        if not location:
+            location = os.getenv('LOCATION', 'Unknown')
+        return weather_wrapper.run(location=location)
+    except Exception as e:
+        return f"Error getting weather: {str(e)}"
+
+
+tools = [yahoo, web_search, current_time, weather, change_device_states, coder_agent]
+
+

{agent → agents/utils}/file_preprocessing.py

@@ -13,7 +13,7 @@ async def preprocess_file(file_name: str):
     if "image" in mime_type:
         return await preprocess_image(file_name)
     elif "video" in mime_type:
-        prompt = "Give a detailed description of the video."
+        pass
     elif "audio" in mime_type:
         return await preprocess_audio(file_name)
     else:

agent/tools.py → agents/utils/smart_home.py

@@ -1,33 +1,9 @@
-from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
-from langchain_community.utilities import OpenWeatherMapAPIWrapper
-from langchain_community.utilities import WikipediaAPIWrapper
-from langchain_community.tools import WikipediaQueryRun
-from langchain_tavily.tavily_search import TavilySearch
-from timezonefinder import TimezoneFinder
-from langchain_core.tools import tool
+from langchain.tools import tool
 from tuya_iot import TuyaOpenAPI
-from dotenv import load_dotenv
-from geopy import Nominatim
-from github import Github
 import phonenumbers
 import pycountry
-import datetime
-import pytz
 import os
 
-load_dotenv()
-
-
-tf = TimezoneFinder()
-web_search = TavilySearch(
-    max_results=5,
-    topic="general",
-    include_answer=True
-)
-yahoo = YahooFinanceNewsTool()
-geolocator = Nominatim(user_agent="my_geocoder")
-wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
-weekday_mapping = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
 endpoint_map = {
     "https://openapi.tuyacn.com": [86],
     "https://openapi.tuyain.com": [91],
@@ -60,156 +36,7 @@ openapi = TuyaOpenAPI(os.getenv('TUYA_ENDPOINT'), os.getenv('TUYA_ACCESS_ID'), o
 connection = openapi.connect(os.getenv('TUYA_USERNAME'), os.getenv('TUYA_PASSWORD'), os.getenv('TUYA_COUNTRY_CODE'), 'TuyaSmart')
 if not connection['success']:
     connection = openapi.connect(os.getenv('TUYA_USERNAME'), os.getenv('TUYA_PASSWORD'), os.getenv('TUYA_COUNTRY_CODE'), 'SmartLife')
-@tool
-def create_repo(repo_name: str, private: bool = False):
-    """Creates a GitHub repository with the given repo_name."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        user.create_repo(repo_name, private=private)
-        return f"Repository '{repo_name}' created successfully!"
-    except Exception as e:
-        return f"Error creating repository: {str(e)}"
 
-@tool
-def commit_file_to_repo(repo_name: str, file_path: str, file_contents: str):
-    """Adds a new file to the GitHub repository or updates the existing one."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(repo_name)
-
-        try:
-            # Check if file exists
-            file = repo.get_contents(file_path)
-            sha = file.sha
-            repo.update_file(file_path, "Updating file", file_contents, sha)
-            return f"File '{file_path}' updated successfully in '{repo_name}'."
-        except:
-            repo.create_file(file_path, "Adding new file", file_contents)
-            return f"File '{file_path}' created successfully in '{repo_name}'."
-    except Exception as e:
-        return f"Error with file operation: {str(e)}"
-
-@tool
-def read_file(repo_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(repo_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 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_files(repo_name: str):
-    """Lists all files in the GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(repo_name)
-        files = repo.get_contents("")
-        file_list = [file.name for file in files]
-        return f"Files in '{repo_name}': {', '.join(file_list)}"
-    except Exception as e:
-        return f"Error listing files: {str(e)}"
-
-@tool
-def delete_file(repo_name: str, file_path: str):
-    """Deletes a file from the GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(repo_name)
-        file = repo.get_contents(file_path)
-        sha = file.sha
-        repo.delete_file(file_path, "Deleting file", sha)
-        return f"File '{file_path}' deleted successfully from '{repo_name}'."
-    except Exception as e:
-        return f"Error deleting file: {str(e)}"
-
-@tool
-def list_branches(repo_name: str):
-    """Lists all branches in a GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(repo_name)
-        branches = repo.get_branches()
-        branch_names = [branch.name for branch in branches]
-        return f"Branches in '{repo_name}': {', '.join(branch_names)}"
-    except Exception as e:
-        return f"Error listing branches: {str(e)}"
-
-@tool
-def create_branch(repo_name: str, branch_name: str):
-    """Creates a new branch in a GitHub repository."""
-    try:
-        github = Github(os.getenv('GITHUB_TOKEN'))
-        user = github.get_user()
-        repo = user.get_repo(repo_name)
-        main_branch = repo.get_branch("main")
-        repo.create_git_ref(ref=f"refs/heads/{branch_name}", sha=main_branch.commit.sha)
-        return f"Branch '{branch_name}' created successfully in '{repo_name}'."
-    except Exception as e:
-        return f"Error creating branch: {str(e)}"
-
-@tool
-def current_time(location: str = None):
-    """Get the current time for a location or current position."""
-    try:
-        if location:
-            location_data = geolocator.geocode(location)
-            if location_data:
-                timezone = pytz.timezone(tf.timezone_at(lat=location_data.latitude, lng=location_data.longitude))
-                location_name = location.capitalize()
-            else:
-                return f"Could not find location: {location}"
-        else:
-            # Use environment location if available
-            lat = os.getenv('LATITUDE')
-            lon = os.getenv('LONGITUDE')
-            if lat and lon:
-                timezone = pytz.timezone(tf.timezone_at(lat=float(lat), lng=float(lon)))
-                location_name = os.getenv('LOCATION', 'Current Location')
-            else:
-                timezone = pytz.UTC
-                location_name = 'UTC'
-
-        current_dt = datetime.datetime.now(timezone)
-        weekday = weekday_mapping[current_dt.weekday()]
-        return f"Location: {location_name}; Current Date and Time: {current_dt.strftime('%Y-%m-%d %H:%M')}, {weekday}."
-    except Exception as e:
-        return f"Error getting current time: {str(e)}"
-
-@tool
-def weather(location: str = None):
-    """Get the current weather for a location or current position."""
-    try:
-        weather_wrapper = OpenWeatherMapAPIWrapper(
-            openweathermap_api_key=os.getenv('OPENWEATHERMAP_API_KEY')
-        )
-        if not location:
-            location = os.getenv('LOCATION', 'Unknown')
-        return weather_wrapper.run(location=location)
-    except Exception as e:
-        return f"Error getting weather: {str(e)}"
-
-@tool
 def get_device_list():
     """
     Get the list of devices bound to the smart home.
@@ -232,9 +59,4 @@ def change_device_states(id:str, commands:list[dict]):
     State format: {'code': ..., 'value': ...}
     """
     response = openapi.post('/v1.0/devices/{}/commands'.format(id), {"commands": commands})
-    return "success" if response["success"] else "failure"
-
-coder_tools = [web_search, create_repo, create_branch, commit_file_to_repo, read_file, list_files, list_repos, list_branches, delete_file]
-supervisor_tools = [yahoo, web_search, current_time, weather]
-deep_research_tools = [web_search, yahoo, wikipedia]
-smart_home_tools = [get_device_list, change_device_states]
+    return "success" if response["success"] else "failure"

app.py

@@ -1,7 +1,7 @@
-from agent.file_preprocessing import preprocess_file, preprocess_audio
+from agents.utils.file_preprocessing import preprocess_file, preprocess_audio
 from fastapi import FastAPI, UploadFile, File, HTTPException
 from langchain_core.messages import HumanMessage
-from agent.multi_agent import Assistant
+from agents.multi_agent import Assistant
 from typing import Any, Dict
 from fastapi import Form
 from pathlib import Path