Client #

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

向 Chat 类型数据集添加示例（行）。

参数:

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]) – 示例的创建时间戳。

返回:

创建的示例

返回类型:

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

为现有提示创建提交。

参数:

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

返回:

prompt 提交的 url。

返回类型:

str

Raises:

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

在 LangSmith API 上创建比较实验。

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

参数:

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]) – 与数据集关联的附加元数据。

返回:

创建的数据集。

返回类型:

Raises:

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]) – 示例的附件（可选）。

返回:

创建的示例。

返回类型:

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 (Run) – 要从中创建示例的 Run 对象。
dataset_id (Optional[Union[UUID, str]]) – 数据集的 ID。
dataset_name (Optional[str]) – 数据集的名称。
created_at (Optional[datetime.datetime]) – 示例的创建时间戳。

返回:

创建的示例

返回类型:

在数据集中创建示例。

参数:

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（可选）。

Raises:

ValueError – 如果同时提供了 ‘examples’ 和旧式参数。

返回:

LangSmith JSON 响应。包括 ‘count’ 和 ‘example_ids’。

返回类型:

Changed in version 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": [...

在 LangSmith API 中创建反馈。

参数:

run_id (Optional[Union[UUID, str]]) – 要为其提供反馈的 run 的 ID（可选）。必须提供 run_id 或 project_id 中的一个。
key (str) – 此反馈相关的指标或 ‘方面’ 的名称。
score (Optional[Union[float, int, bool]]) – 在指标或方面对此 run 进行评分的分数（可选）。
value (Optional[Union[float, int, bool, str, dict]]) – 此反馈的显示值或非数字值（可选）。
correction (Optional[dict]) – 此 run 的正确的真实值（可选）。
comment (Optional[str]) – 关于此反馈的评论（可选），例如对分数的理由或 LLM 评判的思维链轨迹。
source_info (Optional[Dict[str, Any]]) – 关于此反馈来源的信息（可选）。
feedback_source_type (Union[FeedbackSourceType, str]) –

反馈来源的类型，例如模型（对于模型生成的反馈）
或 API。
source_run_id (Optional[Union[UUID, str]]) – 生成此反馈的 run 的 ID（可选），如果类型为 “model”。
feedback_id (Optional[Union[UUID, str]]) – 要创建的反馈的 ID（可选）。如果未提供，将生成一个随机 UUID。
feedback_config (Optional[FeedbackConfig]) – 指定如何解释具有此键的反馈的配置（可选）。示例包括连续型（具有最小/最大边界）、类别型或自由格式。
stop_after_attempt (int, default=10) – 在放弃之前重试请求的次数（默认值=10）。
project_id (Optional[Union[UUID, str]]) – 要在其上提供反馈的 project_id 的 ID（可选）。必须提供此参数和 run_id 中的一个，且仅能提供一个。
comparative_experiment_id (Optional[Union[UUID, str]]) – 如果此反馈是作为比较实验的一部分记录的，则此参数将反馈与该实验关联（可选）。
feedback_group_id (Optional[Union[UUID, str]]) – 当记录偏好、排名 run 或其他比较反馈时，此参数用于将反馈分组在一起（可选）。
extra (Optional[Dict]) – 反馈的元数据（可选）。
trace_id (Optional[Union[UUID, str]]) – 要为其提供反馈的 run 的跟踪 ID（可选）。启用批量摄取。
**kwargs (Any) – 额外的关键字参数。
error (bool | None)
**kwargs

返回:

创建的反馈对象。

返回类型:

Feedback

从预签名令牌或 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。

Raises:

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]) – 示例的创建时间戳。

返回:

创建的示例

返回类型:

RunWithAnnotationQueueInfo

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

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

参数:

run_id (Union[UUID, str]) – Run 的 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]) – Run 的 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 对象，仅创建空的 prompt。

参数:

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

返回:

创建的 prompt 对象。

返回类型:

Prompt

Raises:

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) – Run 的名称。
inputs (Dict[str, Any]) – Run 的输入值。
run_type (str) – Run 的类型，例如 tool、chain、llm、retriever、embedding、prompt 或 parser。
project_name (Optional[str]) – Run 的项目名称。
revision_id (Optional[Union[UUID, str]]) – Run 的修订 ID。
**kwargs (Any) – 额外的关键字参数。
dangerously_allow_filesystem (bool)
**kwargs

返回:

None

Raises:

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

Raises:

ValueError – 如果未提供 project_name 或 project_id。

返回类型:

None

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

删除提示。

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

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]) – 要从注释队列中删除的 Run 的 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 并返回 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

使用带有现成的 LangChain 评估器的 evaluate API

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]) – 要评估的 Run。
evaluator (RunEvaluator) – 要使用的评估器。
source_info (Optional[Dict[str, Any]]) – 关于评估来源的其他信息，将作为反馈元数据记录。
reference_example (Optional[Union[Example, str, dict, UUID]]) – 用作评估参考的示例。如果未提供，将使用 run 的参考示例。
load_child_runs (bool, default=False) – 在解析 run ID 时是否加载子 run。

返回:

评估创建的反馈对象。

返回类型:

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 的标识符。标识符应采用 “prompt_name” 或 “owner/prompt_name” 格式。
返回:: prompt 对象。
返回类型:: Optional[Prompt]
Raises:: requests.exceptions.HTTPError – 如果未找到 prompt 或发生其他错误。

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

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

参数:

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

返回:

指定索引处的 Run。

返回类型:

Raises:

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

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

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

参数:

id (Optional[List[Union[UUID, str]]]) – 要按运行 ID 筛选的列表。
trace (Optional[Union[UUID, str]]) – 要按跟踪 ID 筛选的跟踪 ID。
parent_run (Optional[Union[UUID, str]]) – 要按父运行 ID 筛选的父运行 ID。
run_type (Optional[str]) – 要按运行类型筛选的运行类型。
project_names (Optional[List[str]]) – 要按项目名称筛选的项目名称列表。
project_ids (Optional[List[Union[UUID, str]]]) – 要按项目 ID 筛选的项目 ID 列表。
reference_example_ids (Optional[List[Union[UUID, str]]]) – 要按参考示例 ID 筛选的参考示例 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) – prompt 的标识符。
返回:: 一个字典，键为“likes”，值为点赞计数。
返回类型:: Dict[str, int]

列出 LangSmith API 上的注释队列。

参数:

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

Yields:

注释队列。

返回类型:

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]) – 要返回的最大版本数。

Yields:

数据集版本。

返回类型:

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]) – 要返回的最大数据集数。

Yields:

数据集。

返回类型:

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]]) – 数据集拆分的列表，它是数据集的划分，例如 ‘train’、‘test’ 或 ‘validation’。仅返回来自指定拆分的示例。
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) – 额外的关键字参数将被忽略。

Yields:

示例。

返回类型:

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]]]) – 用于筛选的 Run 的 ID 列表。
feedback_key (Optional[Sequence[str]]) – 用于筛选的反馈键。例如：“correctness”。查询将执行所有反馈键的并集操作。
feedback_source_type (Optional[Sequence[FeedbackSourceType]]) – 反馈来源类型，例如模型或 API。
limit (Optional[int]) – 返回的最大反馈数。
**kwargs (Any) – 额外的关键字参数。

Yields:

反馈对象。

返回类型:

Iterator[Feedback]

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

列出运行的反馈摄取令牌。

参数:

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

Yields:

反馈摄取令牌。

返回类型:

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]]) – 用于筛选的元数据。

Yields:

项目。

Raises:

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 = 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。

Yields:

每个提交的 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 或 ID 列表。
project_name (Optional[Union[str, Sequence[str]]]) – 用于筛选的项目名称或名称列表。
run_type (Optional[str]) – 用于筛选的 Run 的类型。
trace_id (Optional[Union[UUID, str]]) – 用于筛选的 Trace 的 ID。
reference_example_id (Optional[Union[UUID, str]]) – 用于筛选的参考示例的 ID。
query (可选[str]) – 用于筛选的查询字符串。
filter (可选[str]) – 用于筛选的过滤器字符串。
trace_filter (可选[str]) – 应用于追踪树中 ROOT 运行的过滤器。此参数旨在与常规 filter 参数结合使用，以便您可以通过追踪中根运行的属性来筛选运行。
tree_filter (可选[str]) – 应用于追踪树中其他运行的过滤器，包括同级和子运行。此参数旨在与常规 filter 参数结合使用，以便您可以通过追踪中任何运行的属性来筛选运行。
is_root (可选[bool]) – 是否按根运行筛选。
parent_run_id (可选[Union[UUID, str]]) – 要筛选的父级运行的 ID。
start_time (可选[datetime.datetime]) – 要筛选的开始时间。
error (可选[bool]) – 是否按错误状态筛选。
run_ids (Optional[Sequence[Union[UUID, str]]]) – 用于筛选的 Run 的 ID 列表。
select (可选[Sequence[str]]) – 要选择的字段。
limit (可选[int]) – 要返回的最大运行数。
**kwargs (Any) – 额外的关键字参数。

Yields:

运行。

返回类型:

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 (可选[List[UUID, str]], 可选) – 要筛选的示例的 ID。默认为 None。

返回:

共享示例的列表。

返回类型:

List[ls_schemas.Example]

列出共享项目。

参数:

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

Yields:

共享项目。

返回类型:

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 (可选[List[str]]) – 用于筛选结果的运行 ID 列表。

Yields:

一个共享运行。

返回类型:

Iterator[Run]

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

参数:

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

Raises:

LangsmithAPIError – 如果 API 请求中存在错误。

返回:

None

返回类型:

None

注意

run 对象必须包含 dotted_order 和 trace_id 字段
才能被 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) – prompt 的标识符。
include_model (可选[bool], default=False) – 是否在提示数据中包含模型信息。

返回:

指定格式的提示对象。

返回类型:

Any

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

从 LangSmith API 拉取提示对象。

参数:

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

返回:

prompt 对象。

返回类型:

PromptCommit

Raises:

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

将提示推送到 LangSmith API。

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

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

参数:

prompt_identifier (str) – prompt 的标识符。
object (可选[Any]) – 要推送的 LangChain 对象。
parent_commit_hash (str) – 父提交哈希。默认为“latest”。
is_public (可选[bool]) – 提示是否应公开。如果为 None（默认值），则现有提示保持当前的可见性状态。对于新提示，None 默认为私有。设置为 True 以公开，或设置为 False 以设为私有。
description (可选[str]) – 提示的描述。默认为空字符串。
readme (可选[str]) – 提示的 readme。默认为空字符串。
tags (可选[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 (可选[str]) – 要读取的数据集的名称。
dataset_id (可选[Union[UUID, str]]) – 要读取的数据集的 ID。

返回:

数据集。

返回类型:

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

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

参数:

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

返回:

数据集加载为字典列表。

返回类型:

list[dict]

Raises:

ValueError – 如果未提供 dataset_id 和 dataset_name。

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

检索数据集的共享模式。

参数:

dataset_id (可选[Union[UUID, str]]) – 数据集的 ID。必须提供 dataset_id 或 dataset_name。
dataset_name (可选[str]) – 数据集的名称。必须提供 dataset_id 或 dataset_name。

返回:

数据集的共享架构。

返回类型:

ls_schemas.DatasetShareSchema

Raises:

ValueError – 如果未提供 dataset_id 和 dataset_name。

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

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

参数:

dataset_id (可选[ID_TYPE]) – 数据集的 ID。
dataset_name (Optional[str]) – 数据集的名称。
as_of (可选[datetime.datetime]) – 要检索的数据集的时间戳。
tag (可选[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 (可选[datetime.datetime]) – 数据集版本标签或时间戳，用于检索该时间点的示例。响应示例将仅是标记（或带时间戳）版本时存在的示例。

返回:

示例。

返回类型:

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 (可选[str]) – 要读取的项目的 ID。
project_name (可选[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。

参数:

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 的共享链接。

参数:: run_id (Union[UUID, str]) – Run 的 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

Raises:

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

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

获取 run 的共享状态。

参数:: run_id (Union[UUID, str]) – Run 的 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]#

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

Deprecated since version 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 (可选[Union[UUID, str]]) – 数据集的 ID。必须提供 dataset_id 或 dataset_name。
dataset_name (可选[str]) – 数据集的名称。必须提供 dataset_id 或 dataset_name。

返回:

数据集的共享架构。

返回类型:

ls_schemas.DatasetShareSchema

Raises:

ValueError – 如果未提供 dataset_id 和 dataset_name。

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

获取 run 的共享链接。

参数:

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"),
    ),
]

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

取消点赞一个 prompt。

参数:: prompt_identifier (str) – prompt 的标识符。
返回:: 一个字典，键为“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 的共享链接。

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

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

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) – 要更新的拆分（split）的名称。
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): 示例的拆分，这是数据集的划分，例如 ‘train’、‘test’ 或 ‘validation’。
- attachments_operations (Sequence[AttachmentsOperations | None] | None): 要对附件执行的操作。
- dataset_ids (Sequence[UUID | str] | None): 要将示例移动到的数据集的 ID。

返回:

LangSmith JSON 响应。包括 ‘message’、‘count’ 和 ‘example_ids’。

Changed in version 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]#

使用 multipart 更新示例。

Deprecated since version 0.3.9: 自 0.3.9 版本起已弃用：请改用 Client.update_examples。将在 0.4.0 版本中移除。

参数:

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

返回类型:

在 LangSmith API 中更新一个 feedback。

参数:

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]) – 项目的新名称。仅当项目已被分配 end_time（表示已完成/关闭）时，此项才有效。
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]

Raises:

ValueError – 如果 prompt_identifier 为空。
HTTPError – 如果服务器请求失败。

在 LangSmith API 中更新一个 run。

参数:

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 类型的 Attachment 对象。
**kwargs (Any) – Kwargs 将被忽略。
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]) – 数据集的数据类型。

返回:

上传的数据集。

返回类型:

Raises:

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]#

将一个 dataframe 作为单独的示例上传到 LangSmith API。

参数:

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

返回:

上传的数据集。

返回类型:

Raises:

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]#

使用 multipart 上传示例。

Deprecated since version 0.3.9: 自 0.3.9 版本弃用: 请使用 Client.create_examples 代替。将在 0.4.0 版本中移除。

参数:

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

返回类型:

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

Upsert 示例。

Deprecated since version 0.3.9: 自 0.3.9 版本弃用: 请使用 Client.create_examples 和 Client.update_examples 代替。将在 0.4.0 版本中移除。

参数:

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

返回类型: