fix(streaming): invert logic for assistant stream parsing

Author: RobertCraigie

src/openai/_streaming.py

@@ -59,7 +59,7 @@ def __stream__(self) -> Iterator[_T]:
             if sse.data.startswith("[DONE]"):
                 break
 
-            if sse.event is None or sse.event.startswith("response.") or sse.event.startswith("transcript."):
+            if sse.event == "error":
                 data = sse.json()
                 if is_mapping(data) and data.get("error"):
                     message = None
@@ -75,12 +75,13 @@ def __stream__(self) -> Iterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data=data, cast_to=cast_to, response=response)
 
+            if sse.event and sse.event.startswith('thread.'):
+                # the assistants API uses a different event shape structure
+                yield process_data(data={"data": sse.json(), "event": sse.event}, cast_to=cast_to, response=response)
             else:
                 data = sse.json()
-
-                if sse.event == "error" and is_mapping(data) and data.get("error"):
+                if is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -94,7 +95,7 @@ def __stream__(self) -> Iterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
+                yield process_data(data=data, cast_to=cast_to, response=response)
 
         # Ensure the entire stream is consumed
         for _sse in iterator:
@@ -161,7 +162,7 @@ async def __stream__(self) -> AsyncIterator[_T]:
             if sse.data.startswith("[DONE]"):
                 break
 
-            if sse.event is None or sse.event.startswith("response.") or sse.event.startswith("transcript."):
+            if sse.event == "error":
                 data = sse.json()
                 if is_mapping(data) and data.get("error"):
                     message = None
@@ -177,12 +178,13 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data=data, cast_to=cast_to, response=response)
 
+            if sse.event and sse.event.startswith('thread.'):
+                # the assistants API uses a different event shape structure
+                yield process_data(data={"data": sse.json(), "event": sse.event}, cast_to=cast_to, response=response)
             else:
                 data = sse.json()
-
-                if sse.event == "error" and is_mapping(data) and data.get("error"):
+                if is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -196,7 +198,7 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
+                yield process_data(data=data, cast_to=cast_to, response=response)
 
         # Ensure the entire stream is consumed
         async for _sse in iterator: