HierarchicalStructuredCommunicationFramework

Overview

The Hierarchical Structured Communication Framework implements the “Talk Structurally, Act Hierarchically” approach for LLM multi-agent systems, based on the research paper arXiv:2502.11098. It provides structured communication protocols with specialized agent classes for content generation, evaluation, refinement, and supervision.

These classes are not exported from the top-level swarms package. Import them from swarms.structs.hierarchical_structured_communication_framework.

Installation

pip install -U swarms

Key Components

Agent Classes

Each class subclasses Agent, fills in a role-specific default system prompt when system_prompt is None, and forwards other keyword arguments to Agent.

Class	Default `agent_name`	Extra method
`HierarchicalStructuredCommunicationGenerator`	`TalkHierGenerator`	`generate_with_structure(message, background="", intermediate_output="")` returns a `GeneratorResponseSchema`
`HierarchicalStructuredCommunicationEvaluator`	`TalkHierEvaluator`	`evaluate_with_criterion(content, criterion)` returns an `EvaluatorResponseSchema`. Also accepts `evaluation_criteria` for its default prompt
`HierarchicalStructuredCommunicationRefiner`	`TalkHierRefiner`	`refine_with_feedback(original_content, evaluation_results)` returns a `RefinerResponseSchema`
`HierarchicalStructuredCommunicationSupervisor`	`TalkHierSupervisor`	None

All four default to model_name="gpt-5.4" and verbose=False. The framework itself calls each agent’s plain run() method, not these extra methods.

Main Framework

HierarchicalStructuredCommunicationFramework - Main orchestrator class

Attributes

str

default:"HierarchicalStructuredCommunicationFramework"

Name of the framework instance

str

default:"Talk Structurally, Act Hierarchically Framework"

Description of the framework instance

Optional[Union[Agent, Callable, Any]]

default:"None"

Main supervisor agent. If not provided, a default Agent supervisor is created automatically. The workflow does not currently call the supervisor; its name (supervisor_name) is used as the sender of structured messages.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of generator agents. If not provided, a single default generator agent is created automatically. Only the first generator is used.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of evaluator agents. If not provided (and enable_hierarchical_evaluation=True), a single default evaluator agent is created automatically. Every evaluator runs, one after another, each on one criterion.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of refiner agents. If not provided, a single default refiner agent is created automatically. Only the first refiner is used.

Optional[Union[Agent, Callable, Any]]

default:"None"

Supervisor that summarizes the evaluators’ results. Created automatically if not provided and enable_hierarchical_evaluation=True. Its summary is logged but not passed to the refiner.

int

default:"3"

Maximum number of generate-evaluate-refine loops

OutputType

default:"dict-all-except-first"

Stored on the instance but not used. run() always returns a dictionary.

str

default:"Supervisor"

Name of the default supervisor agent, and the sender name on messages to generators and refiners.

str

default:"EvaluationSupervisor"

Name of the default evaluation supervisor agent, and the sender name on messages to evaluators.

bool

default:"True"

Stored on the instance but not read. Structured messages are always recorded.

bool

default:"True"

When False, no evaluators or evaluation supervisor are created by default and run_hierarchical_evaluation() returns [], so the refiner is skipped and each loop returns the generator’s output.

bool

default:"True"

Stored on the instance but not read.

str

default:"gpt-5.4"

LLM model name used for the default agents the framework creates

bool

default:"False"

Enable verbose logging

bool

default:"False"

Route the default agents’ calls through a local Ollama server instead of a hosted provider. Agents you pass in are not changed.

str

default:"http://localhost:11434/v1"

Base URL for the Ollama server when use_ollama=True.

str

default:"ollama"

API key sent to the Ollama server when use_ollama=True.

Methods

run()

Execute the complete workflow for a given task, looping up to max_loops times (set at construction).

def run(self, task: str, img: str = None, *args, **kwargs) -> dict

Parameters:

task (str): The task to execute
img (str, optional): Accepted but not passed to any agent

Returns: A dictionary with these keys:

Key	Type	Description
`final_result`	`Optional[str]`	The refined output of the last completed loop, or `None` if the first loop failed
`total_loops`	`int`	Number of loops started
`conversation_history`	`List[StructuredMessage]`	Every structured message recorded on this instance
`evaluation_results`	`List[EvaluationResult]`	Every evaluation recorded on this instance
`intermediate_outputs`	`Dict[str, str]`	The latest `"generator"` and `"refiner"` outputs

Each loop calls step() on the original task, so it generates fresh content rather than refining the previous loop’s result. The loop stops early when the average evaluation score of a loop is 8.0 or higher (checked on every loop but the last), or when step() returns an error.

conversation_history, evaluation_results and intermediate_outputs are not cleared between run() calls on the same instance.

run() does not accept a max_loops override — the number of refinement loops is fixed by the max_loops value passed to the constructor.

step()

Execute a single workflow step (generate, evaluate, refine).

def step(self, task: str, img: str = None, *args, **kwargs) -> dict

Parameters:

task (str): The task to execute for one step
img (str, optional): Accepted but not passed to any agent

Returns: A dictionary with generator_result, evaluation_results, refined_result, and conversation_history (or an error key on failure). refined_result falls back to the generator’s output when there are no evaluations or the refiner fails

step() guards against runaway prompts. A task longer than 1,000 characters returns {"error": ...} without calling any agent. The generator’s output is truncated to 2,000 characters, and an output that mostly repeats the task is treated as a failure.

send_structured_message()

Send a structured communication message between agents, following the Message (M_ij) / Background (B_ij) / Intermediate Output (I_ij) protocol.

def send_structured_message(
    self,
    sender: str,
    recipient: str,
    message: str,
    background: str = "",
    intermediate_output: str = "",
) -> StructuredMessage

Parameters:

sender (str): Name of the sending agent
recipient (str): Name of the receiving agent
message (str): Specific task message (M_ij)
background (str, optional): Background context (B_ij)
intermediate_output (str, optional): Intermediate output (I_ij)

Returns: The StructuredMessage that was appended to conversation_history

run_hierarchical_evaluation()

Run the hierarchical evaluation system with supervisor coordination.

def run_hierarchical_evaluation(
    self, content: str, evaluation_criteria: List[str] = None
) -> List[EvaluationResult]

Parameters:

content (str): Content to evaluate
evaluation_criteria (List[str], optional): Criteria to evaluate against. Defaults to ["accuracy", "completeness", "clarity", "relevance"]. Evaluator i gets criterion i % len(evaluation_criteria).

Returns: A list of EvaluationResult objects, one per evaluator that answered. An evaluator that raises is skipped. Returns [] when enable_hierarchical_evaluation=False Each evaluator is asked for a score from 0 to 10 and a confidence from 0 to 1. The framework parses both from the reply: score is the last number after the word “score” that falls in 0-10 (default 7.5), and confidence is the last number after “confidence” that falls in 0-1 (default 0.8). Range hints such as (0-10) are ignored. The full reply is stored as feedback. If an evaluation supervisor exists, it summarizes the results and the summary is logged.

Usage Examples

Quick Start

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework,
    HierarchicalStructuredCommunicationGenerator,
    HierarchicalStructuredCommunicationEvaluator,
    HierarchicalStructuredCommunicationRefiner,
    HierarchicalStructuredCommunicationSupervisor
)

# Create specialized agents
generator = HierarchicalStructuredCommunicationGenerator(
    agent_name="ContentGenerator"
)

evaluator = HierarchicalStructuredCommunicationEvaluator(
    agent_name="QualityEvaluator"
)

refiner = HierarchicalStructuredCommunicationRefiner(
    agent_name="ContentRefiner"
)

supervisor = HierarchicalStructuredCommunicationSupervisor(
    agent_name="WorkflowSupervisor"
)

# Create the framework
framework = HierarchicalStructuredCommunicationFramework(
    name="MyFramework",
    supervisor=supervisor,
    generators=[generator],
    evaluators=[evaluator],
    refiners=[refiner],
    max_loops=3
)

# Run the workflow
result = framework.run("Create a comprehensive analysis of AI trends in 2024")

Basic Usage with Default Supervisor

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework,
    HierarchicalStructuredCommunicationGenerator,
    HierarchicalStructuredCommunicationEvaluator,
    HierarchicalStructuredCommunicationRefiner
)

# Create agents with custom names
generator = HierarchicalStructuredCommunicationGenerator(agent_name="ContentGenerator")
evaluator = HierarchicalStructuredCommunicationEvaluator(agent_name="QualityEvaluator")
refiner = HierarchicalStructuredCommunicationRefiner(agent_name="ContentRefiner")

# Create framework with default supervisor
framework = HierarchicalStructuredCommunicationFramework(
    generators=[generator],
    evaluators=[evaluator],
    refiners=[refiner],
    max_loops=3,
    verbose=True
)

# Execute task
result = framework.run("Write a detailed report on renewable energy technologies")
print(result["final_result"])

Advanced Configuration

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework
)

# Create framework with custom configuration
framework = HierarchicalStructuredCommunicationFramework(
    name="AdvancedFramework",
    max_loops=5,
    enable_structured_communication=True,
    enable_hierarchical_evaluation=True,
    shared_memory=True,
    model_name="claude-sonnet-4-6",
    verbose=True
)

# Run the task (loops up to the max_loops set above)
result = framework.run(
    "Analyze the impact of climate change on global agriculture"
)

How It Works

The framework operates through a structured multi-phase workflow:

Generation Phase: The first generator creates content for the task
Evaluation Phase: Each evaluator scores the content on one criterion, and the evaluation supervisor summarizes the results
Refinement Phase: The first refiner improves the content using every evaluator’s feedback and score
Stopping rule: If the loop’s average evaluation score is 8.0 or higher, the workflow stops
Iteration: Otherwise steps 1-3 repeat on the original task, up to max_loops times

Structured Communication Protocol

The framework uses a formal communication protocol with three components:

Message (M_ij): Direct communication between agents
Background (B_ij): Contextual information shared between agents
Intermediate Output (I_ij): Partial results passed between workflow stages

Features

Structured Communication: Formal protocol for inter-agent messaging
Hierarchical Evaluation: Several evaluators, each on its own criterion, summarized by an evaluation supervisor
Score-based Early Stop: Stops looping once the average parsed evaluation score reaches 8.0
Specialized Agents: Purpose-built agent classes for each workflow role
Local Models: Route the default agents through Ollama with use_ollama=True

Source Code

View the source code on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

HierarchicalStructuredCommunicationFramework

Overview

Installation

Key Components

Agent Classes

Main Framework

Attributes

Methods

run()

step()

send_structured_message()

run_hierarchical_evaluation()

Usage Examples

Quick Start

Basic Usage with Default Supervisor

Advanced Configuration

How It Works

Structured Communication Protocol

Features

Source Code

​Overview

​Installation

​Key Components

​Agent Classes

​Main Framework

​Attributes

​Methods

​run()

​step()

​send_structured_message()

​run_hierarchical_evaluation()

​Usage Examples

​Quick Start

​Basic Usage with Default Supervisor

​Advanced Configuration

​How It Works

​Structured Communication Protocol

​Features

​Source Code

Overview

Installation

Key Components

Agent Classes

Main Framework

Attributes

Methods

run()

step()

send_structured_message()

run_hierarchical_evaluation()

Usage Examples

Quick Start

Basic Usage with Default Supervisor

Advanced Configuration

How It Works

Structured Communication Protocol

Features

Source Code