diff --git a/src/app/deployment/models.py b/src/app/deployment/models.py index 0edcc2bbdd26968b7d86186ae1ab4eaf2a6fbdef..43b9514337a8cb7eb265389fa28ae14f312b35d8 100644 --- a/src/app/deployment/models.py +++ b/src/app/deployment/models.py @@ -87,6 +87,60 @@ class DeploymentConfig: return len(errors) == 0, errors + async def validate_llm_connectivity(self) -> tuple[bool, str, dict]: + """ + 验证 LLM API 连接性和功能 + + 单独验证 LLM 配置的有效性，包括模型可用性和 function_call 支持。 + 当 LLM 的3个核心字段（endpoint、api_key、model）填完后调用。 + + Returns: + tuple[bool, str, dict]: (是否验证成功, 消息, 验证详细信息) + + """ + from .validators import APIValidator + + # 检查必要字段是否完整 + if not (self.llm.endpoint.strip() and self.llm.api_key.strip() and self.llm.model.strip()): + return False, "LLM 基础配置不完整", {} + + validator = APIValidator() + llm_valid, llm_msg, llm_info = await validator.validate_llm_config( + self.llm.endpoint, + self.llm.api_key, + self.llm.model, + self.llm.request_timeout, + ) + + return llm_valid, llm_msg, llm_info + + async def validate_embedding_connectivity(self) -> tuple[bool, str, dict]: + """ + 验证 Embedding API 连接性和功能 + + 单独验证 Embedding 配置的有效性。 + 当 Embedding 的3个核心字段（endpoint、api_key、model）填完后调用。 + + Returns: + tuple[bool, str, dict]: (是否验证成功, 消息, 验证详细信息) + + """ + from .validators import APIValidator + + # 检查必要字段是否完整 + if not (self.embedding.endpoint.strip() and self.embedding.api_key.strip() and self.embedding.model.strip()): + return False, "Embedding 基础配置不完整", {} + + validator = APIValidator() + embed_valid, embed_msg, embed_info = await validator.validate_embedding_config( + self.embedding.endpoint, + self.embedding.api_key, + self.embedding.model, + self.llm.request_timeout, # 使用相同的超时设置 + ) + + return embed_valid, embed_msg, embed_info + def _validate_basic_fields(self) -> list[str]: """验证基础字段""" errors = [] diff --git a/src/app/deployment/ui.py b/src/app/deployment/ui.py index 6f236b9c434b450b33611111e67a53bdfd4ed18c..f0ff7b03b3030b8e328f3e5971e4a4f982bd96d1 100644 --- a/src/app/deployment/ui.py +++ b/src/app/deployment/ui.py @@ -92,12 +92,18 @@ class DeploymentConfigScreen(ModalScreen[bool]): margin: 0 1; background: $error; } + + #llm_validation_status, #embedding_validation_status { + text-style: italic; + } """ def __init__(self) -> None: """初始化部署配置屏幕""" super().__init__() self.config = DeploymentConfig() + self._llm_validation_task: asyncio.Task[None] | None = None + self._embedding_validation_task: asyncio.Task[None] | None = None def compose(self) -> ComposeResult: """组合界面组件""" @@ -168,6 +174,11 @@ class DeploymentConfigScreen(ModalScreen[bool]): classes="form-input", ) + # LLM 验证状态显示 + with Horizontal(classes="form-row"): + yield Label("验证状态:", classes="form-label") + yield Static("未验证", id="llm_validation_status", classes="form-input") + with Horizontal(classes="form-row"): yield Label("最大 Token 数:", classes="form-label") yield Input( @@ -231,6 +242,11 @@ class DeploymentConfigScreen(ModalScreen[bool]): classes="form-input", ) + # Embedding 验证状态显示 + with Horizontal(classes="form-row"): + yield Label("验证状态:", classes="form-label") + yield Static("未验证", id="embedding_validation_status", classes="form-input") + def _compose_deployment_options(self) -> ComposeResult: """组合部署选项组件""" with Vertical(classes="form-section"): @@ -288,6 +304,147 @@ class DeploymentConfigScreen(ModalScreen[bool]): self.query_one("#enable_web", Switch).value = False self.query_one("#enable_rag", Switch).value = False + @on(Input.Changed, "#llm_endpoint, #llm_api_key, #llm_model") + async def on_llm_field_changed(self, event: Input.Changed) -> None: + """处理 LLM 字段变化，检查是否需要自动验证""" + # 取消之前的验证任务 + if self._llm_validation_task and not self._llm_validation_task.done(): + self._llm_validation_task.cancel() + + # 检查是否所有核心字段都已填写 + if self._should_validate_llm(): + # 延迟 1 秒后进行验证，避免用户快速输入时频繁触发 + self._llm_validation_task = asyncio.create_task(self._delayed_llm_validation()) + + @on(Input.Changed, "#embedding_endpoint, #embedding_api_key, #embedding_model") + async def on_embedding_field_changed(self, event: Input.Changed) -> None: + """处理 Embedding 字段变化，检查是否需要自动验证""" + # 取消之前的验证任务 + if self._embedding_validation_task and not self._embedding_validation_task.done(): + self._embedding_validation_task.cancel() + + # 检查是否所有核心字段都已填写 + if self._should_validate_embedding(): + # 延迟 1 秒后进行验证，避免用户快速输入时频繁触发 + self._embedding_validation_task = asyncio.create_task(self._delayed_embedding_validation()) + + def _should_validate_llm(self) -> bool: + """检查是否应该验证 LLM 配置""" + try: + endpoint = self.query_one("#llm_endpoint", Input).value.strip() + api_key = self.query_one("#llm_api_key", Input).value.strip() + model = self.query_one("#llm_model", Input).value.strip() + return bool(endpoint and api_key and model) + except (AttributeError, ValueError): + return False + + def _should_validate_embedding(self) -> bool: + """检查是否应该验证 Embedding 配置""" + try: + endpoint = self.query_one("#embedding_endpoint", Input).value.strip() + api_key = self.query_one("#embedding_api_key", Input).value.strip() + model = self.query_one("#embedding_model", Input).value.strip() + return bool(endpoint and api_key and model) + except (AttributeError, ValueError): + return False + + async def _delayed_llm_validation(self) -> None: + """延迟 LLM 验证""" + try: + await asyncio.sleep(1) # 等待 1 秒 + await self._validate_llm_config() + except asyncio.CancelledError: + pass + + async def _delayed_embedding_validation(self) -> None: + """延迟 Embedding 验证""" + try: + await asyncio.sleep(1) # 等待 1 秒 + await self._validate_embedding_config() + except asyncio.CancelledError: + pass + + async def _validate_llm_config(self) -> None: + """验证 LLM 配置""" + # 更新状态为验证中 + status_widget = self.query_one("#llm_validation_status", Static) + status_widget.update("[yellow]验证中...[/yellow]") + + # 收集当前 LLM 配置 + self._collect_llm_config() + + try: + # 执行验证 + is_valid, message, info = await self.config.validate_llm_connectivity() + + # 更新验证状态 + if is_valid: + status_widget.update(f"[green]✓ {message}[/green]") + # 检查是否支持 function_call + if info.get("supports_function_call"): + self.notify("LLM 验证成功，支持 function_call 功能", severity="information") + else: + self.notify("LLM 验证成功，但不支持 function_call 功能", severity="warning") + else: + status_widget.update(f"[red]✗ {message}[/red]") + self.notify(f"LLM 验证失败: {message}", severity="error") + + except (OSError, ValueError, TypeError) as e: + status_widget.update(f"[red]✗ 验证异常: {e}[/red]") + self.notify(f"LLM 验证过程中出现异常: {e}", severity="error") + + async def _validate_embedding_config(self) -> None: + """验证 Embedding 配置""" + # 更新状态为验证中 + status_widget = self.query_one("#embedding_validation_status", Static) + status_widget.update("[yellow]验证中...[/yellow]") + + # 收集当前 Embedding 配置 + self._collect_embedding_config() + + try: + # 执行验证 + is_valid, message, info = await self.config.validate_embedding_connectivity() + + # 更新验证状态 + if is_valid: + status_widget.update(f"[green]✓ {message}[/green]") + # 显示维度信息 + dimension = info.get("dimension", "未知") + self.notify(f"Embedding 验证成功，向量维度: {dimension}", severity="information") + else: + status_widget.update(f"[red]✗ {message}[/red]") + self.notify(f"Embedding 验证失败: {message}", severity="error") + + except (OSError, ValueError, TypeError) as e: + status_widget.update(f"[red]✗ 验证异常: {e}[/red]") + self.notify(f"Embedding 验证过程中出现异常: {e}", severity="error") + + def _collect_llm_config(self) -> None: + """收集 LLM 配置""" + try: + self.config.llm.endpoint = self.query_one("#llm_endpoint", Input).value.strip() + self.config.llm.api_key = self.query_one("#llm_api_key", Input).value.strip() + self.config.llm.model = self.query_one("#llm_model", Input).value.strip() + self.config.llm.max_tokens = int(self.query_one("#llm_max_tokens", Input).value or "8192") + self.config.llm.temperature = float(self.query_one("#llm_temperature", Input).value or "0.7") + self.config.llm.request_timeout = int(self.query_one("#llm_timeout", Input).value or "300") + except (ValueError, AttributeError): + # 如果转换失败，使用默认值 + pass + + def _collect_embedding_config(self) -> None: + """收集 Embedding 配置""" + try: + embedding_type_value = self.query_one("#embedding_type", Select).value + self.config.embedding.type = str(embedding_type_value) if embedding_type_value else "openai" + self.config.embedding.endpoint = self.query_one("#embedding_endpoint", Input).value.strip() + self.config.embedding.api_key = self.query_one("#embedding_api_key", Input).value.strip() + self.config.embedding.model = self.query_one("#embedding_model", Input).value.strip() + except AttributeError: + # 如果获取失败，使用默认值 + pass + def _collect_config(self) -> bool: """收集用户配置""" try: diff --git a/src/app/deployment/validators.py b/src/app/deployment/validators.py new file mode 100644 index 0000000000000000000000000000000000000000..8603ba4138134a9170eada00ab962fa4ec3539db --- /dev/null +++ b/src/app/deployment/validators.py @@ -0,0 +1,261 @@ +""" +配置验证器 + +提供实际 API 调用验证配置的有效性。 +""" + +import asyncio +from typing import Any + +from openai import APIError, AsyncOpenAI, AuthenticationError, OpenAIError + +from log.manager import get_logger + +# 常量定义 +MAX_MODEL_DISPLAY = 5 + + +class APIValidator: + """API 配置验证器""" + + def __init__(self) -> None: + """初始化验证器""" + self.logger = get_logger(__name__) + + async def validate_llm_config( + self, + endpoint: str, + api_key: str, + model: str, + timeout: int = 30, + ) -> tuple[bool, str, dict[str, Any]]: + """ + 验证 LLM 配置 + + Args: + endpoint: API 端点 + api_key: API 密钥 + model: 模型名称 + timeout: 超时时间（秒） + + Returns: + tuple[bool, str, dict]: (是否验证成功, 错误/成功消息, 额外信息) + + """ + self.logger.info("开始验证 LLM 配置 - 端点: %s, 模型: %s", endpoint, model) + + try: + client = AsyncOpenAI(api_key=api_key, base_url=endpoint, timeout=timeout) + + # 1. 验证模型是否存在 + models_valid, models_msg, available_models = await self._check_model_availability( + client, + model, + ) + if not models_valid: + await client.close() + return False, models_msg, {"available_models": available_models} + + # 2. 验证基本对话功能 + chat_valid, chat_msg = await self._test_basic_chat(client, model) + if not chat_valid: + await client.close() + return False, chat_msg, {"available_models": available_models} + + # 3. 验证 function_call 支持 + func_valid, func_msg, func_info = await self._test_function_call(client, model) + + await client.close() + + if chat_valid: + success_msg = f"LLM 配置验证成功 - 模型: {model}" + if func_valid: + success_msg += " (支持 function_call)" + else: + success_msg += f" (不支持 function_call: {func_msg})" + + return True, success_msg, { + "available_models": available_models, + "supports_function_call": func_valid, + "function_call_info": func_info, + } + + except asyncio.TimeoutError: + return False, f"连接超时 - 无法在 {timeout} 秒内连接到 {endpoint}", {} + except (AuthenticationError, APIError, OpenAIError) as e: + error_msg = f"LLM 配置验证失败: {e!s}" + self.logger.exception(error_msg) + return False, error_msg, {} + + return False, "未知错误", {} + + async def validate_embedding_config( + self, + endpoint: str, + api_key: str, + model: str, + timeout: int = 30, + ) -> tuple[bool, str, dict[str, Any]]: + """ + 验证 Embedding 配置 + + Args: + endpoint: API 端点 + api_key: API 密钥 + model: 模型名称 + timeout: 超时时间（秒） + + Returns: + tuple[bool, str, dict]: (是否验证成功, 错误/成功消息, 额外信息) + + """ + self.logger.info("开始验证 Embedding 配置 - 端点: %s, 模型: %s", endpoint, model) + + try: + client = AsyncOpenAI(api_key=api_key, base_url=endpoint, timeout=timeout) + + # 测试 embedding 功能 + test_text = "这是一个测试文本" + response = await client.embeddings.create(input=test_text, model=model) + + await client.close() + except asyncio.TimeoutError: + return False, f"连接超时 - 无法在 {timeout} 秒内连接到 {endpoint}", {} + except (AuthenticationError, APIError, OpenAIError) as e: + error_msg = f"Embedding 配置验证失败: {e!s}" + self.logger.exception(error_msg) + return False, error_msg, {} + else: + if response.data and len(response.data) > 0: + embedding = response.data[0].embedding + dimension = len(embedding) + return True, f"Embedding 配置验证成功 - 模型: {model}, 维度: {dimension}", { + "model": model, + "dimension": dimension, + "sample_embedding_length": len(embedding), + } + + return False, "Embedding 响应为空", {} + + async def _check_model_availability( + self, + client: AsyncOpenAI, + target_model: str, + ) -> tuple[bool, str, list[str]]: + """检查模型是否可用""" + try: + models_response = await client.models.list() + available_models = [model.id for model in models_response.data] + + if target_model in available_models: + return True, f"模型 {target_model} 可用", available_models + + return ( + False, + ( + f"模型 {target_model} 不可用。可用模型: " + f"{', '.join(available_models[:MAX_MODEL_DISPLAY])}" + f"{'...' if len(available_models) > MAX_MODEL_DISPLAY else ''}" + ), + available_models, + ) + except (AuthenticationError, APIError, OpenAIError) as e: + return False, f"获取模型列表失败: {e!s}", [] + + async def _test_basic_chat( + self, + client: AsyncOpenAI, + model: str, + ) -> tuple[bool, str]: + """测试基本对话功能""" + try: + response = await client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": "请回复'测试成功'"}], + max_tokens=10, + ) + except (AuthenticationError, APIError, OpenAIError) as e: + return False, f"基本对话测试失败: {e!s}" + else: + if response.choices and len(response.choices) > 0: + return True, "基本对话功能正常" + + return False, "对话响应为空" + + async def _test_function_call( + self, + client: AsyncOpenAI, + model: str, + ) -> tuple[bool, str, dict[str, Any]]: + """测试 function_call 支持""" + try: + # 定义一个简单的测试函数 + test_function = { + "name": "get_current_time", + "description": "获取当前时间", + "parameters": {"type": "object", "properties": {}, "required": []}, + } + + response = await client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": "请调用函数获取当前时间"}], + functions=[test_function], # type: ignore[arg-type] + function_call="auto", + max_tokens=50, + ) + except (AuthenticationError, APIError, OpenAIError) as e: + # 如果 functions 参数不支持，尝试 tools 格式 + if "functions" in str(e).lower() or "function_call" in str(e).lower(): + return await self._test_tools_format(client, model) + return False, f"function_call 测试失败: {e!s}", {"supports_functions": False} + else: + if response.choices and len(response.choices) > 0: + choice = response.choices[0] + if hasattr(choice.message, "function_call") and choice.message.function_call: + return True, "支持 function_call", { + "function_name": choice.message.function_call.name, + "supports_functions": True, + } + + # 尝试 tools 格式（OpenAI API 新版本） + return await self._test_tools_format(client, model) + + return False, "function_call 响应为空", {"supports_functions": False} + + async def _test_tools_format( + self, + client: AsyncOpenAI, + model: str, + ) -> tuple[bool, str, dict[str, Any]]: + """测试新版 tools 格式的 function calling""" + try: + test_tool = { + "type": "function", + "function": { + "name": "get_current_time", + "description": "获取当前时间", + "parameters": {"type": "object", "properties": {}, "required": []}, + }, + } + + response = await client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": "请调用函数获取当前时间"}], + tools=[test_tool], # type: ignore[arg-type] + tool_choice="auto", + max_tokens=50, + ) + except (AuthenticationError, APIError, OpenAIError) as e: + return False, f"tools 格式测试失败: {e!s}", {"supports_functions": False} + else: + if response.choices and len(response.choices) > 0: + choice = response.choices[0] + if hasattr(choice.message, "tool_calls") and choice.message.tool_calls: + tool_call = choice.message.tool_calls[0] + return True, "支持 tools 格式的 function_call", { + "function_name": tool_call.function.name, + "supports_functions": True, + "format": "tools", + } + + return False, "不支持 function_call 功能", {"supports_functions": False} diff --git a/tests/app/deployment/test_validate_llm_config.py b/tests/app/deployment/test_validate_llm_config.py new file mode 100644 index 0000000000000000000000000000000000000000..77f2e2e377fc4e4de7b06a8032a875cd9a7297da --- /dev/null +++ b/tests/app/deployment/test_validate_llm_config.py @@ -0,0 +1,137 @@ +""" +API 配置验证功能演示 + +简单演示如何使用新的验证功能。 +使用方法: source .venv/bin/activate && PYTHONPATH=src python tests/app/deployment/test_validate_llm_config.py +""" + +import asyncio +import sys +from typing import Any + +from app.deployment.models import DeploymentConfig, EmbeddingConfig, LLMConfig + + +def _output(message: str = "") -> None: + """输出消息到标准输出""" + sys.stdout.write(f"{message}\n") + sys.stdout.flush() + + +def _output_llm_validation_info(llm_info: dict[str, Any]) -> None: + """输出 LLM 验证信息""" + _output(f" 📱 LLM: {llm_info['message']}") + + if llm_info.get("supports_function_call"): + _output(" 🔧 Function Call: ✅ 支持") + if "function_call_info" in llm_info: + format_type = llm_info["function_call_info"].get("format", "unknown") + _output(f" 📋 支持格式: {format_type}") + else: + _output(" 🔧 Function Call: ❌ 不支持") + + if "available_models" in llm_info: + models = llm_info["available_models"][:3] + _output(f" 📦 可用模型示例: {', '.join(models)}") + + +def _output_embedding_validation_info(embed_info: dict[str, Any]) -> None: + """输出 Embedding 验证信息""" + _output(f" 🔢 Embedding: {embed_info['message']}") + if "dimension" in embed_info: + _output(f" 📐 向量维度: {embed_info['dimension']}") + + +async def main() -> None: + """主演示函数""" + _output("🔧 API 配置验证演示") + _output("=" * 40) + + config = DeploymentConfig( + server_ip="127.0.0.1", + deployment_mode="light", + llm=LLMConfig( + endpoint="http://127.0.0.1:1234/v1", + api_key="lm-studio", + model="qwen/qwen3-30b-a3b-2507", + max_tokens=4096, + temperature=0.7, + request_timeout=30, + ), + embedding=EmbeddingConfig( + type="openai", + endpoint="http://127.0.0.1:1234/v1", + api_key="lm-studio", + model="text-embedding-bge-m3", + ), + ) + + _output("📋 步骤 1: 基础字段验证") + is_valid, errors = config.validate() + if not is_valid: + _output("❌ 基础验证失败:") + for error in errors: + _output(f" • {error}") + return + _output("✅ 基础验证通过") + + _output("\n🌐 步骤 2: API 连接性验证") + _output("⚠️ 注意: 需要有效的 API 密钥才能通过此步骤") + try: + # 分别验证 LLM 和 Embedding 配置 + llm_valid, llm_msg, llm_info = await config.validate_llm_connectivity() + embed_valid, embed_msg, embed_info = await config.validate_embedding_connectivity() + + api_valid = llm_valid and embed_valid + api_errors = [] + + if not llm_valid: + api_errors.append(f"LLM 验证失败: {llm_msg}") + if not embed_valid: + api_errors.append(f"Embedding 验证失败: {embed_msg}") + + validation_info = { + "llm": { + "valid": llm_valid, + "message": llm_msg, + **llm_info, + }, + "embedding": { + "valid": embed_valid, + "message": embed_msg, + **embed_info, + }, + } + + if not api_valid: + _output("❌ API 验证失败:") + for error in api_errors: + _output(f" • {error}") + return + + _output("✅ API 验证成功!") + if "llm" in validation_info: + _output_llm_validation_info(validation_info["llm"]) + if "embedding" in validation_info: + _output_embedding_validation_info(validation_info["embedding"]) + + except (ConnectionError, TimeoutError, ValueError) as e: + _output(f"⚠️ 验证过程异常: {e}") + _output("💡 通常是网络连接或 API 密钥问题") + + +if __name__ == "__main__": + _output("🚀 开始演示...") + _output("💡 运行方法: ") + _output("💡 source .venv/bin/activate && PYTHONPATH=src python tests/app/deployment/test_validate_llm_config.py") + _output() + + asyncio.run(main()) + + _output("\n" + "=" * 40) + _output("📝 验证功能特点:") + _output("✓ API 连接性测试") + _output("✓ 模型可用性检查") + _output("✓ Function Call 支持检测") + _output("✓ Embedding 向量维度验证") + _output("✓ 支持 OpenAI 和兼容 API")