fix ai suggestions

Author: minimAluminiumalism

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py

@@ -190,7 +190,7 @@ def _create_shared_attributes(
                 association_properties.get("base_url") or
                 association_properties.get("server_address")
             )
-        except Exception:
+        except (AttributeError, KeyError, TypeError):
             pass
 
         if not server_address:

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py

@@ -55,18 +55,20 @@ def _set_span_attribute(span: Span, name: str, value: AttributeValue):
         span.set_attribute(name, value)
 
 
-def _get_unified_unknown_model(class_name: str = None, existing_model: str = None) -> str:
+def _get_unified_unknown_model(
+    class_name: Optional[str] = None, existing_model: Optional[str] = None
+) -> str:
     """Get unified unknown model name to ensure consistency across all fallbacks."""
 
     if existing_model:
         existing_lower = existing_model.lower()
-        if existing_model.startswith("deepseek"):
+        if existing_lower.startswith("deepseek"):
             return "deepseek-unknown"
-        elif existing_model.startswith("gpt"):
+        elif existing_lower.startswith("gpt"):
             return "gpt-unknown"
-        elif existing_model.startswith("claude"):
+        elif existing_lower.startswith("claude"):
             return "claude-unknown"
-        elif existing_model.startswith("command"):
+        elif existing_lower.startswith("command"):
             return "command-unknown"
         elif ("ollama" in existing_lower or "llama" in existing_lower):
             return "ollama-unknown"
@@ -130,7 +132,7 @@ def _extract_model_name_from_request(
         association_properties = context_api.get_value("association_properties") or {}
         if (model := association_properties.get("ls_model_name")) is not None:
             return model
-    except Exception:
+    except (AttributeError, KeyError, TypeError):
         pass
 
     # Extract from serialized information for third-party integrations
@@ -253,7 +255,7 @@ def set_chat_request(
     span_holder: SpanHolder,
 ) -> None:
     metadata = kwargs.get("metadata")
-    set_request_params(span, serialized.get("kwargs", {}), span_holder, serialized, metadata)
+    set_request_params(span, kwargs, span_holder, serialized, metadata)
 
     if should_send_prompts():
         for i, function in enumerate(
@@ -471,7 +473,7 @@ def set_chat_response_usage(
                 )
 
 
-def extract_model_name_from_response_metadata(response: LLMResult) -> str:
+def extract_model_name_from_response_metadata(response: LLMResult) -> Optional[str]:
     """Enhanced model name extraction from response metadata with third-party support."""
 
     # Standard extraction from response metadata

packages/opentelemetry-instrumentation-langchain/tests/test_model_extraction.py

@@ -190,3 +190,14 @@ def test_metadata_has_priority_over_class_inference(self):
 
         result = _extract_model_name_from_request(kwargs, self.span_holder, serialized, metadata)
         assert result == "deepseek-v3"  # Should use metadata, not fallback to "deepseek-chat"
+
+    def test_response_metadata_extraction_returns_none_when_no_model_found(self):
+        """Test that extract_model_name_from_response_metadata returns None when no model info is found."""
+        # Response with no model information anywhere
+        response = LLMResult(
+            generations=[[Generation(text="Response without model info")]],
+            llm_output={"other_field": "value"}  # No model-related fields
+        )
+
+        result = extract_model_name_from_response_metadata(response)
+        assert result is None, "Should return None when no model information is found"

packages/opentelemetry-instrumentation-langchain/tests/test_streaming_metrics.py

@@ -1,5 +1,5 @@
 import time
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 from uuid import uuid4
 from langchain_core.outputs import LLMResult, Generation
 from opentelemetry.instrumentation.langchain.callback_handler import TraceloopCallbackHandler
@@ -37,33 +37,37 @@ def test_ttft_metric_recorded_on_first_token(self):
         mock_span = Mock(spec=Span)
         mock_span.attributes = {SpanAttributes.LLM_SYSTEM: "Langchain"}
 
-        # Create span holder with specific start time
-        start_time = time.time()
-        span_holder = SpanHolder(
-            span=mock_span,
-            token=None,
-            context=None,
-            children=[],
-            workflow_name="test",
-            entity_name="test",
-            entity_path="test",
-            start_time=start_time
-        )
-        self.handler.spans[run_id] = span_holder
-
-        # Simulate first token arrival after a small delay
-        time.sleep(0.1)
-        self.handler.on_llm_new_token("Hello", run_id=run_id)
-
-        # Verify TTFT metric was recorded
-        self.ttft_histogram.record.assert_called_once()
-        args = self.ttft_histogram.record.call_args
-        ttft_value = args[0][0]
-        assert ttft_value > 0.05, "TTFT should be greater than 0.05 seconds"
-
-        # Verify attributes
-        attributes = args[1]["attributes"]
-        assert attributes[SpanAttributes.LLM_SYSTEM] == "Langchain"
+        # Use mock time for stable testing
+        with patch('opentelemetry.instrumentation.langchain.callback_handler.time.time') as mock_time, \
+             patch('opentelemetry.instrumentation.langchain.span_utils.time.time') as mock_span_time:
+
+            start_time = 1000.0
+            mock_time.return_value = start_time
+            mock_span_time.return_value = start_time
+
+            span_holder = SpanHolder(
+                span=mock_span,
+                token=None,
+                context=None,
+                children=[],
+                workflow_name="test",
+                entity_name="test",
+                entity_path="test",
+                start_time=start_time
+            )
+            self.handler.spans[run_id] = span_holder
+
+            mock_time.return_value = start_time + 0.1
+            mock_span_time.return_value = start_time + 0.1
+            self.handler.on_llm_new_token("Hello", run_id=run_id)
+
+            self.ttft_histogram.record.assert_called_once()
+            args = self.ttft_histogram.record.call_args
+            ttft_value = args[0][0]
+            assert abs(ttft_value - 0.1) < 0.001, f"TTFT should be approximately 0.1 seconds, got {ttft_value}"
+
+            attributes = args[1]["attributes"]
+            assert attributes[SpanAttributes.LLM_SYSTEM] == "Langchain"
 
     def test_ttft_metric_not_recorded_on_subsequent_tokens(self):
         """Test that TTFT metric is only recorded once."""
@@ -134,37 +138,47 @@ def test_streaming_time_to_generate_metric(self):
         mock_span = Mock(spec=Span)
         mock_span.attributes = {SpanAttributes.LLM_SYSTEM: "Langchain"}
 
-        start_time = time.time()
-        span_holder = SpanHolder(
-            span=mock_span,
-            token=None,
-            context=None,
-            children=[],
-            workflow_name="test",
-            entity_name="test",
-            entity_path="test",
-            start_time=start_time
-        )
-        self.handler.spans[run_id] = span_holder
-
-        # Simulate token arrival
-        time.sleep(0.05)
-        self.handler.on_llm_new_token("Hello", run_id=run_id)
-
-        # Simulate completion after more time
-        time.sleep(0.05)
-        llm_result = LLMResult(
-            generations=[[Generation(text="Hello world")]],
-            llm_output={"model_name": "test-model"}
-        )
-
-        self.handler.on_llm_end(llm_result, run_id=run_id)
-
-        # Verify streaming time metric was recorded
-        self.streaming_time_histogram.record.assert_called_once()
-        args = self.streaming_time_histogram.record.call_args
-        streaming_time = args[0][0]
-        assert streaming_time > 0.04, "Streaming time should be greater than 0.04 seconds"
+        with patch('opentelemetry.instrumentation.langchain.callback_handler.time.time') as mock_time, \
+             patch('opentelemetry.instrumentation.langchain.span_utils.time.time') as mock_span_time:
+
+            start_time = 1000.0
+            mock_time.return_value = start_time
+            mock_span_time.return_value = start_time
+
+            span_holder = SpanHolder(
+                span=mock_span,
+                token=None,
+                context=None,
+                children=[],
+                workflow_name="test",
+                entity_name="test",
+                entity_path="test",
+                start_time=start_time
+            )
+            self.handler.spans[run_id] = span_holder
+
+            first_token_time = start_time + 0.05
+            mock_time.return_value = first_token_time
+            mock_span_time.return_value = first_token_time
+            self.handler.on_llm_new_token("Hello", run_id=run_id)
+
+            completion_time = first_token_time + 0.05
+            mock_time.return_value = completion_time
+            mock_span_time.return_value = completion_time
+            llm_result = LLMResult(
+                generations=[[Generation(text="Hello world")]],
+                llm_output={"model_name": "test-model"}
+            )
+
+            self.handler.on_llm_end(llm_result, run_id=run_id)
+
+            self.streaming_time_histogram.record.assert_called_once()
+            args = self.streaming_time_histogram.record.call_args
+            streaming_time = args[0][0]
+            assert abs(streaming_time - 0.05) < 0.001, (
+                f"Streaming time should be approximately 0.05 seconds, "
+                f"got {streaming_time}"
+            )
 
     def test_exception_metric_recorded_on_error(self):
         """Test that exception metric is recorded on LLM errors."""