客户端#

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

初始化 Client 实例。

参数:

api_url (Optional[str]) – LangSmith API 的 URL。如果未设置，默认为 LANGCHAIN_ENDPOINT 环境变量或 https://api.smith.langchain.com。
api_key (Optional[str]) – LangSmith API 的 API 密钥。默认为 LANGCHAIN_API_KEY 环境变量。
retry_config (Optional[Retry]) – HTTPAdapter 的重试配置。
timeout_ms (Optional[Union[int, Tuple[int, int]]]) – HTTPAdapter 的超时时间。也可以是 (连接超时, 读取超时) 的二元组，用于分别设置。
web_url (Optional[str]) – LangSmith Web 应用的 URL。默认为从 ENDPOINT 自动推断。
session (Optional[requests.Session]) – 用于请求的会话。如果为 None，将创建一个新会话。
auto_batch_tracing (bool, default=True) – 是否自动批量跟踪。
anonymizer (Optional[Callable[[dict], dict]]) – 在发送到 API 之前，用于遮蔽序列化后的运行输入和输出的函数。
hide_inputs (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行输入。如果为 True，则隐藏整个输入。如果为函数，则在创建运行时代入所有运行输入。
hide_outputs (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行输出。如果为 True，则隐藏整个输出。如果为函数，则在创建运行时代入所有运行输出。
hide_metadata (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行元数据。如果为 True，则隐藏整个元数据。如果为函数，则在创建运行时代入所有运行元数据。
info (Optional[ls_schemas.LangSmithInfo]) – 关于 LangSmith API 的信息。如果未提供，将从 API 获取。
api_urls (Optional[Dict[str, str]]) – 写入 API URL 及其对应 API 密钥的字典。适用于多租户设置。数据仅从字典中的第一个 URL 读取。但是，只有运行（Runs）会被写入（POST 和 PATCH）到字典中的所有 URL。反馈、会话、数据集、示例、标注队列和评估结果仅写入第一个 URL。
otel_tracer_provider (Optional[TracerProvider]) – OpenTelemetry 集成的可选跟踪器提供程序。如果未提供，将使用 LangSmith 特定的跟踪器提供程序。
tracing_sampling_rate (Optional[float]) – 跟踪的采样率。如果提供，将覆盖 LANGCHAIN_TRACING_SAMPLING_RATE 环境变量。应为 0 到 1 之间的浮点数，其中 1 表示跟踪所有内容，0 表示不跟踪任何内容。

抛出:

LangSmithUserError – 如果在使用托管服务时未提供 API 密钥。如果同时提供了 api_url 和 api_urls。

属性

`api_url`
`retry_config`
`timeout_ms`
`session`
`tracing_sample_rate`
`tracing_queue`
`compressed_traces`
`otel_exporter`
`api_key`	返回用于身份验证的 API 密钥。
`info`	获取 LangSmith API 的信息。

方法

`__init__`([api_url, api_key, retry_config, ...])	初始化 Client 实例。
`add_runs_to_annotation_queue`(queue_id, *, ...)	将运行添加到具有指定队列 ID 的标注队列。
`aevaluate`(target, /[, data, evaluators, ...])	在给定数据集上评估异步目标系统。
`aevaluate_run`(run, evaluator, *[, ...])	异步评估运行。
`arun_on_dataset`(dataset_name, ...[, ...])	在数据集上异步运行链或语言模型。
`batch_ingest_runs`([create, update, pre_sampled])	在 Langsmith 系统中批量摄取/更新（upsert）多个运行。
`cleanup`()	手动触发后台线程的清理。
`clone_public_dataset`(token_or_url, *[, ...])	将公共数据集克隆到您的 Langsmith 租户。
`create_annotation_queue`(*, name[, ...])	在 LangSmith API 上创建标注队列。
`create_chat_example`(messages[, generations, ...])	将一个示例（行）添加到聊天类型数据集。
`create_commit`(prompt_identifier, object, *)	为现有提示创建提交。
`create_comparative_experiment`(name, ...[, ...])	在 LangSmith API 上创建比较实验。
`create_dataset`(dataset_name, *[, ...])	在 LangSmith API 中创建数据集。
`create_example`([inputs, dataset_id, ...])	在 LangSmith API 中创建数据集示例。
`create_example_from_run`(run[, dataset_id, ...])	从运行中将一个示例（行）添加到数据集。
`create_examples`(*[, dataset_name, ...])	在数据集中创建示例。
`create_feedback`([run_id, key, score, value, ...])	为运行创建反馈。
`create_feedback_from_token`(token_or_url[, ...])	从预签名令牌或 URL 创建反馈。
`create_llm_example`(prompt[, generation, ...])	将一个示例（行）添加到 LLM 类型数据集。
`create_presigned_feedback_token`(run_id, ...)	创建一个预签名 URL 以发送反馈数据。
`create_presigned_feedback_tokens`(run_id, ...)	创建一个预签名 URL 以发送反馈数据。
`create_project`(project_name, *[, ...])	在 LangSmith API 上创建项目。
`create_prompt`(prompt_identifier, *[, ...])	创建新提示。
`create_run`(name, inputs, run_type, *[, ...])	将运行持久化到 LangSmith API。
`delete_annotation_queue`(queue_id)	删除具有指定队列 ID 的标注队列。
`delete_dataset`(*[, dataset_id, dataset_name])	从 LangSmith API 中删除数据集。
`delete_example`(example_id)	按 ID 删除示例。
`delete_examples`(example_ids)	按 ID 删除多个示例。
`delete_feedback`(feedback_id)	按 ID 删除反馈。
`delete_project`(*[, project_name, project_id])	从 LangSmith 删除项目。
`delete_prompt`(prompt_identifier)	删除提示。
`delete_run_from_annotation_queue`(queue_id, ...)	从具有指定队列 ID 和运行 ID 的标注队列中删除运行。
`diff_dataset_versions`([dataset_id, dataset_name])	获取数据集两个版本之间的差异。
`evaluate`(target, /[, data, evaluators, ...])	在给定数据集上评估目标系统。
`evaluate_run`(run, evaluator, *[, ...])	评估一个运行。
`flush`()	根据模式刷新队列或压缩缓冲区。
`flush_compressed_traces`([attempts])	强制刷新当前已缓冲的压缩运行。
`get_prompt`(prompt_identifier)	根据其标识符获取特定提示。
`get_run_from_annotation_queue`(queue_id, *, index)	从指定索引的标注队列中获取运行。
`get_run_stats`(*[, id, trace, parent_run, ...])	获取查询运行的聚合统计信息。
`get_run_url`(*, run[, project_name, project_id])	获取运行的URL。
`get_test_results`(*[, project_id, project_name])	从实验中读取记录级别的信息到Pandas DF中。
`has_dataset`(*[, dataset_name, dataset_id])	检查租户中是否存在数据集。
`has_project`(project_name, *[, project_id])	检查项目是否存在。
`index_dataset`(*, dataset_id[, tag])	启用数据集索引。
`like_prompt`(prompt_identifier)	喜欢一个提示。
`list_annotation_queues`(*[, queue_ids, name, ...])	列出LangSmith API上的标注队列。
`list_dataset_splits`(*[, dataset_id, ...])	获取数据集的分区。
`list_dataset_versions`(*[, dataset_id, ...])	列出数据集版本。
`list_datasets`(*[, dataset_ids, data_type, ...])	列出LangSmith API上的数据集。
`list_examples`([dataset_id, dataset_name, ...])	检索指定数据集的示例行。
`list_feedback`(*[, run_ids, feedback_key, ...])	列出LangSmith API上的反馈对象。
`list_presigned_feedback_tokens`(run_id, *[, ...])	列出运行的预签名反馈令牌。
`list_projects`([project_ids, name, ...])	从LangSmith API列出项目。
`list_prompt_commits`(prompt_identifier, *[, ...])	列出给定提示的提交。
`list_prompts`(*[, limit, offset, is_public, ...])	分页列出提示。
`list_runs`(*[, project_id, project_name, ...])	从LangSmith API列出运行。
`list_shared_examples`(share_token, *[, ...])	获取共享示例。
`list_shared_projects`(*, dataset_share_token)	列出共享项目。
`list_shared_runs`(share_token[, run_ids])	获取共享运行。
`multipart_ingest`([create, update, ...])	在 Langsmith 系统中批量摄取/更新（upsert）多个运行。
`pull_prompt`(prompt_identifier, *[, ...])	拉取提示并将其作为LangChain PromptTemplate返回。
`pull_prompt_commit`(prompt_identifier, *[, ...])	从LangSmith API拉取提示对象。
`push_prompt`(prompt_identifier, *[, object, ...])	将提示推送到LangSmith API。
`read_annotation_queue`(queue_id)	读取指定队列ID的标注队列。
`read_dataset`(*[, dataset_name, dataset_id])	从LangSmith API读取数据集。
`read_dataset_openai_finetuning`([dataset_id, ...])	下载OpenAI Jsonl格式的数据集并将其加载为字典列表。
`read_dataset_shared_schema`([dataset_id, ...])	检索数据集的共享模式。
`read_dataset_version`(*[, dataset_id, ...])	按as_of或确切标签获取数据集版本。
`read_example`(example_id, *[, as_of])	从LangSmith API读取示例。
`read_feedback`(feedback_id)	从LangSmith API读取反馈。
`read_project`(*[, project_id, project_name, ...])	从LangSmith API读取项目。
`read_run`(run_id[, load_child_runs])	从LangSmith API读取运行。
`read_run_shared_link`(run_id)	检索特定运行的共享链接。
`read_shared_dataset`(share_token)	获取共享数据集。
`read_shared_run`(share_token[, run_id])	获取共享运行。
`request_with_retries`(method, pathname, *[, ...])	发送带重试的请求。
`run_is_shared`(run_id)	获取运行的共享状态。
`run_on_dataset`(dataset_name, ...[, ...])	在数据集上运行链或语言模型。
`share_dataset`([dataset_id, dataset_name])	获取数据集的共享链接。
`share_run`(run_id, *[, share_id])	获取运行的共享链接。
`similar_examples`(inputs, /, *, limit, dataset_id)	检索输入与当前输入最匹配的数据集示例。
`sync_indexed_dataset`(, dataset_id, *kwargs)	同步数据集索引。
`unlike_prompt`(prompt_identifier)	取消喜欢一个提示。
`unshare_dataset`(dataset_id)	删除数据集的共享链接。
`unshare_run`(run_id)	删除运行的共享链接。
`update_annotation_queue`(queue_id, *, name[, ...])	更新具有指定queue_id的标注队列。
`update_dataset_splits`(*[, dataset_id, ...])	更新数据集的分区。
`update_dataset_tag`(*[, dataset_id, dataset_name])	更新数据集的标签。
`update_example`(example_id, *[, inputs, ...])	更新特定示例。
`update_examples`(*[, dataset_name, ...])	更新多个示例。
`update_examples_multipart`(*, dataset_id[, ...])	使用多部分更新示例。
`update_feedback`(feedback_id, *[, score, ...])	在LangSmith API中更新反馈。
`update_project`(project_id, *[, name, ...])	更新LangSmith项目。
`update_prompt`(prompt_identifier, *[, ...])	更新提示的元数据。
`update_run`(run_id, *[, name, end_time, ...])	在LangSmith API中更新运行。
`upload_csv`(csv_file, input_keys, output_keys, *)	上传CSV文件到LangSmith API。
`upload_dataframe`(df, name, input_keys, ...)	将数据帧作为单个示例上传到LangSmith API。
`upload_examples_multipart`(*, dataset_id[, ...])	使用多部分上传示例。
`upsert_examples_multipart`(*[, upserts, ...])	插入/更新示例。

初始化 Client 实例。

参数:

api_url (Optional[str]) – LangSmith API 的 URL。如果未设置，默认为 LANGCHAIN_ENDPOINT 环境变量或 https://api.smith.langchain.com。
api_key (Optional[str]) – LangSmith API 的 API 密钥。默认为 LANGCHAIN_API_KEY 环境变量。
retry_config (Optional[Retry]) – HTTPAdapter 的重试配置。
timeout_ms (Optional[Union[int, Tuple[int, int]]]) – HTTPAdapter 的超时时间。也可以是 (连接超时, 读取超时) 的二元组，用于分别设置。
web_url (Optional[str]) – LangSmith Web 应用的 URL。默认为从 ENDPOINT 自动推断。
session (Optional[requests.Session]) – 用于请求的会话。如果为 None，将创建一个新会话。
auto_batch_tracing (bool, default=True) – 是否自动批量跟踪。
anonymizer (Optional[Callable[[dict], dict]]) – 在发送到 API 之前，用于遮蔽序列化后的运行输入和输出的函数。
hide_inputs (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行输入。如果为 True，则隐藏整个输入。如果为函数，则在创建运行时代入所有运行输入。
hide_outputs (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行输出。如果为 True，则隐藏整个输出。如果为函数，则在创建运行时代入所有运行输出。
hide_metadata (Optional[Union[Callable[[dict], dict], bool]]) – 使用此客户端进行跟踪时是否隐藏运行元数据。如果为 True，则隐藏整个元数据。如果为函数，则在创建运行时代入所有运行元数据。
info (Optional[ls_schemas.LangSmithInfo]) – 关于 LangSmith API 的信息。如果未提供，将从 API 获取。
api_urls (Optional[Dict[str, str]]) – 写入 API URL 及其对应 API 密钥的字典。适用于多租户设置。数据仅从字典中的第一个 URL 读取。但是，只有运行（Runs）会被写入（POST 和 PATCH）到字典中的所有 URL。反馈、会话、数据集、示例、标注队列和评估结果仅写入第一个 URL。
otel_tracer_provider (Optional[TracerProvider]) – OpenTelemetry 集成的可选跟踪器提供程序。如果未提供，将使用 LangSmith 特定的跟踪器提供程序。
tracing_sampling_rate (Optional[float]) – 跟踪的采样率。如果提供，将覆盖 LANGCHAIN_TRACING_SAMPLING_RATE 环境变量。应为 0 到 1 之间的浮点数，其中 1 表示跟踪所有内容，0 表示不跟踪任何内容。

抛出:

LangSmithUserError – 如果在使用托管服务时未提供 API 密钥。如果同时提供了 api_url 和 api_urls。

返回类型:

None

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

在给定数据集上评估异步目标系统。

参数:

target (Union[ATARGET_T, AsyncIterable[dict], Runnable, str, uuid.UUID, TracerSession]) – 要评估的目标系统或实验。可以是接受字典并返回字典的异步函数、langchain Runnable、现有实验ID，或由两个实验ID组成的元组。
data (Union[DATA_T, AsyncIterable[Example]]) – 用于评估的数据集。可以是数据集名称、示例列表、异步示例生成器或异步示例可迭代对象。
evaluators (Optional[Sequence[EVALUATOR_T]]) – 对每个示例运行的评估器列表。默认为None。
summary_evaluators (Optional[Sequence[SUMMARY_EVALUATOR_T]]) – 对整个数据集运行的摘要评估器列表。默认为None。
metadata (Optional[dict]) – 要附加到实验的元数据。默认为None。
experiment_prefix (Optional[str]) – 为实验名称提供的前缀。默认为None。
description (Optional[str]) – 实验的描述。
max_concurrency (Optional[int], default=0) – 最大并发评估数量。如果为None则不设置限制。如果为0则不并发。默认为0。
num_repetitions (int, default=1) – 运行评估的次数。数据集中的每个项目都将运行并评估这么多次。默认为1。
blocking (bool, default=True) – 是否阻塞直到评估完成。默认为True。
experiment (Optional[TracerSession]) – 要扩展的现有实验。如果提供，experiment_prefix将被忽略。仅用于高级用法。
upload_results (bool, default=True) – 是否将结果上传到LangSmith。默认为True。
**kwargs (Any) – 传递给评估器的额外关键字参数。

返回:

实验结果的异步迭代器。

返回类型:

AsyncIterator[ExperimentResultRow]

环境

LANGSMITH_TEST_CACHE: 如果设置，API调用将被缓存到磁盘，以在测试期间节省时间和成本。建议将缓存文件提交到您的仓库，以便更快地进行CI/CD运行。需要安装“langsmith[vcr]”包。
cost during testing. Recommended to commit the cache files to your repository for faster CI/CD runs. Requires the ‘langsmith[vcr]’ package to be installed.

示例

准备数据集

import asyncio
from langsmith import Client

client = Client()
dataset = client.clone_public_dataset(
    "https://smith.langchain.com/public/419dcab2-1d66-4b94-8901-0357ead390df/d"
)
dataset_name = "Evaluate Examples"

基本用法

def accuracy(outputs: dict, reference_outputs: dict) -> dict:
    # Row-level evaluator for accuracy.
    pred = outputs["resposen"]
    expected = reference_outputs["answer"]
    return {"score": expected.lower() == pred.lower()}


def precision(outputs: list[dict], reference_outputs: list[dict]) -> dict:
    # Experiment-level evaluator for precision.
    # TP / (TP + FP)
    predictions = [out["response"].lower() for out in outputs]
    expected = [ref["answer"].lower() for ref in reference_outputs]
    # yes and no are the only possible answers
    tp = sum([p == e for p, e in zip(predictions, expected) if p == "yes"])
    fp = sum([p == "yes" and e == "no" for p, e in zip(predictions, expected)])
    return {"score": tp / (tp + fp)}


async def apredict(inputs: dict) -> dict:
    # This can be any async function or just an API call to your app.
    await asyncio.sleep(0.1)
    return {"response": "Yes"}


results = asyncio.run(
    client.aevaluate(
        apredict,
        data=dataset_name,
        evaluators=[accuracy],
        summary_evaluators=[precision],
        experiment_prefix="My Experiment",
        description="Evaluate the accuracy of the model asynchronously.",
        metadata={
            "my-prompt-version": "abcd-1234",
        },
    )
)

使用异步生成器仅评估部分示例

async def example_generator():
    examples = client.list_examples(dataset_name=dataset_name, limit=5)
    for example in examples:
        yield example


results = asyncio.run(
    client.aevaluate(
        apredict,
        data=example_generator(),
        evaluators=[accuracy],
        summary_evaluators=[precision],
        experiment_prefix="My Subset Experiment",
        description="Evaluate a subset of examples asynchronously.",
    )
)

流式传输每个预测以便更轻松、更快速地调试。

results = asyncio.run(
    client.aevaluate(
        apredict,
        data=dataset_name,
        evaluators=[accuracy],
        summary_evaluators=[precision],
        experiment_prefix="My Streaming Experiment",
        description="Streaming predictions for debugging.",
        blocking=False,
    )
)


async def aenumerate(iterable):
    async for elem in iterable:
        print(elem)


asyncio.run(aenumerate(results))

非并发运行

results = asyncio.run(
    client.aevaluate(
        apredict,
        data=dataset_name,
        evaluators=[accuracy],
        summary_evaluators=[precision],
        experiment_prefix="My Experiment Without Concurrency",
        description="This was run without concurrency.",
        max_concurrency=0,
    )
)

使用异步评估器

async def helpfulness(outputs: dict) -> dict:
    # Row-level evaluator for helpfulness.
    await asyncio.sleep(5)  # Replace with your LLM API call
    return {"score": outputs["output"] == "Yes"}


results = asyncio.run(
    client.aevaluate(
        apredict,
        data=dataset_name,
        evaluators=[helpfulness],
        summary_evaluators=[precision],
        experiment_prefix="My Helpful Experiment",
        description="Applying async evaluators example.",
    )
)

评估现有实验

results = asyncio.run(
    client.aevaluate(
        # The target is the ID of the experiment we are evaluating
        target="419dcab2-1d66-4b94-8901-0357ead390df",
        evaluators=[accuracy, helpfulness],
        summary_evaluators=[precision],
    )
)

版本0.2.0中新增。

异步评估运行。

参数:

run (Union[Run, str, UUID]) – 要评估的运行。
evaluator (RunEvaluator) – 要使用的评估器。
source_info (Optional[Dict[str, Any]]) – 有关评估来源的附加信息，作为反馈元数据进行记录。
reference_example (Optional[Union[Example, str, dict, UUID]]) – 用作评估参考的示例。如果未提供，将使用运行的参考示例。
load_child_runs (bool, default=False) – 解析运行ID时是否加载子运行。

返回:

评估创建的评估结果对象。

返回类型:

EvaluationResult

async arun_on_dataset(

dataset_name: str,

llm_or_chain_factory: Any,

*,

evaluation: Any | None = None,

concurrency_level: int = 5,

project_name: str | None = None,

project_metadata: dict[str, Any] | None = None,

dataset_version: datetime | str | None = None,

verbose: bool = False,

input_mapper: Callable[[dict], Any] | None = None,

revision_id: str | None = None,

**kwargs: Any,

) → dict[str, Any][source]#

在数据集上异步运行链或语言模型。

自版本0.1.0起已弃用: 此方法已弃用。请改用langsmith.aevaluate()。

参数:

dataset_name (str)
llm_or_chain_factory (Any)
evaluation (Any | None)
concurrency_level (int)
project_name (str | None)
project_metadata (dict[str, Any] | None)
dataset_version (datetime | str | None)
verbose (bool)
input_mapper (Callable[[dict], Any] | None)
revision_id (str | None)
kwargs (Any)

返回类型:

dict[str, Any]

在 Langsmith 系统中批量摄取/更新（upsert）多个运行。

参数:

create (Optional[Sequence[Union[Run, RunLikeDict]]]) – 表示要创建/发布的运行的Run对象序列或等效字典。
update (Optional[Sequence[Union[Run, RunLikeDict]]]) – 表示已创建并应更新/修补的运行的Run对象序列或等效字典。
pre_sampled (bool, default=False) – 运行是否已进行采样，因此不应再次采样。默认为False。

抛出:

LangsmithAPIError – 如果API请求中发生错误。

返回:

None

返回类型:

None

注意

运行对象必须包含dotted_order和trace_id字段才能被API接受。
to be accepted by the API.

示例

from langsmith import Client
import datetime
from uuid import uuid4

client = Client()
_session = "__test_batch_ingest_runs"
trace_id = uuid4()
trace_id_2 = uuid4()
run_id_2 = uuid4()
current_time = datetime.datetime.now(datetime.timezone.utc).strftime(
    "%Y%m%dT%H%M%S%fZ"
)
later_time = (
    datetime.datetime.now(datetime.timezone.utc) + timedelta(seconds=1)
).strftime("%Y%m%dT%H%M%S%fZ")

runs_to_create = [
    {
        "id": str(trace_id),
        "session_name": _session,
        "name": "run 1",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id)}",
        "trace_id": str(trace_id),
        "inputs": {"input1": 1, "input2": 2},
        "outputs": {"output1": 3, "output2": 4},
    },
    {
        "id": str(trace_id_2),
        "session_name": _session,
        "name": "run 3",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id_2)}",
        "trace_id": str(trace_id_2),
        "inputs": {"input1": 1, "input2": 2},
        "error": "error",
    },
    {
        "id": str(run_id_2),
        "session_name": _session,
        "name": "run 2",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id)}."
        f"{later_time}{str(run_id_2)}",
        "trace_id": str(trace_id),
        "parent_run_id": str(trace_id),
        "inputs": {"input1": 5, "input2": 6},
    },
]
runs_to_update = [
    {
        "id": str(run_id_2),
        "dotted_order": f"{current_time}{str(trace_id)}."
        f"{later_time}{str(run_id_2)}",
        "trace_id": str(trace_id),
        "parent_run_id": str(trace_id),
        "outputs": {"output1": 4, "output2": 5},
    },
]

client.batch_ingest_runs(create=runs_to_create, update=runs_to_update)

cleanup() → None[source]#

手动触发后台线程的清理。

返回类型:: None

clone_public_dataset( token_or_url: str, *, source_api_url: str | None = None, dataset_name: str | None = None, ) → Dataset[source]#

将公共数据集克隆到您的 Langsmith 租户。

此操作是幂等的。如果您已有同名数据集，此函数将不执行任何操作。

参数:

token_or_url (str) – 要克隆的公共数据集的令牌。
source_api_url (Optional[str]) – 数据托管的Langsmith服务器URL。默认为当前客户端的API URL。
dataset_name (Optional[str]) – 要在您的租户中创建的数据集名称。默认为公共数据集的名称。

返回:

克隆的数据集。

返回类型:

Dataset

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

在 LangSmith API 上创建标注队列。

参数:

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

返回:

创建的标注队列对象。

返回类型:

AnnotationQueue

将一个示例（行）添加到聊天类型数据集。

参数:

messages (List[Union[Mapping[str, Any], BaseMessageLike]]) – 示例的输入消息。
generations (Optional[Union[Mapping[str, Any], BaseMessageLike]]) – 示例的输出消息。
dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
created_at (Optional[datetime.datetime]) – 示例的创建时间戳。

返回:

创建的示例。

返回类型:

Example

create_commit( prompt_identifier: str, object: Any, *, parent_commit_hash: str | None = None, ) → str[source]#

为现有提示创建提交。

参数:

prompt_identifier (str) – 提示的标识符。
object (Any) – 要提交的LangChain对象。
parent_commit_hash (Optional[str]) – 父提交的哈希值。默认为最新提交。

返回:

提示提交的URL。

返回类型:

str

抛出:

HTTPError – 如果服务器请求失败。
ValueError – 如果提示不存在。

在 LangSmith API 上创建比较实验。

这些实验比较了两个或更多在共享数据集上的实验结果。

参数:

name (str) – 比较实验的名称。
experiments (Sequence[Union[UUID, str]]) – 要比较的实验ID。
reference_dataset (Optional[Union[UUID, str]]) – 这些实验所比较的数据集ID。
description (Optional[str]) – 比较实验的描述。
created_at (Optional[datetime.datetime]) – 比较实验的创建时间。
metadata (Optional[Dict[str, Any]]) – 比较实验的附加元数据。
id (Optional[Union[UUID, str]]) – 比较实验的ID。

返回:

创建的比较实验对象。

返回类型:

ComparativeExperiment

create_dataset( dataset_name: str, *, description: str | None = None, data_type: DataType = DataType.kv, inputs_schema: dict[str, Any] | None = None, outputs_schema: dict[str, Any] | None = None, transformations: list[DatasetTransformation] | None = None, metadata: dict | None = None, ) → Dataset[source]#

在 LangSmith API 中创建数据集。

参数:

dataset_name (str) – 数据集的名称。
description (Optional[str]) – 数据集的描述。
data_type (DataType, default=DataType.kv) – 数据集的类型。
inputs_schema (Optional[Dict[str, Any]]) – 数据集输入的模式定义。
outputs_schema (Optional[Dict[str, Any]]) – 数据集输出的模式定义。
transformations (Optional[List[DatasetTransformation]]) – 要应用于数据集的转换列表。
metadata (Optional[dict]) – 要与数据集关联的附加元数据。

返回:

创建的数据集。

返回类型:

Dataset

抛出:

requests.HTTPError – 如果创建数据集的请求失败。

在 LangSmith API 中创建数据集示例。

示例是数据集中的行，包含模型或链的输入和预期输出（或其他参考信息）。

参数:

inputs (Mapping[str, Any]) – 示例的输入值。
dataset_id (Optional[Union[UUID, str]]) – 创建示例的数据集ID。
dataset_name (Optional[str]) – 创建示例的数据集名称。
created_at (Optional[datetime.datetime]) – 示例的创建时间戳。
outputs (Optional[Mapping[str, Any]]) – 示例的输出值。
metadata (Optional[Mapping[str, Any]]) – 示例的元数据。
split (Optional[str | List[str]]) – 示例的分区，它们是数据集的划分，例如“train”、“test”或“validation”。
example_id (Optional[Union[UUID, str]]) – 要创建的示例的ID。如果未提供，将创建一个新示例。
source_run_id (Optional[Union[UUID, str]]) – 与此示例关联的源运行的ID。
use_source_run_io (bool) – 是否使用源运行的输入、输出和附件。
use_source_run_attachments (Optional[List[str]]) – 要从源运行中使用的附件。如果use_source_run_io为True，则无论此参数如何，都将使用所有附件。
attachments (Optional[Attachments]) – 示例的附件。

返回:

创建的示例。

返回类型:

Example

create_example_from_run( run: Run, dataset_id: UUID | str | None = None, dataset_name: str | None = None, created_at: datetime | None = None, ) → Example[source]#

从运行中将一个示例（行）添加到数据集。

参数:

运行 (Run) – 用于创建示例的运行。
dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
created_at (Optional[datetime.datetime]) – 示例的创建时间戳。

返回:

创建的示例。

返回类型:

Example

在数据集中创建示例。

参数:

dataset_name (str 或 None) – 用于创建示例的数据集名称。必须精确指定 dataset_name 或 dataset_id 其中之一。
dataset_id (UUID 或 str 或 None) – 用于创建示例的数据集 ID。必须精确指定 dataset_name 或 dataset_id 其中之一
examples (Sequence[ExampleCreate 或 dict]) – 要创建的示例。
dangerously_allow_filesystem (bool) – 是否允许从文件系统上传文件。
**kwargs (Any) –
旧版关键字参数。如果已指定“examples”，则不应再指定此项。
- inputs (Sequence[Mapping[str, Any]]): 示例的输入值。
- outputs (Optional[Sequence[Optional[Mapping[str, Any]]]]): 示例的输出值。
- metadata (Optional[Sequence[Optional[Mapping[str, Any]]]]): 示例的元数据。
- splits (Optional[Sequence[Optional[str 或 List[str]]]]): 示例的拆分，是数据集的划分，例如“train”、“test”或“validation”。
- source_run_ids (Optional[Sequence[Optional[Union[UUID, str]]]]): 与示例关联的源运行 ID。
- ids (Optional[Sequence[Union[UUID, str]]]): 示例的 ID。

抛出:

ValueError – 如果同时提供了“examples”和旧版参数。

返回:

LangSmith JSON 响应。包括“count”和“example_ids”。

返回类型:

UpsertExamplesResponse | dict[str, Any]

0.3.11 版本中的变更：更新为接受参数“examples”，这是一个包含完整示例的单个列表，其中每个元素都是要创建的完整示例。应使用此参数，而不是旧版“inputs”、“outputs”等将每个示例属性分散到不同参数中的参数。

更新为支持创建带附件的示例。

Example

from langsmith import Client

client = Client()

dataset = client.create_dataset("agent-qa")

examples = [
    {
        "inputs": {"question": "what's an agent"},
        "outputs": {"answer": "an agent is..."},
        "metadata": {"difficulty": "easy"},
    },
    {
        "inputs": {
            "question": "can you explain the agent architecture in this diagram?"
        },
        "outputs": {"answer": "this diagram shows..."},
        "attachments": {"diagram": {"mime_type": "image/png", "data": b"..."}},
        "metadata": {"difficulty": "medium"},
    },
    # more examples...
]

response = client.create_examples(dataset_name="agent-qa", examples=examples)
# -> {"example_ids": [...

为运行创建反馈。

注意: 为了使反馈能够在后台批量上传，您必须指定 trace_id。我们强烈建议在对延迟敏感的环境中使用此功能。

参数:

key (str) – 反馈指标的名称。
score (Optional[Union[float, int, bool]]) – 用于对本次运行进行指标或方面评分的分数。
value (Optional[Union[float, int, bool, str, dict]]) – 此反馈的显示值或非数值。
run_id (Optional[Union[UUID, str]]) – 要提供反馈的运行 ID。必须指定 run_id、trace_id 或 project_id 中的至少一个。
trace_id (Optional[Union[UUID, str]]) – 要提供反馈的运行（由 run_id 指定）的跟踪 ID（即根父运行）。如果 run_id 和 trace_id 相同，则只需指定 trace_id。注意：trace_id 是反馈批量和后台摄取所必需的。
correction (Optional[dict]) – 此运行的正确参考真相。
comment (Optional[str]) – 关于此反馈的评论，例如分数的理由或 LLM 评判器的思维链轨迹。
source_info (Optional[Dict[str, Any]]) – 有关此反馈来源的信息。
feedback_source_type (Union[FeedbackSourceType, str]) – 反馈来源的类型，例如模型（用于模型生成的反馈）或 API。
source_run_id (Optional[Union[UUID, str]]) – 如果是“模型”类型，则为生成此反馈的运行 ID。
feedback_id (Optional[Union[UUID, str]]) – 要创建的反馈 ID。如果未提供，将生成一个随机 UUID。
feedback_config (Optional[FeedbackConfig]) – 指定如何解释带有此键的反馈的配置。示例包括连续（带最小/最大边界）、分类或自由形式。
stop_after_attempt (int, default=10) – 在放弃之前重试请求的次数。
project_id (Optional[Union[UUID, str]]) – 要提供反馈的项目（或实验）的 ID。这用于为实验创建摘要指标。如果指定了 project_id，则不能指定 run_id 或 trace_id，反之亦然。
comparative_experiment_id (Optional[Union[UUID, str]]) – 如果此反馈作为比较实验的一部分进行记录，则此项会将反馈与该实验关联起来。
feedback_group_id (Optional[Union[UUID, str]]) – 记录偏好、运行排名或其他比较反馈时，此项用于将反馈分组。
extra (Optional[Dict]) – 反馈的元数据。
**kwargs (Any) – 附加关键字参数。
error (bool 或 None)
**kwargs

返回:

创建的反馈对象。

返回类型:

Feedback

Example

from langsmith import trace, traceable, Client


@traceable
def foo(x):
    return {"y": x * 2}


@traceable
def bar(y):
    return {"z": y - 1}


client = Client()

inputs = {"x": 1}
with trace(name="foobar", inputs=inputs) as root_run:
    result = foo(**inputs)
    result = bar(**result)
    root_run.outputs = result
    trace_id = root_run.id
    child_runs = root_run.child_runs

# Provide feedback for a trace (a.k.a. a root run)
client.create_feedback(
    key="user_feedback",
    score=1,
    trace_id=trace_id,
)

# Provide feedback for a child run
foo_run_id = [run for run in child_runs if run.name == "foo"][0].id
client.create_feedback(
    key="correctness",
    score=0,
    run_id=foo_run_id,
    # trace_id= is optional but recommended to enable batched and backgrounded
    # feedback ingestion.
    trace_id=trace_id,
)

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

参数:

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

抛出:

ValueError – 如果源 API URL 无效。

返回:

None

返回类型:

None

将一个示例（行）添加到 LLM 类型数据集。

参数:

prompt (str) – 示例的输入提示。
generation (Optional[str]) – 示例的输出生成。
dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
created_at (Optional[datetime.datetime]) – 示例的创建时间戳。

返回:

创建的示例。

返回类型:

Example

创建一个预签名 URL 以发送反馈数据。

这对于让基于浏览器的客户端直接向 LangSmith 上传反馈数据而无需访问 API 密钥非常有用。

参数:

run_id (Union[UUID, str]) – 运行的 ID。
feedback_key (str) – 要创建的反馈键。
expiration (Optional[datetime.datetime 或 datetime.timedelta]) – 预签名 URL 的过期时间。可以是 datetime 对象，也可以是相对于当前时间的 timedelta 偏移量。默认为 3 小时。
feedback_config (Optional[FeedbackConfig]) – 如果是首次创建 feedback_key，此参数定义了应如何解释该指标，例如连续分数（带可选边界）或分类值的分布。
feedback_id (Optional[Union[UUID, str]) – 要创建的反馈 ID。如果未提供，将创建一个新的反馈。

返回:

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

返回类型:

FeedbackIngestToken

创建一个预签名 URL 以发送反馈数据。

这对于让基于浏览器的客户端直接向 LangSmith 上传反馈数据而无需访问 API 密钥非常有用。

参数:

run_id (Union[UUID, str]) – 运行的 ID。
feedback_keys (Sequence[str]) – 要创建的反馈键。
expiration (Optional[datetime.datetime 或 datetime.timedelta]) – 预签名 URL 的过期时间。可以是 datetime 对象，也可以是相对于当前时间的 timedelta 偏移量。默认为 3 小时。
feedback_configs (Optional[Sequence[Optional[FeedbackConfig]]]) – 如果是首次创建 feedback_key，此参数定义了应如何解释该指标，例如连续分数（带可选边界）或分类值的分布。

返回:

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

返回类型:

Sequence[FeedbackIngestToken]

在 LangSmith API 上创建项目。

参数:

project_name (str) – 项目的名称。
project_extra (Optional[dict]) – 附加项目信息。
metadata (Optional[dict]) – 与项目关联的附加元数据。
description (Optional[str]) – 项目的描述。
upsert (bool, default=False) – 如果项目已存在，是否更新。
reference_dataset_id (Optional[Union[UUID, str]) – 要与项目关联的参考数据集 ID。

返回:

创建的项目。

返回类型:

TracerSession

create_prompt( prompt_identifier: str, *, description: str | None = None, readme: str | None = None, tags: Sequence[str] | None = None, is_public: bool = False, ) → Prompt[source]#

创建新提示。

不附加提示对象，仅创建空提示。

参数:

prompt_identifier (str) – 提示的标识符。标识符应为 owner/name:hash、name:hash、owner/name 或 name 格式。
description (Optional[str]) – 提示的描述。
readme (Optional[str]) – 提示的自述文件。
tags (Optional[Sequence[str]]) – 提示的标签列表。
is_public (bool) – 提示是否应公开。默认为 False。

返回:

创建的提示对象。

返回类型:

Prompt

抛出:

ValueError – 如果当前租户不是所有者。
HTTPError – 如果服务器请求失败。

create_run(

name: str,

inputs: dict[str, Any],

run_type: Literal['tool', 'chain', 'llm', 'retriever', 'embedding', 'prompt', 'parser'],

*,

project_name: str | None = None,

revision_id: str | None = None,

dangerously_allow_filesystem: bool = False,

**kwargs: Any,

) → None[source]#

将运行持久化到 LangSmith API。

参数:

name (str) – 运行的名称。
inputs (Dict[str, Any]) – 运行的输入值。
run_type (str) – 运行的类型，例如 tool、chain、llm、retriever、embedding、prompt 或 parser。
project_name (Optional[str]) – 运行的项目名称。
revision_id (Optional[Union[UUID, str]]) – 运行的修订 ID。
**kwargs (Any) – 附加关键字参数。
dangerously_allow_filesystem (bool)
**kwargs

返回:

None

抛出:

LangSmithUserError – 如果在使用托管服务时未提供 API 密钥。

返回类型:

None

示例

from langsmith import Client
import datetime
from uuid import uuid4

client = Client()

run_id = uuid4()
client.create_run(
    id=run_id,
    project_name=project_name,
    name="test_run",
    run_type="llm",
    inputs={"prompt": "hello world"},
    outputs={"generation": "hi there"},
    start_time=datetime.datetime.now(datetime.timezone.utc),
    end_time=datetime.datetime.now(datetime.timezone.utc),
    hide_inputs=True,
    hide_outputs=True,
)

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

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

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

delete_dataset( *, dataset_id: UUID | str | None = None, dataset_name: str | None = None, ) → None[source]#

从 LangSmith API 中删除数据集。

参数:

dataset_id (Optional[Union[UUID, str]]) – 要删除的数据集 ID。
dataset_name (Optional[str]) – 要删除的数据集名称。

返回:

None

返回类型:

None

delete_example(example_id: UUID | str) → None[source]#

按 ID 删除示例。

参数:: example_id (Union[UUID, str]) – 要删除的示例 ID。
返回:: None
返回类型:: None

delete_examples( example_ids: Sequence[UUID | str], ) → None[source]#

按 ID 删除多个示例。

参数:: example_ids (Sequence[ID_TYPE]) – 要删除的示例 ID。
返回类型:: None

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

按 ID 删除反馈。

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

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

从 LangSmith 删除项目。

参数:

project_name (Optional[str]) – 要删除的项目名称。
project_id (Optional[str]) – 要删除的项目 ID。

返回:

None

抛出:

ValueError – 如果 project_name 和 project_id 均未提供。

返回类型:

None

delete_prompt(prompt_identifier: str) → None[source]#

删除提示。

参数:: prompt_identifier (str) – 要删除的提示的标识符。
返回:: 如果提示成功删除则为 True，否则为 False。
返回类型:: bool
抛出:: ValueError – 如果当前租户不是该提示的所有者。

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

获取数据集两个版本之间的差异。

参数:

dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
from_version (Union[str, datetime.datetime]) – 差异比较的起始版本。
to_version (Union[str, datetime.datetime]) – 差异比较的结束版本。

返回:

数据集两个版本之间的差异。

返回类型:

DatasetDiffInfo

示例

# Get the difference between two tagged versions of a dataset
from_version = "prod"
to_version = "dev"
diff = client.diff_dataset_versions(
    dataset_name="my-dataset",
    from_version=from_version,
    to_version=to_version,
)

# Get the difference between two timestamped versions of a dataset
from_version = datetime.datetime(2024, 1, 1)
to_version = datetime.datetime(2024, 2, 1)
diff = client.diff_dataset_versions(
    dataset_name="my-dataset",
    from_version=from_version,
    to_version=to_version,
)

evaluate(

target: TARGET_T | Runnable | EXPERIMENT_T | tuple[EXPERIMENT_T, EXPERIMENT_T],

/,

data: DATA_T | None = None,

evaluators: Sequence[EVALUATOR_T] | Sequence[COMPARATIVE_EVALUATOR_T] | None = None,

summary_evaluators: Sequence[SUMMARY_EVALUATOR_T] | None = None,

metadata: dict | None = None,

experiment_prefix: str | None = None,

description: str | None = None,

max_concurrency: int | None = 0,

num_repetitions: int = 1,

blocking: bool = True,

experiment: EXPERIMENT_T | None = None,

upload_results: bool = True,

**kwargs: Any,

) → ExperimentResults | ComparativeExperimentResults[source]#

在给定数据集上评估目标系统。

参数:

target (Union[TARGET_T, Runnable, EXPERIMENT_T, Tuple[EXPERIMENT_T, EXPERIMENT_T]]) – 要评估的目标系统或实验。可以是接受并返回 dict 的函数、langchain Runnable、现有实验 ID，或由两个实验 ID 组成的元组。
data (DATA_T) – 用于评估的数据集。可以是数据集名称、示例列表或示例生成器。
evaluators (Optional[Union[Sequence[EVALUATOR_T], Sequence[COMPARATIVE_EVALUATOR_T]]]) – 要对每个示例运行的评估器列表。评估器签名取决于目标类型。默认为 None。
summary_evaluators (Optional[Sequence[SUMMARY_EVALUATOR_T]]) – 要对整个数据集运行的摘要评估器列表。如果比较两个现有实验，则不应指定此项。默认为 None。
metadata (Optional[dict]) – 要附加到实验的元数据。默认为None。
experiment_prefix (Optional[str]) – 为实验名称提供的前缀。默认为None。
description (Optional[str]) – 实验的自由文本描述。
max_concurrency (Optional[int], default=0) – 最大并发评估数量。如果为None则不设置限制。如果为0则不并发。默认为0。
blocking (bool, default=True) – 是否阻塞直到评估完成。默认为True。
num_repetitions (int, default=1) – 运行评估的次数。数据集中的每个项目都将运行并评估这么多次。默认为1。
experiment (Optional[EXPERIMENT_T]) – 要扩展的现有实验。如果提供，则 experiment_prefix 将被忽略。仅限高级用法。如果 target 是现有实验或由两个实验组成的元组，则不应指定此项。
upload_results (bool, default=True) – 是否将结果上传到LangSmith。默认为True。
**kwargs (Any) – 传递给评估器的额外关键字参数。

返回:

如果 target 是函数、Runnable 或现有实验。ComparativeExperimentResults：如果 target 是由两个现有实验组成的元组。

返回类型:

ExperimentResults

示例

准备数据集

from langsmith import Client

client = Client()
dataset = client.clone_public_dataset(
    "https://smith.langchain.com/public/419dcab2-1d66-4b94-8901-0357ead390df/d"
)
dataset_name = "Evaluate Examples"

基本用法

def accuracy(outputs: dict, reference_outputs: dict) -> dict:
    # Row-level evaluator for accuracy.
    pred = outputs["response"]
    expected = reference_outputs["answer"]
    return {"score": expected.lower() == pred.lower()}

def precision(outputs: list[dict], reference_outputs: list[dict]) -> dict:
    # Experiment-level evaluator for precision.
    # TP / (TP + FP)
    predictions = [out["response"].lower() for out in outputs]
    expected = [ref["answer"].lower() for ref in reference_outputs]
    # yes and no are the only possible answers
    tp = sum([p == e for p, e in zip(predictions, expected) if p == "yes"])
    fp = sum([p == "yes" and e == "no" for p, e in zip(predictions, expected)])
    return {"score": tp / (tp + fp)}


def predict(inputs: dict) -> dict:
    # This can be any function or just an API call to your app.
    return {"response": "Yes"}


results = client.evaluate(
    predict,
    data=dataset_name,
    evaluators=[accuracy],
    summary_evaluators=[precision],
    experiment_prefix="My Experiment",
    description="Evaluating the accuracy of a simple prediction model.",
    metadata={
        "my-prompt-version": "abcd-1234",
    },
)

仅评估部分示例

experiment_name = results.experiment_name
examples = client.list_examples(dataset_name=dataset_name, limit=5)
results = client.evaluate(
    predict,
    data=examples,
    evaluators=[accuracy],
    summary_evaluators=[precision],
    experiment_prefix="My Experiment",
    description="Just testing a subset synchronously.",
)

流式传输每个预测以便更轻松、更快速地调试。

results = client.evaluate(
    predict,
    data=dataset_name,
    evaluators=[accuracy],
    summary_evaluators=[precision],
    description="I don't even have to block!",
    blocking=False,
)
for i, result in enumerate(results):  # doctest: +ELLIPSIS
    pass

使用 evaluate API 和现成的 LangChain 评估器

from langsmith.evaluation import LangChainStringEvaluator
from langchain.chat_models import init_chat_model


def prepare_criteria_data(run: Run, example: Example):
    return {
        "prediction": run.outputs["output"],
        "reference": example.outputs["answer"],
        "input": str(example.inputs),
    }


results = client.evaluate(
    predict,
    data=dataset_name,
    evaluators=[
        accuracy,
        LangChainStringEvaluator("embedding_distance"),
        LangChainStringEvaluator(
            "labeled_criteria",
            config={
                "criteria": {
                    "usefulness": "The prediction is useful if it is correct"
                    " and/or asks a useful followup question."
                },
                "llm": init_chat_model("gpt-4o"),
            },
            prepare_data=prepare_criteria_data,
        ),
    ],
    description="Evaluating with off-the-shelf LangChain evaluators.",
    summary_evaluators=[precision],
)

查看实验的评估结果：… 评估 LangChain 对象

from langchain_core.runnables import chain as as_runnable


@as_runnable
def nested_predict(inputs):
    return {"response": "Yes"}


@as_runnable
def lc_predict(inputs):
    return nested_predict.invoke(inputs)


results = client.evaluate(
    lc_predict,
    data=dataset_name,
    evaluators=[accuracy],
    description="This time we're evaluating a LangChain object.",
    summary_evaluators=[precision],
)

比较评估

results = client.evaluate(
    # The target is a tuple of the experiment IDs to compare
    target=(
        "12345678-1234-1234-1234-123456789012",
        "98765432-1234-1234-1234-123456789012",
    ),
    evaluators=[accuracy],
    summary_evaluators=[precision],
)

评估现有实验

results = client.evaluate(
    # The target is the ID of the experiment we are evaluating
    target="12345678-1234-1234-1234-123456789012",
    evaluators=[accuracy],
    summary_evaluators=[precision],
)

版本0.2.0中新增。

评估一个运行。

参数:

run (Union[Run, RunBase, str, UUID]) – 要评估的运行。
evaluator (RunEvaluator) – 要使用的评估器。
source_info (Optional[Dict[str, Any]]) – 有关评估来源的附加信息，作为反馈元数据进行记录。
reference_example (Optional[Union[Example, str, dict, UUID]]) – 用作评估参考的示例。如果未提供，将使用运行的参考示例。
load_child_runs (bool, default=False) – 解析运行ID时是否加载子运行。

返回:

评估创建的反馈对象。

返回类型:

Feedback

flush() → None[source]#

根据模式刷新队列或压缩缓冲区。

返回类型:: None

flush_compressed_traces(attempts: int = 3) → None[source]#

强制刷新当前已缓冲的压缩运行。

参数:: attempts (int)
返回类型:: None

get_prompt( prompt_identifier: str, ) → Prompt | None[source]#

根据其标识符获取特定提示。

参数:: prompt_identifier (str) – 提示的标识符。标识符应为“prompt_name”或“owner/prompt_name”格式。
返回:: 提示对象。
返回类型:: Optional[Prompt]
抛出:: requests.exceptions.HTTPError – 如果未找到提示或发生其他错误。

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

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

参数:

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

返回:

指定索引处的运行。

返回类型:

RunWithAnnotationQueueInfo

抛出:

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

获取查询运行的聚合统计信息。

接受与 list_runs 类似的查询参数，并根据与查询匹配的运行返回统计信息。

参数:

id (Optional[List[Union[UUID, str]]]) – 要筛选的运行 ID 列表。
trace (Optional[Union[UUID, str]]) – 要筛选的跟踪 ID。
parent_run (Optional[Union[UUID, str]]) – 要筛选的父运行 ID。
run_type (Optional[str]) – 要筛选的运行类型。
project_names (Optional[List[str]]) – 要筛选的项目名称列表。
project_ids (Optional[List[Union[UUID, str]]]) – 要筛选的项目 ID 列表。
reference_example_ids (Optional[List[Union[UUID, str]]]) – 要筛选的参考示例 ID 列表。
start_time (Optional[str]) – 要筛选的开始时间。
end_time (Optional[str]) – 要筛选的结束时间。
error (Optional[bool]) – 按错误状态筛选。
query (Optional[str]) – 要筛选的查询字符串。
filter (Optional[str]) – 要应用的筛选字符串。
trace_filter (Optional[str]) – 要应用的跟踪筛选字符串。
tree_filter (Optional[str]) – 要应用的树筛选字符串。
is_root (Optional[bool]) – 按根运行状态筛选。
data_source_type (Optional[str]) – 要筛选的数据源类型。

返回:

包含运行统计信息的字典。

返回类型:

Dict[str, Any]

get_run_url( *, run: RunBase, project_name: str | None = None, project_id: UUID | str | None = None, ) → str[source]#

获取运行的URL。

不建议在您的代理运行时中使用。更多用于事后与运行交互，以进行数据分析或 ETL 工作负载。

参数:

run (RunBase) – 运行。
project_name (Optional[str]) – 项目的名称。
project_id (Optional[Union[UUID, str]]) – 项目的 ID。

返回:

运行的 URL。

返回类型:

str

get_test_results( *, project_id: ID_TYPE | None = None, project_name: str | None = None, ) → pd.DataFrame[source]#

从实验中读取记录级别的信息到Pandas DF中。

注意：这将获取数据库中存在的任何数据。评估运行完成后，结果不会立即在数据库中可用。

反馈分数将作为实验所有运行的平均值返回。请注意，非数值反馈分数将被省略。

参数:

project_id (Optional[Union[UUID, str]]) – 项目的 ID。
project_name (Optional[str]) – 项目的名称。

返回:

包含测试结果的数据框。

返回类型:

pd.DataFrame

has_dataset( *, dataset_name: str | None = None, dataset_id: UUID | str | None = None, ) → bool[source]#

检查租户中是否存在数据集。

参数:

dataset_name (Optional[str]) – 要检查的数据集名称。
dataset_id (Optional[Union[UUID, str]]) – 要检查的数据集 ID。

返回:

数据集是否存在。

返回类型:

bool

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

检查项目是否存在。

参数:

project_name (str) – 要检查的项目的名称。
project_id (Optional[str]) – 要检查的项目的 ID。

返回:

项目是否存在。

返回类型:

bool

index_dataset(

*,

dataset_id: UUID | str,

tag: str = 'latest',

**kwargs: Any,

) → None[source]#

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

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

参数:

dataset_id (Union[UUID, str]) – 要索引的数据集 ID。
tag (Optional[str]) – 要索引的数据集版本。如果为“latest”，则数据集的任何更新（示例的添加、更新、删除）都将反映在索引中。
**kwargs (Any) – 作为请求体一部分传递的附加关键字参数。

返回:

None

返回类型:

None

like_prompt(prompt_identifier: str) → dict[str, int][source]#

喜欢一个提示。

参数:: prompt_identifier (str) – 提示的标识符。
返回:: 一个字典，键为“likes”，值为点赞计数。
返回类型:: Dict[str, int]

列出LangSmith API上的标注队列。

参数:

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

产生：

标注队列。

返回类型:

Iterator[AnnotationQueue]

获取数据集的分区。

参数:

dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
as_of (Optional[Union[str, datetime.datetime]]) – 要检索拆分的数据集版本。可以是时间戳或字符串标签。默认为“latest”。

返回:

此数据集的拆分名称。

返回类型:

List[str]

列出数据集版本。

参数:

dataset_id (Optional[Union[UUID, str]]) – 数据集的ID。
dataset_name (Optional[str]) – 数据集的名称。
search (Optional[str]) – 搜索查询。
limit (Optional[int]) – 要返回的最大版本数。

产生：

数据集版本。

返回类型:

Iterator[DatasetVersion]

列出LangSmith API上的数据集。

参数:

dataset_ids (Optional[List[Union[UUID, str]]]) – 用于筛选结果的数据集ID列表。
data_type (Optional[str]) – 用于筛选结果的数据集数据类型。
dataset_name (Optional[str]) – 用于筛选结果的数据集名称。
dataset_name_contains (Optional[str]) – 在数据集名称中搜索的子字符串。
metadata (Optional[Dict[str, Any]]) – 用于筛选结果的元数据字典。
limit (Optional[int]) – 要返回的最大数据集数。

产生：

数据集。

返回类型:

Iterator[Dataset]

检索指定数据集的示例行。

参数:

dataset_id (Optional[Union[UUID, str]]) – 用于筛选的数据集ID。默认为None。
dataset_name (Optional[str]) – 用于筛选的数据集名称。默认为None。
example_ids (Optional[Sequence[Union[UUID, str]]) – 用于筛选的示例ID。默认为None。
as_of (Optional[Union[datetime.datetime, str]]) – 用于检索示例的数据集版本标签或时间戳。响应示例将仅包含在指定标签（或时间戳）版本时存在的示例。
splits (Optional[Sequence[str]]) – 数据集拆分列表，例如数据集的“训练”、“测试”或“验证”部分。仅返回指定拆分中的示例。
inline_s3_urls (bool, default=True) – 是否内联S3 URL。默认为True。
offset (int, default=0) – 起始偏移量。默认为0。
limit (Optional[int]) – 要返回的最大示例数。
metadata (Optional[dict]) – 用于筛选的元数据字典。
filter (Optional[str]) – 应用于示例的结构化过滤字符串。
include_attachments (bool, default=False) – 响应中是否包含附件。默认为False。
**kwargs (Any) – 额外关键字参数将被忽略。

产生：

示例。

返回类型:

Iterator[Example]

示例

列出数据集的所有示例

from langsmith import Client

client = Client()

# By Dataset ID
examples = client.list_examples(
    dataset_id="c9ace0d8-a82c-4b6c-13d2-83401d68e9ab"
)
# By Dataset Name
examples = client.list_examples(dataset_name="My Test Dataset")

按ID列出示例

example_ids = [
    "734fc6a0-c187-4266-9721-90b7a025751a",
    "d6b4c1b9-6160-4d63-9b61-b034c585074f",
    "4d31df4e-f9c3-4a6e-8b6c-65701c2fed13",
]
examples = client.list_examples(example_ids=example_ids)

按元数据列出示例

examples = client.list_examples(
    dataset_name=dataset_name, metadata={"foo": "bar"}
)

按结构化过滤器列出示例

examples = client.list_examples(
    dataset_name=dataset_name,
    filter='and(not(has(metadata, \'{"foo": "bar"}\')), exists(metadata, "tenant_id"))',
)

列出LangSmith API上的反馈对象。

参数:

run_ids (Optional[Sequence[Union[UUID, str]]]) – 用于筛选的运行ID。
feedback_key (Optional[Sequence[str]]) – 用于筛选的反馈键。示例：“correctness”。查询将对所有反馈键执行并集操作。
feedback_source_type (Optional[Sequence[FeedbackSourceType]]) – 反馈来源的类型，例如模型或API。
limit (Optional[int]) – 要返回的最大反馈数量。
**kwargs (Any) – 附加关键字参数。

产生：

反馈对象。

返回类型:

Iterator[Feedback]

list_presigned_feedback_tokens( run_id: UUID | str, *, limit: int | None = None, ) → Iterator[FeedbackIngestToken][source]#

列出运行的预签名反馈令牌。

参数:

run_id (Union[UUID, str]) – 用于筛选的运行ID。
limit (Optional[int]) – 要返回的最大令牌数量。

产生：

反馈摄取令牌。

返回类型:

Iterator[FeedbackIngestToken]

从LangSmith API列出项目。

参数:

project_ids (Optional[List[Union[UUID, str]]]) – 用于筛选的项目ID列表，默认为None
name (Optional[str]) – 用于筛选的项目名称，默认为None
name_contains (Optional[str]) – 在项目名称中搜索的字符串，默认为None
reference_dataset_id (Optional[List[Union[UUID, str]]]) – 用于筛选的数据集ID，默认为None
reference_dataset_name (Optional[str]) – 用于筛选的引用数据集名称，默认为None
reference_free (Optional[bool]) – 是否仅筛选与数据集无关的项目。
limit (Optional[int]) – 要返回的最大项目数，默认为None
metadata (Optional[Dict[str, Any]]) – 用于筛选的元数据。

产生：

项目。

抛出:

ValueError – 如果同时提供了 reference_dataset_id 和 reference_dataset_name。

返回类型:

Iterator[TracerSession]

list_prompt_commits( prompt_identifier: str, *, limit: int | None = None, offset: int = 0, include_model: bool | None = False, ) → Iterator[ListedPromptCommit][source]#

列出给定提示的提交。

参数:

prompt_identifier (str) – 提示符的标识符，格式为“owner/repo_name”。
limit (Optional[int]) – 要返回的最大提交数量。如果为None，则返回所有提交。默认为None。
offset (int, default=0) – 在开始返回结果之前要跳过的提交数量。默认为0。
include_model (bool, default=False) – 提交数据中是否包含模型信息。默认为False。

产生：

每个提交的ListedPromptCommit对象。

返回类型:

Iterator[ListedPromptCommit]

注意

此方法使用分页来检索提交。如有必要，它将进行多次API调用以检索所有提交或达到指定的限制。

list_prompts( *, limit: int = 100, offset: int = 0, is_public: bool | None = None, is_archived: bool | None = False, sort_field: PromptSortField = PromptSortField.updated_at, sort_direction: Literal['desc', 'asc'] = 'desc', query: str | None = None, ) → ListPromptsResponse[source]#

分页列出提示。

参数:

limit (int, default=100) – 要返回的最大提示符数量。默认为100。
offset (int, default=0) – 要跳过的提示符数量。默认为0。
is_public (Optional[bool]) – 根据提示符是否公开进行筛选。
is_archived (Optional[bool]) – 根据提示符是否已归档进行筛选。
sort_field (PromptSortField) – 排序字段。默认为“updated_at”。
sort_direction (Literal["desc", "asc"], default="desc") – 排序顺序。默认为“desc”。
query (Optional[str]) – 根据搜索查询筛选提示符。

返回:

包含提示符列表的响应对象。

返回类型:

ListPromptsResponse

从LangSmith API列出运行。

参数:

project_id (Optional[Union[UUID, str], Sequence[Union[UUID, str]]]) – 用于筛选的项目ID。
project_name (Optional[Union[str, Sequence[str]]]) – 用于筛选的项目名称。
run_type (Optional[str]) – 用于筛选的运行类型。
trace_id (Optional[Union[UUID, str]]) – 用于筛选的跟踪ID。
reference_example_id (Optional[Union[UUID, str]]) – 用于筛选的参考示例ID。
query (Optional[str]) – 用于筛选的查询字符串。
filter (Optional[str]) – 用于筛选的过滤字符串。
trace_filter (Optional[str]) – 应用于跟踪树中根运行的过滤器。这旨在与常规filter参数结合使用，以便您可以根据跟踪中根运行的属性筛选运行。
tree_filter (Optional[str]) – 应用于跟踪树中其他运行（包括兄弟运行和子运行）的过滤器。这旨在与常规filter参数结合使用，以便您可以根据跟踪中任何运行的属性筛选运行。
is_root (Optional[bool]) – 是否按根运行筛选。
parent_run_id (Optional[Union[UUID, str]]) – 用于筛选的父运行ID。
start_time (Optional[datetime.datetime]) – 用于筛选的开始时间。
error (Optional[bool]) – 是否按错误状态筛选。
run_ids (Optional[Sequence[Union[UUID, str]]]) – 用于筛选的运行ID。
select (Optional[Sequence[str]]) – 要选择的字段。
limit (Optional[int]) – 要返回的最大运行数量。
**kwargs (Any) – 附加关键字参数。

产生：

运行。

返回类型:

Iterator[Run]

示例

# List all runs in a project
project_runs = client.list_runs(project_name="<your_project>")

# List LLM and Chat runs in the last 24 hours
todays_llm_runs = client.list_runs(
    project_name="<your_project>",
    start_time=datetime.now() - timedelta(days=1),
    run_type="llm",
)

# List root traces in a project
root_runs = client.list_runs(project_name="<your_project>", is_root=1)

# List runs without errors
correct_runs = client.list_runs(project_name="<your_project>", error=False)

# List runs and only return their inputs/outputs (to speed up the query)
input_output_runs = client.list_runs(
    project_name="<your_project>", select=["inputs", "outputs"]
)

# List runs by run ID
run_ids = [
    "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836",
    "9398e6be-964f-4aa4-8ae9-ad78cd4b7074",
]
selected_runs = client.list_runs(id=run_ids)

# List all "chain" type runs that took more than 10 seconds and had
# `total_tokens` greater than 5000
chain_runs = client.list_runs(
    project_name="<your_project>",
    filter='and(eq(run_type, "chain"), gt(latency, 10), gt(total_tokens, 5000))',
)

# List all runs called "extractor" whose root of the trace was assigned feedback "user_score" score of 1
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))',
)

# List all runs that started after a specific timestamp and either have "error" not equal to null or a "Correctness" feedback score equal to 0
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))))',
)

# List all runs where `tags` include "experimental" or "beta" and `latency` is greater than 2 seconds
tagged_runs = client.list_runs(
    project_name="<your_project>",
    filter='and(or(has(tags, "experimental"), has(tags, "beta")), gt(latency, 2))',
)

list_shared_examples( share_token: str, *, example_ids: list[UUID | str] | None = None, ) → list[Example][source]#

获取共享示例。

参数:

share_token (Union[UUID, str]) – 共享数据集的共享令牌或URL。
example_ids (Optional[List[UUID, str]], optional) – 用于筛选的示例ID。默认为None。

返回:

共享示例列表。

返回类型:

List[ls_schemas.Example]

列出共享项目。

参数:

dataset_share_token (str) – 数据集的共享令牌。
project_ids (Optional[List[Union[UUID, str]]]) – 用于筛选结果的项目ID列表，默认为None。
name (Optional[str]) – 用于筛选结果的项目名称，默认为None。
name_contains (Optional[str]) – 在项目名称中搜索的子字符串，默认为None。
limit (Optional[int]) – 要返回的最大项目数，默认为None。

产生：

共享项目。

返回类型:

Iterator[TracerSessionResult]

list_shared_runs( share_token: UUID | str, run_ids: list[str] | None = None, ) → Iterator[Run][source]#

获取共享运行。

参数:

share_token (Union[UUID, str]) – 共享运行的共享令牌或URL。
run_ids (Optional[List[str]]) – 用于筛选结果的运行ID列表。

产生：

共享运行。

返回类型:

Iterator[Run]

在 Langsmith 系统中批量摄取/更新（upsert）多个运行。

参数:

create (Optional[Sequence[Union[ls_schemas.Run, RunLikeDict]]]) – 表示要创建/发布的Run对象序列或等效字典。
update (Optional[Sequence[Union[ls_schemas.Run, RunLikeDict]]]) – 表示已创建且应更新/修补的Run对象序列或等效字典。
pre_sampled (bool, default=False) – 运行是否已进行采样，因此不应再次采样。默认为False。
dangerously_allow_filesystem (bool)

抛出:

LangsmithAPIError – 如果API请求中发生错误。

返回:

None

返回类型:

None

注意

运行对象必须包含dotted_order和trace_id字段才能被API接受。
to be accepted by the API.

示例

from langsmith import Client
import datetime
from uuid import uuid4

client = Client()
_session = "__test_batch_ingest_runs"
trace_id = uuid4()
trace_id_2 = uuid4()
run_id_2 = uuid4()
current_time = datetime.datetime.now(datetime.timezone.utc).strftime(
    "%Y%m%dT%H%M%S%fZ"
)
later_time = (
    datetime.datetime.now(datetime.timezone.utc) + timedelta(seconds=1)
).strftime("%Y%m%dT%H%M%S%fZ")

runs_to_create = [
    {
        "id": str(trace_id),
        "session_name": _session,
        "name": "run 1",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id)}",
        "trace_id": str(trace_id),
        "inputs": {"input1": 1, "input2": 2},
        "outputs": {"output1": 3, "output2": 4},
    },
    {
        "id": str(trace_id_2),
        "session_name": _session,
        "name": "run 3",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id_2)}",
        "trace_id": str(trace_id_2),
        "inputs": {"input1": 1, "input2": 2},
        "error": "error",
    },
    {
        "id": str(run_id_2),
        "session_name": _session,
        "name": "run 2",
        "run_type": "chain",
        "dotted_order": f"{current_time}{str(trace_id)}."
        f"{later_time}{str(run_id_2)}",
        "trace_id": str(trace_id),
        "parent_run_id": str(trace_id),
        "inputs": {"input1": 5, "input2": 6},
    },
]
runs_to_update = [
    {
        "id": str(run_id_2),
        "dotted_order": f"{current_time}{str(trace_id)}."
        f"{later_time}{str(run_id_2)}",
        "trace_id": str(trace_id),
        "parent_run_id": str(trace_id),
        "outputs": {"output1": 4, "output2": 5},
    },
]

client.multipart_ingest(create=runs_to_create, update=runs_to_update)

pull_prompt( prompt_identifier: str, *, include_model: bool | None = False, ) → Any[source]#

拉取提示并将其作为LangChain PromptTemplate返回。

此方法需要 langchain_core。

参数:

prompt_identifier (str) – 提示的标识符。
include_model (Optional[bool], default=False) – 提示数据中是否包含模型信息。

返回:

指定格式的提示符对象。

返回类型:

Any

pull_prompt_commit( prompt_identifier: str, *, include_model: bool | None = False, ) → PromptCommit[source]#

从LangSmith API拉取提示对象。

参数:

prompt_identifier (str) – 提示的标识符。
include_model (bool | None)

返回:

提示对象。

返回类型:

PromptCommit

抛出:

ValueError – 如果未找到提示符的任何提交。

push_prompt( prompt_identifier: str, *, object: Any | None = None, parent_commit_hash: str = 'latest', is_public: bool | None = None, description: str | None = None, readme: str | None = None, tags: Sequence[str] | None = None, ) → str[source]#

将提示推送到LangSmith API。

可用于更新提示符元数据或提示符内容。

如果提示符不存在，则会创建。如果提示符已存在，则会更新。

参数:

prompt_identifier (str) – 提示的标识符。
object (Optional[Any]) – 要推送的LangChain对象。
parent_commit_hash (str) – 父提交哈希。默认为“latest”。
is_public (Optional[bool]) – 提示符是否应为公开。如果为None（默认），则现有提示符的当前可见性状态将保持不变。对于新提示符，None默认为私有。设置为True表示公开，设置为False表示私有。
description (Optional[str]) – 提示符的描述。默认为空字符串。
readme (Optional[str]) – 提示符的readme。默认为空字符串。
tags (Optional[Sequence[str]]) – 提示符的标签列表。默认为空列表。

返回:

提示符的URL。

返回类型:

str

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

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

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

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

从LangSmith API读取数据集。

参数:

dataset_name (Optional[str]) – 要读取的数据集名称。
dataset_id (Optional[Union[UUID, str]]) – 要读取的数据集ID。

返回:

数据集。

返回类型:

Dataset

read_dataset_openai_finetuning( dataset_id: UUID | str | None = None, *, dataset_name: str | None = None, ) → list[source]#

下载OpenAI Jsonl格式的数据集并将其加载为字典列表。

参数:

dataset_id (Optional[Union[UUID, str]]) – 要下载的数据集ID。
dataset_name (Optional[str]) – 要下载的数据集名称。

返回:

作为字典列表加载的数据集。

返回类型:

list[dict]

抛出:

ValueError – 如果既未提供 dataset_id 也未提供 dataset_name。

read_dataset_shared_schema( dataset_id: UUID | str | None = None, *, dataset_name: str | None = None, ) → DatasetShareSchema[source]#

检索数据集的共享模式。

参数:

dataset_id (Optional[Union[UUID, str]]) – 数据集ID。dataset_id 或 dataset_name 必须提供其一。
dataset_name (Optional[str]) – 数据集名称。dataset_id 或 dataset_name 必须提供其一。

返回:

数据集的共享模式。

返回类型:

ls_schemas.DatasetShareSchema

抛出:

ValueError – 如果既未提供 dataset_id 也未提供 dataset_name。

按as_of或确切标签获取数据集版本。

使用此方法解析给定时间戳或给定标签的最近版本。

参数:

dataset_id (Optional[ID_TYPE]) – 数据集ID。
dataset_name (Optional[str]) – 数据集的名称。
as_of (Optional[datetime.datetime]) – 要检索的数据集时间戳。
tag (Optional[str]) – 要检索的数据集标签。

返回:

数据集版本。

返回类型:

DatasetVersion

示例

# Get the latest version of a dataset
client.read_dataset_version(dataset_name="my-dataset", tag="latest")

# Get the version of a dataset <= a given timestamp
client.read_dataset_version(
    dataset_name="my-dataset",
    as_of=datetime.datetime(2024, 1, 1),
)


# Get the version of a dataset with a specific tag
client.read_dataset_version(dataset_name="my-dataset", tag="prod")

read_example( example_id: UUID | str, *, as_of: datetime | None = None, ) → Example[source]#

从LangSmith API读取示例。

参数:

example_id (Union[UUID, str]) – 要读取的示例ID。
as_of (Optional[datetime.datetime]) – 用于检索示例的数据集版本标签或时间戳。响应示例将仅包含在指定标签（或时间戳）版本时存在的示例。

返回:

示例。

返回类型:

Example

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

从LangSmith API读取反馈。

参数:: feedback_id (Union[UUID, str]) – 要读取的反馈ID。
返回:: 反馈。
返回类型:: Feedback

read_project( *, project_id: str | None = None, project_name: str | None = None, include_stats: bool = False, ) → TracerSessionResult[source]#

从LangSmith API读取项目。

参数:

project_id (Optional[str]) – 要读取的项目ID。
project_name (Optional[str]) – 要读取的项目名称。只能提供 project_id 或 project_name 其中之一。
include_stats (bool, default=False) – 响应中是否包含项目的聚合统计信息。

返回:

项目。

返回类型:

TracerSessionResult

read_run( run_id: UUID | str, load_child_runs: bool = False, ) → Run[source]#

从LangSmith API读取运行。

参数:

run_id (Union[UUID, str]) – 要读取的运行ID。
load_child_runs (bool, default=False) – 是否加载嵌套子运行。

返回:

从LangSmith API读取的运行。

返回类型:

Run

示例

from langsmith import Client

# Existing run
run_id = "your-run-id"

client = Client()
stored_run = client.read_run(run_id)

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

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

参数:: run_id (Union[UUID, str]) – 运行的 ID。
返回:: 运行的共享链接，如果链接不可用则为None。
返回类型:: Optional[str]

read_shared_dataset( share_token: str, ) → Dataset[source]#

获取共享数据集。

参数:: share_token (Union[UUID, str]) – 共享数据集的共享令牌或URL。
返回:: 共享数据集。
返回类型:: Dataset

read_shared_run( share_token: UUID | str, run_id: UUID | str | None = None, ) → Run[source]#

获取共享运行。

参数:

share_token (Union[UUID, str]) – 共享运行的共享令牌或URL。
run_id (Optional[Union[UUID, str]]) – 要检索的特定运行ID。如果未提供，将返回完整的共享运行。

返回:

共享运行。

返回类型:

Run

request_with_retries(

method: Literal['GET', 'POST', 'PUT', 'PATCH', 'DELETE'],

pathname: str,

*,

request_kwargs: Mapping | None = None,

stop_after_attempt: int = 1,

retry_on: Sequence[type[BaseException]] | None = None,

to_ignore: Sequence[type[BaseException]] | None = None,

handle_response: Callable[[Response, int], Any] | None = None,

_context: str = '',

**kwargs: Any,

) → Response[source]#

发送带重试的请求。

参数:

method (str) – HTTP请求方法。
pathname (str) – 请求URL的路径名。将附加到API URL。
request_kwargs (Mapping) – 额外的请求参数。
stop_after_attempt (int, default=1) – 尝试次数。
retry_on (Optional[Sequence[Type[BaseException]]]) – 要重试的异常。除了：[LangSmithConnectionError, LangSmithAPIError]。
to_ignore (Optional[Sequence[Type[BaseException]]]) – 要忽略/跳过的异常。
handle_response (Optional[Callable[[requests.Response, int], Any]]) – 处理响应并返回是否继续重试的函数。
_context (str, default="") – 请求的上下文。
**kwargs (Any) – 要传递给请求的额外关键字参数。

返回:

响应对象。

返回类型:

requests.Response

抛出:

LangSmithAPIError – 如果发生服务器错误。
LangSmithUserError – 如果请求失败。
LangSmithConnectionError – 如果发生连接错误。
LangSmithError – 如果请求失败。

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

获取运行的共享状态。

参数:: run_id (Union[UUID, str]) – 运行的 ID。
返回:: 如果运行已共享则为True，否则为False。
返回类型:: bool

run_on_dataset(

dataset_name: str,

llm_or_chain_factory: Any,

*,

evaluation: Any | None = None,

concurrency_level: int = 5,

project_name: str | None = None,

project_metadata: dict[str, Any] | None = None,

dataset_version: datetime | str | None = None,

verbose: bool = False,

input_mapper: Callable[[dict], Any] | None = None,

revision_id: str | None = None,

**kwargs: Any,

) → dict[str, Any][source]#

在数据集上运行链或语言模型。

自版本0.1.0起已弃用: 此方法已弃用。请改用langsmith.aevaluate()。

参数:

dataset_name (str)
llm_or_chain_factory (Any)
evaluation (Any | None)
concurrency_level (int)
project_name (str | None)
project_metadata (dict[str, Any] | None)
dataset_version (datetime | str | None)
verbose (bool)
input_mapper (Callable[[dict], Any] | None)
revision_id (str | None)
kwargs (Any)

返回类型:

dict[str, Any]

share_dataset( dataset_id: UUID | str | None = None, *, dataset_name: str | None = None, ) → DatasetShareSchema[source]#

获取数据集的共享链接。

参数:

dataset_id (Optional[Union[UUID, str]]) – 数据集ID。dataset_id 或 dataset_name 必须提供其一。
dataset_name (Optional[str]) – 数据集名称。dataset_id 或 dataset_name 必须提供其一。

返回:

数据集的共享模式。

返回类型:

ls_schemas.DatasetShareSchema

抛出:

ValueError – 如果既未提供 dataset_id 也未提供 dataset_name。

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

获取运行的共享链接。

参数:

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

返回:

共享运行的URL。

返回类型:

str

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 (Union[UUID, str]) – 要搜索的数据集ID。
filter (Optional[str]) –
应用于搜索结果的过滤字符串。使用与 list_runs() 中的 filter 参数相同的语法。只支持部分操作。默认为None。

例如，您可以使用 and(eq(metadata.some_tag, 'some_value'), neq(metadata.env, 'dev')) 来筛选仅当 some_tag 具有 some_value 且环境不是 dev 时的示例。
**kwargs – 作为请求体一部分传递的额外关键字参数。

返回:

ExampleSearch对象列表。

返回类型:

list[ExampleSearch]

示例

from langsmith import Client

client = Client()
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\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\nYou can also use SQLite Cache which uses a SQLite database:\n\nrm .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"
        },
        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\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\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"),
    ),
]

sync_indexed_dataset(

*,

dataset_id: UUID | str,

**kwargs: Any,

) → None[source]#

同步数据集索引。这每5分钟会自动发生，但您可以调用此方法强制同步。

参数:

dataset_id (Union[UUID, str]) – 要同步的数据集ID。
kwargs (Any)

返回:

None

返回类型:

None

unlike_prompt( prompt_identifier: str, ) → dict[str, int][source]#

取消喜欢一个提示。

参数:: prompt_identifier (str) – 提示的标识符。
返回:: 一个字典，键为“likes”，值为点赞计数。
返回类型:: Dict[str, int]

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

删除数据集的共享链接。

参数:: dataset_id (Union[UUID, str]) – 要取消共享的数据集ID。
返回:: None
返回类型:: None

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

删除运行的共享链接。

参数:: run_id (Union[UUID, str]) – 要取消共享的运行ID。
返回:: None
返回类型:: None

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

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

参数:

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

返回:

None

返回类型:

None

update_dataset_splits( *, dataset_id: UUID | str | None = None, dataset_name: str | None = None, split_name: str, example_ids: list[UUID | str], remove: bool = False, ) → None[source]#

更新数据集的分区。

参数:

dataset_id (Optional[Union[UUID, str]]) – 要更新的数据集ID。
dataset_name (Optional[str]) – 要更新的数据集名称。
split_name (str) – 要更新的拆分名称。
example_ids (List[Union[UUID, str]]) – 要添加到或从拆分中移除的示例ID。
remove (Optional[bool]) – 如果为True，则从拆分中移除示例。如果为False，则将示例添加到拆分中。默认为False。

返回:

None

返回类型:

None

update_dataset_tag( *, dataset_id: UUID | str | None = None, dataset_name: str | None = None, as_of: datetime, tag: str, ) → None[source]#

更新数据集的标签。

如果标签已分配给此数据集的不同版本，则该标签将移动到新版本。as_of 参数用于确定将新标签应用于数据集的哪个版本。它必须是数据集的精确版本才能成功。您可以使用read_dataset_version方法找到要应用标签的精确版本。

参数:

dataset_id (Optional[Union[UUID, str]]) – 要更新的数据集ID。
dataset_name (Optional[str]) – 要更新的数据集名称。
as_of (datetime.datetime) – 应用新标签的数据集时间戳。
tag (str) – 要应用于数据集的新标签。

返回:

None

返回类型:

None

示例

dataset_name = "my-dataset"
# Get the version of a dataset <= a given timestamp
dataset_version = client.read_dataset_version(
    dataset_name=dataset_name, as_of=datetime.datetime(2024, 1, 1)
)
# Assign that version a new tag
client.update_dataset_tags(
    dataset_name="my-dataset",
    as_of=dataset_version.as_of,
    tag="prod",
)

更新特定示例。

参数:

example_id (Union[UUID, str]) – 要更新的示例ID。
inputs (Optional[Dict[str, Any]]) – 要更新的输入值。
outputs (Optional[Mapping[str, Any]]) – 要更新的输出值。
metadata (Optional[Dict]) – 要更新的元数据。
split (Optional[str | List[str]]) – 要更新的数据集拆分，例如“train”、“test”或“validation”。
dataset_id (Optional[Union[UUID, str]]) – 要更新的数据集ID。
attachments_operations (Optional[AttachmentsOperations]) – 要执行的附件操作。
attachments (Optional[Attachments]) – 要添加到示例的附件。

返回:

更新后的示例。

返回类型:

Dict[str, Any]

更新多个示例。

示例预计都属于同一数据集。

参数:

dataset_name (str | None) – 要更新的数据集名称。应精确指定“dataset_name”或“dataset_id”之一。
dataset_id (UUID | str | None) – 要更新的数据集ID。应精确指定“dataset_name”或“dataset_id”之一。
updates (Sequence[ExampleUpdate | dict] | None) – 示例更新。覆盖任何指定的字段，不更新任何未指定的字段。
dangerously_allow_filesystem (bool) – 是否允许使用文件系统路径作为附件。
**kwargs (Any) –
遗留关键字参数。如果指定了“updates”，则不应指定此项。
- example_ids (Sequence[UUID | str]): 要更新的示例ID。
- inputs (Sequence[dict | None] | None): 示例的输入值。
- outputs (Sequence[dict | None] | None): 示例的输出值。
- metadata (Sequence[dict | None] | None): 示例的元数据。
- splits (Sequence[str | list[str] | None] | None): 示例的拆分，即数据集的“训练”、“测试”或“验证”等划分。
- attachments_operations (Sequence[AttachmentsOperations | None] | None): 要对附件执行的操作。
- dataset_ids (Sequence[UUID | str] | None): 要将示例移动到的数据集ID。

返回:

LangSmith JSON响应。包括“message”、“count”和“example_ids”。

版本0.3.9中的变更: 更新为…

返回类型:

dict[str, Any]

Example

from langsmith import Client

client = Client()

dataset = client.create_dataset("agent-qa")

examples = [
    {
        "inputs": {"question": "what's an agent"},
        "outputs": {"answer": "an agent is..."},
        "metadata": {"difficulty": "easy"},
    },
    {
        "inputs": {
            "question": "can you explain the agent architecture in this diagram?"
        },
        "outputs": {"answer": "this diagram shows..."},
        "attachments": {"diagram": {"mime_type": "image/png", "data": b"..."}},
        "metadata": {"difficulty": "medium"},
    },
    # more examples...
]

response = client.create_examples(dataset_name="agent-qa", examples=examples)
example_ids = response["example_ids"]

updates = [
    {
        "id": example_ids[0],
        "inputs": {"question": "what isn't an agent"},
        "outputs": {"answer": "an agent is not..."},
    },
    {
        "id": example_ids[1],
        "attachments_operations": [
            {"rename": {"diagram": "agent_diagram"}, "retain": []}
        ],
    },
]
response = client.update_examples(dataset_name="agent-qa", updates=updates)
# -> {"example_ids": [...

update_examples_multipart( *, dataset_id: UUID | str, updates: list[ExampleUpdate] | None = None, dangerously_allow_filesystem: bool = False, ) → UpsertExamplesResponse[source]#

使用多部分更新示例。

自版本0.3.9起已弃用: 请使用 Client.update_examples 代替。将在0.4.0版本中移除。

参数:

dataset_id (UUID | str)
updates (list[ExampleUpdate] | None)
dangerously_allow_filesystem (bool)

返回类型:

UpsertExamplesResponse

在LangSmith API中更新反馈。

参数:

feedback_id (Union[UUID, str]) – 要更新的意见反馈的ID。
score (Optional[Union[float, int, bool]]) – 用于更新意见反馈的分数。
value (Optional[Union[float, int, bool, str, dict]]) – 用于更新意见反馈的值。
correction (Optional[dict]) – 用于更新意见反馈的修正。
comment (Optional[str]) – 用于更新意见反馈的评论。

返回:

None

返回类型:

None

更新LangSmith项目。

参数:

project_id (Union[UUID, str]) – 要更新的项目的ID。
name (Optional[str]) – 要赋予项目的新名称。仅当项目已分配结束时间（表示已完成/关闭）时，此项才有效。
description (Optional[str]) – 要赋予项目的新描述。
metadata (Optional[dict]) – 与项目关联的附加元数据。
project_extra (Optional[dict]) – 附加项目信息。
end_time (Optional[datetime.datetime]) – 项目完成的时间。

返回:

更新后的项目。

返回类型:

TracerSession

更新提示的元数据。

要更新提示的内容，请改用 push_prompt 或 create_commit。

参数:

prompt_identifier (str) – 要更新的提示符的标识符。
description (Optional[str]) – 提示符的新描述。
readme (Optional[str]) – 提示符的新自述文件。
tags (Optional[Sequence[str]]) – 提示符的新标签列表。
is_public (Optional[bool]) – 提示符的新公开状态。
is_archived (Optional[bool]) – 提示符的新归档状态。

返回:

服务器返回的更新后的提示符数据。

返回类型:

Dict[str, Any]

抛出:

ValueError – 如果提示符标识符为空。
HTTPError – 如果服务器请求失败。

在LangSmith API中更新运行。

参数:

run_id (Union[UUID, str]) – 要更新的运行的ID。
name (Optional[str]) – 运行的名称。
end_time (Optional[datetime.datetime]) – 运行的结束时间。
error (Optional[str]) – 运行的错误消息。
inputs (Optional[Dict]) – 运行的输入值。
outputs (Optional[Dict]) – 运行的输出值。
events (Optional[Sequence[dict]]) – 运行的事件。
extra (Optional[Dict]) – 运行的额外信息。
tags (Optional[List[str]]]) – 运行的标签。
attachments (Optional[Dict[str, Attachment]]]) – 要添加到运行的附件字典。键是附件名称，值是包含数据和MIME类型的附件对象。
**kwargs (Any) – 关键字参数被忽略。
dangerously_allow_filesystem (bool)
**kwargs

返回:

None

返回类型:

None

示例

from langsmith import Client
import datetime
from uuid import uuid4

client = Client()
project_name = "__test_update_run"

start_time = datetime.datetime.now()
revision_id = uuid4()
run: dict = dict(
    id=uuid4(),
    name="test_run",
    run_type="llm",
    inputs={"text": "hello world"},
    project_name=project_name,
    api_url=os.getenv("LANGCHAIN_ENDPOINT"),
    start_time=start_time,
    extra={"extra": "extra"},
    revision_id=revision_id,
)
# Create the run
client.create_run(**run)
run["outputs"] = {"output": ["Hi"]}
run["extra"]["foo"] = "bar"
run["name"] = "test_run_updated"
# Update the run
client.update_run(run["id"], **run)

upload_csv( csv_file: str | tuple[str, BytesIO], input_keys: Sequence[str], output_keys: Sequence[str], *, name: str | None = None, description: str | None = None, data_type: DataType | None = DataType.kv, ) → Dataset[source]#

上传CSV文件到LangSmith API。

参数:

csv_file (Union[str, Tuple[str, io.BytesIO]]) – 要上传的CSV文件。如果是字符串，则应为路径。如果是元组，则应为包含文件名和BytesIO对象的元组。
input_keys (Sequence[str]) – 输入键。
output_keys (Sequence[str]) – 输出键。
name (Optional[str]) – 数据集的名称。
description (Optional[str]) – 数据集的描述。
data_type (Optional[ls_schemas.DataType]) – 数据集的类型。

返回:

上传的数据集。

返回类型:

Dataset

抛出:

ValueError – 如果csv_file不是字符串或元组。

示例

from langsmith import Client
import os

client = Client()

csv_file = "path/to/your/myfile.csv"
input_keys = ["column1", "column2"]  # replace with your input column names
output_keys = ["output1", "output2"]  # replace with your output column names

dataset = client.upload_csv(
    csv_file=csv_file,
    input_keys=input_keys,
    output_keys=output_keys,
    name="My CSV Dataset",
    description="Dataset created from a CSV file",
    data_type="kv",  # The default
)

upload_dataframe( df: pd.DataFrame, name: str, input_keys: Sequence[str], output_keys: Sequence[str], *, description: str | None = None, data_type: ls_schemas.DataType | None = DataType.kv, ) → ls_schemas.Dataset[source]#

将数据帧作为单个示例上传到LangSmith API。

参数:

df (pd.DataFrame) – 要上传的数据框。
name (str) – 数据集的名称。
input_keys (Sequence[str]) – 输入键。
output_keys (Sequence[str]) – 输出键。
description (Optional[str]) – 数据集的描述。
data_type (Optional[DataType]) – 数据集的类型。

返回:

上传的数据集。

返回类型:

Dataset

抛出:

ValueError – 如果csv_file不是字符串或元组。

示例

from langsmith import Client
import os
import pandas as pd

client = Client()

df = pd.read_parquet("path/to/your/myfile.parquet")
input_keys = ["column1", "column2"]  # replace with your input column names
output_keys = ["output1", "output2"]  # replace with your output column names

dataset = client.upload_dataframe(
    df=df,
    input_keys=input_keys,
    output_keys=output_keys,
    name="My Parquet Dataset",
    description="Dataset created from a parquet file",
    data_type="kv",  # The default
)

upload_examples_multipart( *, dataset_id: UUID | str, uploads: list[ExampleCreate] | None = None, dangerously_allow_filesystem: bool = False, ) → UpsertExamplesResponse[source]#

使用多部分上传示例。

从版本0.3.9开始废弃：请改用 Client.create_examples。将在0.4.0版本中移除。

参数:

dataset_id (UUID | str)
uploads (list[ExampleCreate] | None)
dangerously_allow_filesystem (bool)

返回类型:

UpsertExamplesResponse

upsert_examples_multipart( *, upserts: list[ExampleUpsertWithAttachments] | None = None, dangerously_allow_filesystem: bool = False, ) → UpsertExamplesResponse[source]#

插入/更新示例。

从版本0.3.9开始废弃：请改用 Client.create_examples 和 Client.update_examples。将在0.4.0版本中移除。

参数:

upserts (list[ExampleUpsertWithAttachments] | None)
dangerously_allow_filesystem (bool)

返回类型:

UpsertExamplesResponse