Enhance StreamListener to support generic type annotations for output

dspy/streaming/streaming_listener.py

@@ -1,3 +1,4 @@
+import inspect
 import re
 from collections import defaultdict
 from queue import Queue
@@ -135,7 +136,12 @@ def receive(self, chunk: ModelResponseStream):
                 return
 
         # Handle custom streamable types
-        if self._output_type and issubclass(self._output_type, Type) and self._output_type.is_streamable():
+        if (
+            self._output_type
+            and inspect.isclass(self._output_type)
+            and issubclass(self._output_type, Type)
+            and self._output_type.is_streamable()
+        ):
             if parsed_chunk := self._output_type.parse_stream_chunk(chunk):
                 return StreamResponse(
                     self.predict_name,

tests/streaming/test_streaming.py

@@ -1932,3 +1932,24 @@ async def non_reasoning_stream(*args, **kwargs):
                 assert final_prediction.reasoning.content == "Let's think step by step about this question."
                 # Verify Reasoning object is str-like
                 assert str(final_prediction.reasoning) == "Let's think step by step about this question."
+
+
+@pytest.mark.anyio
+async def test_stream_listener_with_generic_type_annotation():
+    class TestSignature(dspy.Signature):
+        input_text: str = dspy.InputField()
+        output_list: list[str] | int = dspy.OutputField()
+
+    predict = dspy.Predict(TestSignature)
+    listener = dspy.streaming.StreamListener(signature_field_name="output_list", predict=predict)
+
+    assert listener._output_type == list[str] | int
+
+    mock_chunk = ModelResponseStream(
+        model="gpt-4o-mini",
+        choices=[StreamingChoices(delta=Delta(content="test"))],
+    )
+
+    with dspy.context(adapter=dspy.JSONAdapter()):
+        result = listener.receive(mock_chunk)
+        assert result is None