feat: hybrid router and async pinecone upgrades

[jamescalam]

Will close #492

Also resolves some previously unnoticed bugs in async PineconeIndex and pytest coverage

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Test Coverage Completeness Ensure that the newly added HybridRouter is thoroughly tested across all scenarios, including edge cases, to validate its integration and functionality. def get_test_routers(): routers = [SemanticRouter] if importlib.util.find_spec("pinecone_text") is not None: routers.append(HybridRouter) return routers @pytest.mark.parametrize( "index_cls,router_cls", [ (index, router) for index in get_test_indexes() for router in get_test_routers() ], ) class TestSemanticRouter: @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_initialization(self, openai_encoder, routes, index_cls, router_cls): index = init_index(index_cls) _ = router_cls( encoder=openai_encoder, routes=routes, top_k=10, index=index, auto_sync="local", ) @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_second_initialization_sync(self, openai_encoder, routes, index_cls, router_cls): index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # allow for index to be populated assert route_layer.is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_second_initialization_not_synced( self, openai_encoder, routes, routes_2, index_cls, router_cls ): index = init_index(index_cls) _ = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=index ) if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # allow for index to be populated assert route_layer.is_synced() is False @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_utterance_diff(self, openai_encoder, routes, routes_2, index_cls, router_cls): index = init_index(index_cls) _ = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) route_layer_2 = router_cls( encoder=openai_encoder, routes=routes_2, index=index ) if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # allow for index to be populated diff = route_layer_2.get_utterance_diff(include_metadata=True) assert '+ Route 1: Hello | None | {"type": "default"}' in diff assert '+ Route 1: Hi | None | {"type": "default"}' in diff assert "- Route 1: Hello | None | {}" in diff assert "+ Route 2: Au revoir | None | {}" in diff assert "- Route 2: Hi | None | {}" in diff assert "+ Route 2: Bye | None | {}" in diff assert "+ Route 2: Goodbye | None | {}" in diff assert "+ Route 3: Boo | None | {}" in diff @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_auto_sync_local(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST LOCAL pinecone_index = init_index(index_cls) _ = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, ) time.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated assert route_layer.index.get_utterances() == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_auto_sync_remote(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST REMOTE pinecone_index = init_index(index_cls) _ = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="remote", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated assert route_layer.index.get_utterances() == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_auto_sync_merge_force_local( self, openai_encoder, routes, routes_2, index_cls, router_cls ): if index_cls is PineconeIndex: # TEST MERGE FORCE LOCAL pinecone_index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="merge-force-local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.is_synced() # now confirm utterances are correct local_utterances = route_layer.index.get_utterances() # we sort to ensure order is the same # TODO JB: there is a bug here where if we include_metadata=True it fails local_utterances.sort(key=lambda x: x.to_str(include_metadata=False)) assert local_utterances == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 1", utterance="Hi"), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_auto_sync_merge_force_remote( self, openai_encoder, routes, routes_2, index_cls, router_cls ): if index_cls is PineconeIndex: # TEST MERGE FORCE LOCAL pinecone_index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="merge-force-remote", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.is_synced() # now confirm utterances are correct local_utterances = route_layer.index.get_utterances() # we sort to ensure order is the same local_utterances.sort( key=lambda x: x.to_str(include_metadata=include_metadata(index_cls)) ) assert local_utterances == [ Utterance( route="Route 1", utterance="Hello", metadata={"type": "default"} ), Utterance( route="Route 1", utterance="Hi", metadata={"type": "default"} ), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), Utterance(route="Route 3", utterance="Boo"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_sync(self, openai_encoder, index_cls, router_cls): route_layer = router_cls( encoder=openai_encoder, routes=[], index=init_index(index_cls), auto_sync=None, ) route_layer.sync("remote") time.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_auto_sync_merge(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST MERGE pinecone_index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="merge", ) time.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.is_synced() # now confirm utterances are correct local_utterances = route_layer.index.get_utterances() # we sort to ensure order is the same local_utterances.sort( key=lambda x: x.to_str(include_metadata=include_metadata(index_cls)) ) assert local_utterances == [ Utterance( route="Route 1", utterance="Hello", metadata={"type": "default"} ), Utterance( route="Route 1", utterance="Hi", metadata={"type": "default"} ), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), Utterance(route="Route 3", utterance="Boo"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_sync_lock_prevents_concurrent_sync( self, openai_encoder, routes, index_cls, router_cls ): """Test that sync lock prevents concurrent synchronization operations""" index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync=None, ) # Acquire sync lock route_layer.index.lock(value=True) if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # Attempt to sync while lock is held should raise exception with pytest.raises(Exception): route_layer.sync("local") # Release lock route_layer.index.lock(value=False) if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # Should succeed after lock is released route_layer.sync("local") if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) assert route_layer.is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) def test_sync_lock_auto_releases(self, openai_encoder, routes, index_cls, router_cls): """Test that sync lock is automatically released after sync operations""" index = init_index(index_cls) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync=None, ) # Initial sync should acquire and release lock route_layer.sync("local") if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) # Lock should be released, allowing another sync route_layer.sync("local") # Should not raise exception if index_cls is PineconeIndex: time.sleep(PINECONE_SLEEP) assert route_layer.is_synced() # clear index route_layer.index.index.delete(namespace="", delete_all=True) @pytest.mark.parametrize("index_cls", get_test_indexes()) class TestAsyncSemanticRouter: @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_initialization(self, openai_encoder, routes, index_cls, router_cls): index = init_index(index_cls, init_async_index=True) _ = router_cls( encoder=openai_encoder, routes=routes, top_k=10, index=index, auto_sync="local", ) @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_second_initialization_sync(self, openai_encoder, routes, index_cls, router_cls): index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated assert route_layer.async_is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_second_initialization_not_synced( self, openai_encoder, routes, routes_2, index_cls, router_cls ): index = init_index(index_cls, init_async_index=True) _ = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=index ) if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated assert await route_layer.async_is_synced() is False @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_utterance_diff(self, openai_encoder, routes, routes_2, index_cls, router_cls): index = init_index(index_cls, init_async_index=True) _ = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync="local" ) route_layer_2 = router_cls( encoder=openai_encoder, routes=routes_2, index=index ) if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated diff = await route_layer_2.aget_utterance_diff(include_metadata=True) assert '+ Route 1: Hello | None | {"type": "default"}' in diff assert '+ Route 1: Hi | None | {"type": "default"}' in diff assert "- Route 1: Hello | None | {}" in diff assert "+ Route 2: Au revoir | None | {}" in diff assert "- Route 2: Hi | None | {}" in diff assert "+ Route 2: Bye | None | {}" in diff assert "+ Route 2: Goodbye | None | {}" in diff assert "+ Route 3: Boo | None | {}" in diff @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_auto_sync_local(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST LOCAL pinecone_index = init_index(index_cls, init_async_index=True) _ = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated assert await route_layer.index.aget_utterances() == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_auto_sync_remote(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST REMOTE pinecone_index = init_index(index_cls, init_async_index=True) _ = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="remote", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated assert await route_layer.index.aget_utterances() == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_auto_sync_merge_force_local( self, openai_encoder, routes, routes_2, index_cls, router_cls ): if index_cls is PineconeIndex: # TEST MERGE FORCE LOCAL pinecone_index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="merge-force-local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.async_is_synced() # now confirm utterances are correct local_utterances = await route_layer.index.aget_utterances() # we sort to ensure order is the same # TODO JB: there is a bug here where if we include_metadata=True it fails local_utterances.sort(key=lambda x: x.to_str(include_metadata=False)) assert local_utterances == [ Utterance(route="Route 1", utterance="Hello"), Utterance(route="Route 1", utterance="Hi"), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_auto_sync_merge_force_remote( self, openai_encoder, routes, routes_2, index_cls, router_cls ): if index_cls is PineconeIndex: # TEST MERGE FORCE LOCAL pinecone_index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="merge-force-remote", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert route_layer.async_is_synced() # now confirm utterances are correct local_utterances = await route_layer.index.aget_utterances() # we sort to ensure order is the same local_utterances.sort( key=lambda x: x.to_str(include_metadata=include_metadata(index_cls)) ) assert local_utterances == [ Utterance( route="Route 1", utterance="Hello", metadata={"type": "default"} ), Utterance( route="Route 1", utterance="Hi", metadata={"type": "default"} ), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), Utterance(route="Route 3", utterance="Boo"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_sync(self, openai_encoder, index_cls, router_cls): route_layer = router_cls( encoder=openai_encoder, routes=[], index=init_index(index_cls, init_async_index=True), auto_sync=None, ) await route_layer.async_sync("remote") await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert await route_layer.async_is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_auto_sync_merge(self, openai_encoder, routes, routes_2, index_cls, router_cls): if index_cls is PineconeIndex: # TEST MERGE pinecone_index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="local", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated route_layer = router_cls( encoder=openai_encoder, routes=routes, index=pinecone_index, auto_sync="merge", ) await asyncio.sleep(PINECONE_SLEEP) # allow for index to be populated # confirm local and remote are synced assert await route_layer.async_is_synced() # now confirm utterances are correct local_utterances = await route_layer.index.aget_utterances() # we sort to ensure order is the same local_utterances.sort( key=lambda x: x.to_str(include_metadata=include_metadata(index_cls)) ) assert local_utterances == [ Utterance( route="Route 1", utterance="Hello", metadata={"type": "default"} ), Utterance( route="Route 1", utterance="Hi", metadata={"type": "default"} ), Utterance(route="Route 2", utterance="Au revoir"), Utterance(route="Route 2", utterance="Bye"), Utterance(route="Route 2", utterance="Goodbye"), Utterance(route="Route 2", utterance="Hi"), Utterance(route="Route 3", utterance="Boo"), ], "The routes in the index should match the local routes" @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_sync_lock_prevents_concurrent_sync( self, openai_encoder, routes, index_cls, router_cls ): """Test that sync lock prevents concurrent synchronization operations""" index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync=None, ) # Acquire sync lock await route_layer.index.alock(value=True) if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) # Attempt to sync while lock is held should raise exception with pytest.raises(Exception): await route_layer.async_sync("local") # Release lock await route_layer.index.alock(value=False) if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) # Should succeed after lock is released await route_layer.async_sync("local") if index_cls is PineconeIndex: await asyncio.sleep(PINECONE_SLEEP) assert await route_layer.async_is_synced() @pytest.mark.skipif( os.environ.get("PINECONE_API_KEY") is None, reason="Pinecone API key required" ) @pytest.mark.asyncio async def test_sync_lock_auto_releases(self, openai_encoder, routes, index_cls, router_cls): """Test that sync lock is automatically released after sync operations""" index = init_index(index_cls, init_async_index=True) route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync=None, Dynamic Router Selection Logic Verify that the get_test_routers function correctly identifies and includes the appropriate router classes dynamically, and ensure it handles cases where dependencies are missing. def get_test_routers(): routers = [SemanticRouter] if importlib.util.find_spec("pinecone_text") is not None: routers.append(HybridRouter) return routers

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Score
Possible issue	Add validation for the router_cls parameter to prevent runtime errors with unsupported classes Ensure that router_cls is properly validated or checked before being used in test cases to avoid potential runtime errors if an invalid or unsupported class is passed. tests/unit/test_sync.py [847-851] +if not isinstance(router_cls, (SemanticRouter, HybridRouter)): + raise ValueError("Invalid router class provided") route_layer = router_cls( encoder=openai_encoder, routes=routes, index=index, auto_sync=None, ) Suggestion importance[1-10]: 7 Why: Adding validation for router_cls ensures that only supported classes are used, preventing potential runtime errors. This is a meaningful improvement for robustness, though the exact list of valid classes might need to be confirmed.	7
	Validate index_cls before initializing to avoid invalid index initialization Add a check to ensure index_cls is properly initialized and valid before passing it to init_index to prevent potential initialization failures. tests/unit/test_sync.py [264] +if index_cls not in [PineconeIndex, QdrantIndex, LocalIndex, PostgresIndex]: + raise ValueError("Invalid index class provided") index = init_index(index_cls) Suggestion importance[1-10]: 7 Why: Validating index_cls ensures that only supported index classes are used, reducing the risk of initialization failures. This is a practical enhancement for error prevention and robustness.	7
	Validate PINECONE_SLEEP to ensure it is a positive number before using it in sleep operations Ensure that PINECONE_SLEEP is defined and has a valid value to avoid potential runtime errors during the sleep operation. tests/unit/test_sync.py [268] +if not isinstance(PINECONE_SLEEP, (int, float)) or PINECONE_SLEEP <= 0: + raise ValueError("PINECONE_SLEEP must be a positive number") time.sleep(PINECONE_SLEEP) # allow for index to be populated Suggestion importance[1-10]: 6 Why: Ensuring PINECONE_SLEEP is a valid positive number prevents potential runtime errors during sleep operations. While useful, this check might be redundant if PINECONE_SLEEP is already defined and controlled elsewhere in the codebase.	6
General	Validate routes and routes_2 to ensure they are not empty before initializing the router Add a check to ensure routes and routes_2 are not empty or invalid before passing them to the router initialization to prevent unexpected behavior. tests/unit/test_sync.py [749-753] +if not routes_2: + raise ValueError("Routes cannot be empty") route_layer = router_cls( encoder=openai_encoder, routes=routes_2, index=pinecone_index, auto_sync="merge-force-remote", ) Suggestion importance[1-10]: 6 Why: Adding a validation for routes and routes_2 prevents unexpected behavior due to empty inputs. This is a reasonable improvement for ensuring the correctness of router initialization, though its necessity depends on the broader context of the test setup.	6