Utilities - swarms

Overview

The swarms.utils module provides essential utilities for file operations, logging, output formatting, token counting, and data processing. These utilities support core agent functionality and framework operations.

Logging

initialize_logger

Initialize a Loguru logger with custom formatting and output configuration.

from swarms.utils.loguru_logger import initialize_logger

logger = initialize_logger(log_folder="my_logs")

logger.info("Application started")
logger.error("An error occurred")
logger.debug("Debug information")

str

default:"swarms"

Legacy parameter, kept only for backwards compatibility — it no longer sets the log directory. Logs are always written to {WORKSPACE_DIR}/logs (via get_log_dir()), regardless of what is passed here.

Logger

Configured Loguru logger instance

Features:

Colored console output
Timestamp formatting
Function and line number tracking
Backtrace and diagnostics enabled
Thread-safe enqueuing

Formatting & Output

Formatter

Rich-based formatter for beautiful console output with markdown support.

from swarms.utils.formatter import Formatter

formatter = Formatter(md=True)

# Print formatted panels
formatter.print_panel(
    "Analysis complete",
    title="Status",
    style="bold green"
)

# Print markdown with syntax highlighting
formatter.print_markdown(
    "# Results\n\n```python\nprint('hello')\n```",
    title="Code Output"
)

Constructor

bool

default:"True"

Enable markdown output rendering

Methods

print_panel

Print content in a styled panel.

formatter.print_panel(
    content="Task completed successfully",
    title="Success",
    style="bold green"
)

str

required

Content to display in the panel

str

default:""

Panel title

str

default:"bold blue"

Panel style (color and formatting)

print_markdown

Render markdown content with syntax highlighting.

formatter.print_markdown(
    content="# Analysis\n\nResults are **positive**",
    title="Report",
    border_style="blue"
)

str

required

Markdown content to render

str

default:""

Panel title

str

default:"blue"

Border color style

print_streaming_panel

Display real-time streaming response with live updates.

response = formatter.print_streaming_panel(
    streaming_response=llm_stream,
    title="Agent Response",
    collect_chunks=True
)

Generator

required

Streaming response generator from LLM

str

default:"Agent Streaming Response"

Panel title

str

default:"None"

Panel style (uses random color if None)

bool

default:"False"

Whether to collect individual chunks

Callable

default:"None"

Callback function for each chunk

str

Complete accumulated response text

print_agent_dashboard

Display a live dashboard showing agent statuses.

agents_data = [
    {"name": "Agent-1", "status": "running", "output": "Processing..."},
    {"name": "Agent-2", "status": "completed", "output": "Done!"}
]

formatter.print_agent_dashboard(
    agents_data=agents_data,
    title="Swarm Dashboard",
    is_final=False
)

List[Dict[str, Any]]

required

List of agent information dictionaries with name, status, and output

str

default:"Concurrent Workflow Dashboard"

Dashboard title

bool

default:"False"

Whether this is the final update

Data Structure Formatting

format_dict_to_string

Recursively format a dictionary into a readable, multi-line string.

from swarms.utils import format_dict_to_string

text = format_dict_to_string({"name": "Agent", "config": {"loops": 3}})
print(text)

dict

required

The dictionary to format

int

default:"0"

Current indentation level for nested structures

bool

default:"True"

If True, use "key: value" formatting; if False, use "key value"

format_data_structure

Format any Python data structure (dict, list, tuple, set, or object) into a readable, indented, multi-line string.

from swarms.utils import format_data_structure

text = format_data_structure({"agents": ["A", "B"], "count": 2})
print(text)

any

required

The data structure to format

int

default:"0"

Current indentation level

int

default:"10"

Maximum depth to recurse

exists

Check if a value is not None.

from swarms.utils import exists

exists(None)  # False
exists("value")  # True

File Processing

create_file_in_folder

Create a file with content in a specified folder.

from swarms.utils import create_file_in_folder

file_path = create_file_in_folder(
    folder_path="./reports",
    file_name="analysis.txt",
    content="Financial analysis results..."
)

str

required

Path to the folder (created if doesn’t exist)

str

required

Name of the file to create

Any

required

Content to write to the file

str

Path to the created file

sanitize_file_path

Clean and sanitize file paths for cross-platform compatibility.

from swarms.utils import sanitize_file_path

safe_path = sanitize_file_path("`C:/Users/file<name>.txt`")
# Returns: C__Users_file_name_.txt
# `:`, `/`, `\`, `<`, `>`, `"`, `|`, `?`, `*`, and backticks are all replaced with `_`

str

required

File path to sanitize

str

Sanitized file path safe for all platforms

load_json

Load and parse a JSON string.

from swarms.utils import load_json

json_str = '{"name": "Agent", "status": "active"}'
data = load_json(json_str)
print(data["name"])  # "Agent"

str

required

JSON string to parse

object

Parsed Python object (dict, list, etc.)

zip_workspace

Zip an entire workspace directory.

from swarms.utils import zip_workspace

zip_path = zip_workspace(
    workspace_path="./my_workspace",
    output_filename="workspace_backup"
)

str

required

Path to workspace directory to zip

str

required

Name for output zip file (without .zip extension)

str

Path to created zip file

zip_folders

Zip multiple folders into a single archive.

from swarms.utils import zip_folders

zip_folders(
    folder1_path="./data",
    folder2_path="./logs",
    zip_file_path="combined_backup"
)

str

required

Path to first folder

str

required

Path to second folder

str

required

Output zip file path

Token Management

count_tokens

Count tokens in text using LiteLLM tokenizer.

from swarms.utils import count_tokens

text = "Analyze the financial statements"
token_count = count_tokens(
    text=text,
    model="gpt-4"
)
print(f"Tokens: {token_count}")

str

required

Text to count tokens for

str

default:"gpt-5.4"

Model to use for tokenization

str

default:"gpt-5.4"

Fallback encoder used if tokenizing with model fails

int

Number of tokens in the text

Raises: ValueError if both the primary model and the fallback encoder fail to tokenize the text.

get_supported_models

Get the list of models supported by LiteLLM.

from swarms.utils.litellm_tokenizer import get_supported_models

models = get_supported_models()
print(models)

list

List of supported model name strings

Agent Loading

load_agent_from_markdown

Load agent configuration from markdown file.

from swarms.utils import load_agent_from_markdown

agent = load_agent_from_markdown("agent_config.md")

load_agents_from_markdown

Load multiple agents from markdown files.

from swarms.utils import load_agents_from_markdown

agents = load_agents_from_markdown([
    "agent1.md",
    "agent2.md",
    "agent3.md"
])

MarkdownAgentLoader

Class for loading agents from markdown with advanced options. load_agent_from_markdown and load_agents_from_markdown are thin wrappers around it.

from swarms.utils import MarkdownAgentLoader

loader = MarkdownAgentLoader(max_workers=4)

agent = loader.load_single_agent("agent_config.md")
agents = loader.load_multiple_agents("./agent_configs")

int

default:"None"

Worker count used when loading multiple files concurrently

Method	Description
`load_single_agent(file_path, **kwargs)`	Load one agent from a markdown file
`load_multiple_agents(file_paths, **kwargs)`	Load agents from a directory path or a list of file paths
`parse_markdown_file(file_path)`	Parse a markdown file into a `MarkdownAgentConfig` without building an `Agent`
`parse_yaml_frontmatter(content)`	Parse just the YAML frontmatter out of markdown content

Context Window Management

Conversation.dynamic_auto_chunking

Trim the conversation history from the beginning so the remainder fits within the conversation’s token budget, using a binary search over token counts. It returns a single trimmed string (the tail of the history that fits), not a list of chunks.

This is a method on Conversation, not a standalone helper — there is no dynamic_auto_chunking in swarms.utils. The budget and tokenizer come from the Conversation’s own context_length and tokenizer_model_name, so the method itself takes no arguments.

from swarms.structs.conversation import Conversation

conversation = Conversation(
    context_length=4000,
    tokenizer_model_name="gpt-5.4",
)

conversation.add("user", "...very long document...")
conversation.add("assistant", "...long analysis...")

trimmed = conversation.dynamic_auto_chunking()
print(f"Trimmed length: {len(trimmed)} chars")

str

The conversation history trimmed to fit within context_length tokens. Returns the full history unchanged if it already fits, or if chunking fails.

Relevant Conversation constructor parameters:

int

default:"8192"

Maximum number of tokens allowed in the conversation history

str

default:"gpt-5.4"

Model used for token counting

Output History Formatting

history_output_formatter

Format a Conversation object’s history into one of several output formats.

from swarms.utils import history_output_formatter

formatted = history_output_formatter(
    conversation=conversation_history,
    type="str",
)
print(formatted)

Conversation

required

A conversation object exposing methods like return_messages_as_list(), to_dict(), get_str(), etc.

HistoryOutputType

default:"list"

Output format. One of: "list", "dict"/"dictionary", "string"/"str", "final"/"last", "json", "all", "yaml", "xml", "dict-all-except-first", "str-all-except-first", "dict-final", "list-final"

Raises: ValueError if type is not one of the supported values.

LiteLLM Wrapper

LiteLLM

Wrapper class for LiteLLM with error handling.

from swarms.utils import LiteLLM

llm = LiteLLM(
    model_name="gpt-5.4",
    temperature=0.7,
    max_tokens=1000
)

response = llm.run("Analyze this data")

The constructor parameter is model_name, not model — because LiteLLM.__init__ absorbs unrecognized keywords via **kwargs, passing model= silently fails to set the model instead of raising an error.

NetworkConnectionError

Exception raised for network connection issues.

from swarms.utils import NetworkConnectionError

try:
    response = llm.run(prompt)
except NetworkConnectionError as e:
    print(f"Network error: {e}")
    # Handle retry logic

LiteLLMException

General exception for LiteLLM errors.

from swarms.utils import LiteLLMException

try:
    response = llm.run(prompt)
except LiteLLMException as e:
    print(f"LiteLLM error: {e}")

Workspace Management

WorkspaceManager

Creates a swarm’s autosave directory once and writes to it on demand. The directory is {WORKSPACE_DIR}/swarms/{ClassName}/{name}-{stamp}, created eagerly so dir is usable straight after construction.

from swarms.utils import WorkspaceManager

manager = WorkspaceManager(owner=my_swarm, verbose=True)
print(manager.dir)

Any

required

The swarm instance. Its class name and name attribute pick the directory, and it is the default source for conversation and config data

str

default:"None"

Overrides owner.name in the path

bool

default:"True"

Timestamp in the directory name when True, otherwise a short UUID

bool

default:"False"

Log the directory and each successful write

bool

default:"True"

When False nothing is created or written and dir stays None

Optional[Sequence[str]]

default:"None"

Path segments joined onto the workspace directory in place of the default swarms/{ClassName}/{name}-{stamp} layout — e.g. ("agents", "my-agent-a1b2c3d4e5f6")

Optional[Dict[str, Any]]

default:"None"

Base fields merged into _autosave_metadata on every write, in place of the default {class_name, swarm_name, swarm_id}

Example: Complete Utility Usage

from swarms.utils import (
    initialize_logger,
    create_file_in_folder,
    count_tokens,
    sanitize_file_path,
)
from swarms.utils.formatter import Formatter

# Initialize logging
logger = initialize_logger("my_app")
logger.info("Starting application")

# Create formatter for output
formatter = Formatter(md=True)

# Process some data
markdown_content = """
# Analysis Results

The quarter closed **above** forecast.
"""

# Count tokens
tokens = count_tokens(markdown_content, model="gpt-5.4")
formatter.print_panel(
    f"Report has {tokens} tokens",
    title="Token Count",
    style="bold cyan"
)

# Save to file
safe_path = sanitize_file_path("./reports/analysis_results.txt")
file_path = create_file_in_folder(
    folder_path="./reports",
    file_name="analysis_results.txt",
    content=markdown_content
)

logger.info(f"Saved to: {file_path}")

# Display markdown
formatter.print_markdown(
    markdown_content,
    title="Analysis Report",
    border_style="green"
)

Best Practices

Use logging extensively: Initialize logger in all modules for debugging
Sanitize paths: Always sanitize file paths before file operations
Count tokens: Monitor token usage to stay within model limits
Format output: Use Formatter for consistent, beautiful CLI output
Handle errors: Wrap file operations in try-catch blocks
Chunk large texts: Use Conversation.dynamic_auto_chunking() to keep long histories inside the context window
Stream responses: Use print_streaming_panel for real-time output

​Overview

​Logging

​initialize_logger

​Formatting & Output

​Formatter

​Constructor

​Methods

print_panel

print_markdown

print_streaming_panel

print_agent_dashboard

​Data Structure Formatting

​format_dict_to_string

​format_data_structure

​exists

​File Processing

​create_file_in_folder

​sanitize_file_path

​load_json

​zip_workspace

​zip_folders

​Token Management

​count_tokens

​get_supported_models

​Agent Loading

​load_agent_from_markdown

​load_agents_from_markdown

​MarkdownAgentLoader

​Context Window Management

​Conversation.dynamic_auto_chunking

​Output History Formatting

​history_output_formatter

​LiteLLM Wrapper

​LiteLLM

​NetworkConnectionError

​LiteLLMException

​Workspace Management

​WorkspaceManager

​Example: Complete Utility Usage

​Best Practices

Overview

Logging

initialize_logger

Formatting & Output

Formatter

Constructor

Methods

Data Structure Formatting

format_dict_to_string

format_data_structure

exists

File Processing

create_file_in_folder

sanitize_file_path

load_json

zip_workspace

zip_folders

Token Management

count_tokens

get_supported_models

Agent Loading

load_agent_from_markdown

load_agents_from_markdown

MarkdownAgentLoader

Context Window Management

Conversation.dynamic_auto_chunking

Output History Formatting

history_output_formatter

LiteLLM Wrapper

LiteLLM

NetworkConnectionError

LiteLLMException

Workspace Management

WorkspaceManager

Example: Complete Utility Usage

Best Practices