Your Name
1 month ago
parent
73d23f1995
commit
96c50bb20c
@ -1,109 +1,87 @@
|
||||
import os
|
||||
from typing import List, Any
|
||||
from swarms.structs.agent import Agent
|
||||
from loguru import logger
|
||||
import uuid
|
||||
|
||||
WORKSPACE_DIR = os.getenv("WORKSPACE_DIR")
|
||||
uuid_for_log = str(uuid.uuid4())
|
||||
logger.add(
|
||||
os.path.join(
|
||||
WORKSPACE_DIR,
|
||||
"agents_available",
|
||||
f"agents-available-{uuid_for_log}.log",
|
||||
),
|
||||
level="INFO",
|
||||
colorize=True,
|
||||
backtrace=True,
|
||||
diagnose=True,
|
||||
)
|
||||
|
||||
|
||||
def get_agent_name(agent: Any) -> str:
|
||||
"""Helper function to safely get agent name
|
||||
|
||||
Args:
|
||||
agent (Any): The agent object to get name from
|
||||
|
||||
Returns:
|
||||
str: The agent's name if found, 'Unknown' otherwise
|
||||
"""
|
||||
if isinstance(agent, Agent) and hasattr(agent, "agent_name"):
|
||||
return agent.agent_name
|
||||
return "Unknown"
|
||||
|
||||
|
||||
def get_agent_description(agent: Any) -> str:
|
||||
"""Helper function to get agent description or system prompt preview
|
||||
|
||||
Args:
|
||||
agent (Any): The agent object
|
||||
|
||||
Returns:
|
||||
str: Description or first 100 chars of system prompt
|
||||
"""
|
||||
if not isinstance(agent, Agent):
|
||||
return "N/A"
|
||||
|
||||
if hasattr(agent, "description") and agent.description:
|
||||
return agent.description
|
||||
|
||||
if hasattr(agent, "system_prompt") and agent.system_prompt:
|
||||
return f"{agent.system_prompt[:150]}..."
|
||||
|
||||
return "N/A"
|
||||
from typing import List
|
||||
|
||||
|
||||
def showcase_available_agents(
|
||||
agents: List[Agent],
|
||||
name: str = None,
|
||||
description: str = None,
|
||||
agents: List[Agent] = [],
|
||||
update_agents_on: bool = False,
|
||||
format: str = "XML",
|
||||
) -> str:
|
||||
"""
|
||||
Generate a formatted string showcasing all available agents and their descriptions.
|
||||
Format the available agents in either XML or Table format.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): List of Agent objects to showcase.
|
||||
update_agents_on (bool, optional): If True, updates each agent's system prompt with
|
||||
the showcase information. Defaults to False.
|
||||
agents (List[Agent]): A list of agents to represent
|
||||
name (str, optional): Name of the swarm
|
||||
description (str, optional): Description of the swarm
|
||||
format (str, optional): Output format ("XML" or "Table"). Defaults to "XML"
|
||||
|
||||
Returns:
|
||||
str: Formatted string containing agent information, including names, descriptions
|
||||
and IDs for all available agents.
|
||||
str: Formatted string containing agent information
|
||||
"""
|
||||
logger.info(f"Showcasing {len(agents)} available agents")
|
||||
|
||||
formatted_agents = []
|
||||
header = f"\n####### Agents available in the swarm: {name} ############\n"
|
||||
header += f"{description}\n"
|
||||
row_format = "{:<5} | {:<20} | {:<50}"
|
||||
header_row = row_format.format("ID", "Agent Name", "Description")
|
||||
separator = "-" * 80
|
||||
def truncate(text: str, max_length: int = 130) -> str:
|
||||
return (
|
||||
f"{text[:max_length]}..."
|
||||
if len(text) > max_length
|
||||
else text
|
||||
)
|
||||
|
||||
formatted_agents.append(header)
|
||||
formatted_agents.append(separator)
|
||||
formatted_agents.append(header_row)
|
||||
formatted_agents.append(separator)
|
||||
output = []
|
||||
|
||||
if format.upper() == "TABLE":
|
||||
output.append("\n| ID | Agent Name | Description |")
|
||||
output.append("|-----|------------|-------------|")
|
||||
for idx, agent in enumerate(agents):
|
||||
if not isinstance(agent, Agent):
|
||||
logger.warning(
|
||||
f"Skipping non-Agent object: {type(agent)}"
|
||||
if isinstance(agent, Agent):
|
||||
agent_name = getattr(agent, "agent_name", str(agent))
|
||||
description = getattr(
|
||||
agent,
|
||||
"description",
|
||||
getattr(
|
||||
agent, "system_prompt", "Unknown description"
|
||||
),
|
||||
)
|
||||
continue
|
||||
|
||||
agent_name = get_agent_name(agent)
|
||||
description = (
|
||||
get_agent_description(agent)[:100] + "..."
|
||||
if len(get_agent_description(agent)) > 100
|
||||
else get_agent_description(agent)
|
||||
desc = truncate(description, 50)
|
||||
output.append(
|
||||
f"| {idx + 1} | {agent_name} | {desc} |"
|
||||
)
|
||||
|
||||
formatted_agents.append(
|
||||
row_format.format(idx + 1, agent_name, description)
|
||||
else:
|
||||
output.append(
|
||||
f"| {idx + 1} | {agent} | Unknown description |"
|
||||
)
|
||||
return "\n".join(output)
|
||||
|
||||
# Default XML format
|
||||
output.append("<agents>")
|
||||
if name:
|
||||
output.append(f" <name>{name}</name>")
|
||||
if description:
|
||||
output.append(
|
||||
f" <description>{truncate(description)}</description>"
|
||||
)
|
||||
for idx, agent in enumerate(agents):
|
||||
output.append(f" <agent id='{idx + 1}'>")
|
||||
if isinstance(agent, Agent):
|
||||
agent_name = getattr(agent, "agent_name", str(agent))
|
||||
description = getattr(
|
||||
agent,
|
||||
"description",
|
||||
getattr(
|
||||
agent, "system_prompt", "Unknown description"
|
||||
),
|
||||
)
|
||||
output.append(f" <name>{agent_name}</name>")
|
||||
output.append(
|
||||
f" <description>{truncate(description)}</description>"
|
||||
)
|
||||
else:
|
||||
output.append(f" <name>{agent}</name>")
|
||||
output.append(
|
||||
" <description>Unknown description</description>"
|
||||
)
|
||||
output.append(" </agent>")
|
||||
output.append("</agents>")
|
||||
|
||||
showcase = "\n".join(formatted_agents)
|
||||
|
||||
return showcase
|
||||
return "\n".join(output)
|
||||
|
||||
@ -0,0 +1,53 @@
|
||||
import concurrent.futures
|
||||
from typing import List, Union
|
||||
from swarms.structs.agent import Agent
|
||||
|
||||
|
||||
def update_system_prompts(
|
||||
agents: List[Union[Agent, str]],
|
||||
prompt: str,
|
||||
) -> List[Agent]:
|
||||
"""
|
||||
Update system prompts for a list of agents concurrently.
|
||||
|
||||
Args:
|
||||
agents: List of Agent objects or strings to update
|
||||
prompt: The prompt text to append to each agent's system prompt
|
||||
|
||||
Returns:
|
||||
List of updated Agent objects
|
||||
"""
|
||||
if not agents:
|
||||
return agents
|
||||
|
||||
def update_agent_prompt(agent: Union[Agent, str]) -> Agent:
|
||||
# Convert string to Agent if needed
|
||||
if isinstance(agent, str):
|
||||
agent = Agent(
|
||||
agent_name=agent,
|
||||
system_prompt=prompt, # Initialize with the provided prompt
|
||||
)
|
||||
else:
|
||||
# Preserve existing prompt and append new one
|
||||
existing_prompt = (
|
||||
agent.system_prompt if agent.system_prompt else ""
|
||||
)
|
||||
agent.system_prompt = existing_prompt + "\n" + prompt
|
||||
return agent
|
||||
|
||||
# Use ThreadPoolExecutor for concurrent execution
|
||||
max_workers = min(len(agents), 4) # Reasonable thread count
|
||||
with concurrent.futures.ThreadPoolExecutor(
|
||||
max_workers=max_workers
|
||||
) as executor:
|
||||
futures = []
|
||||
for agent in agents:
|
||||
future = executor.submit(update_agent_prompt, agent)
|
||||
futures.append(future)
|
||||
|
||||
# Collect results as they complete
|
||||
updated_agents = []
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
updated_agents.append(future.result())
|
||||
|
||||
return updated_agents
|
||||
Loading…
Reference in new issue