refresh() — langchain Function Reference

Architecture documentation for the refresh() function in cli.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 5 called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ refresh() — langchain Function Reference

Dependency Diagram

graph TD
  2f2dc685_603e_b75d_c32f_a2fd813a683c["refresh()"]
  74c7ffe0_0dd5_5e6a_45ab_950da26de248["cli.py"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|defined in| 74c7ffe0_0dd5_5e6a_45ab_950da26de248
  0dec95bd_3d27_d221_3b9a_43294e80bf97["main()"]
  0dec95bd_3d27_d221_3b9a_43294e80bf97 -->|calls| 2f2dc685_603e_b75d_c32f_a2fd813a683c
  3da388c2_fd31_691b_e345_84736796bb68["_validate_data_dir()"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|calls| 3da388c2_fd31_691b_e345_84736796bb68
  7135fdee_ad01_ad3e_8651_29e017d47837["_load_augmentations()"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|calls| 7135fdee_ad01_ad3e_8651_29e017d47837
  4a63284f_7279_def9_56e6_2041d56879d8["_model_data_to_profile()"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|calls| 4a63284f_7279_def9_56e6_2041d56879d8
  6a5b4adb_47f1_21e6_9b77_d11b5f9a5f9c["_apply_overrides()"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|calls| 6a5b4adb_47f1_21e6_9b77_d11b5f9a5f9c
  67e62d64_96d5_5560_dfce_06308c08d58a["_write_profiles_file()"]
  2f2dc685_603e_b75d_c32f_a2fd813a683c -->|calls| 67e62d64_96d5_5560_dfce_06308c08d58a
  style 2f2dc685_603e_b75d_c32f_a2fd813a683c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/model-profiles/langchain_model_profiles/cli.py lines 215–327

def refresh(provider: str, data_dir: Path) -> None:  # noqa: C901, PLR0915
    """Download and merge model profile data for a specific provider.

    Args:
        provider: Provider ID from models.dev (e.g., `'anthropic'`, `'openai'`).
        data_dir: Directory containing `profile_augmentations.toml` and where
            `profiles.py` will be written.
    """
    # Validate and canonicalize data directory path
    data_dir = _validate_data_dir(data_dir)

    api_url = "https://models.dev/api.json"

    print(f"Provider: {provider}")
    print(f"Data directory: {data_dir}")
    print()

    # Download data from models.dev
    print(f"Downloading data from {api_url}...")
    try:
        response = httpx.get(api_url, timeout=30)
        response.raise_for_status()
    except httpx.TimeoutException:
        msg = f"Request timed out connecting to {api_url}"
        print(f"❌ {msg}", file=sys.stderr)
        sys.exit(1)
    except httpx.HTTPStatusError as e:
        msg = f"HTTP error {e.response.status_code} from {api_url}"
        print(f"❌ {msg}", file=sys.stderr)
        sys.exit(1)
    except httpx.RequestError as e:
        msg = f"Failed to connect to {api_url}: {e}"
        print(f"❌ {msg}", file=sys.stderr)
        sys.exit(1)

    try:
        all_data = response.json()
    except json.JSONDecodeError as e:
        msg = f"Invalid JSON response from API: {e}"
        print(f"❌ {msg}", file=sys.stderr)
        sys.exit(1)

    # Basic validation
    if not isinstance(all_data, dict):
        msg = "Expected API response to be a dictionary"
        print(f"❌ {msg}", file=sys.stderr)
        sys.exit(1)

    provider_count = len(all_data)
    model_count = sum(len(p.get("models", {})) for p in all_data.values())
    print(f"Downloaded {provider_count} providers with {model_count} models")

    # Extract data for this provider
    if provider not in all_data:
        msg = f"Provider '{provider}' not found in models.dev data"
        print(msg, file=sys.stderr)
        sys.exit(1)

    provider_data = all_data[provider]
    models = provider_data.get("models", {})
    print(f"Extracted {len(models)} models for {provider}")

    # Load augmentations
    print("Loading augmentations...")
    provider_aug, model_augs = _load_augmentations(data_dir)

    # Merge and convert to profiles
    profiles: dict[str, dict[str, Any]] = {}
    for model_id, model_data in models.items():
        base_profile = _model_data_to_profile(model_data)
        profiles[model_id] = _apply_overrides(
            base_profile, provider_aug, model_augs.get(model_id)
        )

    # Include new models defined purely via augmentations
    extra_models = set(model_augs) - set(models)
    if extra_models:
        print(f"Adding {len(extra_models)} models from augmentations only...")
    for model_id in sorted(extra_models):
        profiles[model_id] = _apply_overrides({}, provider_aug, model_augs[model_id])