feat: slackbot code expert

python/examples/advanced_agents/slackbot_code_expert/.env.example

@@ -0,0 +1,3 @@
+OPENAI_API_KEY=
+COMPOSIO_API_KEY=
+CHAT_ID=

python/examples/advanced_agents/slackbot_code_expert/.gitignore

@@ -0,0 +1,4 @@
+.env
+__pycache__/
+*.py[cod]
+*$py.class

python/examples/advanced_agents/slackbot_code_expert/README.md

@@ -0,0 +1,5 @@
+# Slackbot: Code Expert In Your Slack Channel
+
+## Overview
+
+Slack Expert serves as a code-savvy assistant, capable of answering questions related to a codebase. When a user asks a question, Slack Expert initially tries to answer using OpenAI chat completions. If additional code-specific context is required, it queries the codebase using Composio tools, refines the response, and sends it back to the user.

python/examples/advanced_agents/slackbot_code_expert/chat/codebase_agent.py

@@ -0,0 +1,98 @@
+import os
+from dotenv import load_dotenv
+from composio_crewai import Action, ComposioToolSet, App
+from composio.client.exceptions import NoItemsFound
+from crewai import Agent, Crew, Task, Process
+from langchain_openai import ChatOpenAI
+load_dotenv()
+
+openai_api_key = os.getenv('OPENAI_API_KEY')
+composio_api_key = os.getenv('COMPOSIO_API_KEY')
+
+llm = ChatOpenAI(model="gpt-4o")
+
+def index_code(dir_path: str, embedding_type: str, force_index: bool) -> str:
+    composio_toolset = ComposioToolSet(
+        api_key=composio_api_key,
+        metadata={
+        App.CODE_ANALYSIS_TOOL: {
+            "dir_to_index_path" : dir_path,
+        },  
+    })
+    tools = composio_toolset.get_tools(actions=[Action.CODE_ANALYSIS_TOOL_CREATE_CODE_MAP])
+    indexing_agent = Agent(
+        role="Indexing Agent",
+        goal="Index the code in the given directory",
+        backstory="You're an AI assistant that indexes code for easy retrieval and search.",
+        verbose=True,
+        llm=llm,
+        tools=tools,
+        allow_delegation=False,
+    )
+
+    task_description = f"""
+    1. Index the code in the following directory:
+       "{dir_path}"
+    2. Use the embedding type: "{embedding_type}"
+    3. Force index: {"Yes" if force_index else "No"}
+    4. Return the index ID of the indexed code.
+    """
+
+    process_indexing_request = Task(
+        description=task_description,
+        agent=indexing_agent,
+        expected_output="The index ID of the indexed code.",
+    )
+
+    indexing_processing_crew = Crew(
+        agents=[indexing_agent],
+        tasks=[process_indexing_request],
+        verbose=1,
+        process=Process.sequential,
+    )
+
+    result = indexing_processing_crew.kickoff()
+    return result
+
+def find_code_snippet(dir_path: str, query: str) -> str:
+    composio_toolset = ComposioToolSet(
+    api_key=composio_api_key,
+    metadata={
+    App.CODE_ANALYSIS_TOOL:{
+            "dir_to_index_path" : dir_path,
+        }
+    })
+    tools = composio_toolset.get_tools(actions=[Action.CODE_ANALYSIS_TOOL_GET_RELEVANT_CODE])
+    search_agent = Agent(
+        role="Search Agent",
+        goal="Search the code in the given directory for the provided query",
+        backstory="You're an AI assistant that searches code for relevant snippets based on a query.",
+        verbose=False,
+        llm=llm,
+        tools=tools,
+        allow_delegation=False,
+    )
+
+    task_description = f"""
+    1. Search the code in the following directory:
+       "{dir_path}"
+    2. Search for the following query: "{query}"
+    3. Return the relevant code snippet.
+    """
+
+    process_search_request = Task(
+        description=task_description,
+        agent=search_agent,
+        expected_output="The relevant code snippet.",
+    )
+
+    search_processing_crew = Crew(
+        agents=[search_agent],
+        tasks=[process_search_request],
+        verbose=0,
+        process=Process.sequential,
+    )
+
+    result = search_processing_crew.kickoff()
+    return result
+

python/examples/advanced_agents/slackbot_code_expert/chat/main.py

@@ -0,0 +1,52 @@
+from dotenv import load_dotenv
+load_dotenv()
+import os
+import sys
+from openai import OpenAI
+from db import ChatDB
+
+from .codebase_agent import find_code_snippet
+from .tools import code_search_tool
+
+# Add the parent directory to sys.path
+current_dir = os.path.dirname(os.path.abspath(__file__))
+parent_dir = os.path.dirname(current_dir)
+sys.path.append(parent_dir)
+
+client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+chat_db = ChatDB(db_path=os.path.join(parent_dir, 'db', 'db.json'))
+
+def chatbot(messages: list):
+    # Users query is passed in the last message, if it requires more context to answer the question it uses code_search_tool to find the relevant code snippet
+    completion = client.chat.completions.create(
+        model="gpt-4o-mini",
+        messages=messages,
+        tools=code_search_tool,
+    )
+    # Checking if response from above completion contains tool calls, if so it uses the tool to find the code snippet
+    if completion.choices[0].message.tool_calls:
+        tool_call = completion.choices[0].message.tool_calls[0]
+        function_name = tool_call.function.name
+        arguments = eval(tool_call.function.arguments)
+        if function_name == "find_code_snippet":
+            # Using the tool to find the code snippet
+            result = find_code_snippet(**arguments)
+            # print(f"Code snippet result: {result}")
+            question = messages[-1]['content']
+            prompt = f"Query: {question} \nCode snippet for context: {result}" 
+            completion = client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=messages+[{"role": "user", "content": prompt}],
+            )
+
+            response = completion.choices[0].message.content
+            print("If(Bot): ", response)
+            messages.append({"role": "assistant", "content": response})
+            return response
+    else:
+        # If no tool calls (ie if agent already has context to answer the question), just return the response
+        response = completion.choices[0].message.content
+        print("Else(Bot): ", response)
+        messages.append({"role": "assistant", "content": response})
+        return response

python/examples/advanced_agents/slackbot_code_expert/chat/tools.py

@@ -0,0 +1,23 @@
+code_search_tool = [
+    {
+        "type": "function",
+        "function": {
+            "name": "find_code_snippet",
+            "description": "Search the code in a given directory for a provided query and return the relevant code snippet",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "dir_path": {
+                        "type": "string",
+                        "description": "The directory path where the code is located",
+                    },
+                    "query": {
+                        "type": "string",
+                        "description": "The query to search for in the code",
+                    },
+                },
+                "required": ["dir_path", "query"],
+            },
+        }
+    }
+]

python/examples/advanced_agents/slackbot_code_expert/db/__init__.py

@@ -0,0 +1 @@
+from .main import ChatDB

python/examples/advanced_agents/slackbot_code_expert/db/main.py

@@ -0,0 +1,36 @@
+from tinydb import TinyDB, Query
+import uuid
+
+class ChatDB:
+    def __init__(self, db_path='db.json'):
+        self.db = TinyDB(db_path)
+
+    def create_chat(self, chat_name):
+        chat_id = str(uuid.uuid4())
+        messages = [{
+            "role": "system",
+            "content": f"You are an assistant that answers questions based on the codebase located at {chat_name}. Respond to users' messages in a simple & concise manner. Don't use filler words. If you don't find enough context to respond to the user find/query code using find_code_snippet tool."
+        }]
+        self.db.insert({'chat_id': chat_id, 'chat_name': chat_name, 'messages': messages})
+        return chat_id
+
+
+    def add_message(self, chat_id, content, role):
+        Chat = Query()
+        chat = self.db.get(Chat.chat_id == chat_id)
+        if chat:
+            chat['messages'].append({'role': role, 'content': content})
+            self.db.update({'messages': chat['messages']}, Chat.chat_id == chat_id)
+        else:
+            raise ValueError(f"Chat ID not found: {chat_id}")
+
+
+    def get_chat(self, chat_id):
+        Chat = Query()
+        return self.db.get(Chat.chat_id == chat_id)
+
+    def list_all_chats(self):
+        return self.db.all()
+
+
+

python/examples/advanced_agents/slackbot_code_expert/main.py

@@ -0,0 +1,58 @@
+import os
+from composio.client.collections import TriggerEventData
+from composio_crewai import ComposioToolSet, Action
+from langchain_openai import ChatOpenAI
+from dotenv import load_dotenv
+from typing import Any, Dict
+from db.main import ChatDB
+from chat.main import chatbot
+import os
+
+load_dotenv()
+chat_id = os.getenv('CHAT_ID')
+
+
+chat_db = ChatDB(db_path='./db/db.json')
+
+load_dotenv()
+
+llm = ChatOpenAI(model="gpt-4o")
+
+
+# Trigger instance
+composio_toolset = ComposioToolSet(api_key=os.environ.get("COMPOSIO_API_KEY"))
+listener = composio_toolset.create_trigger_listener()
+
+@listener.callback(filters={"trigger_name": "SLACKBOT_RECEIVE_MESSAGE"})
+def callback_new_message(event: TriggerEventData) -> None:
+    print("Received new message")
+    payload = event.payload
+    print(f"\n\npayload :: {payload}")
+    try:
+        # if its bot message, ignore
+        bot_id = payload['bot_id']
+        if(bot_id):
+            return None
+
+        # get user message & channel id
+        message_text = payload['text']
+        channel_id = payload['channel']
+
+        # add message to db
+        chat_db.add_message(chat_id, message_text, "user")
+        messages = chat_db.get_chat(chat_id)['messages']
+        response = chatbot(messages)
+
+
+        composio_toolset.execute_action(
+            action=Action.SLACKBOT_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL,
+            params={"channel": channel_id, "text": response},
+        )
+    except Exception as e:
+        print(f"Error accessing payload data: {e}")
+
+    return "result"
+
+
+print("Slack trigger listener activated!")
+listener.listen()