async conversation api support

Signed-off-by: Elena Kolevska <[email protected]>

Author: elena-kolevska

dapr/aio/clients/grpc/client.py

@@ -29,6 +29,7 @@
 
 from google.protobuf.message import Message as GrpcMessage
 from google.protobuf.empty_pb2 import Empty as GrpcEmpty
+from google.protobuf.any_pb2 import Any as GrpcAny
 
 import grpc.aio  # type: ignore
 from grpc.aio import (  # type: ignore
@@ -75,9 +76,12 @@
     InvokeMethodRequest,
     BindingRequest,
     TransactionalStateOperation,
+    ConversationInput,
 )
 from dapr.clients.grpc._response import (
     BindingResponse,
+    ConversationResponse,
+    ConversationResult,
     DaprResponse,
     GetSecretResponse,
     GetBulkSecretResponse,
@@ -1711,6 +1715,65 @@ async def purge_workflow(self, instance_id: str, workflow_component: str) -> Dap
         except grpc.aio.AioRpcError as err:
             raise DaprInternalError(err.details())
 
+    async def converse_alpha1(
+        self,
+        name: str,
+        inputs: List[ConversationInput],
+        *,
+        # Force remaining args to be keyword-only
+        context_id: Optional[str] = None,
+        parameters: Optional[Dict[str, GrpcAny]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        scrub_pii: Optional[bool] = None,
+        temperature: Optional[float] = None,
+    ) -> ConversationResponse:
+        """Invoke an LLM using the conversation API (Alpha).
+
+        Args:
+            name: Name of the LLM component to invoke
+            inputs: List of conversation inputs
+            context_id: Optional ID for continuing an existing chat
+            parameters: Optional custom parameters for the request
+            metadata: Optional metadata for the component
+            scrub_pii: Optional flag to scrub PII from outputs
+            temperature: Optional temperature setting (0.0 to 1.0) where
+                        lower values give more deterministic responses and
+                        higher values enable more creative responses
+
+        Returns:
+            ConversationResponse containing the conversation results
+
+        Raises:
+            DaprInternalError: If the Dapr runtime returns an error
+        """
+        inputs_pb = [
+            api_v1.ConversationInput(message=inp.message, role=inp.role, scrubPII=inp.scrub_pii)
+            for inp in inputs
+        ]
+
+        request = api_v1.ConversationRequest(
+            name=name,
+            inputs=inputs_pb,
+            contextID=context_id,
+            parameters=parameters or {},
+            metadata=metadata or {},
+            scrubPII=scrub_pii,
+            temperature=temperature,
+        )
+
+        try:
+            response = await self._stub.ConverseAlpha1(request)
+
+            outputs = [
+                ConversationResult(result=output.result, parameters=output.parameters)
+                for output in response.outputs
+            ]
+
+            return ConversationResponse(context_id=response.contextID, outputs=outputs)
+
+        except Exception as ex:
+            raise DaprInternalError(f'Error invoking conversation API: {str(ex)}')
+
     async def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

examples/conversation/conversation.py

@@ -30,4 +30,4 @@
     )
 
     for output in response.outputs:
-        print(f'Result: {output.result}')
+        print(f'Result: {output.result}')

tests/clients/test_dapr_grpc_client_async.py

@@ -23,13 +23,13 @@
 
 from dapr.aio.clients.grpc.client import DaprGrpcClientAsync
 from dapr.aio.clients import DaprClient
-from dapr.clients.exceptions import DaprGrpcError
+from dapr.clients.exceptions import DaprGrpcError, DaprInternalError
 from dapr.common.pubsub.subscription import StreamInactiveError
 from dapr.proto import common_v1
 from .fake_dapr_server import FakeDaprSidecar
 from dapr.conf import settings
 from dapr.clients.grpc._helpers import to_bytes
-from dapr.clients.grpc._request import TransactionalStateOperation
+from dapr.clients.grpc._request import TransactionalStateOperation, ConversationInput
 from dapr.clients.grpc._state import StateOptions, Consistency, Concurrency, StateItem
 from dapr.clients.grpc._crypto import EncryptOptions, DecryptOptions
 from dapr.clients.grpc._response import (
@@ -1113,6 +1113,58 @@ async def test_decrypt_file_data_read_chunks(self):
             self.assertEqual(await resp.read(5), b'hello')
             self.assertEqual(await resp.read(5), b' dapr')
 
+    async def test_converse_alpha1_basic(self):
+        dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
+
+        inputs = [ConversationInput(message="Hello", role="user"),
+            ConversationInput(message="How are you?", role="user")]
+
+        response = await dapr.converse_alpha1(name="test-llm", inputs=inputs)
+
+        # Check response structure
+        self.assertIsNotNone(response)
+        self.assertEqual(len(response.outputs), 2)
+        self.assertEqual(response.outputs[0].result, "Response to: Hello")
+        self.assertEqual(response.outputs[1].result, "Response to: How are you?")
+        await dapr.close()
+
+    async def test_converse_alpha1_with_options(self):
+        dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
+
+        inputs = [ConversationInput(message="Hello", role="user", scrub_pii=True)]
+
+        response = await dapr.converse_alpha1(name="test-llm", inputs=inputs, context_id="chat-123",
+            temperature=0.7, scrub_pii=True, metadata={"key": "value"})
+
+        self.assertIsNotNone(response)
+        self.assertEqual(len(response.outputs), 1)
+        self.assertEqual(response.outputs[0].result, "Response to: Hello")
+        await dapr.close()
+
+    async def test_converse_alpha1_error_handling(self):
+        dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
+
+        # Setup server to raise an exception
+        self._fake_dapr_server.raise_exception_on_next_call(
+            status_pb2.Status(code=code_pb2.INVALID_ARGUMENT, message="Invalid argument"))
+
+        inputs = [ConversationInput(message="Hello", role="user")]
+
+        with self.assertRaises(DaprInternalError) as context:
+            await dapr.converse_alpha1(name="test-llm", inputs=inputs)
+        self.assertTrue("Invalid argument" in str(context.exception))
+        await dapr.close()
+
+    async def test_converse_alpha1_empty_inputs(self):
+        dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
+
+        # Test with empty inputs list
+        response = await dapr.converse_alpha1(name="test-llm", inputs=[])
+
+        self.assertIsNotNone(response)
+        self.assertEqual(len(response.outputs), 0)
+        await dapr.close()
+
 
 if __name__ == '__main__':
     unittest.main()