import concurrent import csv from swarms import Agent from swarm_models import OpenAIChat from swarms_memory import ChromaDB from dotenv import load_dotenv from swarms.utils.parse_code import extract_code_from_markdown from swarms.utils.file_processing import create_file from swarms.utils.loguru_logger import logger # Load ENV load_dotenv() # Gemini gemini = OpenAIChat() # memory memory = ChromaDB(output_dir="swarm_hackathon") def execute_concurrently(callable_functions: callable, max_workers=5): """ Executes callable functions concurrently using multithreading. Parameters: - callable_functions: A list of tuples, each containing the callable function and its arguments. For example: [(function1, (arg1, arg2), {'kwarg1': val1}), (function2, (), {})] - max_workers: The maximum number of threads to use. Returns: - results: A list of results returned by the callable functions. If an error occurs in any function, the exception object will be placed at the corresponding index in the list. """ results = [None] * len(callable_functions) def worker(fn, args, kwargs, index): try: result = fn(*args, **kwargs) results[index] = result except Exception as e: results[index] = e with concurrent.futures.ThreadPoolExecutor( max_workers=max_workers ) as executor: futures = [] for i, (fn, args, kwargs) in enumerate(callable_functions): futures.append( executor.submit(worker, fn, args, kwargs, i) ) # Wait for all threads to complete concurrent.futures.wait(futures) return results # Adjusting the function to extract specific column values def extract_and_create_agents( csv_file_path: str, target_columns: list ): """ Reads a CSV file, extracts "Project Name" and "Lightning Proposal" for each row, creates an Agent for each, and adds it to the swarm network. Parameters: - csv_file_path: The path to the CSV file. - target_columns: A list of column names to extract values from. """ try: agents = [] with open(csv_file_path, mode="r", encoding="utf-8") as file: reader = csv.DictReader(file) for row in reader: project_name = row[target_columns[0]] lightning_proposal = row[target_columns[1]] # Example of creating and adding an agent based on the project name and lightning proposal agent_name = f"{project_name} agent" print(agent_name) # For demonstration # Create the agent logger.info("Creating agent...") # Design agent logger.info("Creating design agent...") design_agent = Agent( llm=gemini, agent_name="Design Agent", max_loops=1, stopping_token="", sop=None, system_prompt=( "Transform an app idea into step by step very" " simple algorithmic psuedocode so it can be" " implemented simply." ), long_term_memory=memory, ) # Log the agent logger.info( f"Code Agent created: {agent_name} with long term" " memory" ) agent = Agent( llm=gemini, agent_name=agent_name, max_loops=1, code_interpreter=True, stopping_token="", sop=None, system_prompt=( "Transform an app idea into a very simple" " python app in markdown. Return all the" " python code in a single markdown file." " Return only code and nothing else." ), long_term_memory=memory, ) # Testing agent logger.info(f"Testing_agent agent: {agent_name}") agent = Agent( llm=gemini, agent_name=agent_name + " testing", max_loops=1, stopping_token="", sop=None, system_prompt=( "Create unit tests using pytest based on the" " code you see, only return unit test code in" " python using markdown, only return the code" " and nothing else." ), long_term_memory=memory, ) # Log the agent logger.info( f"Agent created: {agent_name} with long term" " memory" ) agents.append(agent) # Design agent design_agent_output = design_agent.run( ( "Create the algorithmic psuedocode for the" f" {lightning_proposal} in markdown and" " return it" ), None, ) logger.info( "Algorithmic psuedocode created:" f" {design_agent_output}" ) # Create the code for each project output = agent.run( ( "Create the code for the" f" {lightning_proposal} in python using the" " algorithmic psuedocode" f" {design_agent_output} and wrap it in" " markdown and return it" ), None, ) print(output) # Parse the output output = extract_code_from_markdown(output) # Create the file output = create_file(output, f"{project_name}.py") # Testing agent testing_agent_output = agent.run( ( "Create the unit tests for the" f" {lightning_proposal} in python using the" f" code {output} and wrap it in markdown and" " return it" ), None, ) print(testing_agent_output) # Parse the output testing_agent_output = extract_code_from_markdown( testing_agent_output ) # Create the file testing_agent_output = create_file( testing_agent_output, f"test_{project_name}.py" ) # Log the project created logger.info( f"Project {project_name} created: {output} at" f" file path {project_name}.py" ) print(output) # Log the unit tests created logger.info( f"Unit tests for {project_name} created:" f" {testing_agent_output} at file path" f" test_{project_name}.py" ) print( f"Agent {agent_name} created and added to the" " swarm network" ) return agents except Exception as e: logger.error( "An error occurred while extracting and creating" f" agents: {e}" ) return None # CSV csv_file = "presentation.csv" # Specific columns to extract target_columns = ["Project Name", "Project Description"] # Use the adjusted function specific_column_values = extract_and_create_agents( csv_file, target_columns ) # Display the extracted column values print(specific_column_values) # Concurrently execute the function logger.info( "Concurrently executing the swarm for each hackathon project..." ) output = execute_concurrently( [ (extract_and_create_agents, (csv_file, target_columns), {}), ], max_workers=5, ) print(output)