fix: Improve streaming response type detection for third-party model … (#3637)

Author: Wendong-Fan

camel/agents/chat_agent.py

@@ -4289,12 +4289,22 @@ def _stream_response(
                 return
 
             # Handle streaming response
-            if isinstance(response, Stream) or inspect.isgenerator(response):
+            # Check for Stream, generator, or third-party wrappers
+            if (
+                isinstance(response, Stream)
+                or inspect.isgenerator(response)
+                or (
+                    hasattr(response, '__iter__')
+                    and hasattr(response, '__enter__')
+                    and not hasattr(response, 'get_final_completion')
+                    and not isinstance(response, ChatCompletion)
+                )
+            ):
                 (
                     stream_completed,
                     tool_calls_complete,
                 ) = yield from self._process_stream_chunks_with_accumulator(
-                    response,
+                    response,  # type: ignore[arg-type]
                     content_accumulator,
                     accumulated_tool_calls,
                     tool_call_records,
@@ -4333,11 +4343,9 @@ def _stream_response(
                     # Stream completed without tool calls
                     accumulated_tool_calls.clear()
                     break
-            elif hasattr(response, '__enter__') and hasattr(
-                response, '__exit__'
-            ):
+            elif hasattr(response, 'get_final_completion'):
                 # Handle structured output stream (ChatCompletionStreamManager)
-                with response as stream:
+                with response as stream:  # type: ignore[union-attr]
                     parsed_object = None
 
                     for event in stream:
@@ -4426,7 +4434,9 @@ def _stream_response(
                         return
             else:
                 # Handle non-streaming response (fallback)
-                model_response = self._handle_batch_response(response)
+                model_response = self._handle_batch_response(
+                    response  # type: ignore[arg-type]
+                )
                 yield self._convert_to_chatagent_response(
                     model_response,
                     tool_call_records,
@@ -5098,11 +5108,16 @@ async def _astream_response(
                 return
 
             # Handle streaming response
-            # Note: Also check for async generators since some model backends
-            # (e.g., GeminiModel) wrap AsyncStream in async generators for
-            # additional processing
-            if isinstance(response, AsyncStream) or inspect.isasyncgen(
-                response
+            # Check for AsyncStream, async generator, or third-party wrappers
+            if (
+                isinstance(response, AsyncStream)
+                or inspect.isasyncgen(response)
+                or (
+                    hasattr(response, '__aiter__')
+                    and hasattr(response, '__aenter__')
+                    and not hasattr(response, 'get_final_completion')
+                    and not isinstance(response, ChatCompletion)
+                )
             ):
                 stream_completed = False
                 tool_calls_complete = False
@@ -5111,7 +5126,7 @@ async def _astream_response(
                 async for (
                     item
                 ) in self._aprocess_stream_chunks_with_accumulator(
-                    response,
+                    response,  # type: ignore[arg-type]
                     content_accumulator,
                     accumulated_tool_calls,
                     tool_call_records,
@@ -5158,12 +5173,10 @@ async def _astream_response(
                     # Stream completed without tool calls
                     accumulated_tool_calls.clear()
                     break
-            elif hasattr(response, '__aenter__') and hasattr(
-                response, '__aexit__'
-            ):
+            elif hasattr(response, 'get_final_completion'):
                 # Handle structured output stream
                 # (AsyncChatCompletionStreamManager)
-                async with response as stream:
+                async with response as stream:  # type: ignore[union-attr]
                     parsed_object = None
 
                     async for event in stream:
@@ -5254,7 +5267,9 @@ async def _astream_response(
                         return
             else:
                 # Handle non-streaming response (fallback)
-                model_response = self._handle_batch_response(response)
+                model_response = self._handle_batch_response(
+                    response  # type: ignore[arg-type]
+                )
                 yield self._convert_to_chatagent_response(
                     model_response,
                     tool_call_records,