GroupChat Documentation¶
The GroupChat is a sophisticated multi-agent system that enables interactive conversations between users and AI agents using @mentions. This system allows users to direct tasks to specific agents and facilitates collaborative responses when multiple agents are mentioned.
Features¶
| Feature | Description |
|---|---|
| @mentions Support | Direct tasks to specific agents using @agent_name syntax |
| Multi-Agent Collaboration | Multiple mentioned agents can see and respond to each other's tasks |
| Enhanced Collaborative Prompts | Agents are trained to acknowledge, build upon, and synthesize each other's responses |
| Speaker Functions | Control the order in which agents respond (round robin, random, priority, custom) |
| Dynamic Speaker Management | Change speaker functions and priorities during runtime |
| Random Dynamic Speaker | Advanced speaker function that follows @mentions in agent responses |
| Parallel and Sequential Strategies | Support for both parallel and sequential agent execution |
| Callable Function Support | Supports both Agent instances and callable functions as chat participants |
| Comprehensive Error Handling | Custom error classes for different scenarios |
| Conversation History | Maintains a complete history of the conversation |
| Flexible Output Formatting | Configurable output format for conversation history |
| Interactive Terminal Mode | Full REPL interface for real-time chat with agents |
Installation¶
Methods Reference¶
Constructor (__init__)¶
Description: Initializes a new GroupChat instance with the specified configuration.
Arguments:
| Parameter | Type | Description | Default |
|---|---|---|---|
id |
str | Unique identifier for the chat | auto-generated key with prefix "swarms-" |
name |
str | Name of the group chat | "GroupChat" |
description |
str | Description of the chat's purpose | "A group chat for multiple agents" |
agents |
List[Union[Agent, Callable]] | List of participating agents | empty list |
max_loops |
int | Maximum conversation turns | 1 |
output_type |
str | Type of output format ("dict", "str", "all", etc.) | "dict" |
interactive |
bool | Whether to enable interactive terminal session | False |
speaker_function |
Union[str, Callable] | Function to determine speaking order. String options: "round-robin-speaker", "random-speaker", "priority-speaker", "random-dynamic-speaker" | None (defaults to round_robin_speaker) |
speaker_state |
dict | Initial state/config for the speaker function (e.g., {"priorities": {...}} for priority-speaker, {"strategy": "sequential"} for random-dynamic-speaker) | None (defaults to {"current_index": 0}) |
speaker_fn |
Callable | Legacy speaker function for backward compatibility (deprecated, use speaker_function instead) | None |
rules |
str | Rules to inject into the conversation context | "" (empty string) |
time_enabled |
bool | Whether to enable timestamps in conversation history | True |
Example:
if name == "main":
# Example agents
agent1 = Agent(
agent_name="Financial-Analysis-Agent",
system_prompt="You are a financial analyst specializing in investment strategies.",
model_name="gpt-4.1",
max_loops=1,
)
agent2 = Agent(
agent_name="Tax-Adviser-Agent",
system_prompt="You are a tax adviser who provides clear and concise guidance on tax-related queries.",
model_name="gpt-4.1",
max_loops=1,
)
agents = [agent1, agent2]
chat = GroupChat(
name="Investment Advisory",
description="Financial and tax analysis group",
agents=agents,
speaker_function="round-robin-speaker", # or use a custom callable
)
history = chat.run(
"How to optimize tax strategy for investments?"
)
print(history.model_dump_json(indent=2))
## Speaker Functions
GroupChat supports multiple built-in speaker functions that control the order in which agents respond. You can use them by passing the function name as a string to `speaker_function`, or provide a custom callable.
### Built-in Speaker Functions
#### 1. Round Robin Speaker (`"round-robin-speaker"`)
Agents speak in a fixed order, cycling through the list sequentially.
```python
from swarms import Agent, GroupChat
chat = GroupChat(
agents=[agent1, agent2, agent3],
speaker_function="round-robin-speaker"
)
2. Random Speaker ("random-speaker")¶
Agents are selected randomly to speak.
3. Priority Speaker ("priority-speaker")¶
Agents are selected based on priority weights. Requires speaker_state with priorities.
chat = GroupChat(
agents=[agent1, agent2, agent3],
speaker_function="priority-speaker",
speaker_state={
"priorities": {
"agent1": 0.5, # 50% chance
"agent2": 0.3, # 30% chance
"agent3": 0.2 # 20% chance
}
}
)
4. Random Dynamic Speaker ("random-dynamic-speaker")¶
First agent is randomly selected, then follows @mentions in responses. Supports "sequential" or "parallel" strategies.
chat = GroupChat(
agents=[agent1, agent2, agent3],
speaker_function="random-dynamic-speaker",
speaker_state={"strategy": "sequential"} # or "parallel"
)
Custom Speaker Function Example¶
You can also provide a custom callable function. The function should accept a list of agent names and optional keyword arguments, and return the selected agent name(s).
from typing import List, Union
def custom_speaker(agents: List[str], **kwargs) -> str:
"""
Custom speaker function that selects agents based on task keywords.
Args:
agents: List of available agent names
**kwargs: Additional arguments (e.g., task, response, history)
Returns:
str: Selected agent name
"""
task = kwargs.get("task", "").lower()
# Map keywords to agents
if "data" in task or "analyze" in task:
if "analyst" in agents:
return "analyst"
elif "research" in task or "information" in task:
if "researcher" in agents:
return "researcher"
elif "write" in task or "content" in task:
if "writer" in agents:
return "writer"
# Default to first agent
return agents[0] if agents else None
# Use the custom speaker function
chat = GroupChat(
agents=[analyst, researcher, writer],
speaker_function=custom_speaker
)
For dynamic speaker functions that follow @mentions, you can create a function that returns a list of agents:
def custom_dynamic_speaker(
agents: List[str],
response: str = "",
strategy: str = "sequential",
**kwargs
) -> Union[str, List[str]]:
"""
Custom dynamic speaker function that follows @mentions.
Args:
agents: List of available agent names
response: The previous agent's response (may contain @mentions)
strategy: "sequential" or "parallel"
**kwargs: Additional arguments
Returns:
Union[str, List[str]]: Selected agent name(s)
"""
import re
# Extract @mentions from response
mentions = re.findall(r"@(\w+)", response)
valid_mentions = [m for m in mentions if m in agents]
if valid_mentions:
if strategy == "sequential":
return valid_mentions[0]
else: # parallel
return valid_mentions
# If no mentions, return first agent
return agents[0] if agents else None
# Check for direct mentions
mentioned = agent.agent_name.lower() in last_message
# Check if agent hasn't spoken recently
not_recent_speaker = not any(
agent.agent_name in msg
for msg in history[-3:]
)
return expertise_relevant or mentioned or not_recent_speaker
Usage¶
chat = GroupChat( agents=[agent1, agent2], speaker_fn=custom_speaker )
## Advanced Examples
### Multi-Agent Analysis Team
```python
# Create specialized agents
data_analyst = Agent(
agent_name="Data-Analyst",
system_prompt="You analyze numerical data and patterns",
model_name="gpt-4.1",
)
market_expert = Agent(
agent_name="Market-Expert",
system_prompt="You provide market insights and trends",
model_name="gpt-4.1",
)
strategy_advisor = Agent(
agent_name="Strategy-Advisor",
system_prompt="You formulate strategic recommendations",
model_name="gpt-4.1",
)
# Create analysis team
analysis_team = GroupChat(
name="Market Analysis Team",
description="Comprehensive market analysis group",
agents=[data_analyst, market_expert, strategy_advisor],
speaker_fn=expertise_based,
max_loops=15
)
# Run complex analysis
history = analysis_team.run("""
Analyze the current market conditions:
1. Identify key trends
2. Evaluate risks
3. Recommend investment strategy
""")
Parallel Processing¶
# Define multiple analysis tasks
tasks = [
"Analyze tech sector trends",
"Evaluate real estate market",
"Review commodity prices",
"Assess global economic indicators"
]
# Run tasks concurrently
histories = chat.concurrent_run(tasks)
# Process results
for task, history in zip(tasks, histories):
print(f"\nAnalysis for: {task}")
for turn in history.turns:
for response in turn.responses:
print(f"{response.agent_name}: {response.message}")
Best Practices¶
| Category | Recommendations |
|---|---|
| Agent Design | - Give agents clear, specific roles - Use detailed system prompts - Set appropriate context lengths - Enable retries for reliability |
| Speaker Functions | - Match function to use case - Consider conversation flow - Handle edge cases - Add appropriate logging |
| Error Handling | - Use try-except blocks - Log errors appropriately - Implement retry logic - Provide fallback responses |
| Performance | - Use concurrent processing for multiple tasks - Monitor context lengths - Implement proper cleanup - Cache responses when appropriate |
API Reference¶
GroupChat Methods¶
| Method | Description | Arguments | Returns |
|---|---|---|---|
| run | Run single conversation | task: str | ChatHistory |
| batched_run | Run multiple sequential tasks | tasks: List[str] | List[ChatHistory] |
| concurrent_run | Run multiple parallel tasks | tasks: List[str] | List[ChatHistory] |
| get_recent_messages | Get recent messages | n: int = 3 | List[str] |
Agent Methods¶
| Method | Description | Returns |
|---|---|---|
| run | Process single task | str |
| generate_response | Generate LLM response | str |
| save_context | Save conversation context | None |