|
import os |
|
import requests |
|
import matplotlib.pyplot as plt |
|
import numpy as np |
|
import gradio as gr |
|
|
|
|
|
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 {} |
|
response = requests.get(self.task, headers=headers) |
|
if response.status_code == 200: |
|
self.results = response.json().get('data')[0] |
|
else: |
|
self.results = "Error: Unable to fetch data" |
|
print(f"Agent {self.id} received: {self.results}") |
|
|
|
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(api_url, num_tasks): |
|
return [api_url] * num_tasks |
|
|
|
|
|
def plot_pentagonal_and_mirrored(results): |
|
fig, ax = plt.subplots() |
|
ax.set_aspect('equal') |
|
|
|
|
|
angle = 2 * np.pi / 5 |
|
radius = 1 |
|
pentagon_points = np.array([(radius * np.cos(i * angle), radius * np.sin(i * angle)) for i in range(5)]) |
|
|
|
|
|
for i in range(5): |
|
ax.plot([pentagon_points[i][0], pentagon_points[(i + 1) % 5][0]], |
|
[pentagon_points[i][1], pentagon_points[(i + 1) % 5][1]], 'k-') |
|
|
|
|
|
center = np.array([0, 0]) |
|
points = [ |
|
center, |
|
(center + pentagon_points[0]) / 2, |
|
(center + pentagon_points[1]) / 2, |
|
(center + pentagon_points[2]) / 2, |
|
(center + pentagon_points[3]) / 2, |
|
(center + pentagon_points[4]) / 2, |
|
(pentagon_points[0] + pentagon_points[1]) / 2, |
|
(pentagon_points[1] + pentagon_points[2]) / 2, |
|
(pentagon_points[2] + pentagon_points[3]) / 2 |
|
] |
|
|
|
|
|
for i, point in enumerate(points[:len(results)]): |
|
ax.plot(point[0], point[1], 'bo') |
|
ax.text(point[0], point[1], results[i], fontsize=9, ha='right') |
|
|
|
|
|
mirrored_x = [-point[0], point[0]] |
|
mirrored_y = [-point[1], point[1]] |
|
|
|
for mx in mirrored_x: |
|
for my in mirrored_y: |
|
if (mx, my) != (point[0], point[1]): |
|
ax.plot(mx, my, 'ro') |
|
ax.text(mx, my, results[i], fontsize=9, ha='right') |
|
|
|
plt.show() |
|
|
|
|
|
def run_swarm(api_url, api_key, num_agents, num_tasks): |
|
|
|
swarm = Swarm(num_agents=num_agents, fractal_pattern="Pentagonal", api_key=api_key) |
|
tasks = generate_tasks(api_url, num_tasks) |
|
swarm.assign_tasks(tasks) |
|
swarm.execute() |
|
|
|
|
|
results = swarm.gather_results() |
|
|
|
|
|
print("\nAll results retrieved by the swarm:") |
|
for i, result in enumerate(results): |
|
print(f"Result {i + 1}: {result}") |
|
|
|
|
|
if results: |
|
plot_pentagonal_and_mirrored(results) |
|
|
|
return results |
|
|
|
|
|
def gradio_interface(api_url, api_key, num_agents, num_tasks): |
|
results = run_swarm(api_url, api_key, num_agents, num_tasks) |
|
return results |
|
|
|
iface = gr.Interface( |
|
fn=gradio_interface, |
|
inputs=[ |
|
gr.Textbox(label="API URL", placeholder="Enter the API URL"), |
|
gr.Textbox(label="API Key (Optional)", placeholder="Enter the API Key"), |
|
gr.Number(label="Number of Agents", value=9, precision=0), |
|
gr.Number(label="Number of API Calls", value=9, precision=0) |
|
], |
|
outputs=gr.Textbox(label="Results"), |
|
title="Swarm API Call and Plotter", |
|
description="Enter the API URL, API Key (Optional), number of agents, and number of API calls. The results will be plotted in a pentagonal pattern with mirrored points." |
|
) |
|
|
|
iface.launch(share=True) |
|
|