调整配置目录结构

tangwang
1 parent 1f071951
Showing 8 changed files with 98 additions and 30 deletions Show diff stats
config/config_loader.py
config/schema/customer1_config.yaml -> config/schema/customer1/config.yaml
config/schema/customer1/query_rewrite.dict
example_usage.py
indexer/data_transformer.py
main.py
scripts/run_tests.py
test_aggregation_functionality.py
@@ -104,10 +104,53 @@ class ConfigLoader:
     def __init__(self, config_dir: str = "config/schema"):
         self.config_dir = Path(config_dir)
+    
+    def _load_rewrite_dictionary(self, customer_id: str) -> Dict[str, str]:
+        """
+        Load query rewrite dictionary from external file.
+        
+        Args:
+            customer_id: Customer identifier
+            
+        Returns:
+            Dictionary mapping query terms to rewritten queries
+        """
+        dict_file = self.config_dir / customer_id / "query_rewrite.dict"
+        
+        if not dict_file.exists():
+            # Dictionary file is optional, return empty dict if not found
+            return {}
+        
+        rewrite_dict = {}
+        try:
+            with open(dict_file, 'r', encoding='utf-8') as f:
+                for line_num, line in enumerate(f, 1):
+                    line = line.strip()
+                    # Skip empty lines and comments
+                    if not line or line.startswith('#'):
+                        continue
+                    
+                    # Parse tab-separated format
+                    parts = line.split('\t')
+                    if len(parts) != 2:
+                        print(f"Warning: Invalid format in {dict_file} line {line_num}: {line}")
+                        continue
+                    
+                    key, value = parts
+                    rewrite_dict[key.strip()] = value.strip()
+        except Exception as e:
+            print(f"Error loading rewrite dictionary from {dict_file}: {e}")
+            return {}
+        
+        return rewrite_dict
     def load_customer_config(self, customer_id: str) -> CustomerConfig:
         """
         Load customer configuration from YAML file.
+        
+        Supports two directory structures:
+        1. New structure: config/schema/{customer_id}/config.yaml
+        2. Old structure: config/schema/{customer_id}_config.yaml (for backward compatibility)
         Args:
             customer_id: Customer identifier (used to find config file)
@@ -118,8 +161,13 @@ class ConfigLoader:
         Raises:
             ConfigurationError: If config file not found or invalid
         """
-        config_file = self.config_dir / f"{customer_id}_config.yaml"
-
+        # Try new directory structure first
+        config_file = self.config_dir / customer_id / "config.yaml"
+        
+        # Fall back to old structure if new one doesn't exist
+        if not config_file.exists():
+            config_file = self.config_dir / f"{customer_id}_config.yaml"
+        
         if not config_file.exists():
             raise ConfigurationError(f"Configuration file not found: {config_file}")
@@ -146,13 +194,17 @@ class ConfigLoader:
         # Parse query config
         query_config_data = config_data.get("query_config", {})
+        
+        # Load rewrite dictionary from external file instead of config
+        rewrite_dictionary = self._load_rewrite_dictionary(customer_id)
+        
         query_config = QueryConfig(
             supported_languages=query_config_data.get("supported_languages", ["zh", "en"]),
             default_language=query_config_data.get("default_language", "zh"),
             enable_translation=query_config_data.get("enable_translation", True),
             enable_text_embedding=query_config_data.get("enable_text_embedding", True),
             enable_query_rewrite=query_config_data.get("enable_query_rewrite", True),
-            rewrite_dictionary=query_config_data.get("rewrite_dictionary", {}),
+            rewrite_dictionary=rewrite_dictionary,
             translation_api_key=query_config_data.get("translation_api_key"),
             translation_service=query_config_data.get("translation_service", "deepl")
         )
@@ -342,13 +394,18 @@ class ConfigLoader:
     def save_config(self, config: CustomerConfig, output_path: Optional[str] = None) -> None:
         """
         Save customer configuration to YAML file.
+        
+        Note: rewrite_dictionary is saved separately to query_rewrite.dict file
         Args:
             config: Configuration to save
-            output_path: Optional output path (defaults to config dir)
+            output_path: Optional output path (defaults to new directory structure)
         """
         if output_path is None:
-            output_path = self.config_dir / f"{config.customer_id}_config.yaml"
+            # Use new directory structure by default
+            customer_dir = self.config_dir / config.customer_id
+            customer_dir.mkdir(parents=True, exist_ok=True)
+            output_path = customer_dir / "config.yaml"
         # Convert config back to dictionary format
         config_dict = {
@@ -366,7 +423,7 @@ class ConfigLoader:
                 "enable_translation": config.query_config.enable_translation,
                 "enable_text_embedding": config.query_config.enable_text_embedding,
                 "enable_query_rewrite": config.query_config.enable_query_rewrite,
-                "rewrite_dictionary": config.query_config.rewrite_dictionary,
+                # rewrite_dictionary is stored in separate file, not in config
                 "translation_api_key": config.query_config.translation_api_key,
                 "translation_service": config.query_config.translation_service,
             },
@@ -383,6 +440,25 @@ class ConfigLoader:
         with open(output_path, 'w', encoding='utf-8') as f:
             yaml.dump(config_dict, f, default_flow_style=False, allow_unicode=True)
+        
+        # Save rewrite dictionary to separate file
+        self._save_rewrite_dictionary(config.customer_id, config.query_config.rewrite_dictionary)
+    
+    def _save_rewrite_dictionary(self, customer_id: str, rewrite_dict: Dict[str, str]) -> None:
+        """
+        Save rewrite dictionary to external file.
+        
+        Args:
+            customer_id: Customer identifier
+            rewrite_dict: Dictionary to save
+        """
+        customer_dir = self.config_dir / customer_id
+        customer_dir.mkdir(parents=True, exist_ok=True)
+        dict_file = customer_dir / "query_rewrite.dict"
+        
+        with open(dict_file, 'w', encoding='utf-8') as f:
+            for key, value in rewrite_dict.items():
+                f.write(f"{key}\t{value}\n")
     def _field_to_dict(self, field: FieldConfig) -> Dict[str, Any]:
         """Convert FieldConfig to dictionary."""
@@ -246,12 +246,6 @@ query_config:
   enable_text_embedding: true
   enable_query_rewrite: true
-  # Query Rewrite Dictionary
-  rewrite_dictionary:
-    "芭比": "brand:芭比 OR name:芭比娃娃"
-    "玩具": "category:玩具"
-    "消防": "category:消防 OR name:消防"
-
   # Translation API (DeepL)
   translation_service: "deepl"
   translation_api_key: null  # Set via environment variable
@@ -266,3 +260,4 @@ spu_config:
   enabled: false
   spu_field: null
   inner_hits_size: 3
+
@@ -0,0 +1,4 @@
+芭比	brand:芭比 OR name:芭比娃娃
+玩具	category:玩具
+消防	category:消防 OR name:消防
+
@@ -6,6 +6,7 @@ RequestContext使用示例
 import sys
 import os
+import time
 # 添加项目根目录到Python路径
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -25,7 +26,6 @@ def example_basic_usage():
         # 步骤1: 查询解析
         context.start_stage(RequestContextStage.QUERY_PARSING)
         # 这里调用 query_parser.parse(query, context=context)
-        import time
         time.sleep(0.05)  # 模拟处理时间
         context.end_stage(RequestContextStage.QUERY_PARSING)
@@ -97,7 +97,6 @@ def example_with_searcher():
         # 模拟搜索流程
         context.start_stage(RequestContextStage.QUERY_PARSING)
-        import time
         time.sleep(0.04)
         context.end_stage(RequestContextStage.QUERY_PARSING)
         context.store_query_analysis(
@@ -181,7 +180,6 @@ def example_performance_analysis():
         (RequestContextStage.RERANKING, 42.7)
     ]
-    import time
     for stage, duration_ms in stages_with_durations:
         context.start_stage(stage)
         time.sleep(duration_ms / 1000.0)  # 转换为秒
@@ -6,6 +6,7 @@ Handles field mapping, type conversion, and embedding generation.
 import pandas as pd
 import numpy as np
+import datetime
 from typing import Dict, Any, List, Optional
 from config import CustomerConfig, FieldConfig, FieldType
 from embeddings import BgeEncoder, CLIPImageEncoder
@@ -304,7 +305,6 @@ class DataTransformer:
             elif isinstance(value, str):
                 # Try to parse string datetime and convert to ISO format
                 try:
-                    import datetime
                     # Handle common datetime formats
                     formats = [
                         '%Y-%m-%d %H:%M:%S',    # 2020-07-07 16:44:09
@@ -12,19 +12,21 @@ import sys
 import os
 import argparse
 import json
+import pandas as pd
+import uvicorn
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from config import ConfigLoader
+from utils import ESClient
+from indexer import DataTransformer, IndexingPipeline
+from embeddings import BgeEncoder, CLIPImageEncoder
+from search import Searcher
+
 def cmd_ingest(args):
     """Run data ingestion."""
-    from config import ConfigLoader
-    from utils import ESClient
-    from indexer import DataTransformer, IndexingPipeline
-    from embeddings import BgeEncoder, CLIPImageEncoder
-    import pandas as pd
-
     print(f"Starting ingestion for customer: {args.customer}")
     # Load config
@@ -63,8 +65,6 @@ def cmd_ingest(args):
 def cmd_serve(args):
     """Start API service."""
-    import uvicorn
-
     os.environ['CUSTOMER_ID'] = args.customer
     os.environ['ES_HOST'] = args.es_host
@@ -83,10 +83,6 @@ def cmd_serve(args):
 def cmd_search(args):
     """Test search from command line."""
-    from config import ConfigLoader
-    from utils import ESClient
-    from search import Searcher
-
     # Load config
     config_loader = ConfigLoader("config/schema")
     config = config_loader.load_customer_config(args.customer)
@@ -118,7 +118,6 @@ class TestRunner:
         # 检查Python环境
         try:
-            import sys
             python_version = sys.version
             self.logger.info(f"Python版本: {python_version}")
             checks.append(("Python", True, f"版本 {python_version}"))
@@ -5,6 +5,7 @@ Simple test script to verify aggregation functionality without external dependen
 import sys
 import os
+import inspect
 # Add the project root to the Python path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -110,7 +111,6 @@ def test_searcher_integration():
         return False
     # We can't easily test the full searcher without ES, but we can check the method signature
-    import inspect
     search_method = getattr(Searcher, 'search', None)
     if search_method: