You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
swarms/swarms/structs/collaborative_utils.py

78 lines
2.8 KiB

import traceback
from loguru import logger
from swarms.structs.deep_discussion import one_on_one_debate
from swarms.structs.omni_agent_types import AgentListType, AgentType
def talk_to_agent(
current_agent: AgentType,
agents: AgentListType,
task: str,
agent_name: str,
max_loops: int = 1,
output_type: str = "str-all-except-first",
):
"""
Initiate a one-on-one debate between the current agent and a named target agent
from a provided list, using a specified task or message as the debate topic.
This function searches through the provided list of agents for an agent whose
'agent_name' matches the specified `agent_name`. If found, it runs a turn-based
debate between `current_agent` and the target agent, using the `one_on_one_debate`
utility for a specified number of conversational loops.
Args:
current_agent (AgentType): The agent initiating the debate.
agents (AgentListType): The list of agent objects (must have 'agent_name' attributes).
task (str): The task, question, or message that serves as the debate topic.
agent_name (str): The name of the agent to engage in the debate with (must match 'agent_name').
max_loops (int, optional): Number of debate turns per agent. Defaults to 1.
output_type (str, optional): The format for the debate's output as returned by
`one_on_one_debate`. Defaults to "str-all-except-first".
Returns:
list: The formatted conversation history generated by `one_on_one_debate`.
Raises:
ValueError: If no agent with the specified name exists in the agent list.
Exception: Propagates any error encountered during debate setup or execution.
Example:
>>> talk_to_agent(
... current_agent=alice,
... agents=[alice, bob],
... task="Summarize and critique the given proposal.",
... agent_name="bob",
... max_loops=2
... )
"""
try:
target_agent = None
for agent in agents:
if (
hasattr(agent, "agent_name")
and agent.agent_name == agent_name
):
target_agent = agent
break
if target_agent is None:
raise ValueError(
f"Agent '{agent_name}' not found in agent list."
)
# Initiate a one-on-one debate between the current agent and the target agent.
return one_on_one_debate(
max_loops=max_loops,
agents=[current_agent, target_agent],
task=task,
output_type=output_type,
)
except Exception as e:
logger.error(
f"Error talking to agent: {e} Traceback: {traceback.format_exc()}"
)
raise e