Source code for langchain.hub
"""与LangChain Hub进行接口交互。"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any, Optional
from langchain_core.load.dump import dumps
from langchain_core.load.load import loads
from langchain_core.prompts import BasePromptTemplate
if TYPE_CHECKING:
from langchainhub import Client
def _get_client(api_url: Optional[str] = None, api_key: Optional[str] = None) -> Client:
try:
from langchainhub import Client
except ImportError as e:
raise ImportError(
"Could not import langchainhub, please install with `pip install "
"langchainhub`."
) from e
# Client logic will also attempt to load URL/key from environment variables
return Client(api_url, api_key=api_key)
[docs]def push(
repo_full_name: str,
object: Any,
*,
api_url: Optional[str] = None,
api_key: Optional[str] = None,
parent_commit_hash: Optional[str] = "latest",
new_repo_is_public: bool = True,
new_repo_description: str = "",
) -> str:
"""将对象推送到中心,并返回在浏览器中查看的URL。
:param repo_full_name: 要推送到的仓库的完整名称,格式为`owner/repo`。
:param object: 要序列化并推送到中心的LangChain。
:param api_url: LangChain Hub API的URL。如果设置了api密钥,则默认为托管的API服务,否则为本地主机实例。
:param api_key: 用于与LangChain Hub API进行身份验证的API密钥。
:param parent_commit_hash: 要推送到的父提交的提交哈希。默认为自动选择最新提交。
:param new_repo_is_public: 仓库是否应为公共。默认为True(默认为公共)。
:param new_repo_description: 仓库的描述。默认为空字符串。
"""
client = _get_client(api_url=api_url, api_key=api_key)
manifest_json = dumps(object)
message = client.push(
repo_full_name,
manifest_json,
parent_commit_hash=parent_commit_hash,
new_repo_is_public=new_repo_is_public,
new_repo_description=new_repo_description,
)
return message
[docs]def pull(
owner_repo_commit: str,
*,
api_url: Optional[str] = None,
api_key: Optional[str] = None,
) -> Any:
"""从hub中拉取一个对象,并将其作为LangChain对象返回。
:param owner_repo_commit: 要从中拉取的仓库的完整名称,格式为`owner/repo:commit_hash`。
:param api_url: LangChain Hub API的URL。如果设置了api密钥,则默认为托管的API服务,如果没有设置,则为本地主机实例。
:param api_key: 用于与LangChain Hub API进行身份验证的API密钥。
"""
client = _get_client(api_url=api_url, api_key=api_key)
if hasattr(client, "pull_repo"):
# >= 0.1.15
res_dict = client.pull_repo(owner_repo_commit)
obj = loads(json.dumps(res_dict["manifest"]))
if isinstance(obj, BasePromptTemplate):
if obj.metadata is None:
obj.metadata = {}
obj.metadata["lc_hub_owner"] = res_dict["owner"]
obj.metadata["lc_hub_repo"] = res_dict["repo"]
obj.metadata["lc_hub_commit_hash"] = res_dict["commit_hash"]
return obj
# Then it's < 0.1.15
resp: str = client.pull(owner_repo_commit)
return loads(resp)