File size: 3,934 Bytes
6cbcd8c 5d7ede0 6cbcd8c a02dc5c 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c a02dc5c 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c 5d7ede0 6cbcd8c a02dc5c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
import os
import requests
import gradio as gr
# Define Agent and Swarm classes based on fractal geometry
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().get('data')[0]
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]
# Generate tasks for the swarm
def generate_tasks(api_url, num_tasks):
return [api_url] * num_tasks
# Function to run the swarm and gather results
def run_swarm(api_url, api_key, num_agents, num_tasks):
# Create a swarm with a fractal pattern (Pentagonal spread)
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()
# 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
# Gradio interface
def gradio_interface(api_url, api_key, num_agents, num_tasks):
results = run_swarm(api_url, api_key, num_agents, num_tasks)
return "\n".join(str(result) for result in results)
# Default values for the inputs
default_api_url = "https://meowfacts.herokuapp.com/"
default_num_agents = 5
default_num_tasks = 2
iface = gr.Interface(
fn=gradio_interface,
inputs=[
gr.Textbox(label="API URL", placeholder="Enter the API URL", value=default_api_url),
gr.Textbox(label="API Key (Optional)", placeholder="Enter the API Key"),
gr.Number(label="Number of Agents", value=default_num_agents, precision=0),
gr.Number(label="Number of API Calls", value=default_num_tasks, precision=0)
],
outputs=gr.Textbox(label="Results"),
title="Swarm API Call and Result Gatherer",
description="""
This Gradio app demonstrates a swarm of agents making API calls and gathering results.
- The swarm is created based on a fractal geometry pattern.
- Each agent makes an API call to the specified URL and retrieves data.
- The results from all agents are gathered and displayed.
- Enter the API URL, API Key (optional), number of agents, and number of API calls to see the process in action.
- By default, the app uses the 'Meow Facts' API with 5 agents and 2 API calls.
"""
)
iface.launch() |