KEY INSIGHT
Role specialization enables focused capability building, but requires careful boundary design to avoid both under-specialization (agents do everything poorly) and over-specialization (agents lack context for decisions).
Multi-agent systems derive power from specialization. Each agent focuses on a narrow domain, developing deep capabilities within that scope. A code analysis agent builds expertise in static analysis, AST traversal, and code style detection. A data visualization agent specializes in chart selection, color theory, and accessibility compliance.
Specialization boundaries require careful design. Poor boundaries create either gaps (tasks fall between agent responsibilities) or redundancy (multiple agents handle overlapping concerns). Boundaries should align with natural domain divisions and match how humans conceptualize task categories.
Each role requires defined inputs, outputs, and behavioral expectations. Input specification tells the agent what requests it receives and what context accompanies them. Output specification defines deliverables and their structure. Behavioral expectations encode constraints—safety requirements, quality thresholds, latency limits.
Role documentation includes not just capabilities but also limitations. An agent that cannot handle certain input formats should communicate that clearly. Agents that provide probabilistic outputs should quantify uncertainty. Clear limitation communication prevents misuse.
Role composition enables complex workflows. A pipeline might chain data extraction, transformation, and loading agents. A parallel execution might run multiple analysis agents on the same input and merge results. Role composition requires clear contracts between roles.
Agents operating in specialized roles may still need general context. A product recommendation agent needs user history, inventory state, and business rules—not just product knowledge. Specialization does not mean isolation; agents must access cross-cutting concerns.
```python
from dataclasses import dataclass
from typing import Optional, Any, Protocol
from abc import ABC, abstractmethod
@dataclass
class AgentCapability:
"""Describes what an agent can do"""
input_types: list[str] # Supported input content types
output_type: str # What the agent produces
limitations: list[str] # Known constraints
quality_bounds: dict[str, tuple[float, float]] # Min/max quality metrics
@dataclass
class AgentRole(ABC):
"""Base class for agent roles"""
name: str
description: str
capabilities: list[AgentCapability]
@abstractmethod
async def process(self, request: dict) -> dict:
"""Handle incoming request"""
pass
@abstractmethod
async def describe_limits(self) -> str:
"""Return human-readable limitation description"""
pass
class DataExtractionRole(AgentRole):
"""Specialized role for extracting structured data from unstructured sources"""
def __init__(self, model):
self.model = model
def __init__(self, model):
self.name = "data_extractor"
self.description = "Extracts structured data from unstructured text"
self.capabilities = [
AgentCapability(
input_types=["text/plain", "text/html", "application/pdf"],
output_type="application/json",
limitations=[
"Cannot process images directly (use OCR agent first)",
"Maximum input length: 50,000 tokens",
"Confidence below 0.7 triggers human review flag"
],
quality_bounds={
"precision": (0.85, 1.0),
"recall": (0.75, 0.95)
}
)
]
self.model = model
async def process(self, request: dict) -> dict:
schema = request.get("target_schema")
source = request.get("source_text")
extraction_prompt = f"""Extract data according to this schema:
{schema}
Source text:
{source}
Return JSON matching the schema exactly. Include confidence scores per field."""
result = await self.model.generate(extraction_prompt)
# Post-process to include metadata
return {
"extracted_data": result,
"confidence": self._calculate_confidence(result),
"schema_version": request.get("schema_version"),
"agent": self.name
}
def _calculate_confidence(self, result: dict) -> float:
# Simplified confidence calculation
field_count = len(result)
filled_fields = sum(1 for v in result.values() if v is not None)
return filled_fields / field_count if field_count > 0 else 0.0
async def describe_limits(self) -> str:
return (
"This agent extracts structured data from text sources. "
"It cannot process images directly—use OCR preprocessing first. "
"Inputs exceeding 50,000 tokens are truncated. "
"Results with confidence below 0.7 are flagged for human review."
)
class DataValidationRole(AgentRole):
"""Specialized role for validating extracted data against business rules"""
def __init__(self, model):
self.name = "validator"
self.description = "Validates data against business rules and constraints"
self.capabilities = [
AgentCapability(
input_types=["application/json"],
output_type="application/json",
limitations=[
"Requires explicit validation rules in request",
"Cannot infer business logic—rules must be provided"
],
quality_bounds={
"rule_coverage": (0.9, 1.0),
"false_positive_rate": (0.0, 0.05)
}
)
]
self.model = model
async def process(self, request: dict) -> dict:
data = request.get("data")
rules = request.get("validation_rules", [])
validation_results = []
for rule in rules:
result = self._apply_rule(data, rule)
validation_results.append(result)
passed = sum(1 for r in validation_results if r["passed"])
return {
"validation_results": validation_results,
"passed_count": passed,
"failed_count": len(rules) - passed,
"overall_valid": passed == len(rules),
"agent": self.name
}
def _apply_rule(self, data: dict, rule: dict) -> dict:
field = rule["field"]
condition = rule["condition"]
value = data.get(field)
# Rule evaluation logic
if condition["type"] == "range":
passed = condition["min"] <= value <= condition["max"]
elif condition["type"] == "enum":
passed = value in condition["allowed"]
else:
passed = True
return {
"field": field,
"passed": passed,
"actual_value": value,
"expected": condition
}
async def describe_limits(self) -> str:
return (
"This agent validates data against provided rules. "
"It does not infer business rules—all validation criteria must be "
"explicitly defined in the request."
)
# Role composition example
class PipelineOrchestrator:
def __init__(self, extractor: DataExtractionRole, validator: DataValidationRole):
self.extractor = extractor
self.validator = validator
async def process(self, source_text: str, schema: dict, rules: list) -> dict:
# Stage 1: Extraction
extraction_request = {
"source_text": source_text,
"target_schema": schema
}
extraction_result = await self.extractor.process(extraction_request)
if extraction_result["confidence"] < 0.7:
return {
"status": "review_required",
"reason": "low_confidence",
"extracted_data": extraction_result["extracted_data"]
}
# Stage 2: Validation
validation_request = {
"data": extraction_result["extracted_data"],
"validation_rules": rules
}
validation_result = await self.validator.process(validation_request)
return {
"status": "complete" if validation_result["overall_valid"] else "failed",
"data": extraction_result["extracted_data"],
"validation": validation_result
}
```