Multi-Agent Systems
When creating Agents in AWS Bedrock, the simple way is to have one agent handle all of the requests, but a more effective way to deploy agents is to use multiple specialised agents, that each have different prompts, access to your data, their own tooling. You can even use different models.
Having this separation allows you to have greater control over how your agents respond, and you can have teams work on their own specific area of expertise, and share these specialised agents over multiple implementations.
As an example, you may have a specialised agent that has specialised knowledge of Infrastructure as code, and may even have access to your code or coding standards, and instead of having that full prompt and backing documentation in your main agent, its separated out to its own agent, and when a request needs that information, the call is delegated to the specialised agent. When you start adding more and more specific uses to your agent, having it all in one prompt can become unwieldy. Every request loads all of the prompts for all of your specific areas, even if the response doesn't actually need it. So you may load information about Infrastructure as Code, even if the question is about accounting for example. Having specialised agents means this knowledge is only used when it is required.
Architecture
Handling multiple agents is done through an agent that knows about each of the specialised agents and is able to route between them. Each of the agents are created in the same way as usual, with an orchestration agent that has knowledge of each of the specialised agents.
Setting up the agents
In this example, we'll create an orchestration agent that will answer most questions, but pass on any AWS queries to an AWS specialise agent. At the moment this has a different prompt, but could also have access to AWS technical documentation via an MCP server for example.
First we set up some initial config values for our script, the main part here is the configs for each of the agents. These are the details passed into a bedrock_agent.create_agent call.
REGION = os.getenv('AWS_REGION', 'us-east-1')
AGENT_ROLE_ARN = os.getenv('AGENT_ROLE_ARN') # Must be set in environment
FOUNDATION_MODEL = "anthropic.claude-3-5-sonnet-20241022-v2:0"
# Agent configurations
ORCHESTRATOR_CONFIG = {
"name": "simple-orchestrator",
"instruction": """You are an orchestration agent. Analyze user queries and determine
if they require technical AWS expertise. For technical questions, delegate to the
specialist. For general questions, answer directly."""
}
SPECIALIST_CONFIG = {
"name": "aws-specialist",
"instruction": """You are an AWS technical specialist. Provide detailed, production-ready
guidance on AWS services, architecture, best practices, and troubleshooting. Include
code examples when relevant."""
}
# File to store agent IDs
CONFIG_FILE = "agent_ids.json"
Deploying the Agents
In our example, we will set up a class that is used to deploy the 2 agents.
This class can be called to create the agents, then deploy those agents. It also contains functions to clean up agents.
When creating the agents we do the following:
- call create_agent with the agent name, foundation model name, instructions, role and description
- call prepare_agent with the returned agent id
- create an agent alias (such as latest) for this version
We then store the agent id and agent alias id for the agents we have created.
- In our class we call this function twice, once for each agent, and then store the agent id and alias id in a config file for later use.
class AgentDeployer:
"""Handles one-time creation of Bedrock agents"""
def __init__(self):
self.bedrock_agent = boto3.client('bedrock-agent', region_name=REGION)
def create_agent(self, name: str, instruction: str) -> Dict[str, str]:
"""
Create a Bedrock agent
Returns:
Dict with agent_id and agent_alias_id
"""
print(f"Creating agent: {name}")
# Create the agent
response = self.bedrock_agent.create_agent(
agentName=name,
foundationModel=FOUNDATION_MODEL,
instruction=instruction,
agentResourceRoleArn=AGENT_ROLE_ARN,
description=f"Agent: {name}"
)
agent_id = response['agent']['agentId']
print(f" ✓ Agent created: {agent_id}")
# Prepare agent (required before use)
self.bedrock_agent.prepare_agent(agentId=agent_id)
print(f" ✓ Agent prepared")
# Create alias for stable endpoint
alias_response = self.bedrock_agent.create_agent_alias(
agentId=agent_id,
agentAliasName='latest'
)
alias_id = alias_response['agentAlias']['agentAliasId']
print(f" ✓ Alias created: {alias_id}")
return {
'agent_id': agent_id,
'agent_alias_id': alias_id
}
def deploy_all(self) -> Dict:
"""Deploy all agents and save configuration"""
print("="*70)
print("DEPLOYMENT PHASE: Creating Bedrock Agents")
print("="*70)
if not AGENT_ROLE_ARN:
raise ValueError("AGENT_ROLE_ARN environment variable must be set")
# Create specialist
specialist = self.create_agent(
SPECIALIST_CONFIG['name'],
SPECIALIST_CONFIG['instruction']
)
time.sleep(3) # Wait for specialist to be ready
# Create orchestrator
orchestrator = self.create_agent(
ORCHESTRATOR_CONFIG['name'],
ORCHESTRATOR_CONFIG['instruction']
)
# Save configuration
config = {
'orchestrator': orchestrator,
'specialist': specialist
}
with open(CONFIG_FILE, 'w') as f:
json.dump(config, f, indent=2)
print(f"\n✓ Configuration saved to {CONFIG_FILE}")
print("="*70)
return config
def delete_agent(self, agent_id: str, alias_id: str, name: str):
"""Delete an agent and its alias"""
print(f"Deleting agent: {name}")
try:
# Delete alias
self.bedrock_agent.delete_agent_alias(
agentId=agent_id,
agentAliasId=alias_id
)
print(f" ✓ Alias deleted")
# Delete agent
self.bedrock_agent.delete_agent(agentId=agent_id)
print(f" ✓ Agent deleted")
except Exception as e:
print(f" ✗ Error: {e}")
def delete_all(self):
"""Delete all agents"""
print("="*70)
print("CLEANUP: Deleting Bedrock Agents")
print("="*70)
if not os.path.exists(CONFIG_FILE):
print("No configuration file found")
return
with open(CONFIG_FILE, 'r') as f:
config = json.load(f)
# Delete orchestrator
if 'orchestrator' in config:
self.delete_agent(
config['orchestrator']['agent_id'],
config['orchestrator']['agent_alias_id'],
'orchestrator'
)
# Delete specialist
if 'specialist' in config:
self.delete_agent(
config['specialist']['agent_id'],
config['specialist']['agent_alias_id'],
'specialist'
)
# Remove config file
os.remove(CONFIG_FILE)
print(f"\n✓ Configuration file removed")
print("="*70)
Using the agents
Now that the agent backends are deployed on bedrock, we can set up a wrapper class to call these agents. For the specialist agent, this is straightforward.
Our class loads in the agent_id and agent_alias_id from our config, and has an invoke function.
This invoke function passes a query to the specialised agent using the loaded config.
class SpecialistAgent:
"""Runtime wrapper for specialist agent"""
def __init__(self, agent_id: str, agent_alias_id: str):
self.agent_id = agent_id
self.agent_alias_id = agent_alias_id
self.runtime = boto3.client('bedrock-agent-runtime', region_name=REGION)
def invoke(self, query: str, session_id: str) -> Dict[str, Any]:
"""Invoke the specialist agent"""
response = self.runtime.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=session_id,
inputText=query
)
# Parse streaming response
text = ""
for event in response['completion']:
if 'chunk' in event:
chunk = event['chunk']
if 'bytes' in chunk:
text += chunk['bytes'].decode('utf-8')
return {
'response': text,
'success': True
}
Orchestrator Agent
The orchestrator agent runs slightly differently, where it will decide if the specialist is required, and send the query to the specialised agent or answer the query itself.
The class still calls invoke_agent itself, or calls the invoke within the specialised agent, depending on the query.
class Orchestrator:
"""
Orchestrator implementing Strands SDK patterns
Strands Pattern:
1. Analyze query to determine if delegation is needed
2. Delegate to specialist if technical
3. Synthesize and return response
"""
def __init__(self, agent_id: str, agent_alias_id: str):
self.agent_id = agent_id
self.agent_alias_id = agent_alias_id
self.runtime = boto3.client('bedrock-agent-runtime', region_name=REGION)
self.specialist = None
def register_specialist(self, specialist: SpecialistAgent):
"""Register the specialist agent (Strands pattern)"""
self.specialist = specialist
def invoke(self, query: str, session_id: Optional[str] = None) -> Dict[str, Any]:
"""
Orchestrate the query using Strands delegation pattern
"""
if not session_id:
session_id = str(uuid.uuid4())
# Strands Pattern: Analyze query
is_technical = self._is_technical_query(query)
if is_technical and self.specialist:
# Strands Pattern: Delegate to specialist
print(f"\n→ Delegating to specialist agent")
result = self.specialist.invoke(query, session_id)
result['delegated'] = True
return result
else:
# Direct orchestrator response
print(f"\n→ Orchestrator handling directly")
return self._invoke_orchestrator(query, session_id)
def _is_technical_query(self, query: str) -> bool:
"""Determine if query requires specialist (Strands pattern)"""
technical_keywords = [
'aws', 'lambda', 'ec2', 's3', 'dynamodb', 'deploy',
'architecture', 'infrastructure', 'cloudformation', 'terraform',
'docker', 'kubernetes', 'ci/cd', 'security', 'iam'
]
return any(keyword in query.lower() for keyword in technical_keywords)
def _invoke_orchestrator(self, query: str, session_id: str) -> Dict[str, Any]:
"""Invoke the orchestrator agent directly"""
response = self.runtime.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=session_id,
inputText=query
)
# Parse streaming response
text = ""
for event in response['completion']:
if 'chunk' in event:
chunk = event['chunk']
if 'bytes' in chunk:
text += chunk['bytes'].decode('utf-8')
return {
'response': text,
'success': True,
'delegated': False
}
Code to call the agents
To complete the example, we can set up a MultiAgent class that will load and use the orchestrator and specialised agent.
This isn't specifically necessary to use the multi-agent setup, it is just as an example of how you can call each of the agents.
This class will do the following:
- function to load the configs
- register the specialised agent in the class
- load the specialised agent class with the relevant agent ids
- load the orchestration agent class with the relevant agent ids
- When a query is run, call the orchestrator invoke function to handle the query
This allows us to then have a main function where we can call the AgentDeployer to create or destroy the agents, using the AgentDeployer class if we pass in specific command line parameters.
Otherwise, we can call the MultiAgentSytem class, and initialise our agents, then pass in our query (either as a passed in parameter, or we can run in interactive mode, which we've set up so that users can type in a query, then once we hit enter, the query is sent to the orchestrator, and continues accepting requests until you type quit, exit, or q.
class MultiAgentSystem:
"""Complete multi-agent system"""
def __init__(self):
self.config = self._load_config()
self.orchestrator = None
self.specialist = None
def _load_config(self) -> Dict:
"""Load agent configuration"""
if not os.path.exists(CONFIG_FILE):
raise FileNotFoundError(
f"{CONFIG_FILE} not found. Run with --create first."
)
with open(CONFIG_FILE, 'r') as f:
return json.load(f)
def initialize(self):
"""Initialize the system with pre-deployed agents"""
print("="*70)
print("RUNTIME PHASE: Loading Pre-Deployed Agents")
print("="*70)
# Initialize specialist
specialist_config = self.config['specialist']
self.specialist = SpecialistAgent(
specialist_config['agent_id'],
specialist_config['agent_alias_id']
)
print(f"✓ Specialist loaded: {specialist_config['agent_id']}")
# Initialize orchestrator
orchestrator_config = self.config['orchestrator']
self.orchestrator = Orchestrator(
orchestrator_config['agent_id'],
orchestrator_config['agent_alias_id']
)
print(f"✓ Orchestrator loaded: {orchestrator_config['agent_id']}")
# Register specialist with orchestrator (Strands pattern)
self.orchestrator.register_specialist(self.specialist)
print(f"✓ Specialist registered with orchestrator")
print("="*70)
def query(self, user_query: str) -> str:
"""Process a query through the system"""
print(f"\nQuery: {user_query}")
print("-"*70)
result = self.orchestrator.invoke(user_query)
print(f"\n{result['response']}")
print("-"*70)
return result['response']
def interactive(self):
"""Run in interactive mode"""
print("\nInteractive Mode - Type 'quit' to exit")
print("="*70)
while True:
try:
query = input("\n> ").strip()
if query.lower() in ['quit', 'exit', 'q']:
print("\nGoodbye!")
break
if not query:
continue
self.query(query)
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
print(f"\nError: {e}")
# =============================================================================
# MAIN
# =============================================================================
def main():
parser = argparse.ArgumentParser(description='Simple Multi-Agent System')
parser.add_argument('--create', action='store_true',
help='Create Bedrock agents (deployment phase)')
parser.add_argument('--delete', action='store_true',
help='Delete Bedrock agents')
parser.add_argument('--query', type=str,
help='Single query to process')
parser.add_argument('--interactive', action='store_true',
help='Interactive mode')
args = parser.parse_args()
try:
if args.create:
# DEPLOYMENT PHASE
deployer = AgentDeployer()
deployer.deploy_all()
print("\n✓ Agents created! Now run with --interactive or --query")
elif args.delete:
# CLEANUP PHASE
deployer = AgentDeployer()
deployer.delete_all()
print("\n✓ Agents deleted")
else:
# RUNTIME PHASE
system = MultiAgentSystem()
system.initialize()
if args.query:
system.query(args.query)
elif args.interactive:
system.interactive()
else:
print("\nUse --interactive for interactive mode or --query 'your question'")
print("Run --help for all options")
except FileNotFoundError as e:
print(f"\nError: {e}")
print("\nRun with --create first to deploy agents")
except Exception as e:
print(f"\nError: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
Top comments (0)