fix(vlm): make OpenAI keepalive expiry configurable

docs/en/guides/01-configuration.md

@@ -639,11 +639,12 @@ Vision Language Model for semantic extraction (L0/L1 generation).
 | `thinking` | bool | Enable thinking mode for VolcEngine models (default: `false`) |
 | `max_concurrent` | int | Maximum concurrent semantic LLM calls (default: `32`) |
 | `max_retries` | int | Maximum retry attempts for transient VLM provider errors (default: `3`; `0` disables retry) |
-| `credentials` | array | Ordered VLM credential/model list, with index 0 having the highest priority. Each item can override `provider`, `model`, `api_key`, `api_base`, `api_version`, `extra_headers`, `extra_request_body`, `stream`, and `reasoning_effort` |
+| `credentials` | array | Ordered VLM credential/model list, with index 0 having the highest priority. Each item can override `provider`, `model`, `api_key`, `api_base`, `api_version`, `extra_headers`, `extra_request_body`, `stream`, `reasoning_effort`, and `keepalive_expiry` |
 | `failback_timeout_seconds` | float | Time threshold for attempting a step back toward a higher-priority credential after failover (default: `600`) |
 | `failback_request_count` | int | Successful requests on a lower-priority credential before attempting a step back (default: `50`) |
 | `backup` | object | Optional backup VLM configuration (same shape as `vlm`) for automatic failover when the primary fails with retryable errors such as rate limits, `5xx` responses, or connection/timeout failures. Only one level of failover is supported — the backup itself cannot define a nested `backup` |
 | `timeout` | float | Per-request HTTP timeout in seconds passed to the underlying OpenAI/LiteLLM client. Increase for slow endpoints (e.g., DashScope, local inference). Must be `> 0` (default: `600.0`) |
+| `keepalive_expiry` | float | Idle connection lifetime in seconds for OpenAI-compatible VLM clients. Set to `0` to disable idle connection reuse; unset uses the OpenAI SDK default. Must be `>= 0` |
 | `extra_headers` | object | Custom HTTP headers for compatible HTTP providers. `kimi` also accepts header overrides, but already injects the required subscription headers by default |
 | `extra_request_body` | object | Extra JSON body fields for OpenAI-compatible completion requests, useful for provider-specific options such as Ollama `{"think": false}` |
 | `stream` | bool | Enable streaming mode (for OpenAI-compatible providers, default: `false`) |

docs/zh/guides/01-configuration.md

@@ -608,11 +608,12 @@ provider，并设置 `storage.vectordb.sparse_weight > 0`。自托管模型的
 | `thinking` | bool | 启用思考模式（仅对部分火山模型生效，默认：`false`） |
 | `max_concurrent` | int | 语义处理阶段 LLM 最大并发调用数（默认：`32`） |
 | `max_retries` | int | VLM provider 瞬时错误的最大重试次数（默认：`3`；`0` 表示禁用重试） |
-| `credentials` | array | 有序 VLM 凭据/模型列表，索引 0 优先级最高。每项可单独覆盖 `provider`、`model`、`api_key`、`api_base`、`api_version`、`extra_headers`、`extra_request_body`、`stream` 和 `reasoning_effort` |
+| `credentials` | array | 有序 VLM 凭据/模型列表，索引 0 优先级最高。每项可单独覆盖 `provider`、`model`、`api_key`、`api_base`、`api_version`、`extra_headers`、`extra_request_body`、`stream`、`reasoning_effort` 和 `keepalive_expiry` |
 | `failback_timeout_seconds` | float | 切换到低优先级 credential 后，尝试逐级切回的时间阈值（默认：`600`） |
 | `failback_request_count` | int | 低优先级 credential 成功处理多少次请求后尝试逐级切回（默认：`50`） |
 | `backup` | object | 可选的备用 VLM 配置（结构与 `vlm` 相同），当主 VLM 遇到限流、`5xx`、超时或连接失败等可重试错误时自动切换。仅支持 1 层备用 — 备用 VLM 本身不能再嵌套 `backup` |
 | `timeout` | float | 单次 VLM API 请求的 HTTP 超时时间（秒），传递给底层 OpenAI/LiteLLM 客户端。慢端点（如 DashScope、本地推理）可调大。必须 `> 0`（默认：`600.0`） |
+| `keepalive_expiry` | float | OpenAI 兼容 VLM 客户端的空闲连接保留秒数。设为 `0` 可禁用空闲连接复用；不设置时使用 OpenAI SDK 默认值。必须 `>= 0` |
 | `extra_headers` | object | 兼容 HTTP provider 的自定义请求头。`kimi` 默认已注入所需订阅请求头，也支持在这里覆盖或扩展 |
 | `extra_request_body` | object | 传给 OpenAI 兼容 completion 请求的额外 JSON body 字段，可用于 Ollama `{"think": false}` 等 provider 专有参数 |
 | `stream` | bool | 启用流式模式（OpenAI 兼容 provider 可用，默认：`false`） |

openviking/models/vlm/backends/openai_vlm.py

@@ -9,6 +9,8 @@
 from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlparse
 
+import httpx
+
 from openviking.telemetry import tracer
 from openviking.utils.async_client_cache import LoopScopedAsyncClientCache
 from openviking.utils.multimodal import redact_image_data_urls
@@ -85,6 +87,28 @@ def __init__(self, config: Dict[str, Any]):
         self.api_version = config.get("api_version")
         self.reasoning_effort = config.get("reasoning_effort", "low")
 
+    def _http_limits(self) -> httpx.Limits:
+        defaults = openai.DEFAULT_CONNECTION_LIMITS
+        return httpx.Limits(
+            max_connections=defaults.max_connections,
+            max_keepalive_connections=defaults.max_keepalive_connections,
+            keepalive_expiry=self.keepalive_expiry,
+        )
+
+    def _configure_sync_http_client(self, kwargs: Dict[str, Any]) -> None:
+        if self.keepalive_expiry is not None:
+            kwargs["http_client"] = openai.DefaultHttpxClient(
+                limits=self._http_limits(),
+                timeout=self.timeout,
+            )
+
+    def _configure_async_http_client(self, kwargs: Dict[str, Any]) -> None:
+        if self.keepalive_expiry is not None:
+            kwargs["http_client"] = openai.DefaultAsyncHttpxClient(
+                limits=self._http_limits(),
+                timeout=self.timeout,
+            )
+
     def get_client(self):
         """Get sync client"""
         if self._sync_client is None:
@@ -98,6 +122,7 @@ def get_client(self):
                 self.extra_headers,
                 self.timeout,
             )
+            self._configure_sync_http_client(kwargs)
             if self.provider == "azure":
                 self._sync_client = openai.AzureOpenAI(**kwargs)
             else:
@@ -116,6 +141,7 @@ def _build_async_client(self):
             self.extra_headers,
             self.timeout,
         )
+        self._configure_async_http_client(kwargs)
         if self.provider == "azure":
             return openai.AsyncAzureOpenAI(**kwargs)
         return openai.AsyncOpenAI(**kwargs)
@@ -124,6 +150,13 @@ def get_async_client(self):
         """Get an async client scoped to the current event loop."""
         return self._async_client_cache.get(self._build_async_client)
 
+    def close(self):
+        """Close clients and HTTP transports owned by this backend."""
+        close_sync_client = getattr(self._sync_client, "close", None)
+        if close_sync_client is not None:
+            close_sync_client()
+        self._async_client_cache.close_all_with_close()
+
     def _supports_enable_thinking(self) -> bool:
         """Return True for OpenAI-compatible DashScope endpoints that accept enable_thinking."""
         if self.provider != "openai":

openviking/models/vlm/base.py

@@ -73,6 +73,7 @@ def __init__(self, config: Dict[str, Any]):
         self.temperature = config.get("temperature", 0.0)
         self.max_retries = config.get("max_retries", 3)
         self.timeout = config.get("timeout", 600.0)
+        self.keepalive_expiry = config.get("keepalive_expiry")
         self.max_tokens = config.get("max_tokens")
         self.extra_headers = config.get("extra_headers")
         self.extra_request_body = dict(config.get("extra_request_body") or {})
@@ -332,6 +333,9 @@ def reset_token_usage(self) -> None:
         """Reset token usage"""
         self._token_tracker.reset()
 
+    def close(self) -> None:
+        """Release provider resources, if any."""
+
     def _extract_content_from_response(self, response) -> str:
         if isinstance(response, str):
             return response
@@ -759,6 +763,11 @@ def reset_token_usage(self) -> None:
         self.primary.reset_token_usage()
         self.backup.reset_token_usage()
 
+    def close(self) -> None:
+        """Close both provider instances."""
+        self.primary.close()
+        self.backup.close()
+
 
 class MultiCredentialVLM(VLMBase):
     """VLM wrapper that provides failover across multiple ordered credentials.
@@ -1123,3 +1132,8 @@ def reset_token_usage(self) -> None:
         """Reset token usage for all credential instances."""
         for instance in self._vlm_instances:
             instance.reset_token_usage()
+
+    def close(self) -> None:
+        """Close all credential provider instances."""
+        for instance in self._vlm_instances:
+            instance.close()

openviking/service/core.py

@@ -546,6 +546,13 @@ async def close(self) -> None:
             self._queue_manager = None
             logger.info("Queue manager stopped")
 
+        config = getattr(self, "_config", None)
+        vlm_config = getattr(config, "vlm", None)
+        close_vlm = getattr(vlm_config, "close", None)
+        if close_vlm is not None:
+            close_vlm()
+            await asyncio.sleep(0)
+
         if self._vikingdb_manager:
             self._vikingdb_manager.mark_closing()
 

openviking_cli/utils/config/vlm_config.py

@@ -50,6 +50,11 @@ class VLMCredential(BaseModel):
         default=None,
         description="Reasoning effort for OpenAI-compatible reasoning models",
     )
+    keepalive_expiry: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        description="Idle HTTP connection lifetime for OpenAI-compatible providers",
+    )
     max_tokens: Optional[int] = Field(
         default=None,
         gt=0,
@@ -117,6 +122,14 @@ class VLMConfig(BaseModel):
             "high-latency endpoints (e.g., DashScope, local inference servers)."
         ),
     )
+    keepalive_expiry: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        description=(
+            "Idle HTTP connection lifetime in seconds for OpenAI-compatible VLM clients. "
+            "Set to 0 to disable connection reuse; None uses the provider SDK default."
+        ),
+    )
 
     provider: Optional[str] = Field(default=None, description="Provider type")
     backend: Optional[str] = Field(
@@ -283,6 +296,7 @@ def _migrate_legacy_config(self):
             or self.extra_request_body
             or self.stream
             or self.reasoning_effort
+            or self.keepalive_expiry is not None
             or self.forward_api_key is not None
         ):
             if self.provider not in self.providers:
@@ -307,6 +321,11 @@ def _migrate_legacy_config(self):
                 self.providers[self.provider]["stream"] = self.stream
             if self.reasoning_effort and "reasoning_effort" not in self.providers[self.provider]:
                 self.providers[self.provider]["reasoning_effort"] = self.reasoning_effort
+            if (
+                self.keepalive_expiry is not None
+                and "keepalive_expiry" not in self.providers[self.provider]
+            ):
+                self.providers[self.provider]["keepalive_expiry"] = self.keepalive_expiry
 
     def _normalize_credentials(self):
         """Normalize credentials configuration:
@@ -346,6 +365,11 @@ def _normalize_credentials(self):
                     else self.stream
                 ),
                 reasoning_effort=(primary_cfg.get("reasoning_effort") or self.reasoning_effort),
+                keepalive_expiry=(
+                    primary_cfg.get("keepalive_expiry")
+                    if primary_cfg.get("keepalive_expiry") is not None
+                    else self.keepalive_expiry
+                ),
                 max_tokens=self.max_tokens,
             )
             migrated_credentials.append(primary_cred)
@@ -378,6 +402,11 @@ def _normalize_credentials(self):
                 reasoning_effort=(
                     backup_cfg.get("reasoning_effort") or self.backup.reasoning_effort
                 ),
+                keepalive_expiry=(
+                    backup_cfg.get("keepalive_expiry")
+                    if backup_cfg.get("keepalive_expiry") is not None
+                    else self.backup.keepalive_expiry
+                ),
                 max_tokens=self.backup.max_tokens,
             )
             migrated_credentials.append(backup_cred)
@@ -421,6 +450,11 @@ def _normalize_credentials(self):
                         reasoning_effort=(
                             provider_cfg.get("reasoning_effort") or self.reasoning_effort
                         ),
+                        keepalive_expiry=(
+                            provider_cfg.get("keepalive_expiry")
+                            if provider_cfg.get("keepalive_expiry") is not None
+                            else self.keepalive_expiry
+                        ),
                     )
                 )
 
@@ -453,6 +487,8 @@ def _normalize_credentials(self):
                 cred.stream = self.stream
             if not cred.reasoning_effort:
                 cred.reasoning_effort = self.reasoning_effort
+            if cred.keepalive_expiry is None:
+                cred.keepalive_expiry = self.keepalive_expiry
 
     def _has_legacy_provider_config(self) -> bool:
         """Check if there's legacy provider config (not credentials-based)."""
@@ -508,6 +544,8 @@ def _get_provider_config_by_name(self, provider_name: str) -> Dict[str, Any]:
             config["stream"] = self.stream
         if self.reasoning_effort and "reasoning_effort" not in config:
             config["reasoning_effort"] = self.reasoning_effort
+        if self.keepalive_expiry is not None and "keepalive_expiry" not in config:
+            config["keepalive_expiry"] = self.keepalive_expiry
         return config
 
     def _provider_has_usable_credentials(self, provider_name: str, config: Dict[str, Any]) -> bool:
@@ -539,6 +577,8 @@ def _get_provider_config_from_credential(self, cred: VLMCredential) -> Dict[str,
             config["stream"] = cred.stream
         if cred.reasoning_effort:
             config["reasoning_effort"] = cred.reasoning_effort
+        if cred.keepalive_expiry is not None:
+            config["keepalive_expiry"] = cred.keepalive_expiry
         return config
 
     def _match_provider(self, model: str | None = None) -> tuple[Dict[str, Any] | None, str | None]:
@@ -625,13 +665,21 @@ def get_vlm_instance(self) -> Any:
 
         return self._vlm_instance
 
+    def close(self) -> None:
+        """Close and clear the cached VLM instance."""
+        instance = self._vlm_instance
+        self._vlm_instance = None
+        if instance is not None:
+            instance.close()
+
     def _build_vlm_config_dict_for_credential(self, credential: VLMCredential) -> Dict[str, Any]:
         """Build VLM instance config dict for a specific credential."""
         result = {
             "model": credential.model or self.model,
             "temperature": self.temperature,
             "max_retries": self.max_retries,
             "timeout": self.timeout,
+            "keepalive_expiry": credential.keepalive_expiry,
             "provider": credential.provider,
             "thinking": self.thinking,
             "max_tokens": (
@@ -671,6 +719,11 @@ def _build_vlm_config_dict(self) -> Dict[str, Any]:
             "temperature": self.temperature,
             "max_retries": self.max_retries,
             "timeout": self.timeout,
+            "keepalive_expiry": (
+                config.get("keepalive_expiry")
+                if config and config.get("keepalive_expiry") is not None
+                else self.keepalive_expiry
+            ),
             "provider": name,
             "thinking": self.thinking,
             "max_tokens": self.max_tokens,