Initial commit

config_manager.py

@@ -0,0 +1,80 @@
+# config_manager.py
+# -*- coding: utf-8 -*-
+import json
+import os
+import threading
+from llm_adapters import create_llm_adapter
+from embedding_adapters import create_embedding_adapter
+
+
+def load_config(config_file: str) -> dict:
+    """从指定的 config_file 加载配置，若不存在则返回空字典。"""
+    if os.path.exists(config_file):
+        try:
+            with open(config_file, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except:
+            pass
+    return {}
+
+def save_config(config_data: dict, config_file: str) -> bool:
+    """将 config_data 保存到 config_file 中，返回 True/False 表示是否成功。"""
+    try:
+        with open(config_file, 'w', encoding='utf-8') as f:
+            json.dump(config_data, f, ensure_ascii=False, indent=4)
+        return True
+    except:
+        return False
+
+def test_llm_config(interface_format, api_key, base_url, model_name, temperature, max_tokens, timeout, log_func, handle_exception_func):
+    """测试当前的LLM配置是否可用"""
+    def task():
+        try:
+            log_func("开始测试LLM配置...")
+            llm_adapter = create_llm_adapter(
+                interface_format=interface_format,
+                base_url=base_url,
+                model_name=model_name,
+                api_key=api_key,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                timeout=timeout
+            )
+
+            test_prompt = "Please reply 'OK'"
+            response = llm_adapter.invoke(test_prompt)
+            if response:
+                log_func("✅ LLM配置测试成功！")
+                log_func(f"测试回复: {response}")
+            else:
+                log_func("❌ LLM配置测试失败：未获取到响应")
+        except Exception as e:
+            log_func(f"❌ LLM配置测试出错: {str(e)}")
+            handle_exception_func("测试LLM配置时出错")
+
+    threading.Thread(target=task, daemon=True).start()
+
+def test_embedding_config(api_key, base_url, interface_format, model_name, log_func, handle_exception_func):
+    """测试当前的Embedding配置是否可用"""
+    def task():
+        try:
+            log_func("开始测试Embedding配置...")
+            embedding_adapter = create_embedding_adapter(
+                interface_format=interface_format,
+                api_key=api_key,
+                base_url=base_url,
+                model_name=model_name
+            )
+
+            test_text = "测试文本"
+            embeddings = embedding_adapter.embed_query(test_text)
+            if embeddings and len(embeddings) > 0:
+                log_func("✅ Embedding配置测试成功！")
+                log_func(f"生成的向量维度: {len(embeddings)}")
+            else:
+                log_func("❌ Embedding配置测试失败：未获取到向量")
+        except Exception as e:
+            log_func(f"❌ Embedding配置测试出错: {str(e)}")
+            handle_exception_func("测试Embedding配置时出错")
+
+    threading.Thread(target=task, daemon=True).start()