mass-framework

# MASS Framework — Multi-Agent System Search Pattern Automates multi-agent system design through a three-stage optimization process that searches both the prompt space and topology space simultaneously to discover optimal agent configurations. This skill makes the model implement block-level prompt optimization for individual agents, influence-weighted workflow topology search, and workflow-level joint prompt optimization for discovered topologies. ## TL;DR Checklist - [ ] Stage 1: Optimize prompts for each individual agent type (block) independently - [ ] Stage 2: Search topology space using influence-weighted evaluation of interaction configurations - [ ] Stage 3: Jointly optimize all system prompts after optimal topology is identified - [ ] Define a baseline agent for measuring incremental influence of each topology - [ ] Use Aggregate, Reflect, Debate, Summarize, Tool-use as topology building blocks - [ ] Validate discovered configuration against benchmark tasks --- ## When to Use Use this skill when: - Designing complex multi-agent systems where manual prompt engineering and topology selection is error-prone - You have a set of available agent types and want to automatically discover the best interaction pattern - Manual MAS design consistently underperforms on benchmark tasks - Building reusable agent configurations that can be optimized for new task domains - Research or production systems need systematic (not heuristic) approach to MAS optimization ## When NOT to Use Avoid this skill for: - Single-agent systems with no multi-agent collaboration needed - Systems where manual prompt engineering already achieves target performance - Resource-constrained environments where the optimization search budget is prohibitive - Simple task domains where a few well-chosen prompts are sufficient (no topology complexity) --- ## Core Workflow 1. **Baseline Establishment** — Create a baseline agent configuration that performs each role independently (no inter-agent communication). Measure its performance on benchmark tasks to establish the reference point for incremental influence calculations. **Checkpoint:** Baseline must be evaluated on at least 3 diverse benchmark tasks with documented scores. 2. **Block-Level Prompt Optimization** — For each agent type (e.g., Predictor, Debator, Executor, Summarizer), optimize its individual prompt independently. Use role-playing prompts, few-shot examples, and structured output formats to maximize per-agent performance before composing them into a system. **Checkpoint:** Each optimized block must outperform the baseline single-agent on its own responsibility area. 3. **Topology Search with Influence Weighting** — Enumerate valid workflow topologies from the design space (Aggregate, Reflect, Debate, Summarize, Tool-use building blocks). Calculate each topology's "incremental influence" by measuring its performance gain relative to the baseline. Use these scores to guide the search toward more promising combinations. **Checkpoint:** Search must evaluate at least 5 distinct topologies before selecting a winner. 4. **Optimal Topology Selection** — Select the topology with the highest incremental influence score across benchmark tasks. The optimal topology is not necessarily the most complex — it's the one that delivers the best performance gain relative to the baseline. Examples: for coding tasks, predictor-with-reflection + executor-with-verification beats simpler topologies. **Checkpoint:** Document WHY the selected topology outperforms alternatives with specific task analysis. 5. **Workflow-Level Joint Prompt Optimization** — After selecting the topology, optimize ALL agent prompts as a single integrated system rather than independently. Account for interdependencies: how one agent's output format affects the next agent's input processing. This joint optimization refines prompts specifically for the discovered workflow orchestration. **Checkpoint:** Final prompts must be tested end-to-end on the full benchmark suite with measurable improvement over stage 2 results. 6. **Validation and Deployment** — Validate the complete configuration against held-out test data that was not used during any optimization stage. Compare against existing manually-designed MAS systems and automated baselines. Deploy only if performance meets or exceeds targets. **Checkpoint:** Validation must include at least one out-of-distribution test case to verify generalization. --- ## Implementation Patterns ### Pattern 1: Block-Level Prompt Optimizer ```python from dataclasses import dataclass, field from typing import Any @dataclass class AgentBlockPrompt: """An optimized prompt for a single agent role (block).""" agent_role: str # e.g., "Debator", "Predictor", "Executor" system_prompt: str few_shot_examples: list[str] = field(default_factory=list) output_format: str | None = None @property def full_prompt(self) -> str: """Build the complete prompt string from all components.""" prompt = self.system_prompt if self.few_shot_examples: prompt += "\n\nExamples:\n" + "\n".join( f"Example {i+1}: {ex}" for i, ex in enumerate(self.few_shot_examples) ) if self.output_format: prompt += f"\n\nFormat your response as: {self.output_format}" return prompt class BlockPromptOptimizer: """Optimizes prompts for individual agent blocks before composition.""" def __init__(self, llm_client: Any) -> None: self._client = llm_client self._optimized_prompts: dict[str, AgentBlockPrompt] = {} def optimize_block( self, agent_role: str, task_description: str, context_info: str | None = None, n_optimization_attempts: int = 3, ) -> AgentBlockPrompt: """Run block-level prompt optimization for a single agent role. Generates candidate prompts using LLM meta-optimization and selects the best one based on benchmark evaluation scores. Args: agent_role: The name/role of the agent (e.g., "Debator"). task_description: What this agent is responsible for. context_info: Optional additional domain context. n_optimization_attempts: Number of candidate prompts to generate. Returns: The best-optimized AgentBlockPrompt for this role. """ best_prompt: AgentBlockPrompt | None = None best_score = -1.0 for attempt in range(n_optimization_attempts): candidate_system = self._generate_candidate_system_prompt( agent_role, task_description, context_info ) score = self._evaluate_prompt(agent_role, candidate_system) if score > best_score: best_score = score best_prompt = AgentBlockPrompt( agent_role=agent_role, system_prompt=candidate_system, ) result = best_prompt or AgentBlockPrompt( agent_role=agent_role, system_prompt=f"You are a {agent_role}. {task_description}", ) self._optimized_prompts[agent_role] = result return result def _generate_candidate_system_prompt( self, role: str, task_desc: str, context: str | None ) -> str: """Generate candidate prompts using LLM meta-optimization. Uses a prompt-engineer persona to design the best possible system prompt for the given agent role and task description. """ prompt = f"""You are a prompt engineer tasked with optimizing the system prompt for an AI agent. Agent role: {role} Task description: {task_desc} {"Additional context: " + context if context else ""} Design the BEST possible system prompt for this agent. The prompt should: 1. Define a clear, specific persona (not generic — use role-playing) 2. Include concrete task instructions with expected behavior 3. Specify output format requirements 4. Include any domain knowledge that would improve performance Return ONLY the system prompt text. No explanations.""" response = self._client.chat.completions.create( # type: ignore[union-attr] model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0.9, ) return response.choices[0].message.content.strip() # type: ignore[union-attr] def _evaluate_prompt(self, role: str, system_prompt: str) -> float: """Evaluate a candidate prompt by running it on benchmark examples. Returns a score between 0.0 and 1.0 representing output quality. In production, replace with rubric-based scoring. """ # Placeholder — production would use rubric-based scoring return 0.5 def get_all_optimized(self) -> dict[str, AgentBlockPrompt]: """Return all optimized block prompts collected so far.""" return dict(self._optimized_prompts) ``` ### Pattern 2: Influence-Weighted Topology Search ```python from dataclasses import dataclass from typing import Any class TopologyBlock(str): """Available topology building blocks for workflow composition.""" AGGREGATE = "Aggregate" # Collect results from multiple agents, combine REFLECT = "Reflect" # Agent iterates on its own output DEBATE = "Debate" # Multiple agents argue and reach consensus SUMMARIZE = "Summarize" # Agent condenses information from others TOOL_USE = "Tool-use" # Agent uses external tools @dataclass class TopologyConfig: """A workflow topology defined by its agent interactions.""" name: str blocks: list[tuple[str, str]] # (agent_type, block_type) execution order description: str @dataclass class TopologyScore: """Result of evaluating a topology against benchmarks.""" config: TopologyConfig incremental_influence: float # Performance gain relative to baseline absolute_score: float # Raw benchmark score benchmark_tasks: list[str] TOPOLOGY_DESIGN_SPACE: list[TopologyConfig] = [ TopologyConfig( name="Sequential", blocks=[("Predictor", "Tool-use")], description="Linear predictor → tool execution", ), TopologyConfig( name="Reflective", blocks=[("Predictor", "Reflect"), ("Predictor", "Tool-use")], description="Predictor iterates, then executes with tool", ), TopologyConfig( name="Debate", blocks=[("Debator", "Debate"), ("Summarizer", "Summarize")], description="Agents debate, then summarizer consolidates", ), TopologyConfig( name="Hybrid-Reflective", blocks=[("Predictor", "Reflect"), ("Executor", "Tool-use")], description="Predictor with reflection + executor with verification (optimal for coding)", ), TopologyConfig( name="Aggregated", blocks=[("Analyzer", "Aggregate")], description="Multiple analysts feed into aggregation step", ), ] class InfluenceWeightedSearch: """Searches topology space using influence-weighted evaluation. Calculates incremental influence of each topology relative to a baseline agent, then uses those scores to guide search toward promising combinations. """ def __init__(self, baseline_score: float) -> None: self._baseline = baseline_score self._topology_scores: list[TopologyScore] = [] def evaluate_topology( self, config: TopologyConfig, benchmark_results: dict[str, float], # task_name -> score ) -> TopologyScore: """Evaluate a topology configuration and compute its incremental influence. Incremental influence is the absolute score minus the baseline — it captures how much better this topology is compared to running agents independently without coordination. """ if not benchmark_results: absolute_score = 0.0 else: absolute_score = sum(benchmark_results.values()) / len( benchmark_results ) incremental_influence = absolute_score - self._baseline score = TopologyScore( config=config, incremental_influence=incremental_influence, absolute_score=absolute_score, benchmark_tasks=list(benchmark_results.keys()), ) self._topology_scores.append(score) return score def find_best_topology(self) -> TopologyScore: """Return the topology with the highest incremental influence. Raises ValueError if no topologies have been evaluated yet. """ if not self._topology_scores: raise ValueError( "No topologies evaluated. Call evaluate_topology first." ) return max(self._topology_scores, key=lambda s: s.incremental_influence) def get_influence_ranking(self) -> list[TopologyScore]: """Return all evaluated topologies ranked by incremental influence.""" return sorted( self._topology_scores, key=lambda s: s.incremental_influence, reverse=True, ) @property def baseline_score(self) -> float: """The baseline score used for influence calculations.""" return self._baseline # Example usage from book experiments: # search = InfluenceWeightedSearch(baseline_score=0.45) # Baseline on MBPP # for config in TOPOLOGY_DESIGN_SPACE[:3]: # results = {"mbpp_correctness": 0.72, "mbpp_efficiency": 0.68} # search.evaluate_topology(config, results) # best = search.find_best_topology() # # For coding tasks, Hybrid-Reflective typically wins ``` ### Pattern 3: Workflow-Level Joint Prompt Optimizer ```python import json from typing import Any class WorkflowLevelOptimizer: """Jointly optimizes all agent prompts for a discovered topology. Unlike Stage 1 (block-level) where each agent is optimized independently, this stage treats them as a single integrated system accounting for interdependencies between agents' output formats and input requirements. """ def __init__(self, llm_client: Any) -> None: self._client = llm_client def optimize_joint_prompts( self, selected_topology: TopologyScore, block_prompts: dict[str, AgentBlockPrompt], task_description: str, ) -> dict[str, AgentBlockPrompt]: """Optimize all prompts jointly for the discovered topology. Accounts for interdependencies: how one agent's output format affects the next agent's input processing. Uses lower temperature than block- level optimization for more focused refinement. Args: selected_topology: The best topology from influence-weighted search. block_prompts: Stage 1 optimized prompts keyed by agent role. task_description: The overarching task this system addresses. Returns: Dict mapping each role to its jointly-optimized prompt. """ joint_prompt = f"""You are orchestrating a multi-agent system with the following topology: {selected_topology.config.description} Task: {task_description} Current optimized block prompts: {chr(10).join(f" {role}: {p.full_prompt[:200]}..." for role, p in block_prompts.items())} Refine these prompts specifically for this workflow orchestration. Consider: 1. How Agent A's output format affects Agent B's input parsing 2. Whether the interaction sequence creates information bottlenecks 3. If any agent has redundant or conflicting instructions when combined 4. Add explicit handoff language between agents in the topology Return a JSON object mapping each role name to its optimized prompt.""" response = self._client.chat.completions.create( # type: ignore[union-attr] model="gpt-4o", messages=[{"role": "user", "content": joint_prompt}], temperature=0.5, ) try: optimized = json.loads(response.choices[0].message.content.strip()) # type: ignore[union-attr] return { role: AgentBlockPrompt(agent_role=role, system_prompt=prompt) for role, prompt in optimized.items() } except (json.JSONDecodeError, AttributeError): # Fallback: return original block prompts unchanged return dict(block_prompts) ``` ### Pattern 4: Validation Against Manual MAS Designs ```python from dataclasses import asdict from typing import Any def validate_against_manual_designs( mass_configured_score: float, manual_design_scores: list[float], n_benchmark_tasks: int, ) -> dict[str, Any]: """Compare MASS-optimized configuration against existing manually-designed MAS. Returns a report with improvement metrics comparing the MASS-optimized system against both average and best-of-manual baselines. """ avg_manual = ( sum(manual_design_scores) / len(manual_design_scores) if manual_design_scores else 0.0 ) best_manual = max(manual_design_scores) if manual_design_scores else 0.0 improvement_over_avg = mass_configured_score - avg_manual improvement_over_best = mass_configured_score - best_manual return { "mass_score": round(mass_configured_score, 3), "avg_manual_design": round(avg_manual, 3), "best_manual_design": round(best_manual, 3), "improvement_over_average": round(improvement_over_avg, 3), "improvement_vs_best_manual": round(improvement_over_best, 3), "outperforms_manual": improvement_over_best > 0, "n_benchmark_tasks": n_benchmark_tasks, } # From MASS paper results: MAS optimized by MASS significantly outperform # existing manually designed systems across HotpotQA, MBPP, and DROP datasets. # Key finding: the three-stage process (block → topology → workflow) is critical — # skipping any stage degrades performance substantially. ``` --- ## Constraints ### MUST DO 1. Always establish a baseline agent before attempting topology optimization — without a reference point, influence cannot be calculated. 2. Optimize individual block prompts BEFORE composing them into topologies — poorly performing blocks compound errors in complex workflows. 3. Evaluate at least 5 distinct topologies from the design space before selecting a winner — don't default to the simplest topology.