import os import requests import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM # Load model and tokenizer tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-0.5B") model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2-0.5B") class Agent: def __init__(self, id, api_key=None): self.id = id self.task = None self.results = None self.api_key = api_key def execute_task(self): if self.task: print(f"Agent {self.id} is making an API call to '{self.task}'") headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {} try: response = requests.get(self.task, headers=headers) if response.status_code == 200: self.results = response.json() else: self.results = "Error: Unable to fetch data" print(f"Agent {self.id} received: {self.results}") except Exception as e: self.results = f"Error: {str(e)}" print(f"Agent {self.id} encountered an error: {str(e)}") def communicate(self, other_agents): pass class Swarm: def __init__(self, num_agents, fractal_pattern, api_key=None): self.agents = [Agent(i, api_key) for i in range(num_agents)] self.fractal_pattern = fractal_pattern print(f"Swarm created with {num_agents} agents using the {fractal_pattern} pattern.") def assign_tasks(self, tasks): for i, task in enumerate(tasks): self.agents[i % len(self.agents)].task = task print(f"Task assigned to Agent {self.agents[i % len(self.agents)].id}: {task}") def execute(self): for agent in self.agents: agent.execute_task() for agent in self.agents: agent.communicate(self.agents) def gather_results(self): return [agent.results for agent in self.agents if agent.results] def generate_tasks_from_model(prompt, num_tasks): # Generate tasks using the Qwen model inputs = tokenizer(prompt, return_tensors="pt") outputs = model.generate(**inputs, max_length=100, num_return_sequences=num_tasks) tasks = [tokenizer.decode(output, skip_special_tokens=True) for output in outputs] return tasks def run_swarm(prompt, api_key, num_agents, num_tasks): # Generate tasks using the model tasks = generate_tasks_from_model(prompt, num_tasks) # Create a swarm with a fractal pattern (Pentagonal spread) swarm = Swarm(num_agents=num_agents, fractal_pattern="Pentagonal", api_key=api_key) swarm.assign_tasks(tasks) swarm.execute() # Gather results results = swarm.gather_results() # Print all results print("\nAll results retrieved by the swarm:") for i, result in enumerate(results): print(f"Result {i + 1}: {result}") return results def gradio_interface(prompt, api_key, num_agents, num_tasks): results = run_swarm(prompt, api_key, num_agents, num_tasks) return "\n".join(str(result) for result in results) # Default values for the inputs default_prompt = "Generate API calls to fetch random cat facts." default_api_key = "" default_num_agents = 5 default_num_tasks = 2 iface = gr.Interface( fn=gradio_interface, inputs=[ gr.Textbox(label="Prompt", placeholder="Enter the prompt", value=default_prompt), gr.Textbox(label="API Key (Optional)", placeholder="Enter the API Key", value=default_api_key), gr.Number(label="Number of Agents", value=default_num_agents, precision=0), gr.Number(label="Number of Tasks", value=default_num_tasks, precision=0) ], outputs=gr.Textbox(label="Results"), title="Swarm Model Processing and Result Gatherer", description=""" This Gradio app demonstrates a swarm of agents using a language model to generate API calls and gather results. - The language model generates API calls based on the provided prompt. - The swarm is created based on a fractal geometry pattern. - Each agent makes an API call to the generated URLs and retrieves data. - The results from all agents are gathered and displayed. - Enter the prompt, API Key (optional), number of agents, and number of tasks to see the process in action. - By default, the app uses the prompt 'Generate API calls to fetch random cat facts' with 5 agents and 2 tasks. """ ) iface.launch()