mirror of
https://github.com/Mintplex-Labs/langchain-python.git
synced 2026-07-25 20:45:22 -04:00
Roy Williams
6086292252
It's generally considered to be a good practice to pin dependencies to prevent surprise breakages when a new version of a dependency is released. This commit adds the ability to pin dependencies when loading from LangChainHub. Centralizing this logic and using urllib fixes an issue identified by some windows users highlighted in this video - https://youtu.be/aJ6IQUh8MLQ?t=537
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
"""Utilities for loading configurations from langchian-hub."""
|
|
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional, Set, TypeVar, Union
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
|
|
DEFAULT_REF = os.environ.get("LANGCHAIN_HUB_DEFAULT_REF", "master")
|
|
URL_BASE = os.environ.get(
|
|
"LANGCHAIN_HUB_URL_BASE",
|
|
"https://raw.githubusercontent.com/hwchase17/langchain-hub/{ref}/",
|
|
)
|
|
HUB_PATH_RE = re.compile(r"lc(?P<ref>@[^:]+)?://(?P<path>.*)")
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
def try_load_from_hub(
|
|
path: Union[str, Path],
|
|
loader: Callable[[str], T],
|
|
valid_prefix: str,
|
|
valid_suffixes: Set[str],
|
|
**kwargs: Any,
|
|
) -> Optional[T]:
|
|
"""Load configuration from hub. Returns None if path is not a hub path."""
|
|
if not isinstance(path, str) or not (match := HUB_PATH_RE.match(path)):
|
|
return None
|
|
ref, remote_path_str = match.groups()
|
|
ref = ref[1:] if ref else DEFAULT_REF
|
|
remote_path = Path(remote_path_str)
|
|
if remote_path.parts[0] != valid_prefix:
|
|
return None
|
|
if remote_path.suffix[1:] not in valid_suffixes:
|
|
raise ValueError("Unsupported file type.")
|
|
|
|
full_url = urljoin(URL_BASE.format(ref=ref), str(remote_path))
|
|
r = requests.get(full_url, timeout=5)
|
|
if r.status_code != 200:
|
|
raise ValueError(f"Could not find file at {full_url}")
|
|
with tempfile.TemporaryDirectory() as tmpdirname:
|
|
file = Path(tmpdirname) / remote_path.name
|
|
with open(file, "wb") as f:
|
|
f.write(r.content)
|
|
return loader(str(file), **kwargs)
|