Spaces:
Running
Running
| """LLM credential and grounding config resolution. | |
| Resolves the model name and litellm kwargs for OpenSpace's LLM client, | |
| and assembles grounding config from env-var overrides. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import os | |
| import tempfile | |
| from typing import Any, Dict, Optional | |
| logger = logging.getLogger("openspace.host_detection") | |
| def build_llm_kwargs(model: str) -> tuple[str, Dict[str, Any]]: | |
| """Build litellm kwargs and resolve model for OpenSpace's LLM client. | |
| Resolution order (highest → lowest priority): | |
| Tier 1 — Explicit ``OPENSPACE_LLM_*`` env vars:: | |
| OPENSPACE_LLM_API_KEY → litellm ``api_key`` | |
| OPENSPACE_LLM_API_BASE → litellm ``api_base`` | |
| OPENSPACE_LLM_EXTRA_HEADERS → litellm ``extra_headers`` (JSON string) | |
| OPENSPACE_LLM_CONFIG → arbitrary litellm kwargs (JSON string) | |
| Tier 2 — Auto-detect from host agent config file:: | |
| ~/.nanobot/config.json → providers.{matched}.apiKey / apiBase | |
| Tier 3 — Provider-native env vars inherited from the parent process | |
| (e.g. ``OPENROUTER_API_KEY``). Read by litellm automatically. | |
| Returns: | |
| ``(resolved_model, llm_kwargs_dict)`` | |
| """ | |
| from openspace.host_detection.nanobot import try_read_nanobot_config | |
| kwargs: Dict[str, Any] = {} | |
| resolved_model = model | |
| source = "inherited env" | |
| # --- Tier 2: auto-detect from host config (filled first, may be overridden) --- | |
| host_config = try_read_nanobot_config(model) | |
| if host_config: | |
| host_model = host_config.pop("_model", None) | |
| forced_provider = host_config.pop("_forced_provider", None) | |
| if not resolved_model and host_model: | |
| resolved_model = host_model | |
| # If the host config forces a gateway provider (e.g. openrouter) | |
| # and the model name doesn't already carry that prefix, prepend | |
| # it so that litellm uses the correct request format (OpenAI- | |
| # compatible for gateways vs native for direct providers). | |
| _GATEWAY_PROVIDERS = {"openrouter", "aihubmix", "siliconflow"} | |
| if ( | |
| forced_provider | |
| and forced_provider in _GATEWAY_PROVIDERS | |
| and resolved_model | |
| and not resolved_model.lower().startswith(f"{forced_provider}/") | |
| ): | |
| resolved_model = f"{forced_provider}/{resolved_model}" | |
| logger.info( | |
| "Prepended gateway prefix: model=%r (forced_provider=%s)", | |
| resolved_model, forced_provider, | |
| ) | |
| kwargs.update(host_config) | |
| source = "nanobot config" | |
| # --- Tier 1: explicit env vars override everything --- | |
| api_key = os.environ.get("OPENSPACE_LLM_API_KEY") | |
| if api_key: | |
| kwargs["api_key"] = api_key | |
| source = "OPENSPACE_LLM_* env" | |
| api_base = os.environ.get("OPENSPACE_LLM_API_BASE") | |
| if api_base: | |
| kwargs["api_base"] = api_base | |
| extra_headers_raw = os.environ.get("OPENSPACE_LLM_EXTRA_HEADERS") | |
| if extra_headers_raw: | |
| try: | |
| headers = json.loads(extra_headers_raw) | |
| if isinstance(headers, dict): | |
| kwargs["extra_headers"] = headers | |
| except json.JSONDecodeError: | |
| logger.warning("Invalid JSON in OPENSPACE_LLM_EXTRA_HEADERS: %r", extra_headers_raw) | |
| llm_config_raw = os.environ.get("OPENSPACE_LLM_CONFIG") | |
| if llm_config_raw: | |
| try: | |
| llm_config = json.loads(llm_config_raw) | |
| if isinstance(llm_config, dict): | |
| kwargs.update(llm_config) | |
| source = "OPENSPACE_LLM_CONFIG env" | |
| except json.JSONDecodeError: | |
| logger.warning("Invalid JSON in OPENSPACE_LLM_CONFIG: %r", llm_config_raw) | |
| # Default model fallback | |
| if not resolved_model: | |
| resolved_model = "openrouter/anthropic/claude-sonnet-4.5" | |
| if kwargs: | |
| safe = { | |
| k: (v[:8] + "..." if k == "api_key" and isinstance(v, str) and len(v) > 8 else v) | |
| for k, v in kwargs.items() | |
| } | |
| logger.info("LLM kwargs resolved (source=%s): %s", source, safe) | |
| return resolved_model, kwargs | |
| def build_grounding_config_path() -> Optional[str]: | |
| """Resolve grounding config: inline JSON > file path > None. | |
| Supports: | |
| * ``OPENSPACE_CONFIG_JSON`` — inline JSON string (written to a temp file) | |
| * ``OPENSPACE_CONFIG_PATH`` — path to a JSON config file | |
| Granular env-var overrides (``OPENSPACE_SHELL_*``, ``OPENSPACE_SKILLS_*``, | |
| etc.) are merged before writing. | |
| Returns: | |
| Path to the resolved config file, or None. | |
| """ | |
| config_json_raw = os.environ.get("OPENSPACE_CONFIG_JSON", "").strip() | |
| overrides: Dict[str, Any] = {} | |
| if config_json_raw: | |
| try: | |
| overrides = json.loads(config_json_raw) | |
| if not isinstance(overrides, dict): | |
| logger.warning("OPENSPACE_CONFIG_JSON is not a dict, ignoring") | |
| overrides = {} | |
| else: | |
| logger.info("Loaded inline config from OPENSPACE_CONFIG_JSON") | |
| except json.JSONDecodeError as e: | |
| logger.warning("Invalid JSON in OPENSPACE_CONFIG_JSON: %s", e) | |
| # --- Granular env-var overrides --- | |
| conda_env = os.environ.get("OPENSPACE_SHELL_CONDA_ENV", "").strip() | |
| if conda_env: | |
| overrides.setdefault("shell", {})["conda_env"] = conda_env | |
| shell_wd = os.environ.get("OPENSPACE_SHELL_WORKING_DIR", "").strip() | |
| if shell_wd: | |
| overrides.setdefault("shell", {})["working_dir"] = shell_wd | |
| skills_dirs_raw = os.environ.get("OPENSPACE_SKILLS_DIRS", "").strip() | |
| if skills_dirs_raw: | |
| dirs = [d.strip() for d in skills_dirs_raw.split(",") if d.strip()] | |
| if dirs: | |
| overrides.setdefault("skills", {})["skill_dirs"] = dirs | |
| mcp_servers_raw = os.environ.get("OPENSPACE_MCP_SERVERS_JSON", "").strip() | |
| if mcp_servers_raw: | |
| try: | |
| servers = json.loads(mcp_servers_raw) | |
| if isinstance(servers, dict): | |
| overrides["mcpServers"] = servers | |
| except json.JSONDecodeError as e: | |
| logger.warning("Invalid JSON in OPENSPACE_MCP_SERVERS_JSON: %s", e) | |
| log_level = os.environ.get("OPENSPACE_LOG_LEVEL", "").strip().upper() | |
| if log_level and log_level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): | |
| overrides["log_level"] = log_level | |
| if overrides: | |
| try: | |
| fd, tmp_path = tempfile.mkstemp(suffix=".json", prefix="openspace_cfg_") | |
| with os.fdopen(fd, "w", encoding="utf-8") as f: | |
| json.dump(overrides, f, ensure_ascii=False) | |
| logger.info( | |
| "Grounding config overrides written to %s (%d keys)", | |
| tmp_path, len(overrides), | |
| ) | |
| return tmp_path | |
| except Exception as e: | |
| logger.warning("Failed to write config overrides: %s", e) | |
| return os.environ.get("OPENSPACE_CONFIG_PATH") | |