mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-03-22 13:47:19 +08:00
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from enum import Enum
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
LEMONADE_DOCKER_BASE_URL = 'http://host.docker.internal:8000/api/v1/'
|
|
_LEMONADE_PROVIDER_NAME = 'lemonade'
|
|
_LEMONADE_MODEL_PREFIX = 'lemonade/'
|
|
|
|
|
|
class StorageProvider(str, Enum):
|
|
"""Storage provider types for event and shared event storage."""
|
|
|
|
AWS = 'aws'
|
|
GCP = 'gcp'
|
|
FILESYSTEM = 'filesystem'
|
|
|
|
|
|
def get_storage_provider() -> StorageProvider:
|
|
"""Get the storage provider based on environment variables.
|
|
|
|
Determines the storage provider from environment configuration:
|
|
- SHARED_EVENT_STORAGE_PROVIDER: Primary setting, supports 'aws', 'gcp', 'google_cloud'
|
|
- FILE_STORE: Legacy fallback, supports 'google_cloud'
|
|
|
|
Returns:
|
|
StorageProvider: The configured storage provider (AWS, GCP, or FILESYSTEM)
|
|
"""
|
|
provider = os.environ.get('SHARED_EVENT_STORAGE_PROVIDER', '').lower()
|
|
|
|
# If not explicitly set, fall back to FILE_STORE
|
|
if not provider:
|
|
provider = os.environ.get('FILE_STORE', '').lower()
|
|
|
|
if provider == 'aws':
|
|
return StorageProvider.AWS
|
|
elif provider in ('gcp', 'google_cloud'):
|
|
return StorageProvider.GCP
|
|
else:
|
|
return StorageProvider.FILESYSTEM
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def is_running_in_docker() -> bool:
|
|
"""Best-effort detection for Docker containers."""
|
|
docker_env_markers = (
|
|
Path('/.dockerenv'),
|
|
Path('/run/.containerenv'),
|
|
)
|
|
if any(marker.exists() for marker in docker_env_markers):
|
|
return True
|
|
|
|
if os.environ.get('DOCKER_CONTAINER') == 'true':
|
|
return True
|
|
|
|
try:
|
|
with Path('/proc/self/cgroup').open('r', encoding='utf-8') as cgroup_file:
|
|
for line in cgroup_file:
|
|
if any(token in line for token in ('docker', 'containerd', 'kubepods')):
|
|
return True
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
return False
|
|
|
|
|
|
def is_lemonade_provider(
|
|
model: str | None,
|
|
custom_provider: str | None = None,
|
|
) -> bool:
|
|
provider = (custom_provider or '').strip().lower()
|
|
if provider == _LEMONADE_PROVIDER_NAME:
|
|
return True
|
|
return (model or '').startswith(_LEMONADE_MODEL_PREFIX)
|
|
|
|
|
|
def get_effective_llm_base_url(
|
|
model: str | None,
|
|
base_url: str | None,
|
|
custom_provider: str | None = None,
|
|
) -> str | None:
|
|
"""Return the runtime LLM base URL with provider-specific overrides."""
|
|
if (
|
|
base_url in (None, '')
|
|
and is_lemonade_provider(model, custom_provider)
|
|
and is_running_in_docker()
|
|
):
|
|
return LEMONADE_DOCKER_BASE_URL
|
|
return base_url
|