Refactor configuration management to use Pydantic for better validation and structure

- Replaced the existing config loading mechanism with a Pydantic-based Settings model.
- Updated all modules to access configuration values through the new settings instance.
- Enhanced the knowledge base and chat configurations for improved clarity and usability.
- Added new fields for API keys and base URLs, ensuring they are loaded from environment variables or .env files.
- Improved the display of configuration information in the console for better user experience.
- Removed deprecated functions and streamlined the codebase for maintainability.

Author: MisonL

config.ini.example

@@ -45,7 +45,8 @@ KB_REPLACE_WHITESPACE = False # 是否在文本预处理时将所有连续的空
 KB_REMOVE_SPACES = False # 是否在文本预处理时移除所有空格 (请谨慎使用)
 KB_REMOVE_URLS = False # 是否在文本预处理时移除URL和电子邮件地址
 KB_USE_QA_SEGMENTATION = False # 是否启用QA对分割模式 (将文档按预设的问答对格式进行切分)
-KB_SPLITTER_SEPARATORS = ### # 文本分割器使用的分隔符，可以设置多个，用逗号隔开
+KB_SPLITTER_SEPARATORS = ### 
+# 文本分割器使用的分隔符。如果要使用多个，请用英文逗号隔开，例如: ###,---,===
 KB_CHUNK_SIZE = 1500 # 文本切块的最大长度 (单位：字符)
 KB_CHUNK_OVERLAP = 150 # 文本切块之间的重叠长度 (单位：字符)
 KB_EMBEDDING_BATCH_SIZE = 32 # 向量化处理时，每批处理的文本数量 (可根据显存或API限制调整)
@@ -58,7 +59,7 @@ DEFAULT_RERANK_PROVIDER = siliconflow # 默认使用的Rerank模型 (必须是
 
 [CHAT]
 # 聊天机器人核心功能配置
-CHAT_RETRIEVAL_METHOD = HYBRID_SEARCH # 检索方法, 可选: SEMANTIC_SEARCH (向量搜索), FULL_TEXT_SEARCH (关键词搜索), HYBRID_SEARCH (混合搜索)
+CHAT_RETRIEVAL_METHOD = HYBRID_SEARCH # 检索方法。可选值: SEMANTIC_SEARCH (或 "向量检索"), FULL_TEXT_SEARCH (或 "全文检索"), HYBRID_SEARCH (或 "混合检索")
 CHAT_VECTOR_WEIGHT = 0.3 # 混合搜索中，向量搜索结果的权重 (与关键词权重相加建议为1)
 CHAT_KEYWORD_WEIGHT = 0.7 # 混合搜索中，关键词搜索结果的权重 (与向量权重相加建议为1)
 CHAT_RERANK_ENABLED = False # 是否启用Rerank精排模型对检索结果进行二次排序

main.py

@@ -62,7 +62,8 @@ def display_banner():
     welcome_text.append("Mison", style="default")
     welcome_text.append("  ·  邮箱: ", style="bold")
     welcome_text.append("1360962086@qq.com", style="default")
-    welcome_text.append("  ·  GitHub: ", style="bold")
+    welcome_text.append("\n") # 换行
+    welcome_text.append("GitHub: ", style="bold")
     # 使用正确的 GitHub 仓库地址
     github_url = "https://github.com/MisonL/PyRAG-Kit"
     welcome_text.append(github_url, style=f"link {github_url}")

requirements.txt

@@ -14,3 +14,5 @@ rank_bm25
 questionary
 pyfiglet
 prompt_toolkit
+pydantic
+pydantic-settings

scripts/embed_knowledge_base.py

@@ -19,10 +19,10 @@
 from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
-from rich.console import Group
+from rich import box
 
 # 从 src 导入重构后的模块
-from src.utils.config import KB_CONFIG, CHAT_CONFIG, API_CONFIG
+from src.utils.config import settings, KB_CONFIG, API_CONFIG
 from src.providers.factory import ModelProviderFactory
 from src.retrieval.retriever import VectorStore # VectorStore现在从这里导入
 from src.ui.display_utils import CONSOLE_WIDTH, get_relative_path
@@ -126,35 +126,68 @@ def process_documents(vector_store: VectorStore):
 # 4. 辅助与主函数 (HELPERS & MAIN)
 # =================================================================
 def display_config_and_confirm():
-    """显示全面的配置信息并请求用户确认。"""
+    """以美观的表格形式显示全面的配置信息，并请求用户确认。"""
     console = Console()
 
     def mask_api_key(key: Optional[str]) -> str:
-        if not key: return "[dim]未设置[/dim]"
-        return f"{key[:10]}..."
-
-    kb_table = Table(title="[bold green]知识库构建配置 (KB_CONFIG)[/bold green]", show_header=False, box=None, padding=(0, 1))
-    kb_table.add_column(style="cyan")
-    kb_table.add_column(style="bold white")
-    active_embedding = KB_CONFIG['active_embedding_configuration']
-    embedding_model_details = KB_CONFIG['embedding_configurations'][active_embedding]
-    kb_table.add_row("激活的嵌入模型:", f"{active_embedding} ({embedding_model_details['provider']}: {embedding_model_details['model_name']})")
-    kb_table.add_row("知识库目录:", get_relative_path(KB_CONFIG['kb_dir']))
-    kb_table.add_row("输出文件:", get_relative_path(KB_CONFIG['output_file']))
+        """对API密钥进行脱敏处理，使其更安全地显示。"""
+        if not key or key == "lm-studio":
+            return "[dim]未设置或无需设置[/dim]"
+        if len(key) > 12:
+            return f"[white]{key[:6]}...{key[-4:]}[/white]"
+        return "[white]已设置[/white]"
+
+    table = Table(
+        box=box.ROUNDED,
+        padding=(0, 2),
+        title="[bold yellow]向量化脚本配置总览[/bold yellow]",
+        show_header=False,
+        width=CONSOLE_WIDTH
+    )
+    # 参数名列: 蓝色
+    table.add_column(justify="right", style="cyan", no_wrap=True, width=28)
+    # 参数值列: 亮白色
+    table.add_column(style="bright_white")
+
+    # --- 知识库配置 ---
+    table.add_row("[bold green]知识库配置[/bold green]", "")
+    table.add_row("知识库目录", f"[bold cyan]{get_relative_path(settings.knowledge_base_path)}[/bold cyan]")
+    table.add_row("输出文件路径", f"[bold cyan]{get_relative_path(settings.pkl_path)}[/bold cyan]")
+    table.add_row("文本切分块大小 (Chunk Size)", f"[bold magenta]{settings.kb_chunk_size}[/bold magenta]")
+    table.add_row("文本切分重叠量 (Overlap)", f"[bold magenta]{settings.kb_chunk_overlap}[/bold magenta]")
+    table.add_row("切分分隔符 (Separators)", f"[bold bright_white]{settings.kb_splitter_separators}[/bold bright_white]")
+    table.add_section()
+
+    # --- 模型与API配置 ---
+    table.add_row("[bold green]模型与API配置[/bold green]", "")
+    active_embedding_key = settings.default_embedding_provider
+    embedding_model_details = settings.embedding_configurations[active_embedding_key]
+    provider = embedding_model_details.provider
 
-    api_table = Table(title="[bold green]相关API配置[/bold green]", show_header=False, box=None, padding=(0, 1))
-    api_table.add_column(style="cyan")
-    api_table.add_column(style="bold white")
-    provider = embedding_model_details['provider']
-    key_name = f"{provider.upper()}_API_KEY"
-    url_name = f"{provider.upper()}_BASE_URL"
-    if key_name in API_CONFIG:
-        api_table.add_row(f"{key_name}:", mask_api_key(API_CONFIG.get(key_name)))
-    if url_name in API_CONFIG and API_CONFIG.get(url_name):
-        api_table.add_row(f"{url_name}:", API_CONFIG.get(url_name))
-
-    console.print(Panel(Group(kb_table, api_table), title="[bold yellow]向量化脚本配置总览[/bold yellow]", border_style="blue", width=CONSOLE_WIDTH))
-    console.print("[yellow]配置信息来源于 [bold]src/utils/config.py[/bold] 和 [bold].env[/bold] 文件。[/yellow]")
+    table.add_row("激活的嵌入模型 (Provider)", f"[bold green]{active_embedding_key}[/bold green] ([dim]{provider}[/dim])")
+    table.add_row("模型名称 (Model Name)", f"[bold bright_white]{embedding_model_details.model_name}[/bold bright_white]")
+
+    # 动态获取API Key和Base URL的字段名
+    key_field_name = f"{provider.lower()}_api_key"
+    
+    # 处理不一致的URL字段名
+    url_field_name = ""
+    if provider == "openai":
+        url_field_name = "openai_api_base"
+    elif hasattr(settings, f"{provider.lower()}_base_url"):
+        url_field_name = f"{provider.lower()}_base_url"
+
+    api_key_value = getattr(settings, key_field_name, None)
+    # mask_api_key 函数已内置样式，无需额外添加
+    table.add_row(f"对应的 API Key ({key_field_name.upper()})", mask_api_key(api_key_value))
+
+    if url_field_name:
+        base_url_value = getattr(settings, url_field_name, None)
+        if base_url_value:
+            table.add_row(f"对应的 Base URL ({url_field_name.upper()})", f"[bold cyan]{base_url_value}[/bold cyan]")
+
+    console.print(table)
+    console.print("[yellow]配置信息来源于 [bold]config.ini[/bold], [bold].env[/bold] 文件或 [bold]环境变量[/bold]。[/yellow]")
 
     choice = console.input("是否使用以上配置继续处理？ ([bold green]y[/bold green]/[bold red]n[/bold red]): ").lower()
     if choice not in ['y', 'yes']:

src/chat/core.py

@@ -20,7 +20,7 @@
 # 使用相对导入
 from ..providers.factory import ModelProviderFactory
 from ..providers.__base__.model_provider import LargeLanguageModel
-from ..utils.config import CHAT_CONFIG, KB_CONFIG, LOG_PATH
+from ..utils.config import CHAT_CONFIG, KB_CONFIG, LOG_PATH, PKL_PATH, settings
 from ..ui.config_menu import launch_config_editor
 from ..ui.display_utils import display_chat_config
 from ..retrieval.retriever import VectorStore, retrieve_documents
@@ -59,7 +59,7 @@ def __init__(self, console: Console):
         self.reload_llm() # 初始加载
 
     def _load_vector_store(self) -> Optional[VectorStore]:
-        file_path = KB_CONFIG["output_file"]
+        file_path = str(PKL_PATH)
         try:
             with open(file_path, "rb") as f:
                 data = pickle.load(f)
@@ -78,7 +78,7 @@ def _load_vector_store(self) -> Optional[VectorStore]:
     def reload_llm(self) -> bool:
         """重新加载或初始化LLM模型。"""
         try:
-            active_llm_key = CHAT_CONFIG['active_llm_configuration']
+            active_llm_key = settings.default_llm_provider
             self.console.print(f"[dim]正在加载LLM: [bold cyan]{active_llm_key}[/bold cyan]...[/dim]")
             self.llm_model = ModelProviderFactory.get_llm_provider(active_llm_key)
             if self.llm_model:
@@ -103,7 +103,7 @@ def _identify_intent(self, user_query: str) -> str:
 
     def _retrieve_knowledge(self, query: str) -> List[Dict[str, Any]]:
         if not self.vector_store: return []
-        self.console.print(f"[dim]正在使用 '[yellow]{CHAT_CONFIG['retrieval_method'].value}[/yellow]' 模式检索...[/dim]")
+        self.console.print(f"[dim]正在使用 '[yellow]{settings.chat_retrieval_method.value}[/yellow]' 模式检索...[/dim]")
         return retrieve_documents(query, self.vector_store, self.console)
 
     def _generate_answer_stream(self, user_query: str, intent: str, retrieved_docs: List[Dict[str, Any]]) -> "Generator[str, None, None]":
@@ -181,7 +181,7 @@ def start_chat_session():
 
     if bot.llm_model:
         display_chat_config(console)
-        console.print(f"我是你的智能客服（由 [bold green]{CHAT_CONFIG['active_llm_configuration']}[/bold green] 支持），请输入问题（输入'[bold]/quit[/bold]'或'[bold]/config[/bold]'）：")
+        console.print(f"我是你的智能客服（由 [bold green]{settings.default_llm_provider}[/bold green] 支持），请输入问题（输入'[bold]/quit[/bold]'或'[bold]/config[/bold]'）：")
 
         while True:
             try:
@@ -197,7 +197,7 @@ def start_chat_session():
                         console.print("[yellow]检测到LLM配置变更，正在重载模型...[/yellow]")
                         bot.reload_llm()
                     display_chat_config(console)
-                    console.print(f"我是你的智能客服（由 [bold green]{CHAT_CONFIG['active_llm_configuration']}[/bold green] 支持），请输入问题（输入'[bold]/quit[/bold]'或'[bold]/config[/bold]'）：")
+                    console.print(f"我是你的智能客服（由 [bold green]{settings.default_llm_provider}[/bold green] 支持），请输入问题（输入'[bold]/quit[/bold]'或'[bold]/config[/bold]'）：")
                     continue
 
                 answer_stream = bot.chat(user_query)

src/providers/factory.py

@@ -7,12 +7,7 @@
     RerankModel,
     TextEmbeddingModel,
 )
-from src.utils.config import (
-    CHAT_CONFIG,
-    EMBEDDING_CONFIGS,
-    LLM_CONFIGS,
-    RERANK_CONFIGS,
-)
+from src.utils.config import settings
 
 class ModelProviderFactory:
     """模型提供商工厂"""
@@ -48,44 +43,44 @@ def _get_provider_class(provider_name: str) -> Type:
     @staticmethod
     def get_llm_provider(provider_key: str) -> LargeLanguageModel:
         """获取一个语言模型提供商实例"""
-        if provider_key not in LLM_CONFIGS:
+        if provider_key not in settings.llm_configurations:
             raise ValueError(f"在LLM配置中未找到key: {provider_key}")
 
-        config = LLM_CONFIGS[provider_key]
-        provider_name = config["provider"]
-        model_name = config["model_name"]
+        config = settings.llm_configurations[provider_key]
+        provider_name = config.provider
+        model_name = config.model_name
 
         ProviderClass = ModelProviderFactory._get_provider_class(provider_name)
         return ProviderClass(model_name=model_name)
 
     @staticmethod
     def get_embedding_provider(provider_key: str) -> TextEmbeddingModel:
         """获取一个文本向量化模型提供商实例"""
-        if provider_key not in EMBEDDING_CONFIGS:
+        if provider_key not in settings.embedding_configurations:
             raise ValueError(f"在Embedding配置中未找到key: {provider_key}")
 
-        config = EMBEDDING_CONFIGS[provider_key]
-        provider_name = config["provider"]
-        model_name = config["model_name"]
+        config = settings.embedding_configurations[provider_key]
+        provider_name = config.provider
+        model_name = config.model_name
 
         ProviderClass = ModelProviderFactory._get_provider_class(provider_name)
         return ProviderClass(model_name=model_name)
 
     @staticmethod
     def get_rerank_provider(provider_key: str) -> RerankModel:
         """获取一个Rerank模型提供商实例"""
-        if provider_key not in RERANK_CONFIGS:
+        if provider_key not in settings.rerank_configurations:
             raise ValueError(f"在Rerank配置中未找到key: {provider_key}")
 
-        config = RERANK_CONFIGS[provider_key]
+        config = settings.rerank_configurations[provider_key]
         # Rerank提供商的key可能与LLM/Embedding提供商的key冲突（如siliconflow）
         # 因此，我们在这里使用一个特殊的key，或者直接在配置中指定provider_map的key
         # 为了简单起见，我们假设rerank的provider name是唯一的
-        provider_name = config["provider"]
+        provider_name = config.provider
         if provider_name == "siliconflow":
             provider_name = "siliconflow_rerank" # 映射到唯一的rerank provider
 
-        model_name = config["model_name"]
+        model_name = config.model_name
 
         ProviderClass = ModelProviderFactory._get_provider_class(provider_name)
         return ProviderClass(model_name=model_name)

src/providers/grok.py

@@ -4,7 +4,7 @@
 import requests
 
 from src.providers.__base__.model_provider import LargeLanguageModel
-from src.utils.config import API_CONFIG
+from src.utils.config import settings
 
 
 class GrokProvider(LargeLanguageModel):
@@ -14,10 +14,10 @@ class GrokProvider(LargeLanguageModel):
 
     def __init__(self, model_name: str):
         self._model_name = model_name
-        self._api_key = API_CONFIG.get("GROK_API_KEY")
+        self._api_key = settings.grok_api_key
         if not self._api_key:
             raise ValueError("Grok配置不完整：缺少 GROK_API_KEY。")
-        self._base_url = API_CONFIG.get("GROK_BASE_URL", "https://api.x.ai/v1")
+        self._base_url = str(settings.grok_base_url)
 
     def invoke(
         self,

src/providers/jina.py

@@ -4,7 +4,7 @@
 import requests
 
 from src.providers.__base__.model_provider import RerankModel
-from src.utils.config import API_CONFIG
+from src.utils.config import settings
 
 
 class JinaProvider(RerankModel):
@@ -14,7 +14,7 @@ class JinaProvider(RerankModel):
 
     def __init__(self, model_name: str):
         self._model_name = model_name
-        self._api_key = API_CONFIG.get("JINA_API_KEY")
+        self._api_key = settings.jina_api_key
         if not self._api_key:
             raise ValueError("错误：Jina Rerank 提供商需要 API 密钥。")
         self._base_url = "https://api.jina.ai/v1/rerank"

src/providers/volcengine.py

@@ -11,7 +11,7 @@
     LargeLanguageModel,
     TextEmbeddingModel,
 )
-from src.utils.config import API_CONFIG
+from src.utils.config import settings
 
 
 class VolcengineProvider(LargeLanguageModel, TextEmbeddingModel):
@@ -21,9 +21,9 @@ class VolcengineProvider(LargeLanguageModel, TextEmbeddingModel):
 
     def __init__(self, model_name: str):
         self._model_name = model_name
-        self._access_key = API_CONFIG.get("VOLC_ACCESS_KEY")
-        self._secret_key = API_CONFIG.get("VOLC_SECRET_KEY")
-        self._base_url = API_CONFIG.get("VOLC_BASE_URL")
+        self._access_key = settings.volc_access_key
+        self._secret_key = settings.volc_secret_key
+        self._base_url = str(settings.volc_base_url)
 
         if not all([self._access_key, self._secret_key, self._base_url]):
             raise ValueError("火山引擎配置不完整：缺少 VOLC_ACCESS_KEY, VOLC_SECRET_KEY, 或 VOLC_BASE_URL。")

src/retrieval/retriever.py

@@ -11,7 +11,7 @@
 from rich.console import Console
 
 # 使用相对导入来引用同一 src 目录下的模块
-from ..utils.config import CHAT_CONFIG, KB_CONFIG, RetrievalMethod
+from ..utils.config import CHAT_CONFIG, KB_CONFIG, RetrievalMethod, settings
 from ..providers.factory import ModelProviderFactory
 
 # =================================================================
@@ -74,12 +74,12 @@ def rerank(self, documents: List[Dict]) -> List[Dict]:
         return sorted(documents, key=lambda x: x["score"], reverse=True)
 
 def retrieve_documents(query: str, vector_store: VectorStore, console: Console) -> List[Dict]:
-    retrieval_method = CHAT_CONFIG["retrieval_method"]
-    top_k = CHAT_CONFIG["top_k"]
-    score_threshold = CHAT_CONFIG["score_threshold"]
+    retrieval_method = settings.chat_retrieval_method
+    top_k = settings.chat_top_k
+    score_threshold = settings.chat_score_threshold
 
     # 语义搜索
-    active_embedding_key = KB_CONFIG['active_embedding_configuration']
+    active_embedding_key = settings.default_embedding_provider
     embedding_provider = ModelProviderFactory.get_embedding_provider(active_embedding_key)
     query_embedding = np.array(embedding_provider.embed_documents([query])[0])
     semantic_results = vector_store.semantic_search(query_embedding, top_k, score_threshold)
@@ -103,8 +103,8 @@ def retrieve_documents(query: str, vector_store: VectorStore, console: Console)
         ranked_results = sorted(full_text_results, key=lambda x: x.get("keyword_score", 0), reverse=True)
 
     # 使用Reranker（如果启用）
-    if CHAT_CONFIG["rerank_enabled"]:
-        active_rerank_key = CHAT_CONFIG['active_rerank_configuration']
+    if settings.chat_rerank_enabled:
+        active_rerank_key = settings.default_rerank_provider
         rerank_provider = ModelProviderFactory.get_rerank_provider(active_rerank_key)
         if rerank_provider and ranked_results:
             console.print(f"[dim]正在使用 '{active_rerank_key}' 进行重排...[/dim]")