Merge pull request #91 from pipecat-ai/mb/add-context-getter

Add a `get_current_context` function to access the latest context

CHANGELOG.md

@@ -40,6 +40,16 @@ node_config = {
 }
 ```
 
+- Added a new function called `get_current_context` which provides access to
+  the LLM context.
+
+Example usage:
+
+```python
+# Access current conversation context
+context = flow_manager.get_current_context()
+```
+
 ## [0.0.12] - 2025-01-30
 
 ### Added

src/pipecat_flows/manager.py

@@ -162,6 +162,21 @@ async def initialize(self) -> None:
             self.initialized = False
             raise FlowInitializationError(f"Failed to initialize flow: {str(e)}") from e
 
+    def get_current_context(self) -> List[dict]:
+        """Get the current conversation context.
+
+        Returns:
+            List of messages in the current context, including system messages,
+            user messages, and assistant responses.
+
+        Raises:
+            FlowError: If context aggregator is not available
+        """
+        if not self._context_aggregator:
+            raise FlowError("No context aggregator available")
+
+        return self._context_aggregator.user()._context.messages
+
     def register_action(self, action_type: str, handler: Callable) -> None:
         """Register a handler for a specific action type.
 

tests/test_manager.py

@@ -1094,3 +1094,26 @@ async def test_transition_configuration_exclusivity(self):
         self.assertIn(
             "cannot have both transition_to and transition_callback", str(context.exception)
         )
+
+    async def test_get_current_context(self):
+        """Test getting current conversation context."""
+        flow_manager = FlowManager(
+            task=self.mock_task,
+            llm=self.mock_llm,
+            context_aggregator=self.mock_context_aggregator,
+        )
+        await flow_manager.initialize()
+
+        # Mock the context messages
+        mock_messages = [{"role": "system", "content": "Test message"}]
+        self.mock_context_aggregator.user()._context.messages = mock_messages
+
+        # Test getting context
+        context = flow_manager.get_current_context()
+        self.assertEqual(context, mock_messages)
+
+        # Test error when context aggregator is not available
+        flow_manager._context_aggregator = None
+        with self.assertRaises(FlowError) as context:
+            flow_manager.get_current_context()
+        self.assertIn("No context aggregator available", str(context.exception))