Utilities

langsmith.utils

Generic utility functions.

FUNCTION	DESCRIPTION
tracing_is_enabled	Return True if tracing is enabled.
test_tracking_is_disabled	Return True if testing is enabled.
xor_args	Validate specified keyword args are mutually exclusive.
raise_for_status_with_text	Raise an error with the response text.
get_enum_value	Get the value of a string enum.
log_once	Log a message at the specified level, but only once.
get_messages_from_inputs	Extract messages from the given inputs dictionary.
get_message_generation_from_outputs	Retrieve the message generation from the given outputs.
get_prompt_from_inputs	Retrieve the prompt from the given inputs.
get_llm_generation_from_outputs	Get the LLM generation from the outputs.
get_docker_compose_command	Get the correct docker compose command for this system.
convert_langchain_message	Convert a LangChain message to an example.
is_base_message_like	Check if the given object is similar to BaseMessage.
is_env_var_truish	Check if the given environment variable is truish.
get_env_var	Retrieve an environment variable from a list of namespaces.
get_tracer_project	Get the project name for a LangSmith tracer.
filter_logs	Temporarily adds specified filters to a logger.
get_cache_dir	Get the testing cache directory.
filter_request_headers	Filter request headers based on ignore_hosts and allow_hosts.
with_cache	Use a cache for requests.
with_optional_cache	Use a cache for requests.
deepish_copy	Deep copy a value with a compromise for uncopyable objects.
is_version_greater_or_equal	Check if the current version is greater or equal to the target version.
parse_prompt_identifier	Parse a string in the format of owner/name:hash, name:hash, owner/name, or name.
get_api_url	Get the LangSmith API URL from the environment or the given value.
get_api_key	Get the API key from the environment or the given value.
get_workspace_id	Get workspace ID.
get_host_url	Get the host URL based on the web URL or API URL.
is_truish	Check if the value is truish.

LangSmithError

Bases: Exception

An error occurred while communicating with the LangSmith API.

LangSmithAPIError

Bases: LangSmithError

Internal server error while communicating with LangSmith.

LangSmithRequestTimeout

Bases: LangSmithError

Client took too long to send request body.

LangSmithUserError

Bases: LangSmithError

User error caused an exception when communicating with LangSmith.

LangSmithRateLimitError

Bases: LangSmithError

You have exceeded the rate limit for the LangSmith API.

LangSmithAuthError

Bases: LangSmithError

Couldn't authenticate with the LangSmith API.

LangSmithNotFoundError

Bases: LangSmithError

Couldn't find the requested resource.

LangSmithConflictError

Bases: LangSmithError

The resource already exists.

LangSmithConnectionError

Bases: LangSmithError

Couldn't connect to the LangSmith API.

LangSmithExceptionGroup

Bases: LangSmithError

Port of ExceptionGroup for Py < 3.11.

METHOD	DESCRIPTION
__init__	Initialize.

__init__

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

Initialize.

LangSmithWarning

Bases: UserWarning

Base class for warnings.

LangSmithMissingAPIKeyWarning

Bases: LangSmithWarning

Warning for missing API key.

FilterPoolFullWarning

Bases: Filter

Filter urrllib3 warnings logged when the connection pool isn't reused.

METHOD	DESCRIPTION
__init__	Initialize the FilterPoolFullWarning filter.
filter	urllib3.connectionpool:Connection pool is full, discarding connection: ...

__init__

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

Initialize the FilterPoolFullWarning filter.

PARAMETER	DESCRIPTION
name	The name of the filter. Defaults to "". TYPE: str DEFAULT: ''
host	The host to filter. Defaults to "". TYPE: str DEFAULT: ''

filter

filter(record) -> bool

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

FilterLangSmithRetry

Bases: Filter

Filter for retries from this lib.

METHOD	DESCRIPTION
filter	Filter retries from this library.

filter

filter(record) -> bool

Filter retries from this library.

LangSmithRetry

Bases: Retry

Wrapper to filter logs with this name.

ContextThreadPoolExecutor

Bases: ThreadPoolExecutor

ThreadPoolExecutor that copies the context to the child thread.

METHOD	DESCRIPTION
submit	Submit a function to the executor.
map	Return an iterator equivalent to stdlib map.

submit

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

Submit a function to the executor.

PARAMETER	DESCRIPTION
func	The function to submit. TYPE: Callable[..., T]
*args	The positional arguments to the function. TYPE: Any DEFAULT: ()
**kwargs	The keyword arguments to the function. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Future[T]	Future[T]: The future for the function.

map

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

Return an iterator equivalent to stdlib map.

Each function will receive its own copy of the context from the parent thread.

PARAMETER	DESCRIPTION
fn	A callable that will take as many arguments as there are passed iterables. TYPE: Callable[..., T]
timeout	The maximum number of seconds to wait. If None, then there is no limit on the wait time. TYPE: Optional[float] DEFAULT: None
chunksize	The size of the chunks the iterable will be broken into before being passed to a child process. This argument is only used by ProcessPoolExecutor; it is ignored by ThreadPoolExecutor. TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
Iterator[T]	An iterator equivalent to: map(func, *iterables) but the calls may
Iterator[T]	be evaluated out-of-order.

RAISES	DESCRIPTION
TimeoutError	If the entire result iterator could not be generated before the given timeout.
Exception	If fn(*args) raises for any values.

tracing_is_enabled

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

Return True if tracing is enabled.

test_tracking_is_disabled

test_tracking_is_disabled() -> bool

Return True if testing is enabled.

xor_args

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

Validate specified keyword args are mutually exclusive.

raise_for_status_with_text

raise_for_status_with_text(response: Union[Response, Response]) -> None

Raise an error with the response text.

get_enum_value

get_enum_value(enu: Union[Enum, str]) -> str

Get the value of a string enum.

log_once cached

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

Log a message at the specified level, but only once.

get_messages_from_inputs

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

Extract messages from the given inputs dictionary.

PARAMETER	DESCRIPTION
inputs	The inputs dictionary. TYPE: Mapping[str, Any]

RETURNS	DESCRIPTION
list[dict[str, Any]]	List[Dict[str, Any]]: A list of dictionaries representing the extracted messages.

RAISES	DESCRIPTION
ValueError	If no message(s) are found in the inputs dictionary.

get_message_generation_from_outputs

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

Retrieve the message generation from the given outputs.

PARAMETER	DESCRIPTION
outputs	The outputs dictionary. TYPE: Mapping[str, Any]

RETURNS	DESCRIPTION
dict[str, Any]	Dict[str, Any]: The message generation.

RAISES	DESCRIPTION
ValueError	If no generations are found or if multiple generations are present.

get_prompt_from_inputs

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

Retrieve the prompt from the given inputs.

PARAMETER	DESCRIPTION
inputs	The inputs dictionary. TYPE: Mapping[str, Any]

RETURNS	DESCRIPTION
str	The prompt. TYPE: str

RAISES	DESCRIPTION
ValueError	If the prompt is not found or if multiple prompts are present.

get_llm_generation_from_outputs

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

Get the LLM generation from the outputs.

get_docker_compose_command cached

get_docker_compose_command() -> list[str]

Get the correct docker compose command for this system.

convert_langchain_message

convert_langchain_message(message: BaseMessageLike) -> dict

Convert a LangChain message to an example.

is_base_message_like

is_base_message_like(obj: object) -> bool

Check if the given object is similar to BaseMessage.

PARAMETER	DESCRIPTION
obj	The object to check. TYPE: object

RETURNS	DESCRIPTION
bool	True if the object is similar to BaseMessage, False otherwise. TYPE: bool

is_env_var_truish

is_env_var_truish(value: Optional[str]) -> bool

Check if the given environment variable is truish.

get_env_var cached

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

Retrieve an environment variable from a list of namespaces.

PARAMETER	DESCRIPTION
name	The name of the environment variable. TYPE: str
default	The default value to return if the environment variable is not found. Defaults to None. TYPE: Optional[str] DEFAULT: None
namespaces	A tuple of namespaces to search for the environment variable. Defaults to ("LANGSMITH", "LANGCHAINs"). TYPE: Tuple DEFAULT: ('LANGSMITH', 'LANGCHAIN')

RETURNS	DESCRIPTION
Optional[str]	Optional[str]: The value of the environment variable if found, otherwise the default value.

get_tracer_project cached

get_tracer_project(return_default_value=True) -> Optional[str]

Get the project name for a LangSmith tracer.

filter_logs

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

Temporarily adds specified filters to a logger.

Parameters: - logger: The logger to which the filters will be added. - filters: A sequence of logging.Filter objects to be temporarily added to the logger.

get_cache_dir

get_cache_dir(cache: Optional[str]) -> Optional[str]

Get the testing cache directory.

PARAMETER	DESCRIPTION
cache	The cache path. TYPE: Optional[str]

RETURNS	DESCRIPTION
Optional[str]	Optional[str]: The cache path if provided, otherwise the value
Optional[str]	from the LANGSMITH_TEST_CACHE environment variable.

filter_request_headers

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

Filter request headers based on ignore_hosts and allow_hosts.

with_cache

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

Use a cache for requests.

with_optional_cache

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

Use a cache for requests.

deepish_copy

deepish_copy(val: T) -> T

Deep copy a value with a compromise for uncopyable objects.

PARAMETER	DESCRIPTION
val	The value to be deep copied. TYPE: T

RETURNS	DESCRIPTION
T	The deep copied value.

is_version_greater_or_equal

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

Check if the current version is greater or equal to the target version.

parse_prompt_identifier

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

Parse a string in the format of owner/name:hash, name:hash, owner/name, or name.

PARAMETER	DESCRIPTION
identifier	The prompt identifier to parse. TYPE: str

RETURNS	DESCRIPTION
tuple[str, str, str]	Tuple[str, str, str]: A tuple containing (owner, name, hash).

RAISES	DESCRIPTION
ValueError	If the identifier doesn't match the expected formats.

get_api_url

get_api_url(api_url: Optional[str]) -> str

Get the LangSmith API URL from the environment or the given value.

get_api_key

get_api_key(api_key: Optional[str]) -> Optional[str]

Get the API key from the environment or the given value.

get_workspace_id

get_workspace_id(workspace_id: Optional[str]) -> Optional[str]

Get workspace ID.

get_host_url cached

get_host_url(web_url: Optional[str], api_url: str)

Get the host URL based on the web URL or API URL.

is_truish

is_truish(val: Any) -> bool

Check if the value is truish.

PARAMETER	DESCRIPTION
val	The value to check. TYPE: Any

RETURNS	DESCRIPTION
bool	True if the value is truish, False otherwise. TYPE: bool