工具

langsmith.utils

通用实用函数。

函数	描述
tracing_is_enabled	如果启用了跟踪，则返回 True。
test_tracking_is_disabled	如果启用了测试，则返回 True。
xor_args	验证指定的关键字参数是互斥的。
raise_for_status_with_text	抛出一个带有响应文本的错误。
get_enum_value	获取字符串枚举的值。
log_once	以指定级别记录一条消息，但只记录一次。
get_messages_from_inputs	从给定的输入字典中提取消息。
get_message_generation_from_outputs	从给定的输出中检索消息生成。
get_prompt_from_inputs	从给定的输入中检索提示。
get_llm_generation_from_outputs	从输出中获取 LLM 生成。
get_docker_compose_command	获取此系统的正确 docker compose 命令。
convert_langchain_message	将 LangChain 消息转换为示例。
is_base_message_like	检查给定的对象是否类似于 BaseMessage。
is_env_var_truish	检查给定的环境变量是否为真值。
get_env_var	从命名空间列表中检索环境变量。
get_tracer_project	获取 LangSmith 跟踪器的项目名称。
filter_logs	临时向记录器添加指定的过滤器。
get_cache_dir	获取测试缓存目录。
filter_request_headers	根据 ignore_hosts 和 allow_hosts 过滤请求头。
with_cache	为请求使用缓存。
with_optional_cache	为请求使用缓存。
deepish_copy	对一个值进行深拷贝，并对不可拷贝的对象做出妥协。
is_version_greater_or_equal	检查当前版本是否大于或等于目标版本。
parse_prompt_identifier	解析格式为 owner/name:hash、name:hash、owner/name 或 name 的字符串。
get_api_url	从环境或给定值中获取 LangSmith API URL。
get_api_key	从环境或给定值中获取 API 密钥。
get_workspace_id	获取工作区 ID。
get_host_url	根据 web URL 或 API URL 获取主机 URL。
is_truish	检查值是否为真值。

LangSmithError

基类： Exception

与 LangSmith API 通信时发生错误。

LangSmithAPIError

基类：LangSmithError

与 LangSmith 通信时出现内部服务器错误。

LangSmithRequestTimeout

基类：LangSmithError

客户端发送请求体用时过长。

LangSmithUserError

基类：LangSmithError

用户错误导致与 LangSmith 通信时出现异常。

LangSmithRateLimitError

基类：LangSmithError

您已超过 LangSmith API 的速率限制。

LangSmithAuthError

基类：LangSmithError

无法通过 LangSmith API 进行身份验证。

LangSmithNotFoundError

基类：LangSmithError

找不到请求的资源。

LangSmithConflictError

基类：LangSmithError

资源已存在。

LangSmithConnectionError

基类：LangSmithError

无法连接到 LangSmith API。

LangSmithExceptionGroup

基类：LangSmithError

为低于 3.11 版本的 Python 提供的 ExceptionGroup 移植版。

方法	描述
__init__	初始化。

__init__

__init__(*args: Any, exceptions: Sequence[Exception], **kwargs: Any) -> None

初始化。

LangSmithWarning

基类：UserWarning

警告的基类。

LangSmithMissingAPIKeyWarning

基类：LangSmithWarning

缺少 API 密钥的警告。

FilterPoolFullWarning

基类：Filter

过滤在连接池未被重用时记录的 urrllib3 警告。

方法	描述
__init__	初始化 FilterPoolFullWarning 过滤器。
filter	urllib3.connectionpool:Connection pool is full, discarding connection: ...

__init__

__init__(name: str = '', host: str = '') -> None

初始化 FilterPoolFullWarning 过滤器。

参数	描述
name	过滤器的名称。默认为 ""。 类型: str 默认值: ''
host	要过滤的主机。默认为 ""。 类型: str 默认值: ''

filter

filter(record) -> bool

urllib3.connectionpool:Connection pool is full, discarding connection: ...

FilterLangSmithRetry

基类：Filter

用于此库重试的过滤器。

方法	描述
filter	过滤来自此库的重试。

filter

filter(record) -> bool

过滤来自此库的重试。

LangSmithRetry

基类：Retry

用于过滤具有此名称的日志的包装器。

ContextThreadPoolExecutor

基类：ThreadPoolExecutor

将上下文复制到子线程的 ThreadPoolExecutor。

方法	描述
submit	向执行器提交一个函数。
map	返回一个与 stdlib map 等效的迭代器。

submit

submit(func: Callable[P, T], *args: args, **kwargs: kwargs) -> Future[T]

向执行器提交一个函数。

参数	描述
func	要提交的函数。 类型： Callable[..., T]
*args	函数的位置参数。 类型: Any 默认值: ()
**kwargs	函数的关键字参数。 类型： Any 默认值： {}

返回	描述
Future[T]	Future[T]：函数的未来对象。

map

map(
    fn: Callable[..., T],
    *iterables: Iterable[Any],
    timeout: float | None = None,
    chunksize: int = 1,
) -> Iterator[T]

返回一个与 stdlib map 等效的迭代器。

每个函数都将从父线程接收到自己的上下文副本。

参数	描述
fn	一个可调用对象，它将接受与传递的迭代器数量一样多的参数。 类型： Callable[..., T]
timeout	等待的最大秒数。如果为 None，则等待时间没有限制。 类型： float | None 默认值： None
chunksize	在传递给子进程之前，迭代器将被分解成的块的大小。此参数仅由 ProcessPoolExecutor 使用；ThreadPoolExecutor 会忽略它。 类型： int 默认值： 1

返回	描述
Iterator[T]	一个等效于：map(func, *iterables) 的迭代器，但调用可能
Iterator[T]	以乱序方式评估。

引发	描述
TimeoutError	如果在给定的超时时间内无法生成整个结果迭代器。
Exception	如果 fn(*args) 对任何值引发异常。

tracing_is_enabled

tracing_is_enabled(ctx: dict | None = None) -> bool | Literal['local']

如果启用了跟踪，则返回 True。

test_tracking_is_disabled

test_tracking_is_disabled() -> bool

如果启用了测试，则返回 True。

xor_args

xor_args(*arg_groups: tuple[str, ...]) -> Callable

验证指定的关键字参数是互斥的。

raise_for_status_with_text

raise_for_status_with_text(response: Response | Response) -> None

抛出一个带有响应文本的错误。

get_enum_value

get_enum_value(enu: Enum | str) -> str

获取字符串枚举的值。

log_once cached

log_once(level: int, message: str) -> None

以指定级别记录一条消息，但只记录一次。

get_messages_from_inputs

get_messages_from_inputs(inputs: Mapping[str, Any]) -> list[dict[str, Any]]

从给定的输入字典中提取消息。

参数	描述
inputs	输入字典。 类型： Mapping[str, Any]

返回	描述
list[dict[str, Any]]	List[Dict[str, Any]]：一个表示提取消息的字典列表。

引发	描述
ValueError	如果在输入字典中找不到消息。

get_message_generation_from_outputs

get_message_generation_from_outputs(outputs: Mapping[str, Any]) -> dict[str, Any]

从给定的输出中检索消息生成。

参数	描述
outputs	输出字典。 类型： Mapping[str, Any]

返回	描述
dict[str, Any]	Dict[str, Any]：消息生成。

引发	描述
ValueError	如果未找到生成或存在多个生成。

get_prompt_from_inputs

get_prompt_from_inputs(inputs: Mapping[str, Any]) -> str

从给定的输入中检索提示。

参数	描述
inputs	输入字典。 类型： Mapping[str, Any]

返回	描述
str	提示。 类型： str

引发	描述
ValueError	如果未找到提示或存在多个提示。

get_llm_generation_from_outputs

get_llm_generation_from_outputs(outputs: Mapping[str, Any]) -> str

从输出中获取 LLM 生成。

get_docker_compose_command cached

get_docker_compose_command() -> list[str]

获取此系统的正确 docker compose 命令。

convert_langchain_message

convert_langchain_message(message: BaseMessageLike) -> dict

将 LangChain 消息转换为示例。

is_base_message_like

is_base_message_like(obj: object) -> bool

检查给定的对象是否类似于 BaseMessage。

参数	描述
obj	要检查的对象。 类型： object

返回	描述
bool	如果对象类似于 BaseMessage 则为 True，否则为 False。 类型： bool

is_env_var_truish

is_env_var_truish(value: str | None) -> bool

检查给定的环境变量是否为真值。

get_env_var cached

get_env_var(
    name: str,
    default: str | None = None,
    *,
    namespaces: tuple = ("LANGSMITH", "LANGCHAIN"),
) -> str | None

从命名空间列表中检索环境变量。

参数	描述
name	环境变量的名称。 类型： str
default	如果找不到环境变量，则返回的默认值。默认为 None。 类型: str | None 默认值: None
namespaces	用于搜索环境变量的命名空间元组。默认为 ("LANGSMITH", "LANGCHAINs")。 类型： Tuple 默认值： ('LANGSMITH', 'LANGCHAIN')

返回	描述
str | None	Optional[str]：如果找到，则为环境变量的值，否则为默认值。

get_tracer_project cached

get_tracer_project(return_default_value=True) -> str | None

获取 LangSmith 跟踪器的项目名称。

filter_logs

filter_logs(logger: Logger, filters: Sequence[Filter]) -> Generator[None, None, None]

临时向记录器添加指定的过滤器。

参数： - logger：将添加过滤器的记录器。 - filters：要临时添加到记录器的 logging.Filter 对象序列。

get_cache_dir

get_cache_dir(cache: str | None) -> str | None

获取测试缓存目录。

参数	描述
cache	缓存路径。 类型： str | None

返回	描述
str | None	Optional[str]：如果提供了缓存路径，则为缓存路径，否则
str | None	为 LANGSMITH_TEST_CACHE 环境变量的值。

filter_request_headers

filter_request_headers(
    request: Any,
    *,
    ignore_hosts: Sequence[str] | None = None,
    allow_hosts: Sequence[str] | None = None,
) -> Any

根据 ignore_hosts 和 allow_hosts 过滤请求头。

with_cache

with_cache(
    path: str | Path,
    ignore_hosts: Sequence[str] | None = None,
    allow_hosts: Sequence[str] | None = None,
) -> Generator[None, None, None]

为请求使用缓存。

with_optional_cache

with_optional_cache(
    path: str | Path | None,
    ignore_hosts: Sequence[str] | None = None,
    allow_hosts: Sequence[str] | None = None,
) -> Generator[None, None, None]

为请求使用缓存。

deepish_copy

deepish_copy(val: T) -> T

对一个值进行深拷贝，并对不可拷贝的对象做出妥协。

参数	描述
val	要深拷贝的值。 类型： T

返回	描述
T	深拷贝的值。

is_version_greater_or_equal

is_version_greater_or_equal(current_version: str, target_version: str) -> bool

检查当前版本是否大于或等于目标版本。

parse_prompt_identifier

parse_prompt_identifier(identifier: str) -> tuple[str, str, str]

解析格式为 owner/name:hash、name:hash、owner/name 或 name 的字符串。

参数	描述
identifier	要解析的提示标识符。 类型： str

返回	描述
tuple[str, str, str]	Tuple[str, str, str]：一个包含 (owner, name, hash) 的元组。

引发	描述
ValueError	如果标识符与预期格式不匹配。

get_api_url

get_api_url(api_url: str | None) -> str

从环境或给定值中获取 LangSmith API URL。

get_api_key

get_api_key(api_key: str | None) -> str | None

从环境或给定值中获取 API 密钥。

get_workspace_id

get_workspace_id(workspace_id: str | None) -> str | None

获取工作区 ID。

get_host_url cached

get_host_url(web_url: str | None, api_url: str)

根据 web URL 或 API URL 获取主机 URL。

is_truish

is_truish(val: Any) -> bool

检查值是否为真值。

参数	描述
val	要检查的值。 类型： Any

返回	描述
bool	如果值为真值则为 True，否则为 False。 类型： bool