CrewAI

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "debug streamlit",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "./.venv/Lib/site-packages/streamlit",  // /home/xx/tmp/venv/bin/streamlit",
+            "args": [
+                "run",
+                "app.py"
+            ],
+            "justMyCode": false
+        }
+    ]
+}

app.py

@@ -0,0 +1,62 @@
+from src.agents.agents import CustomHandler
+from configfile import Config
+from src.llmconfig.groqllm import GroqLLM
+from src.streamlitUI.loadui import LoadStreamlitUI
+import os
+import streamlit as st
+from crewai import Crew, Process, Agent, Task
+from langchain_core.callbacks import BaseCallbackHandler
+from langchain_groq import ChatGroq  # Assuming this is the correct import for Groq
+from typing import Any, Dict
+from src.task.tasks import CrewAITasks
+from src.agents.agents import CrewAIAgents
+
+
+# MAIN Function START
+if __name__ == "__main__":
+    # config
+    obj_config = Config()
+    # load ui
+    ui = LoadStreamlitUI()
+    user_input = ui.load_streamlit_ui()
+    # Configure LLM
+    obj_llm_config = GroqLLM(user_controls_input=user_input)
+    llm = obj_llm_config.groq_llm_config()
+        
+        
+    # Initialize the message log in session state if not already present
+    if "messages" not in st.session_state:
+        st.session_state["messages"] = [{"role": "assistant", "content": "What code do you want us to write?"}]
+
+    # Display existing messages
+    for msg in st.session_state.messages:
+        st.chat_message(msg["role"]).write(msg["content"])
+
+    # Handle user input
+    if prompt := st.chat_input():
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        st.chat_message("user").write(prompt)
+        
+        
+        # agents 
+        obj_crewai_agents = CrewAIAgents(llm=llm)
+        lst_agents = obj_crewai_agents.crewai_agents()
+
+        # Define tasks for each agent
+        obj_crewai_tasks = CrewAITasks(llm=llm,prompt=prompt,lst_agents=lst_agents)
+        lst_tasks = obj_crewai_tasks.create_tasks()
+        
+        # Set up the crew and process tasks hierarchically
+        project_crew = Crew(
+            tasks=lst_tasks,
+            agents=lst_agents,
+            process=Process.hierarchical,
+            manager_llm=llm,
+            manager_callbacks=[CustomHandler("Crew Manager")]
+        )
+        final = project_crew.kickoff()
+
+        # Display the final result
+        result = f"## Here is the Final Result \n\n {final}"
+        st.session_state.messages.append({"role": "assistant", "content": result})
+        st.chat_message("assistant").write(result)

configfile.ini

@@ -0,0 +1,6 @@
+[DEFAULT]
+PAGE_TITLE = CrewAI Studio
+LLM_OPTIONS = Groq, Huggingface
+USECASE_OPTIONS = MultiAgent Coder
+GROQ_MODEL_OPTIONS = mixtral-8x7b-32768, llama3-8b-8192, llama3-70b-8192, gemma-7b-i
+

configfile.py

@@ -0,0 +1,20 @@
+from configparser import ConfigParser
+
+
+class Config:
+  def __init__(self, config_file="configfile.ini"):
+    self.config = ConfigParser()
+    self.config.read(config_file)
+
+  def get_llm_options(self):
+    return self.config["DEFAULT"].get("LLM_OPTIONS").split(", ")
+
+  def get_usecase_options(self):
+    return self.config["DEFAULT"].get("USECASE_OPTIONS").split(", ")
+
+  def get_groq_model_options(self):
+    return self.config["DEFAULT"].get("GROQ_MODEL_OPTIONS").split(", ")
+
+  def get_page_title(self):
+    return self.config["DEFAULT"].get("PAGE_TITLE")
+

requirements.txt

@@ -0,0 +1,4 @@
+streamlit
+langchain-core
+langchain-groq  
+crewai

src/agents/agents.py

@@ -0,0 +1,56 @@
+from langchain_core.callbacks import BaseCallbackHandler
+from crewai import Agent
+#from llm import groq
+
+class CustomHandler(BaseCallbackHandler):
+    """A custom handler for logging interactions within the process chain."""
+
+    def __init__(self, agent_name: str) -> None:
+        super().__init__()
+        self.agent_name = agent_name
+
+    def on_chain_start(self, serialized, outputs, **kwargs) -> None:
+        """Log the start of a chain with user input."""
+        from streamlit import session_state, chat_message
+        session_state.messages.append({"role": "assistant", "content": outputs['input']})
+        chat_message("assistant").write(outputs['input'])
+
+    def on_agent_action(self, serialized, inputs, **kwargs) -> None:
+        """Log the action taken by an agent during a chain run."""
+        from streamlit import session_state, chat_message
+        session_state.messages.append({"role": "assistant", "content": inputs['input']})
+        chat_message("assistant").write(inputs['input'])
+
+    def on_chain_end(self, outputs, **kwargs) -> None:
+        """Log the end of a chain with the output generated by an agent."""
+        from streamlit import session_state, chat_message
+        session_state.messages.append({"role": self.agent_name, "content": outputs['output']})
+        chat_message(self.agent_name).write(outputs['output'])
+        
+class CrewAIAgents:
+    def __init__(self,llm):
+        self.llm = llm
+        
+        
+    def crewai_agents(self):
+        project_manager = Agent(
+            role='Project Manager',
+            backstory='''You are the project manager. 
+                        You consider the task and break it down into smaller tasks to be performed by the team.
+                        You do not write code.''',
+            goal='Generate actionable steps for task completion.',
+            llm=self.llm,
+            callbacks=[CustomHandler("Project Manager")],
+        )
+        coder = Agent(
+            role='Python Coder',
+            backstory='''You are a Senior Python Developer responsible for writing clean, efficient and robust Python code
+                        that is easy to read and understand.
+                        You write code using object oriented programming principles and follow best practices.
+                        You produce functional, feature complete code.''',
+            goal='Develop high-quality, well-structured Python code.',
+            llm=self.llm,
+            callbacks=[CustomHandler("Coder")],
+        )
+        
+        return [project_manager,coder]

src/llmconfig/groqllm.py

@@ -0,0 +1,19 @@
+import os
+import streamlit as st
+# Initialize the Groq model for use with agents
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_groq import ChatGroq
+
+
+
+class GroqLLM:
+    def __init__(self, user_controls_input):
+        self.user_controls_input = user_controls_input
+
+    def groq_llm_config(self):
+        groq = ChatGroq(
+            temperature=0.5,
+            model="llama3-70b-8192",
+            # api_key="" # Optional if not set as an environment variable
+        )
+        return groq

src/streamlitUI/loadui.py

@@ -0,0 +1,35 @@
+import streamlit as st
+from crewai import Crew, Process
+import streamlit as st
+from configfile import Config  # Import the Config class
+
+
+class LoadStreamlitUI:
+    def __init__(self):
+        self.config = Config()  # Create a Config instance
+        self.user_controls = {}
+
+    def load_streamlit_ui(self):
+        st.set_page_config(page_title= "🚣🏻‍♂️ " + self.config.get_page_title(), layout="wide")
+        st.header("🚣🏻‍♂️ " + self.config.get_page_title())
+
+        with st.sidebar:
+            # Get options from config
+            llm_options = self.config.get_llm_options()
+            usecase_options = self.config.get_usecase_options()
+
+            # LLM selection
+            self.user_controls["selected_llm"] = st.selectbox("Select LLM", llm_options)
+
+            if self.user_controls["selected_llm"] == 'Groq':
+                # Model selection
+                model_options = self.config.get_groq_model_options()
+                self.user_controls["selected_groq_model"] = st.selectbox("Select Model", model_options)
+                # API key input
+                self.user_controls["GROQ_API_KEY"] = st.session_state["GROQ_API_KEY"] = st.text_input("API Key",
+                                                                                                      type="password")
+            # Use case selection
+            self.user_controls["selected_usecase"] = st.selectbox("Select Usecases", usecase_options)
+            st.session_state["chat_with_history"] = st.sidebar.toggle("Chat With History")
+
+        return self.user_controls

src/task/tasks.py

@@ -0,0 +1,25 @@
+from crewai import Task
+
+class CrewAITasks:
+    def __init__(self,llm,prompt,lst_agents) -> None:
+        self.prompt = prompt
+        self.llm = llm 
+        self.lst_agents = lst_agents
+    
+
+    def create_tasks(self):
+        task_plan = Task(
+            description=f"""Consider how you would go about the task, '{self.prompt}'.
+                        Create a plan to complete the task.
+                        The final step should always require delivering feature complete code""",
+            agent=self.lst_agents[0],
+            expected_output="A detailed plan for the team to complete the task.",
+        )
+
+        task_code = Task(
+            description="Write feature complete code that is simple, efficient and adheres to object oriented principles.",
+            agent=self.lst_agents[1],
+            expected_output="Well-written and structured code that is feature complete, simple, efficient and adheres to object oriented principles.",
+        )
+
+        return [task_plan, task_code]

src/utility/callbacks.py

@@ -0,0 +1,24 @@
+from langchain_core.callbacks import BaseCallbackHandler
+import streamlit as st
+
+class CustomHandler(BaseCallbackHandler):
+    """A custom handler for logging interactions within the process chain."""
+
+    def __init__(self, agent_name: str) -> None:
+        super().__init__()
+        self.agent_name = agent_name
+
+    def on_chain_start(self, serialized, outputs, **kwargs) -> None:
+        """Log the start of a chain with user input."""
+        st.session_state.messages.append({"role": "assistant", "content": outputs['input']})
+        st.chat_message("assistant").write(outputs['input'])
+
+    def on_agent_action(self, serialized, inputs, **kwargs) -> None:
+        """Log the action taken by an agent during a chain run."""
+        st.session_state.messages.append({"role": "assistant", "content": inputs['input']})
+        st.chat_message("assistant").write(inputs['input'])
+
+    def on_chain_end(self, outputs, **kwargs) -> None:
+        """Log the end of a chain with the output generated by an agent."""
+        st.session_state.messages.append({"role": self.agent_name, "content": outputs['output']})
+        st.chat_message(self.agent_name).write(outputs['output'])