mirror of https://github.com/hwchase17/langchain
add more import checks (#11033)
Harrison Chase
11 months ago
committed by
GitHub
parent
efb7c459a2
commit
e355606b11
@ -0,0 +1,11 @@
|
||||
from typing import Sequence, TypedDict
|
||||
|
||||
from langchain.schema import BaseMessage
|
||||
|
||||
|
||||
class ChatSession(TypedDict):
|
||||
"""Chat Session represents a single
|
||||
conversation, channel, or other group of messages."""
|
||||
|
||||
messages: Sequence[BaseMessage]
|
||||
"""The LangChain chat messages loaded from the source."""
|
||||
@ -1,54 +1,4 @@
|
||||
"""Utilities for loading configurations from langchain-hub."""
|
||||
from langchain.utils.loading import try_load_from_hub
|
||||
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
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(f"Unsupported file type, must be one of {valid_suffixes}.")
|
||||
|
||||
# Using Path with URLs is not recommended, because on Windows
|
||||
# the backslash is used as the path separator, which can cause issues
|
||||
# when working with URLs that use forward slashes as the path separator.
|
||||
# Instead, use PurePosixPath to ensure that forward slashes are used as the
|
||||
# path separator, regardless of the operating system.
|
||||
full_url = urljoin(URL_BASE.format(ref=ref), PurePosixPath(remote_path).__str__())
|
||||
|
||||
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)
|
||||
# For backwards compatibility
|
||||
__all__ = ["try_load_from_hub"]
|
||||
|
||||
@ -0,0 +1,54 @@
|
||||
"""Utilities for loading configurations from langchain-hub."""
|
||||
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
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(f"Unsupported file type, must be one of {valid_suffixes}.")
|
||||
|
||||
# Using Path with URLs is not recommended, because on Windows
|
||||
# the backslash is used as the path separator, which can cause issues
|
||||
# when working with URLs that use forward slashes as the path separator.
|
||||
# Instead, use PurePosixPath to ensure that forward slashes are used as the
|
||||
# path separator, regardless of the operating system.
|
||||
full_url = urljoin(URL_BASE.format(ref=ref), PurePosixPath(remote_path).__str__())
|
||||
|
||||
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)
|
||||
Loading…
Reference in New Issue