Supercharging Amazon Bedrock Flows with AWS Multi-Agent Orchestrator

I've been working with Amazon Bedrock Flows lately - it's this really cool feature from Bedrock that lets you visually design your AI workflows. If you're like me and prefer visuals over code, you'll love it! Drag, drop, connect nodes, and boom - you've got an AI flow running.

But after using it for a while, I hit some limitations that Bedrock Flows alone couldn't solve.

Two big ones stood out:

No memory across conversations (each request starts fresh)
No built-in way to manage multiple flows

Let me show you how I solved these using the Multi-Agent Orchestrator framework.

The Memory Problem 🧠

As of December 3, 2024, Amazon Bedrock Flow doesn't have built-in memory support. That means if you're building a conversational AI, your flow can't remember what was said earlier in the conversation - not ideal!

But here's where the Bedrock Flows Agent comes to the rescue. It adds chat history support to your flows, making them actually conversational. Pretty neat, right?

Case 1: Single Flow Agent 🎯

For our example we have created a tech agent flow as followed:

Image not found

The tech_agent prompt node will extract history and question from the Flow input.

Here is the prompt we used for this tech_agent node:

1
2
3
4
5
6
7
8
9
10
You are an AWS Services Expert Agent, a specialized AI assistant 
with comprehensive knowledge of AWS cloud services. 
Your primary objective is to provide expert-level guidance, 
architectural recommendations, and technical solutions 
for complex cloud infrastructure challenges.

Conversation History:
{{history}}

Answer the following question: {{question}}

In this example we are not going to use a classifier since we only have one agent and we don’t need to use the full orchestrator logic.

Here is a diagram that shows how everything connects together within the multi-agent-orchestrator:

Image not found

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator
from multi_agent_orchestrator.classifiers import ClassifierResult
from multi_agent_orchestrator.agents import (
BedrockFlowsAgent, 
BedrockFlowsAgentOptions)

orchestrator = MultiAgentOrchestrator()

# implement a custom Flow input encoder to include our question and history as object
def flow_input_encoder(agent:Agent, input: str, **kwargs) -> Any:
    global tech_support_flow
    if agent == tech_support_flow:
        chat_history:List[ConversationMessage] = kwargs.get('chat_history', [])

        chat_history_string = '\n'.join(f"{message.role}:{message.content[0].get('text')}" for message in chat_history)

        return {
            "question": input,
            "history":chat_history_string
        }
    else:
        return input # return input as a string

# Create a tech agent flow
tech_support_flow = BedrockFlowsAgent(BedrockFlowsAgentOptions(
    name="tech-agent",
    description="Handles technical requests about AWS services",
    flowIdentifier='YOUR_FLOW_ID',
    flowAliasIdentifier='YOUR_FLOW_ALIAS_ID',
    flow_input_encoder=flow_input_encoder, # our custom flow input encoder
    enableTrace=True  # Good for debugging!
))
orchestrator.add_agent(tech_support_flow)

# Call it directly using agent_process_request
response = orchestrator.agent_process_request(
    input_text="What is AWS Lambda?",
    user_id="123",
    session_id="abc",
    ClassifierResult(selected_agent=tech_support_flow, confidence=1.0)
)

print(response) # AWS Lambda is a serverless computing service provided by Amazon Web Services (AWS)....

# Call it again with a follow up question
response_next = await orchestrator.agent_process_request(
    input_text="is it cost efficient?",
    user_id="123",
    session_id="abc",
    ClassifierResult(selected_agent=tech_support_flow, confidence=1.0)
)
print(response_next) # Yes, AWS Lambda can be a cost-efficient solution in many cases. Here's why...

Our tech agent flow, enhanced by chat history context, accurately recognized that the question was related to AWS Lambda. This allowed it to provide the correct response to our follow-up query.

Pretty impressive, right? 🤩

Case 2: Multiple Flows with Orchestrator 🔄

Let’s create another Bedrock Flows, for health request:

Image not found

Similar to the tech_agent flow, question and history are extracted from the Flow input node.

Here is the prompt we defined for this health_agent node:

1
2
3
4
5
6
7
You are a Health Agent, an advanced AI assistant specializing in 
comprehensive health support and personalized wellness guidance.

Here is the conversation history:
{{history}}

Answer the following question: {{question}}

When you have multiple flows, you'll want to use the orchestrator to route requests to the right flow:

Image not found

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
from multi_agent_orchestrator.agents import (
                BedrockFlowsAgent,
                BedrockFlowsAgentOptions)
 
# implement a custom Flow input encoder to include our question and history as object
def flow_input_encoder(agent:Agent, input: str, **kwargs) -> Any:
    global tech_support_flow, health_agent_flow
    if agent == tech_support_flow or \
    agent == health_agent_flow:
        chat_history:List[ConversationMessage] = kwargs.get('chat_history', [])

        chat_history_string = '\n'.join(f"{message.role}:{message.content[0].get('text')}" for message in chat_history)

        return {
            "question": input,
            "history":chat_history_string
        }
    else:
        return input # return input as a string

# Create the orchestrator
orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
    LOG_AGENT_CHAT=True,
    LOG_CLASSIFIER_OUTPUT=True
))

# Create multiple flow agents
tech_support_flow = BedrockFlowsAgent(BedrockFlowsAgentOptions(
    name="tech-agent",
    description="Handles technical requests about AWS services",
    flowIdentifier='YOUR_FLOW_ID',
    flowAliasIdentifier='YOUR_FLOW_ALIAS_ID',
    flow_input_encoder=flow_input_encoder, # our custom flow input encoder
))

health_agent_flow = BedrockFlowsAgent(BedrockFlowsAgentOptions(
    name="health-agent",
    description="Specialized in comprehensive health support and personalized wellness guidance.",
    flowIdentifier='HEALTH_FLOW_ID',
    flowAliasIdentifier='HEALTH_FLOW_ALIAS',
    flow_input_encoder=flow_input_encoder, # our custom flow input encoder
))

# Add all agents to the orchestrator
orchestrator.add_agent(tech_support_flow)
orchestrator.add_agent(health_agent_flow)

# Use it!
response = await orchestrator.route_request(
    "How much sleep do I really need for optimal health?",
    user_id="123",
    session_id="abc"
)

response = await orchestrator.route_request(
    "What is AWS Lambda?",
    user_id="123",
    session_id="abc"
)

The Result? 🎉

Now you have:

Flows that integrates conversation history
Smart routing between different flows
A clean way to mix flows with other agent types

Wrap Up

By using the Multi-Agent Orchestrator framework, we've basically given our Bedrock Flows superpowers! The flows can now remember conversations and work together seamlessly. No more awkward "sorry, what were we talking about?" moments or confusion about which flow should handle what.

Give it a try and let me know how it works for you! 🚀

Wants to know more?

Head out to our documentation with practical examples:

Any opinions in this post are those of the individual author and may not reflect the opinions of AWS.

Select your cookie preferences

Site Terms, Privacy, and more.

Supercharging Amazon Bedrock Flows with AWS Multi-Agent Orchestrator

The Memory Problem 🧠

Case 1: Single Flow Agent 🎯

Case 2: Multiple Flows with Orchestrator 🔄

The Result? 🎉

Wrap Up

Wants to know more?

Comments