Merge pull request #125 from brnaba-aws/main

Added support for Amazon Bedrock Flows

docs/astro.config.mjs

@@ -70,7 +70,8 @@ export default defineConfig({
 						{ label: 'Chain Agent', link: '/agents/built-in/chain-agent' },
 						{ label: 'Comprehend Filter Agent', link: '/agents/built-in/comprehend-filter-agent' },
 						{ label: 'Amazon Bedrock Translator Agent', link: '/agents/built-in/bedrock-translator-agent' },
-						{ label: 'Amazon Bedrock Inline Agent', link: '/agents/built-in/bedrock-inline-agent' }
+						{ label: 'Amazon Bedrock Inline Agent', link: '/agents/built-in/bedrock-inline-agent' },
+						{ label: 'Bedrock Flows Agent', link: '/agents/built-in/bedrock-flows-agent' },
 					  ]
 					},
 					{ label: 'Custom Agents', link: '/agents/custom-agents' },

docs/src/content/docs/agents/built-in/bedrock-flows-agent.mdx

@@ -0,0 +1,133 @@
+---
+title: Amazon Bedrock Flows Agent
+description: Documentation for the BedrockFlowsAgent in the Multi-Agent Orchestrator
+---
+
+## Overview
+
+The **Bedrock Flows Agent** is a specialized agent class in the Multi-Agent Orchestrator that integrates directly with [Amazon Bedrock Flows](https://aws.amazon.com/bedrock/flows/).
+This integration enables you to orchestrate your Bedrock Flows alongside other agent types (Bedrock Agent, Lex, Bedrock API...), providing a unified and flexible approach to agents orchestration.
+
+## Key Features
+
+- Support for cross-region Bedrock Flows invocation
+- Support for multiple flow input output type via flow input/output encoder/decoder callbacks
+
+## Creating a BedrockFlowsAgent
+
+### Basic Example
+
+import { Tabs, TabItem } from '@astrojs/starlight/components';
+
+<Tabs syncKey="runtime">
+  <TabItem label="TypeScript" icon="seti:typescript" color="blue">
+    ```typescript
+    import { BedrockFlowsAgent } from 'multi-agent-orchestrator';
+
+    const techFlowAgent = new BedrockFlowsAgent({
+      name: 'tech-flow-agent',
+      description: 'Specialized in AWS services',
+      flowIdentifier: 'AEXAMPLID',
+      flowAliasIdentifier: 'AEXAMPLEALIASID',
+      enableTrace:true
+    });
+    ```
+  </TabItem>
+  <TabItem label="Python" icon="seti:python">
+    ```python
+    from multi_agent_orchestrator.agents import BedrockFlowsAgent, BedrockFlowsAgentOptions
+
+    tech_flow_agent = BedrockFlowsAgent(BedrockFlowsAgentOptions(
+        name="tech-flow-agent",
+        description="Specializes in handling tech questions about AWS services",
+        flowIdentifier='AEXAMPLID',
+        flowAliasIdentifier='AEXAMPLEALIASID',
+        enableTrace=True
+    ))
+    ```
+  </TabItem>
+</Tabs>
+
+### Flow Input Encoder callback
+
+Amazon [Bedrock Flows Input](https://docs.aws.amazon.com/bedrock/latest/userguide/flows-nodes.html) supports multiple type of document output:
+- String
+- Number
+- Boolean
+- Object
+- Array
+
+In the default definition of the BedrockFlowsAgent, the output document type is a string.
+If you need to send an object, array, number or a boolean to your Flow input, you can use the flow input callback to transform the input payload based on your needs.
+
+Here are an example for TS and python:
+
+<Tabs syncKey="runtime">
+  <TabItem label="TypeScript" icon="seti:typescript" color="blue">
+    ```typescript
+
+// implementation of the custom flowInputEncoder callback
+const flowInputEncoder = (
+    agent: Agent,
+    input: string,
+    kwargs: {
+        userId?: string,
+        sessionId?: string,
+        chatHistory?: any[],
+        [key: string]: any  // This allows any additional properties
+      }
+) => {
+      if (agent.name == 'tech-flow-agent'){
+          return {
+            "question":input,
+          };
+      } else {
+          return input
+      }
+}
+
+// passing flowInputEncoder to our BedrockFlowsAgent
+const techFlowAgent = new BedrockFlowsAgent({
+      name: 'tech-flow-agent',
+      description: 'Specialized in AWS services',
+      flowIdentifier: 'AEXAMPLID',
+      flowAliasIdentifier: 'AEXAMPLEALIASID',
+      flowInputEncoder: flowInputEncoder
+      enableTrace:true
+    });
+
+
+    ```
+  </TabItem>
+  <TabItem label="Python" icon="seti:python">
+    ```python
+  # implementation of the custom flowInputEncoder callback
+  def flow_input_encoder(agent:Agent, input: str, **kwargs) -> Any:
+      if agent.name == 'tech-flow-agent':
+        # return a dict
+        return {
+          "question": input
+        }
+      else:
+          return input #input as string
+
+  # passing flowInputEncoder to our BedrockFlowsAgent
+    tech_flow_agent = BedrockFlowsAgent(BedrockFlowsAgentOptions(
+        name="tech-flow-agent",
+        description="Specializes in handling tech questions about AWS services",
+        flowIdentifier='AEXAMPLID',
+        flowAliasIdentifier='AEXAMPLEALIASID',
+        flow_input_encoder=flow_input_encoder,
+        enableTrace=True
+    ))
+      ```
+  </TabItem>
+</Tabs>
+
+## Sample Code
+
+You can find sample code for using the BedrockFlowsAgent in both TypeScript and Python:
+
+- [TypeScript Sample](https://github.com/awslabs/multi-agent-orchestrator/tree/main/examples/bedrock-flows/typescript)
+- [Python Sample](https://github.com/awslabs/multi-agent-orchestrator/tree/main/examples/bedrock-flows/python)
+

examples/bedrock-flows/python/main.py

@@ -0,0 +1,87 @@
+import asyncio
+import uuid
+import sys
+from typing import Any, List
+from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
+from multi_agent_orchestrator.classifiers import ClassifierResult
+from multi_agent_orchestrator.agents import AgentResponse, Agent, BedrockFlowsAgent, BedrockFlowsAgentOptions
+from multi_agent_orchestrator.types import ConversationMessage, ParticipantRole
+
+async def handle_request(_orchestrator: MultiAgentOrchestrator,agent:Agent, _user_input:str, _user_id:str, _session_id:str):
+    classifier_result = ClassifierResult(selected_agent=agent, confidence=1.0)
+    response:AgentResponse = await _orchestrator.agent_process_request(
+        _user_input,
+        _user_id,
+        _session_id,
+        classifier_result)
+
+    print(response.output.content[0].get('text'))
+
+
+def flow_input_encoder(agent:Agent, input: str, **kwargs) -> Any:
+    global flow_tech_agent
+    if agent == flow_tech_agent:
+        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
+
+def flow_output_decode(agent:Agent, response: Any, **kwargs) -> Any:
+    global flow_tech_agent
+    if agent == flow_tech_agent:
+        return ConversationMessage(
+            role=ParticipantRole.ASSISTANT.value,
+            content=[{'text': response}]
+        )
+    else:
+        return ConversationMessage(
+            role=ParticipantRole.ASSISTANT.value,
+            content=[{'text': response}]
+        )
+
+if __name__ == "__main__":
+
+    # Initialize the orchestrator with some options
+    orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
+        LOG_AGENT_CHAT=True,
+        LOG_CLASSIFIER_CHAT=True,
+        LOG_CLASSIFIER_RAW_OUTPUT=True,
+        LOG_CLASSIFIER_OUTPUT=True,
+        LOG_EXECUTION_TIMES=True,
+        MAX_RETRIES=3,
+        USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=False,
+        MAX_MESSAGE_PAIRS_PER_AGENT=10
+    ))
+
+    flow_tech_agent = BedrockFlowsAgent(BedrockFlowsAgentOptions(
+        name="tech-agent",
+        description="Specializes in handling tech questions about AWS services",
+        flowIdentifier='BEDROCK-FLOW-ID',
+        flowAliasIdentifier='BEDROCK-FLOW-ALIAS-ID',
+        enableTrace=False,
+        flow_input_encoder=flow_input_encoder,
+        flow_output_decoder=flow_output_decode
+    ))
+    orchestrator.add_agent(flow_tech_agent)
+
+    USER_ID = "user123"
+    SESSION_ID = str(uuid.uuid4())
+
+    print("Welcome to the interactive Multi-Agent system. Type 'quit' to exit.")
+
+    while True:
+        # Get user input
+        user_input = input("\nYou: ").strip()
+
+        if user_input.lower() == 'quit':
+            print("Exiting the program. Goodbye!")
+            sys.exit()
+
+        # Run the async function
+        asyncio.run(handle_request(orchestrator, flow_tech_agent, user_input, USER_ID, SESSION_ID))

examples/bedrock-flows/readme.md

@@ -0,0 +1,30 @@
+## BedrockFlowsAgent example
+This example, in Python and Typescript, show how to integrate your Bedrock Flows into the multi-agent orchestrator.
+
+This is the flow we used for our testing.
+
+### tech-agent-flow
+
+In this flow we connected an input node to a prompt node and the output of the prompt is connected to an output node.
+
+
+![tech-agent-flow](./tech-agent-flow.png)
+
+
+The prompt node has 2 inputs:
+- question (current question)
+- history (previous conversation)
+
+![prompt-node-configuration](./prompt-config.png)
+
+
+📝 Note
+
+📅 As of December 2, 2024, Bedrock Flows does not include a memory feature to retain previous interactions.
+
+In this example, we demonstrate:
+- 1️⃣ How to integrate your flow into a multi-agent orchestrator.
+- 2️⃣ How to incorporate conversation history into your flow to provide a smoother user experience and generate more accurate results.
+
+🚀 Let's get started!
+