Spaces:
Sleeping
Sleeping
| """ | |
| ADAS Feature Extractor | |
| Extracts ADAS features from user queries | |
| """ | |
| import json | |
| from typing import List, Optional | |
| from openai import OpenAI | |
| # ADAS feature keywords mapping | |
| ADAS_FEATURES_KEYWORDS = { | |
| "DISTRONIC": ["distronic", "distance assist", "adaptive cruise", "acc", "cruise control", "following distance"], | |
| "Active Lane Change Assist": ["lane change", "lca", "lane change assist", "change lane", "lane switching"], | |
| "Active Steering Assist": ["steering assist", "lane keeping", "lka", "lane keep", "steering", "lane centering"], | |
| "Active Stop-and-Go Assist": ["stop and go", "traffic jam", "low speed", "stop-and-go", "traffic assist"] | |
| } | |
| class ADASFeatureExtractor: | |
| """ADAS feature extractor""" | |
| def __init__(self, use_llm: bool = False, client: Optional[OpenAI] = None): | |
| """ | |
| Args: | |
| use_llm: Whether to use LLM extraction (more accurate but slower) | |
| client: OpenAI client (if use_llm=True) | |
| """ | |
| self.use_llm = use_llm | |
| self.client = client | |
| def extract(self, query: str) -> List[str]: | |
| """ | |
| Extract ADAS features from query | |
| Args: | |
| query: User query text | |
| Returns: | |
| List[str]: List of extracted ADAS features | |
| """ | |
| if self.use_llm and self.client: | |
| return self._extract_with_llm(query) | |
| else: | |
| return self._extract_with_keywords(query) | |
| def _extract_with_keywords(self, query: str) -> List[str]: | |
| """Extract features using keyword matching (fast method)""" | |
| query_lower = query.lower() | |
| matched_features = [] | |
| for feature, keywords in ADAS_FEATURES_KEYWORDS.items(): | |
| if any(kw in query_lower for kw in keywords): | |
| matched_features.append(feature) | |
| return matched_features | |
| def _extract_with_llm(self, query: str) -> List[str]: | |
| """Extract features using LLM (more accurate method)""" | |
| if not self.client: | |
| return self._extract_with_keywords(query) | |
| try: | |
| available_features = list(ADAS_FEATURES_KEYWORDS.keys()) | |
| prompt = f""" | |
| Extract ADAS features mentioned in this query: "{query}" | |
| Available features: | |
| {chr(10).join(f'- {f}' for f in available_features)} | |
| Return a JSON object with a "features" array containing the feature names. | |
| If no features are mentioned, return an empty array. | |
| """ | |
| response = self.client.chat.completions.create( | |
| model="gpt-4o-mini", | |
| messages=[ | |
| {"role": "system", "content": "You are an expert in ADAS systems. Extract ADAS features from user queries."}, | |
| {"role": "user", "content": prompt} | |
| ], | |
| response_format={"type": "json_object"}, | |
| temperature=0.1 | |
| ) | |
| result = json.loads(response.choices[0].message.content) | |
| features = result.get("features", []) | |
| # Validate extracted features against available list | |
| valid_features = [f for f in features if f in available_features] | |
| return valid_features if valid_features else self._extract_with_keywords(query) | |
| except Exception as e: | |
| print(f"⚠️ Error in LLM feature extraction: {e}") | |
| return self._extract_with_keywords(query) | |