AsyncClient#

用于与 LangSmith API 交互的异步客户端。

初始化异步客户端。

方法

`__init__`([api_url, api_key, timeout_ms, ...])	初始化异步客户端。
`aclose`()	关闭异步客户端。
`add_runs_to_annotation_queue`(queue_id, *, ...)	将运行添加到具有指定队列 ID 的标注队列。
`create_annotation_queue`(*, name[, ...])	在 LangSmith API 上创建标注队列。
`create_dataset`(dataset_name, **kwargs)	创建数据集。
`create_example`(inputs[, outputs, ...])	创建示例。
`create_feedback`(run_id, key[, score, value, ...])	为运行创建反馈。
`create_feedback_from_token`(token_or_url[, ...])	从预签名令牌或 URL 创建反馈。
`create_presigned_feedback_token`(run_id, ...)	创建用于发送反馈数据的预签名 URL。
`create_project`(project_name, **kwargs)	创建项目。
`create_run`(name, inputs, run_type, *[, ...])	创建运行。
`delete_annotation_queue`(queue_id)	删除具有指定队列 ID 的标注队列。
`delete_dataset`(dataset_id)	删除数据集。
`delete_feedback`(feedback_id)	按 ID 删除反馈。
`delete_project`(*[, project_name, project_id])	从 LangSmith 删除项目。
`delete_run_from_annotation_queue`(queue_id, ...)	从具有指定队列 ID 和运行 ID 的标注队列中删除运行。
`get_run_from_annotation_queue`(queue_id, *, index)	从指定索引的标注队列中获取运行。
`index_dataset`(*, dataset_id[, tag])	启用数据集索引。
`list_annotation_queues`(*[, queue_ids, name, ...])	列出 LangSmith API 上的标注队列。
`list_datasets`(**kwargs)	列出数据集。
`list_examples`(*[, dataset_id, dataset_name])	列出示例。
`list_feedback`(*[, run_ids, feedback_key, ...])	列出反馈。
`list_runs`(*[, project_id, project_name, ...])	列出 LangSmith API 中的运行。
`read_annotation_queue`(queue_id)	读取具有指定队列 ID 的标注队列。
`read_dataset`([dataset_name, dataset_id])	读取数据集。
`read_example`(example_id)	读取示例。
`read_feedback`(feedback_id)	读取反馈。
`read_project`([project_name, project_id])	读取项目。
`read_run`(run_id)	读取运行。
`read_run_shared_link`(run_id)	异步检索特定运行的共享链接。
`run_is_shared`(run_id)	异步获取运行的共享状态。
`share_run`(run_id, *[, share_id])	异步获取运行的共享链接。
`similar_examples`(inputs, /, *, limit, dataset_id)	检索输入最匹配当前输入的数据集示例。
`update_annotation_queue`(queue_id, *, name[, ...])	更新具有指定 queue_id 的标注队列。
`update_run`(run_id, **kwargs)	更新运行。

参数:

api_url (Optional[str])
api_key (Optional[str])
timeout_ms (Optional[Union[int, Tuple[Optional[int], Optional[int], Optional[int], Optional[int]]]])
retry_config (Optional[Mapping[str, Any]])
web_url (Optional[str])

初始化异步客户端。

参数:

api_url (str | None)
api_key (str | None)
timeout_ms (int | Tuple[int | None, int | None, int | None, int | None] | None)
retry_config (Mapping[str, Any] | None)
web_url (str | None)

async aclose()[source]#: 关闭异步客户端。

async add_runs_to_annotation_queue(queue_id: UUID | str, *, run_ids: List[UUID | str]) → None[source]#

将运行添加到具有指定队列 ID 的标注队列。

参数:

queue_id (Union[UUID, str]) – 标注队列的 ID。
run_ids (List[Union[UUID, str]]) – 要添加到标注队列的运行的 ID 列表。

返回值:

None

返回类型:

None

async create_annotation_queue(*, name: str, description: str | None = None, queue_id: UUID | str | None = None) → AnnotationQueue[source]#

在 LangSmith API 上创建标注队列。

参数:

name (str) – 标注队列的名称。
description (Optional[str]) – 标注队列的描述。
queue_id (Optional[Union[UUID, str]]) – 标注队列的 ID。

返回值:

创建的标注队列对象。

返回类型:

AnnotationQueue

async create_dataset(dataset_name: str, **kwargs: Any) → Dataset[source]#

创建数据集。

参数:

dataset_name (str)
kwargs (Any)

返回类型:

Dataset

async create_example(inputs: Dict[str, Any], outputs: Dict[str, Any] | None = None, dataset_id: UUID | str | None = None, dataset_name: str | None = None, **kwargs: Any) → Example[source]#

创建示例。

参数:

inputs (Dict[str, Any])
outputs (Dict[str, Any] | None)
dataset_id (UUID | str | None)
dataset_name (str | None)
kwargs (Any)

返回类型:

Example

为运行创建反馈。

参数:

run_id (Optional[ls_client.ID_TYPE]) – 要为其提供反馈的运行的 ID。对于项目级别的反馈可以为 None。
key (str) – 此反馈所关于的指标或方面的名称。
score (Optional[float]) – 对此运行在指标或方面的评分。
value (Optional[Any]) – 此反馈的显示值或非数值。
comment (Optional[str]) – 关于此反馈的评论。
**kwargs – 包含在反馈数据中的其他关键字参数。

返回值:

创建的反馈对象。

返回类型:

ls_schemas.Feedback

Raises:

httpx.HTTPStatusError – 如果 API 请求失败。

从预签名令牌或 URL 创建反馈。

参数:

token_or_url (Union[str, uuid.UUID]) – 从中创建反馈的令牌或 URL。
score (Union[float, int, bool, None], optional) – 反馈的评分。默认为 None。
value (Union[float, int, bool, str, dict, None], optional) – 反馈的值。默认为 None。
correction (Union[dict, None], optional) – 反馈的更正。默认为 None。
comment (Union[str, None], optional) – 反馈的评论。默认为 None。
metadata (可选[dict], 可选) – 反馈的附加元数据。默认为 None。

Raises:

ValueError – 如果源 API URL 无效。

返回值:

此方法不返回任何内容。

返回类型:

None

创建用于发送反馈数据的预签名 URL。

这对于为基于浏览器的客户端提供一种将反馈数据直接上传到 LangSmith 的方法非常有用，而无需访问 API 密钥。

参数:

run_id (UUID | str)
feedback_key (str)
expiration (datetime | timedelta | None) – 预签名 URL 的过期时间。可以是 datetime 对象或从现在开始的 timedelta 偏移量。默认为 3 小时。
feedback_config (FeedbackConfig | None) – FeedbackConfig 或 None。如果首次创建 feedback_key，则此参数定义应如何解释指标，例如连续分数（带有可选边界）或分类值的分布。
feedback_id (UUID | str | None) – 要创建的反馈的 ID。如果未提供，则将创建新的反馈。

返回值:

用于上传反馈数据的预签名 URL。

返回类型:

FeedbackIngestToken

async create_project(project_name: str, **kwargs: Any) → TracerSession[source]#

创建项目。

参数:

project_name (str)
kwargs (Any)

返回类型:

TracerSession

async create_run(name: str, inputs: Dict[str, Any], run_type: str, *, project_name: str | None = None, revision_id: UUID | str | None = None, **kwargs: Any) → None[source]#

创建运行。

参数:

name (str)
inputs (Dict[str, Any])
run_type (str)
project_name (str | None)
revision_id (UUID | str | None)
kwargs (Any)

返回类型:

None

async delete_annotation_queue(queue_id: UUID | str) → None[source]#

删除具有指定队列 ID 的标注队列。

参数:: queue_id (Union[UUID, str]) – 要删除的注释队列的 ID。
返回值:: None
返回类型:: None

async delete_dataset(dataset_id: UUID | str) → None[source]#

删除数据集。

参数:: dataset_id (UUID | str)
返回类型:: None

async delete_feedback(feedback_id: UUID | str) → None[source]#

按 ID 删除反馈。

参数:: feedback_id (Union[UUID, str]) – 要删除的反馈的 ID。
返回值:: None
返回类型:: None

async delete_project(*, project_name: str | None = None, project_id: str | None = None) → None[source]#

从 LangSmith 删除项目。

参数:

project_name (str or None, default=None) – 要删除的项目的名称。
project_id (str or None, default=None) – 要删除的项目的 ID。

返回类型:

None

async delete_run_from_annotation_queue(queue_id: UUID | str, *, run_id: UUID | str) → None[source]#

从具有指定队列 ID 和运行 ID 的标注队列中删除运行。

参数:

queue_id (Union[UUID, str]) – 标注队列的 ID。
run_id (Union[UUID, str]) – 要从注释队列中删除的运行的 ID。

返回值:

None

返回类型:

None

async get_run_from_annotation_queue(queue_id: UUID | str, *, index: int) → RunWithAnnotationQueueInfo[source]#

从指定索引的标注队列中获取运行。

参数:

queue_id (Union[UUID, str]) – 标注队列的 ID。
index (int) – 要检索的运行的索引。

返回值:

指定索引处的运行。

返回类型:

RunWithAnnotationQueueInfo

Raises:

LangSmithNotFoundError – 如果在给定索引处未找到运行。
LangSmithError – 对于其他 API 相关错误。

index_dataset(*, dataset_id: UUID | str, tag: str = 'latest', **kwargs: Any) → None[source]#

启用数据集索引。示例按其输入进行索引。

这使得可以使用 client.similar_examples() 通过输入搜索相似的示例。

参数:

dataset_id (UUID) – 要索引的数据集的 ID。
tag (str, optional) – 要索引的数据集的版本。如果为 “latest”，则对数据集的任何更新（添加、更新、删除示例）都将反映在索引中。
kwargs (Any)

返回值:

None

Raises:

requests.HTTPError –

返回类型:

None

列出 LangSmith API 上的标注队列。

参数:

queue_ids (Optional[List[Union[UUID, str]]]) – 要按 ID 筛选的队列的 ID 列表（可选）。
name (Optional[str]) – 要按名称筛选的队列的名称（可选）。
name_contains (Optional[str]) – 队列名称应包含的子字符串（可选）。
limit (Optional[int]) – 要返回的最大队列数（可选）。

Yields:

注释队列。

返回类型:

AsyncIterator[AnnotationQueue]

async list_datasets(**kwargs: Any) → AsyncIterator[Dataset][source]#

列出数据集。

参数:: kwargs (Any)
返回类型:: AsyncIterator[Dataset]

async list_examples(*, dataset_id: UUID | str | None = None, dataset_name: str | None = None, **kwargs: Any) → AsyncIterator[Example][source]#

列出示例。

参数:

dataset_id (UUID | str | None)
dataset_name (str | None)
kwargs (Any)

返回类型:

AsyncIterator[Example]

async list_feedback(*, run_ids: Sequence[UUID | str] | None = None, feedback_key: Sequence[str] | None = None, feedback_source_type: Sequence[FeedbackSourceType] | None = None, limit: int | None = None, **kwargs: Any) → AsyncIterator[Feedback][source]#

列出反馈。

参数:

run_ids (Sequence[UUID | str] | None)
feedback_key (Sequence[str] | None)
feedback_source_type (Sequence[FeedbackSourceType] | None)
limit (int | None)
kwargs (Any)

返回类型:

AsyncIterator[Feedback]

列出 LangSmith API 中的运行。

参数:

project_id (UUID 或 None, default=None) – 要按项目 ID 筛选的项目 ID（或多个 ID）。
project_name (str 或 None, default=None) – 要按项目名称筛选的项目名称（或多个名称）。
run_type (str 或 None, default=None) – 要按运行类型筛选的运行类型。
trace_id (UUID 或 None, default=None) – 要按跟踪 ID 筛选的跟踪 ID。
reference_example_id (UUID 或 None, default=None) – 要按参考示例 ID 筛选的参考示例 ID。
query (str 或 None, default=None) – 要按查询字符串筛选的查询字符串。
filter (str 或 None, default=None) – 要按筛选字符串筛选的筛选字符串。
trace_filter (str 或 None, default=None) – 应用于跟踪树中 ROOT 运行的筛选器。旨在与常规 filter 参数结合使用，以便您可以通过跟踪中根运行的属性来筛选运行。
tree_filter (str 或 None, default=None) – 应用于跟踪树中其他运行（包括同级和子运行）的筛选器。旨在与常规 filter 参数结合使用，以便您可以通过跟踪中任何运行的属性来筛选运行。
is_root (bool 或 None, default=None) – 是否按根运行筛选。
parent_run_id (UUID 或 None, default=None) – 要按父运行 ID 筛选的父运行 ID。
start_time (datetime 或 None, default=None) – 要按开始时间筛选的开始时间。
error (bool 或 None, default=None) – 是否按错误状态筛选。
run_ids (List[str 或 UUID] 或 None, default=None) – 要按运行 ID 筛选的运行 ID（或多个 ID）。
limit (int 或 None, default=None) – 要返回的最大运行数。
**kwargs (Any) – 额外的关键字参数。
Yields
------
Run – 运行。
Examples
--------
project (列出项目中的根跟踪)
code-block: (..) – python: project_runs = client.list_runs(project_name=”<your_project>”)
hours (列出过去 24 小时内的 LLM 和 Chat 运行)
code-block: –
python: todays_llm_runs = client.list_runs(

project_name=”<your_project>”, start_time=datetime.now() - timedelta(days=1), run_type=”llm”,

)
project
code-block: – python: root_runs = client.list_runs(project_name=”<your_project>”, is_root=1)
errors (列出没有错误的运行)
code-block: – python: correct_runs = client.list_runs(project_name=”<your_project>”, error=False)
query) (列出运行并仅返回其输入/输出 (以加速查询))
code-block: –
python: input_output_runs = client.list_runs(

project_name=”<your_project>”, select=[“inputs”, “outputs”]

)
ID (按运行 ID 列出运行)
code-block: –
python: run_ids = [

”a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836”, “9398e6be-964f-4aa4-8ae9-ad78cd4b7074”,

] selected_runs = client.list_runs(id=run_ids)
had (列出所有运行时间超过 10 秒且总 token 数大于 5000 的 “chain” 类型运行)
5000 (total_tokens greater than)
code-block: –
python: chain_runs = client.list_runs(

project_name=”<your_project>”, filter=’and(eq(run_type, “chain”), gt(latency, 10), gt(total_tokens, 5000))’,

)
1 (列出所有名为 “extractor” 的运行，其跟踪的根运行被分配了反馈 “user_score” 分数为 1)
code-block: –
python: good_extractor_runs = client.list_runs(

project_name=”<your_project>”, filter=’eq(name, “extractor”)’, trace_filter=’and(eq(feedback_key, “user_score”), eq(feedback_score, 1))’,

)
0 (列出所有在特定时间戳之后开始，并且 “error” 不等于 null 或 “Correctness” 反馈分数为 0 的运行)
code-block: –
python: complex_runs = client.list_runs(

project_name=”<your_project>”, filter=’and(gt(start_time, “2023-07-15T12:34:56Z”), or(neq(error, null), and(eq(feedback_key, “Correctness”), eq(feedback_score, 0.0))))’,

)
seconds (列出所有标签包括 “experimental” 或 “beta” 且延迟大于 2 秒的运行)
code-block: –
python: tagged_runs = client.list_runs(

project_name=”<your_project>”, filter=’and(or(has(tags, “experimental”), has(tags, “beta”)), gt(latency, 2))’,

)
select (Sequence[str] | None)

返回类型:

AsyncIterator[Run]

async read_annotation_queue(queue_id: UUID | str) → AnnotationQueue[source]#

读取具有指定队列 ID 的标注队列。

参数:: queue_id (Union[UUID, str]) – 要读取的注释队列的 ID。
返回值:: 注释队列对象。
返回类型:: AnnotationQueue

async read_dataset(dataset_name: str | None _= None, dataset_id: UUID | str | None _= None) → Dataset[source]#

读取数据集。

参数:

dataset_name (str | None)
dataset_id (UUID | str | None)

返回类型:

Dataset

async read_example(example_id: UUID | str) → Example[source]#

读取示例。

参数:: example_id (UUID | str)
返回类型:: Example

async read_feedback(feedback_id: UUID | str) → Feedback[source]#

读取反馈。

参数:: feedback_id (UUID | str)
返回类型:: 反馈

async read_project(project_name: str | None _= None, project_id: UUID | str | None _= None) → TracerSession[source]#

读取项目。

参数:

project_name (str | None)
project_id (UUID | str | None)

返回类型:

TracerSession

async read_run(run_id: UUID | str) → Run[source]#

读取运行。

参数:: run_id (UUID | str)
返回类型:: 运行

async read_run_shared_link(run_id: UUID | str) → str | None[source]#

异步检索特定运行的共享链接。

参数:: run_id (ID_TYPE) – 运行的 ID。
返回值:: 运行的共享链接，如果链接不可用，则为 None。
返回类型:: Optional[str]
Raises:: httpx.HTTPStatusError – 如果 API 请求失败。

async run_is_shared(run_id: UUID | str) → bool[source]#

异步获取运行的共享状态。

参数:: run_id (UUID | str)
返回类型:: bool

async share_run(run_id: UUID | str, *, share_id: UUID | str | None _= None) → str[source]#

异步获取运行的共享链接。

参数:

run_id (ID_TYPE) – 要共享的运行的 ID。
share_id (Optional[ID_TYPE], optional) – 自定义共享 ID。如果未提供，将生成一个随机 UUID。

返回值:

共享运行的 URL。

返回类型:

str

Raises:

httpx.HTTPStatusError – 如果 API 请求失败。

similar_examples(inputs: dict, /, *, limit: int, dataset_id: UUID | str, filter: str | None _= None, **kwargs: Any) → List[ExampleSearch][source]#

检索输入最匹配当前输入的数据集示例。

注意：数据集必须启用少样本索引。请参阅 client.index_dataset()。

参数:

inputs (dict) – 用作搜索查询的输入。必须与数据集输入模式匹配。必须是 JSON 可序列化的。
limit (int) – 要返回的最大示例数。
dataset_id (str 或 UUID) – 要在其上搜索的数据集的 ID。
filter (str, optional) – 应用于搜索结果的筛选字符串。使用与 list_runs() 中的 filter 参数相同的语法。仅支持部分操作。默认为 None。
kwargs (Any) – 作为请求正文一部分传递的额外关键字参数。

返回值:

ExampleSearch 对象列表。

返回类型:

List[ExampleSearch]

Example

from langsmith import Client

client = Client()
await client.similar_examples(
    {"question": "When would i use the runnable generator"},
    limit=3,
    dataset_id="...",
)

[
    ExampleSearch(
        inputs={'question': 'How do I cache a Chat model? What caches can I use?'},
        outputs={'answer': 'You can use LangChain\'s caching layer for Chat Models. This can save you money by reducing the number of API calls you make to the LLM provider, if you\'re often requesting the same completion multiple times, and speed up your application.\n\n```python\n\nfrom langchain.cache import InMemoryCache\nlangchain.llm_cache = InMemoryCache()\n\n# The first time, it is not yet in cache, so it should take longer\nllm.predict(\'Tell me a joke\')\n\n```\n\nYou can also use SQLite Cache which uses a SQLite database:\n\n```python\n  rm .langchain.db\n\nfrom langchain.cache import SQLiteCache\nlangchain.llm_cache = SQLiteCache(database_path=".langchain.db")\n\n# The first time, it is not yet in cache, so it should take longer\nllm.predict(\'Tell me a joke\') \n```\n'},
        metadata=None,
        id=UUID('b2ddd1c4-dff6-49ae-8544-f48e39053398'),
        dataset_id=UUID('01b6ce0f-bfb6-4f48-bbb8-f19272135d40')
    ),
    ExampleSearch(
        inputs={'question': "What's a runnable lambda?"},
        outputs={'answer': "A runnable lambda is an object that implements LangChain's `Runnable` interface and runs a callbale (i.e., a function). Note the function must accept a single argument."},
        metadata=None,
        id=UUID('f94104a7-2434-4ba7-8293-6a283f4860b4'),
        dataset_id=UUID('01b6ce0f-bfb6-4f48-bbb8-f19272135d40')
    ),
    ExampleSearch(
        inputs={'question': 'Show me how to use RecursiveURLLoader'},
        outputs={'answer': 'The RecursiveURLLoader comes from the langchain.document_loaders.recursive_url_loader module. Here\'s an example of how to use it:\n\n```python\nfrom langchain.document_loaders.recursive_url_loader import RecursiveUrlLoader\n\n# Create an instance of RecursiveUrlLoader with the URL you want to load\nloader = RecursiveUrlLoader(url="https://example.com")\n\n# Load all child links from the URL page\nchild_links = loader.load()\n\n# Print the child links\nfor link in child_links:\n    print(link)\n```\n\nMake sure to replace "https://example.com" with the actual URL you want to load. The load() method returns a list of child links found on the URL page. You can iterate over this list to access each child link.'},
        metadata=None,
        id=UUID('0308ea70-a803-4181-a37d-39e95f138f8c'),
        dataset_id=UUID('01b6ce0f-bfb6-4f48-bbb8-f19272135d40')
    ),
]

async update_annotation_queue(queue_id: UUID | str, *, name: str, description: str | None _= None) → None[source]#

更新具有指定 queue_id 的标注队列。

参数:

queue_id (Union[UUID, str]) – 要更新的注释队列的 ID。
name (str) – 注释队列的新名称。
description (Optional[str]) – 注释队列的新描述。默认为 None。

返回值:

None

返回类型:

None

async update_run(run_id: UUID | str, **kwargs: Any) → None[source]#

更新运行。

参数:

run_id (UUID | str)
kwargs (Any)

返回类型:

None