codekingpro/portable-devtools
114k
1"""The Async LangSmith Client."""2 3from __future__ import annotations4 5import asyncio6import datetime7import json8import random9import uuid10import warnings11from collections.abc import AsyncGenerator, AsyncIterator, Mapping, Sequence12from functools import partial13from typing import (14 Any,15 Literal,16 Optional,17 Union,18 cast,19)20 21import httpx22 23from langsmith import client as ls_client24from langsmith import schemas as ls_schemas25from langsmith import utils as ls_utils26from langsmith._internal import _profiles27from langsmith._internal._hub import (28 HUB,29 PLATFORM_HUB,30 REPO_HANDLE_PATTERN,31 build_commit_url,32 resolve_owner_for_url,33 validate_parent_commit,34)35from langsmith.prompt_cache import AsyncPromptCache, async_prompt_cache_singleton36 37ID_TYPE = Union[uuid.UUID, str]38 39 40class AsyncClient:41 """Async Client for interacting with the LangSmith API."""42 43 __slots__ = (44 "_retry_config",45 "_client",46 "_web_url",47 "_settings",48 "_cache",49 "_custom_headers",50 "_api_key",51 "_oauth_access_token",52 "_profile_auth",53 "_profile_auth_headers",54 )55 56 _custom_headers: dict[str, str]57 _api_key: Optional[str]58 _oauth_access_token: Optional[str]59 _profile_auth: Optional[_profiles.ProfileAuth]60 _profile_auth_headers: dict[str, str]61 62 def _compute_headers(self) -> dict[str, str]:63 headers = {**self._custom_headers}64 # Required headers that should not be overridden65 headers["Content-Type"] = "application/json"66 if self._api_key:67 headers[ls_client.X_API_KEY] = self._api_key68 elif self._profile_auth_headers:69 headers.update(self._profile_auth_headers)70 elif self._oauth_access_token:71 headers["Authorization"] = f"Bearer {self._oauth_access_token}"72 return headers73 74 @property75 def headers(self) -> dict[str, str]:76 """Return the custom headers used for API requests."""77 return self._custom_headers78 79 @headers.setter80 def headers(self, value: Optional[dict[str, str]]) -> None:81 self._custom_headers = value or {}82 self._client.headers = httpx.Headers(self._compute_headers())83 84 @property85 def _headers(self) -> dict[str, str]:86 """Return the merged headers used for API requests."""87 return dict(self._client.headers)88 89 @property90 def api_key(self) -> Optional[str]:91 """Return the API key used for authentication."""92 return self._api_key93 94 @api_key.setter95 def api_key(self, value: Optional[str]) -> None:96 self._api_key = value97 self._client.headers = httpx.Headers(self._compute_headers())98 99 def __init__(100 self,101 api_url: Optional[str] = None,102 api_key: Optional[str] = None,103 timeout_ms: Optional[104 Union[105 int, tuple[Optional[int], Optional[int], Optional[int], Optional[int]]106 ]107 ] = None,108 retry_config: Optional[Mapping[str, Any]] = None,109 web_url: Optional[str] = None,110 headers: Optional[dict[str, str]] = None,111 disable_prompt_cache: bool = False,112 cache: Optional[Union[bool, AsyncPromptCache]] = None,113 ):114 """Initialize the async client.115 116 Args:117 api_url: URL for the LangSmith API.118 api_key: API key for the LangSmith API.119 timeout_ms: Timeout for requests in milliseconds.120 retry_config: Retry configuration.121 web_url: URL for the LangSmith web app.122 headers: Additional HTTP headers to include in all requests.123 124 These headers will be merged with the default headers125 (Content-Type, x-api-key, etc.). Custom headers will not override126 the default required headers.127 disable_prompt_cache: Disable prompt caching for this client.128 cache: **[Deprecated]** Control prompt caching behavior.129 130 This parameter is deprecated. Use `configure_global_async_prompt_cache()` to131 configure caching, or `disable_prompt_cache=True` to disable it.132 133 - `True`: Enable caching with the global singleton134 - `False`: Disable caching (equivalent to `disable_prompt_cache=True`)135 - `AsyncCache(...)`/`AsyncPromptCache(...)`: Use a custom cache instance136 """137 self._retry_config = retry_config or {"max_retries": 3}138 self._custom_headers = headers or {}139 env_api_url = ls_client._get_langsmith_env_var_uncached("ENDPOINT")140 env_api_key = ls_client._get_langsmith_env_var_uncached("API_KEY")141 profile_config = _profiles.load_profile_client_config()142 api_url_ = (143 api_url if api_url is not None else env_api_url or profile_config.api_url144 )145 explicit_or_env_api_key = api_key if api_key is not None else env_api_key146 profile_auth_enabled = api_key is None and env_api_key is None147 use_profile_oauth = profile_auth_enabled and profile_config.has_oauth148 api_key_ = (149 explicit_or_env_api_key150 if explicit_or_env_api_key is not None151 else None152 if use_profile_oauth153 else profile_config.api_key154 )155 self._oauth_access_token = (156 profile_config.oauth_access_token if use_profile_oauth else None157 )158 api_key = ls_utils.get_api_key(api_key_)159 api_url = ls_utils.get_api_url(api_url_)160 self._profile_auth = None161 self._profile_auth_headers = {}162 if use_profile_oauth:163 self._profile_auth = _profiles.ProfileAuth(164 profile_config,165 api_key_header=ls_client.X_API_KEY,166 )167 self._profile_auth_headers = self._profile_auth.current_auth_headers()168 self._oauth_access_token = self._profile_auth.oauth_access_token169 self._api_key = api_key170 _headers = self._compute_headers()171 ls_client._validate_api_key_if_hosted(172 api_url,173 api_key174 or self._oauth_access_token175 or (176 "profile-auth"177 if self._profile_auth is not None and self._profile_auth.has_auth178 else None179 ),180 )181 182 if isinstance(timeout_ms, int):183 timeout_: Union[tuple, float] = (timeout_ms / 1000, None, None, None)184 elif isinstance(timeout_ms, tuple):185 timeout_ = tuple([t / 1000 if t is not None else None for t in timeout_ms])186 else:187 timeout_ = 10188 self._client = httpx.AsyncClient(189 base_url=api_url, headers=_headers, timeout=timeout_190 )191 self._web_url = web_url192 self._settings: Optional[ls_schemas.LangSmithSettings] = None193 194 # Initialize prompt cache195 # Handle backwards compatibility for deprecated `cache` parameter196 if cache is not None and disable_prompt_cache:197 import warnings198 199 warnings.warn(200 "Both 'cache' and 'disable_prompt_cache' were provided. "201 "The 'cache' parameter is deprecated and will be removed in a future version. "202 "Using 'cache' parameter value.",203 DeprecationWarning,204 stacklevel=2,205 )206 207 if cache is not None:208 import warnings209 210 warnings.warn(211 "The 'cache' parameter is deprecated and will be removed in a future version. "212 "Use 'configure_global_async_prompt_cache()' to configure the global cache, or "213 "'disable_prompt_cache=True' to disable caching for this client.",214 DeprecationWarning,215 stacklevel=2,216 )217 # Handle old cache parameter218 if cache is False:219 self._cache: Optional[AsyncPromptCache] = None220 elif cache is True:221 self._cache = async_prompt_cache_singleton222 else:223 # Custom AsyncPromptCache instance provided224 self._cache = cache225 elif not disable_prompt_cache:226 # Use the global singleton instance227 self._cache = async_prompt_cache_singleton228 else:229 self._cache = None230 231 async def __aenter__(self) -> AsyncClient:232 """Enter the async client."""233 if self._cache is not None:234 await self._cache.start()235 return self236 237 async def __aexit__(self, exc_type, exc_val, exc_tb):238 """Exit the async client."""239 await self.aclose()240 241 async def aclose(self):242 """Close the async client."""243 if self._cache is not None:244 await self._cache.stop()245 await self._client.aclose()246 247 def __repr__(self) -> str:248 """Return a string representation of the instance.249 250 Returns:251 The string representation of the instance.252 """253 return f"AsyncClient (API URL: {self._api_url})"254 255 @property256 def _api_url(self):257 return str(self._client.base_url)258 259 @property260 def _host_url(self) -> str:261 """The web host url."""262 return ls_utils.get_host_url(self._web_url, self._api_url)263 264 async def _ensure_profile_auth(self) -> None:265 if self._api_key or self._profile_auth is None:266 return267 if self._profile_auth.needs_refresh():268 auth_headers = await asyncio.to_thread(self._profile_auth.get_auth_headers)269 else:270 auth_headers = self._profile_auth.current_auth_headers()271 self._profile_auth_headers = auth_headers272 self._oauth_access_token = self._profile_auth.oauth_access_token273 self._client.headers = httpx.Headers(self._compute_headers())274 275 async def _arequest_with_retries(276 self,277 method: str,278 endpoint: str,279 **kwargs: Any,280 ) -> httpx.Response:281 """Make an async HTTP request with retries."""282 max_retries = cast(int, self._retry_config.get("max_retries", 3))283 284 # Python requests library used by the normal Client filters out params with None values285 # The httpx library does not. Filter them out here to keep behavior consistent286 if "params" in kwargs:287 params = kwargs["params"]288 filtered_params = {k: v for k, v in params.items() if v is not None}289 kwargs["params"] = filtered_params290 291 await self._ensure_profile_auth()292 if self._profile_auth is not None and "headers" in kwargs:293 kwargs["headers"] = self._profile_auth.prepare_request_headers(294 kwargs["headers"]295 )296 297 for attempt in range(max_retries):298 try:299 try:300 response = await self._client.request(method, endpoint, **kwargs)301 ls_utils.raise_for_status_with_text(response)302 return response303 except httpx.HTTPStatusError as e:304 response = e.response305 if response.status_code in {425, 500, 502, 503, 504}:306 raise ls_utils.LangSmithAPIError(307 f"Server error ({response.status_code}) caused failure to"308 f" {method} {endpoint} in"309 f" LangSmith API. {repr(e)}"310 ) from e311 elif response.status_code == 408:312 raise ls_utils.LangSmithRequestTimeout(313 f"Client took too long to send request to {method}{endpoint}"314 ) from e315 elif response.status_code == 429:316 raise ls_utils.LangSmithRateLimitError(317 f"Rate limit exceeded for {endpoint}. {repr(e)}"318 ) from e319 elif response.status_code == 401:320 raise ls_utils.LangSmithAuthError(321 f"Authentication failed for {endpoint}. {repr(e)}"322 ) from e323 elif response.status_code == 404:324 raise ls_utils.LangSmithNotFoundError(325 f"Resource not found for {endpoint}. {repr(e)}"326 ) from e327 elif response.status_code == 409:328 raise ls_utils.LangSmithConflictError(329 f"Conflict for {endpoint}. {repr(e)}"330 ) from e331 else:332 raise ls_utils.LangSmithError(333 f"Failed to {method} {endpoint} in LangSmith API. {repr(e)}"334 ) from e335 except httpx.RequestError as e:336 raise ls_utils.LangSmithConnectionError(337 f"Request error: {repr(e)}"338 ) from e339 except (340 ls_utils.LangSmithConnectionError,341 ls_utils.LangSmithRequestTimeout,342 ls_utils.LangSmithAPIError,343 ):344 if attempt == max_retries - 1:345 raise346 sleep_time = 2**attempt + (random.random() * 0.5)347 await asyncio.sleep(sleep_time)348 raise ls_utils.LangSmithAPIError(349 "Unexpected error connecting to the LangSmith API"350 )351 352 async def _aget_paginated_list(353 self,354 path: str,355 params: Optional[dict[str, Any]] = None,356 ) -> AsyncIterator[dict[str, Any]]:357 """Get a paginated list of items."""358 params = params or {}359 offset = params.get("offset", 0)360 params["limit"] = params.get("limit", 100)361 while True:362 params["offset"] = offset363 response = await self._arequest_with_retries("GET", path, params=params)364 items = response.json()365 if not items:366 break367 for item in items:368 yield item369 if len(items) < params["limit"]:370 break371 offset += len(items)372 373 async def _aget_cursor_paginated_list(374 self,375 path: str,376 *,377 body: Optional[dict] = None,378 request_method: str = "POST",379 data_key: str = "runs",380 ) -> AsyncIterator[dict]:381 """Get a cursor paginated list of items."""382 params_ = body.copy() if body else {}383 while True:384 response = await self._arequest_with_retries(385 request_method,386 path,387 content=ls_client._dumps_json(params_),388 )389 response_body = response.json()390 if not response_body:391 break392 if not response_body.get(data_key):393 break394 for run in response_body[data_key]:395 yield run396 cursors = response_body.get("cursors")397 if not cursors:398 break399 if not cursors.get("next"):400 break401 params_["cursor"] = cursors["next"]402 403 async def create_run(404 self,405 name: str,406 inputs: dict[str, Any],407 run_type: str,408 *,409 project_name: Optional[str] = None,410 revision_id: Optional[ls_client.ID_TYPE] = None,411 **kwargs: Any,412 ) -> None:413 """Create a run."""414 run_create = {415 "name": name,416 "id": kwargs.get("id") or uuid.uuid4(),417 "inputs": inputs,418 "run_type": run_type,419 "session_name": project_name or ls_utils.get_tracer_project(),420 "revision_id": revision_id,421 **kwargs,422 }423 await self._arequest_with_retries(424 "POST", "/runs", content=ls_client._dumps_json(run_create)425 )426 427 async def update_run(428 self,429 run_id: ls_client.ID_TYPE,430 **kwargs: Any,431 ) -> None:432 """Update a run."""433 data = {**kwargs, "id": ls_client._as_uuid(run_id)}434 await self._arequest_with_retries(435 "PATCH",436 f"/runs/{ls_client._as_uuid(run_id)}",437 content=ls_client._dumps_json(data),438 )439 440 async def read_run(self, run_id: ls_client.ID_TYPE) -> ls_schemas.Run:441 """Read a run."""442 response = await self._arequest_with_retries(443 "GET",444 f"/runs/{ls_client._as_uuid(run_id)}",445 )446 return ls_schemas.Run(**response.json())447 448 async def list_runs(449 self,450 *,451 project_id: Optional[452 Union[ls_client.ID_TYPE, Sequence[ls_client.ID_TYPE]]453 ] = None,454 project_name: Optional[Union[str, Sequence[str]]] = None,455 run_type: Optional[str] = None,456 trace_id: Optional[ls_client.ID_TYPE] = None,457 reference_example_id: Optional[ls_client.ID_TYPE] = None,458 query: Optional[str] = None,459 filter: Optional[str] = None,460 trace_filter: Optional[str] = None,461 tree_filter: Optional[str] = None,462 is_root: Optional[bool] = None,463 parent_run_id: Optional[ls_client.ID_TYPE] = None,464 start_time: Optional[datetime.datetime] = None,465 error: Optional[bool] = None,466 run_ids: Optional[Sequence[ls_client.ID_TYPE]] = None,467 select: Optional[Sequence[str]] = None,468 limit: Optional[int] = None,469 **kwargs: Any,470 ) -> AsyncIterator[ls_schemas.Run]:471 """List runs from the LangSmith API.472 473 Args:474 project_id: The ID(s) of the project to filter by.475 project_name: The name(s) of the project to filter by.476 run_type: The type of the runs to filter by.477 trace_id: The ID of the trace to filter by.478 reference_example_id: The ID of the reference example to filter by.479 query: The query string to filter by.480 filter: The filter string to filter by.481 trace_filter: Filter to apply to the ROOT run in the trace tree.482 483 This is meant to be used in conjunction with the regular `filter`484 parameter to let you filter runs by attributes of the root run within a485 trace.486 tree_filter: Filter to apply to OTHER runs in the trace tree, including487 sibling and child runs.488 489 This is meant to be used in conjunction with the regular `filter`490 parameter to let you filter runs by attributes of any run within a491 trace.492 is_root: Whether to filter by root runs.493 parent_run_id: The ID of the parent run to filter by.494 start_time: The start time to filter by.495 error: Whether to filter by error status.496 run_ids: The IDs of the runs to filter by.497 select: The fields to select.498 limit: The maximum number of runs to return.499 **kwargs: Additional keyword arguments.500 501 Yields:502 The runs.503 504 Examples:505 ```python506 # List all runs in a project507 project_runs = client.list_runs(project_name="<your_project>")508 509 # List LLM and Chat runs in the last 24 hours510 todays_llm_runs = client.list_runs(511 project_name="<your_project>",512 start_time=datetime.now() - timedelta(days=1),513 run_type="llm",514 )515 516 # List root traces in a project517 root_runs = client.list_runs(project_name="<your_project>", is_root=1)518 519 # List runs without errors520 correct_runs = client.list_runs(project_name="<your_project>", error=False)521 522 # List runs and only return their inputs/outputs (to speed up the query)523 input_output_runs = client.list_runs(524 project_name="<your_project>", select=["inputs", "outputs"]525 )526 527 # List runs by run ID528 run_ids = [529 "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836",530 "9398e6be-964f-4aa4-8ae9-ad78cd4b7074",531 ]532 selected_runs = client.list_runs(id=run_ids)533 534 # List all "chain" type runs that took more than 10 seconds and had535 # `total_tokens` greater than 5000536 chain_runs = client.list_runs(537 project_name="<your_project>",538 filter='and(eq(run_type, "chain"), gt(latency, 10), gt(total_tokens, 5000))',539 )540 541 # List all runs called "extractor" whose root of the trace was assigned feedback "user_score" score of 1542 good_extractor_runs = client.list_runs(543 project_name="<your_project>",544 filter='eq(name, "extractor")',545 trace_filter='and(eq(feedback_key, "user_score"), eq(feedback_score, 1))',546 )547 548 # List all runs that started after a specific timestamp and either have "error" not equal to null or a "Correctness" feedback score equal to 0549 complex_runs = client.list_runs(550 project_name="<your_project>",551 filter='and(gt(start_time, "2023-07-15T12:34:56Z"), or(neq(error, null), and(eq(feedback_key, "Correctness"), eq(feedback_score, 0.0))))',552 )553 554 # List all runs where `tags` include "experimental" or "beta" and `latency` is greater than 2 seconds555 tagged_runs = client.list_runs(556 project_name="<your_project>",557 filter='and(or(has(tags, "experimental"), has(tags, "beta")), gt(latency, 2))',558 )559 ```560 """ # noqa: E501561 project_ids = []562 if isinstance(project_id, (uuid.UUID, str)):563 project_ids.append(project_id)564 elif isinstance(project_id, list):565 project_ids.extend(project_id)566 if project_name is not None:567 if isinstance(project_name, str):568 project_name = [project_name]569 projects = await asyncio.gather(570 *[self.read_project(project_name=name) for name in project_name]571 )572 project_ids.extend([project.id for project in projects])573 574 if select and "child_run_ids" in select:575 warnings.warn(576 "The child_run_ids field is deprecated and will be removed in following versions",577 DeprecationWarning,578 )579 580 body_query: dict[str, Any] = {581 "session": project_ids if project_ids else None,582 "run_type": run_type,583 "reference_example": (584 [reference_example_id] if reference_example_id else None585 ),586 "query": query,587 "filter": filter,588 "trace_filter": trace_filter,589 "tree_filter": tree_filter,590 "is_root": is_root,591 "parent_run": parent_run_id,592 "start_time": start_time.isoformat() if start_time else None,593 "error": error,594 "id": run_ids,595 "trace": trace_id,596 "select": select,597 "limit": limit,598 **kwargs,599 }600 if project_ids:601 body_query["session"] = [602 str(ls_client._as_uuid(id_)) for id_ in project_ids603 ]604 body = {k: v for k, v in body_query.items() if v is not None}605 ix = 0606 async for run in self._aget_cursor_paginated_list("/runs/query", body=body):607 yield ls_schemas.Run(**run)608 ix += 1609 if limit is not None and ix >= limit:610 break611 612 async def share_run(613 self, run_id: ls_client.ID_TYPE, *, share_id: Optional[ls_client.ID_TYPE] = None614 ) -> str:615 """Get a share link for a run asynchronously.616 617 Args:618 run_id (ID_TYPE): The ID of the run to share.619 share_id: Custom share ID.620 621 If not provided, a random UUID will be generated.622 623 Returns:624 The URL of the shared run.625 626 Raises:627 httpx.HTTPStatusError: If the API request fails.628 """629 run_id_ = ls_client._as_uuid(run_id, "run_id")630 data = {631 "run_id": str(run_id_),632 "share_token": str(share_id or uuid.uuid4()),633 }634 response = await self._arequest_with_retries(635 "PUT",636 f"/runs/{run_id_}/share",637 content=ls_client._dumps_json(data),638 )639 ls_utils.raise_for_status_with_text(response)640 share_token = response.json()["share_token"]641 return f"{self._host_url}/public/{share_token}/r"642 643 async def run_is_shared(self, run_id: ls_client.ID_TYPE) -> bool:644 """Get share state for a run asynchronously."""645 link = await self.read_run_shared_link(ls_client._as_uuid(run_id, "run_id"))646 return link is not None647 648 async def read_run_shared_link(self, run_id: ls_client.ID_TYPE) -> Optional[str]:649 """Retrieve the shared link for a specific run asynchronously.650 651 Args:652 run_id (ID_TYPE): The ID of the run.653 654 Returns:655 Optional[str]: The shared link for the run, or None if the link is not656 available.657 658 Raises:659 httpx.HTTPStatusError: If the API request fails.660 """661 response = await self._arequest_with_retries(662 "GET",663 f"/runs/{ls_client._as_uuid(run_id, 'run_id')}/share",664 )665 ls_utils.raise_for_status_with_text(response)666 result = response.json()667 if result is None or "share_token" not in result:668 return None669 return f"{self._host_url}/public/{result['share_token']}/r"670 671 async def create_project(672 self,673 project_name: str,674 **kwargs: Any,675 ) -> ls_schemas.TracerSession:676 """Create a project."""677 data = {"name": project_name, **kwargs}678 response = await self._arequest_with_retries(679 "POST", "/sessions", content=ls_client._dumps_json(data)680 )681 return ls_schemas.TracerSession(**response.json())682 683 async def read_project(684 self,685 project_name: Optional[str] = None,686 project_id: Optional[ls_client.ID_TYPE] = None,687 ) -> ls_schemas.TracerSession:688 """Read a project."""689 if project_id:690 response = await self._arequest_with_retries(691 "GET", f"/sessions/{ls_client._as_uuid(project_id)}"692 )693 elif project_name:694 response = await self._arequest_with_retries(695 "GET", "/sessions", params={"name": project_name}696 )697 else:698 raise ValueError("Either project_name or project_id must be provided")699 700 data = response.json()701 if isinstance(data, list):702 if not data:703 raise ls_utils.LangSmithNotFoundError(704 f"Project {project_name} not found"705 )706 return ls_schemas.TracerSession(**data[0])707 return ls_schemas.TracerSession(**data)708 709 async def delete_project(710 self, *, project_name: Optional[str] = None, project_id: Optional[str] = None711 ) -> None:712 """Delete a project from LangSmith.713 714 Args:715 project_name: The name of the project to delete.716 project_id: The ID of the project to delete.717 """718 if project_id is None and project_name is None:719 raise ValueError("Either project_name or project_id must be provided")720 if project_id is None:721 project = await self.read_project(project_name=project_name)722 project_id = str(project.id)723 if not project_id:724 raise ValueError("Project not found")725 await self._arequest_with_retries(726 "DELETE",727 f"/sessions/{ls_client._as_uuid(project_id)}",728 )729 730 async def create_dataset(731 self,732 dataset_name: str,733 **kwargs: Any,734 ) -> ls_schemas.Dataset:735 """Create a dataset."""736 data = {"name": dataset_name, **kwargs}737 response = await self._arequest_with_retries(738 "POST", "/datasets", content=ls_client._dumps_json(data)739 )740 return ls_schemas.Dataset(**response.json())741 742 async def read_dataset(743 self,744 dataset_name: Optional[str] = None,745 dataset_id: Optional[ls_client.ID_TYPE] = None,746 ) -> ls_schemas.Dataset:747 """Read a dataset."""748 if dataset_id:749 response = await self._arequest_with_retries(750 "GET", f"/datasets/{ls_client._as_uuid(dataset_id)}"751 )752 elif dataset_name:753 response = await self._arequest_with_retries(754 "GET", "/datasets", params={"name": dataset_name}755 )756 else:757 raise ValueError("Either dataset_name or dataset_id must be provided")758 759 data = response.json()760 if isinstance(data, list):761 if not data:762 raise ls_utils.LangSmithNotFoundError(763 f"Dataset {dataset_name} not found"764 )765 return ls_schemas.Dataset(**data[0])766 return ls_schemas.Dataset(**data)767 768 async def delete_dataset(self, dataset_id: ls_client.ID_TYPE) -> None:769 """Delete a dataset."""770 await self._arequest_with_retries(771 "DELETE",772 f"/datasets/{ls_client._as_uuid(dataset_id)}",773 )774 775 async def list_datasets(776 self,777 **kwargs: Any,778 ) -> AsyncIterator[ls_schemas.Dataset]:779 """List datasets."""780 async for dataset in self._aget_paginated_list("/datasets", params=kwargs):781 yield ls_schemas.Dataset(**dataset)782 783 async def create_example(784 self,785 inputs: dict[str, Any],786 outputs: Optional[dict[str, Any]] = None,787 dataset_id: Optional[ls_client.ID_TYPE] = None,788 dataset_name: Optional[str] = None,789 **kwargs: Any,790 ) -> ls_schemas.Example:791 """Create an example."""792 if dataset_id is None and dataset_name is None:793 raise ValueError("Either dataset_id or dataset_name must be provided")794 if dataset_id is None:795 dataset = await self.read_dataset(dataset_name=dataset_name)796 dataset_id = dataset.id797 798 data = {799 "inputs": inputs,800 "outputs": outputs,801 "dataset_id": str(dataset_id),802 **kwargs,803 }804 response = await self._arequest_with_retries(805 "POST", "/examples", content=ls_client._dumps_json(data)806 )807 return ls_schemas.Example(**response.json())808 809 async def read_example(self, example_id: ls_client.ID_TYPE) -> ls_schemas.Example:810 """Read an example."""811 response = await self._arequest_with_retries(812 "GET", f"/examples/{ls_client._as_uuid(example_id)}"813 )814 return ls_schemas.Example(**response.json())815 816 async def list_examples(817 self,818 *,819 dataset_id: Optional[ls_client.ID_TYPE] = None,820 dataset_name: Optional[str] = None,821 **kwargs: Any,822 ) -> AsyncIterator[ls_schemas.Example]:823 """List examples."""824 params = kwargs.copy()825 if dataset_id:826 params["dataset"] = ls_client._as_uuid(dataset_id)827 elif dataset_name:828 dataset = await self.read_dataset(dataset_name=dataset_name)829 params["dataset"] = dataset.id830 831 async for example in self._aget_paginated_list("/examples", params=params):832 yield ls_schemas.Example(**example)833 834 async def create_feedback(835 self,836 run_id: Optional[ls_client.ID_TYPE],837 key: str,838 score: Optional[float] = None,839 value: Union[float, int, bool, str, dict, None] = None,840 comment: Optional[str] = None,841 **kwargs: Any,842 ) -> ls_schemas.Feedback:843 """Create feedback for a run.844 845 Args:846 run_id: The ID of the run to provide feedback for.847 848 Can be `None` for project-level feedback.849 key: The name of the metric or aspect this feedback is about.850 score: The score to rate this run on the metric or aspect.851 value: The display value or non-numeric value for this feedback.852 comment: A comment about this feedback.853 **kwargs: Additional keyword arguments to include in the feedback data.854 855 Returns:856 The created feedback object.857 858 Raises:859 httpx.HTTPStatusError: If the API request fails.860 """ # noqa: E501861 data = {862 "run_id": ls_client._ensure_uuid(run_id, accept_null=True),863 "key": key,864 "score": score,865 "value": value,866 "comment": comment,867 **kwargs,868 }869 response = await self._arequest_with_retries(870 "POST", "/feedback", content=ls_client._dumps_json(data)871 )872 return ls_schemas.Feedback(**response.json())873 874 async def create_feedback_from_token(875 self,876 token_or_url: Union[str, uuid.UUID],877 score: Union[float, int, bool, None] = None,878 *,879 value: Union[float, int, bool, str, dict, None] = None,880 correction: Union[dict, None] = None,881 comment: Union[str, None] = None,882 metadata: Optional[dict] = None,883 ) -> None:884 """Create feedback from a presigned token or URL.885 886 Args:887 token_or_url: The token or URL from which to create feedback.888 score: The score of the feedback.889 value: The value of the feedback.890 correction: The correction of the feedback.891 comment: The comment of the feedback.892 metadata: Additional metadata for the feedback.893 894 Raises:895 ValueError: If the source API URL is invalid.896 897 Returns:898 This method does not return anything.899 """900 source_api_url, token_uuid = ls_client._parse_token_or_url(901 token_or_url, self._api_url, num_parts=1902 )903 if source_api_url != self._api_url:904 raise ValueError(f"Invalid source API URL. {source_api_url}")905 response = await self._arequest_with_retries(906 "POST",907 f"/feedback/tokens/{ls_client._as_uuid(token_uuid)}",908 content=ls_client._dumps_json(909 {910 "score": score,911 "value": value,912 "correction": correction,913 "comment": comment,914 "metadata": metadata,915 # TODO: Add ID once the API supports it.916 }917 ),918 )919 ls_utils.raise_for_status_with_text(response)920 921 async def create_presigned_feedback_token(922 self,923 run_id: ls_client.ID_TYPE,924 feedback_key: str,925 *,926 expiration: Optional[datetime.datetime | datetime.timedelta] = None,927 feedback_config: Optional[ls_schemas.FeedbackConfig] = None,928 feedback_id: Optional[ls_client.ID_TYPE] = None,929 ) -> ls_schemas.FeedbackIngestToken:930 """Create a pre-signed URL to send feedback data to.931 932 This is useful for giving browser-based clients a way to upload933 feedback data directly to LangSmith without accessing the934 API key.935 936 Args:937 run_id (Union[UUID, str]): The ID of the run to provide feedback for.938 feedback_key: The name of the metric or aspect this feedback is about.939 expiration: The expiration time of the pre-signed URL.940 941 Either a datetime or a timedelta offset from now.942 943 Default to 3 hours.944 feedback_config: `FeedbackConfig` or `None`.945 946 If creating a feedback_key for the first time, this defines how the947 metric should be interpreted, such as a continuous score (w/ optional948 bounds), or distribution over categorical values.949 feedback_id: The ID of the feedback to create.950 951 If not provided, a new feedback will be created.952 953 Returns:954 The pre-signed URL for uploading feedback data.955 """956 body: dict[str, Any] = {957 "run_id": run_id,958 "feedback_key": feedback_key,959 "feedback_config": feedback_config,960 "id": feedback_id or str(uuid.uuid4()),961 }962 if expiration is None:963 body["expires_in"] = ls_schemas.TimeDeltaInput(964 days=0,965 hours=3,966 minutes=0,967 )968 elif isinstance(expiration, datetime.datetime):969 body["expires_at"] = expiration.isoformat()970 elif isinstance(expiration, datetime.timedelta):971 body["expires_in"] = ls_schemas.TimeDeltaInput(972 days=expiration.days,973 hours=expiration.seconds // 3600,974 minutes=(expiration.seconds % 3600) // 60,975 )976 else:977 raise ValueError(978 f"Invalid expiration type: {type(expiration)}. "979 "Expected datetime.datetime or datetime.timedelta."980 )981 982 response = await self._arequest_with_retries(983 "POST",984 "/feedback/tokens",985 content=ls_client._dumps_json(body),986 )987 return ls_schemas.FeedbackIngestToken(**response.json())988 989 async def read_feedback(990 self, feedback_id: ls_client.ID_TYPE991 ) -> ls_schemas.Feedback:992 """Read feedback."""993 response = await self._arequest_with_retries(994 "GET", f"/feedback/{ls_client._as_uuid(feedback_id)}"995 )996 return ls_schemas.Feedback(**response.json())997 998 async def list_feedback(999 self,1000 *,1001 run_ids: Optional[Sequence[ls_client.ID_TYPE]] = None,1002 feedback_key: Optional[Sequence[str]] = None,1003 feedback_source_type: Optional[Sequence[ls_schemas.FeedbackSourceType]] = None,1004 limit: Optional[int] = None,1005 **kwargs: Any,1006 ) -> AsyncIterator[ls_schemas.Feedback]:1007 """List feedback."""1008 params = {1009 "run": (1010 [str(ls_client._as_uuid(id_)) for id_ in run_ids] if run_ids else None1011 ),1012 "limit": min(limit, 100) if limit is not None else 100,1013 **kwargs,1014 }1015 if feedback_key is not None:1016 params["key"] = feedback_key1017 if feedback_source_type is not None:1018 params["source"] = feedback_source_type1019 ix = 01020 async for feedback in self._aget_paginated_list("/feedback", params=params):1021 yield ls_schemas.Feedback(**feedback)1022 ix += 11023 if limit is not None and ix >= limit:1024 break1025 1026 async def delete_feedback(self, feedback_id: ID_TYPE) -> None:1027 """Delete a feedback by ID.1028 1029 Args:1030 feedback_id (Union[UUID, str]): The ID of the feedback to delete.1031 """1032 response = await self._arequest_with_retries(1033 "DELETE", f"/feedback/{ls_client._as_uuid(feedback_id, 'feedback_id')}"1034 )1035 ls_utils.raise_for_status_with_text(response)1036 1037 # Annotation Queue API1038 1039 async def list_annotation_queues(1040 self,1041 *,1042 queue_ids: Optional[list[ID_TYPE]] = None,1043 name: Optional[str] = None,1044 name_contains: Optional[str] = None,1045 limit: Optional[int] = None,1046 ) -> AsyncIterator[ls_schemas.AnnotationQueue]:1047 """List the annotation queues on the LangSmith API.1048 1049 Args:1050 queue_ids (Optional[List[Union[UUID, str]]]): The IDs of the queues to1051 filter by.1052 name: The name of the queue to filter by.1053 name_contains: The substring that the queue name should contain.1054 limit: The maximum number of queues to return.1055 1056 Yields:1057 The annotation queues.1058 """1059 params: dict = {1060 "ids": (1061 [1062 ls_client._as_uuid(id_, f"queue_ids[{i}]")1063 for i, id_ in enumerate(queue_ids)1064 ]1065 if queue_ids is not None1066 else None1067 ),1068 "name": name,1069 "name_contains": name_contains,1070 "limit": min(limit, 100) if limit is not None else 100,1071 }1072 ix = 01073 async for feedback in self._aget_paginated_list(1074 "/annotation-queues", params=params1075 ):1076 yield ls_schemas.AnnotationQueue(**feedback)1077 ix += 11078 if limit is not None and ix >= limit:1079 break1080 1081 async def create_annotation_queue(1082 self,1083 *,1084 name: str,1085 description: Optional[str] = None,1086 queue_id: Optional[ID_TYPE] = None,1087 rubric_instructions: Optional[str] = None,1088 rubric_items: Optional[list[ls_schemas.AnnotationQueueRubricItem]] = None,1089 ) -> ls_schemas.AnnotationQueue:1090 """Create an annotation queue on the LangSmith API.1091 1092 Args:1093 name: The name of the annotation queue.1094 description: The description of the annotation queue.1095 queue_id (Optional[Union[UUID, str]]): The ID of the annotation queue.1096 rubric_instructions: The rubric instructions for the annotation queue.1097 rubric_items: The feedback configs to assign to this queue's rubric.1098 1099 Returns:1100 The created annotation queue object.1101 """1102 body: dict[str, Any] = {1103 "name": name,1104 "description": description,1105 "id": str(queue_id) if queue_id is not None else str(uuid.uuid4()),1106 "rubric_instructions": rubric_instructions,1107 }1108 if rubric_items is not None:1109 body["rubric_items"] = rubric_items1110 response = await self._arequest_with_retries(1111 "POST",1112 "/annotation-queues",1113 json={k: v for k, v in body.items() if v is not None},1114 )1115 ls_utils.raise_for_status_with_text(response)1116 return ls_schemas.AnnotationQueue(1117 **response.json(),1118 )1119 1120 async def read_annotation_queue(1121 self, queue_id: ID_TYPE1122 ) -> ls_schemas.AnnotationQueue:1123 """Read an annotation queue with the specified `queue_id`.1124 1125 Args:1126 queue_id (Union[UUID, str]): The ID of the annotation queue to read.1127 1128 Returns:1129 The annotation queue object.1130 """1131 # TODO: Replace when actual endpoint is added1132 return await self.list_annotation_queues(queue_ids=[queue_id]).__anext__()1133 1134 async def update_annotation_queue(1135 self,1136 queue_id: ID_TYPE,1137 *,1138 name: Optional[str] = None,1139 description: Optional[str] = None,1140 rubric_instructions: Optional[str] = None,1141 rubric_items: Optional[list[ls_schemas.AnnotationQueueRubricItem]] = None,1142 ) -> None:1143 """Update an annotation queue with the specified `queue_id`.1144 1145 Args:1146 queue_id (Union[UUID, str]): The ID of the annotation queue to update.1147 name: The new name for the annotation queue.1148 description: The new description for the annotation queue.1149 rubric_instructions: The new rubric instructions for the queue.1150 rubric_items: The feedback configs to assign to this queue's rubric.1151 """1152 body: dict[str, Any] = {}1153 if name is not None:1154 body["name"] = name1155 if description is not None:1156 body["description"] = description1157 if rubric_instructions is not None:1158 body["rubric_instructions"] = rubric_instructions1159 if rubric_items is not None:1160 body["rubric_items"] = rubric_items1161 response = await self._arequest_with_retries(1162 "PATCH",1163 f"/annotation-queues/{ls_client._as_uuid(queue_id, 'queue_id')}",1164 json=body,1165 )1166 ls_utils.raise_for_status_with_text(response)1167 1168 async def delete_annotation_queue(self, queue_id: ID_TYPE) -> None:1169 """Delete an annotation queue with the specified `queue_id`.1170 1171 Args:1172 queue_id (Union[UUID, str]): The ID of the annotation queue to delete.1173 """1174 response = await self._arequest_with_retries(1175 "DELETE",1176 f"/annotation-queues/{ls_client._as_uuid(queue_id, 'queue_id')}",1177 headers={"Accept": "application/json", **self._client.headers},1178 )1179 ls_utils.raise_for_status_with_text(response)1180 1181 async def add_runs_to_annotation_queue(1182 self, queue_id: ID_TYPE, *, run_ids: list[ID_TYPE]1183 ) -> None:1184 """Add runs to an annotation queue with the specified `queue_id`.1185 1186 Args:1187 queue_id (Union[UUID, str]): The ID of the annotation queue.1188 run_ids (list[Union[UUID, str]]): The IDs of the runs to be added to the1189 annotation queue.1190 """1191 response = await self._arequest_with_retries(1192 "POST",1193 f"/annotation-queues/{ls_client._as_uuid(queue_id, 'queue_id')}/runs",1194 json=[1195 str(ls_client._as_uuid(id_, f"run_ids[{i}]"))1196 for i, id_ in enumerate(run_ids)1197 ],1198 )1199 ls_utils.raise_for_status_with_text(response)1200