使用 SDK 查询追踪

推荐使用 SDK 中的 list_runs 方法，或 API 中的 /runs/query 端点，来查询运行（LangSmith 追踪中的跨度数据）。LangSmith 以运行（跨度）数据格式中规定的简单格式存储追踪数据。本页面介绍：

使用过滤参数：使用 SDK 参数进行基于关键字的过滤。
使用过滤查询语言：使用 LangSmith 的过滤语法执行复杂查询。
使用子运行谓词查询追踪树：结合服务端缩小范围和本地子运行遍历。
速率限制：每个租户的限制，以及遵守这些限制的最佳实践。

如果你要导出大量追踪数据，建议使用批量数据导出功能。该功能更适合处理大量数据，并支持自动重试以及跨分区并行处理。

使用过滤参数

对于简单查询，不必依赖查询语法。你可以使用过滤参数参考中指定的过滤参数。

前提条件运行以下代码片段之前，请先初始化客户端。

from langsmith import Client

client = Client()

import { Client, Run } from "langsmith";

const client = new Client();

import com.langchain.smith.client.LangsmithClient;
import com.langchain.smith.client.okhttp.LangsmithOkHttpClient;

LangsmithClient client = LangsmithOkHttpClient.fromEnv();

下面是一些使用关键字参数列出运行的示例：

列出项目中的所有运行

project_runs = client.list_runs(project_name="<your_project>")

// 下载项目中的运行
const projectRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
})) {
  projectRuns.push(run);
};

import com.langchain.smith.models.runs.RunQueryParams;

RunQueryParams projectRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .build();

列出过去 24 小时内的 LLM 和聊天运行

todays_llm_runs = client.list_runs(
    project_name="<your_project>",
    start_time=datetime.now() - timedelta(days=1),
    run_type="llm",
)

const todaysLlmRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
  startTime: new Date(Date.now() - 1000 * 60 * 60 * 24),
  runType: "llm",
})) {
  todaysLlmRuns.push(run);
};

OffsetDateTime now = OffsetDateTime.now();
OffsetDateTime twentyFourHoursAgo = now.minus(24, ChronoUnit.HOURS);

RunQueryParams todaysLlmRuns = RunQueryParams.builder()
    .runType(RunQueryParams.RunType.LLM)
    .startTime(twentyFourHoursAgo)
    .addSession("<your_project>")
    .limit(50L)
    .build();

列出项目中的根运行

根运行是没有父运行的运行。它们的 is_root 值为 True。你可以使用此字段筛选根运行。

root_runs = client.list_runs(
    project_name="<your_project>",
    is_root=True
)

const rootRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
  isRoot: 1,
})) {
  rootRuns.push(run);
};

import com.langchain.smith.models.runs.RunQueryParams;

RunQueryParams rootRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .isRoot(true)
    .build();

列出没有错误的运行

correct_runs = client.list_runs(project_name="<your_project>", error=False)

const correctRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
  error: false,
})) {
  correctRuns.push(run);
};

import com.langchain.smith.models.runs.RunQueryParams;

RunQueryParams noErrorRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .error(false)
    .build();

按运行 ID 列出运行

忽略其他参数如果按照上述方式提供运行 ID 列表，系统将忽略其他过滤参数（例如 project_name、run_type 等），并直接返回与给定 ID 匹配的运行。

如果你有运行 ID 列表，可以直接列出这些运行：

run_ids = ['a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836','9398e6be-964f-4aa4-8ae9-ad78cd4b7074']
selected_runs = client.list_runs(id=run_ids)

const runIds = [
  "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836",
  "9398e6be-964f-4aa4-8ae9-ad78cd4b7074",
];
const selectedRuns: Run[] = [];
for await (const run of client.listRuns({
  id: runIds,
})) {
  selectedRuns.push(run);
};

import com.langchain.smith.models.runs.RunQueryParams;

RunQueryParams runIdsRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .id(runIds)
    .build();

按 ID 获取单个运行

要按 ID 获取单个运行（追踪），请使用 read_run 方法。当你有特定的追踪 ID（例如来自 LangSmith 分享链接 https://smith.langchain.com/public/<trace-id>/r），并希望获取其完整数据时，此方法非常有用。

run_id = "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836"
run = client.read_run(run_id)

# 访问运行数据
print(run.inputs)
print(run.outputs)
print(run.name)

const runId = "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836";
const run = await client.readRun(runId);

// 访问运行数据
console.log(run.inputs);
console.log(run.outputs);
console.log(run.name);

import com.langchain.smith.models.runs.RunQueryParams;

RunQueryParams runIdRun = RunQueryParams.builder()
    .addSession("<your_project>")
    .addId(runId)
    .build();

使用 LangGraph 在本地重放追踪如果你在使用带检查点功能的 LangGraph，可以从 LangSmith 获取追踪，并在本地重放以进行调试。详情请参阅 LangGraph 的时间旅行与重放文档，了解如何从检查点恢复执行。

使用过滤查询语言

对于更复杂的查询，可以使用过滤查询语言。以下示例涵盖最常见的模式。有关完整的运算符和字段参考（包括所有比较器、可过滤字段、值格式规则以及快速参考示例表），请参阅追踪查询语法：过滤查询语言。

列出对话线程中的所有根运行

这是获取对话线程中运行的方式。有关如何设置线程的更多信息，请参阅设置线程操作指南。线程通过设置共享线程 ID 进行分组。LangSmith UI 支持使用以下任一元数据键：session_id 或 thread_id。会话 ID 也称为追踪项目 ID。以下查询会匹配其中任意一个键。

group_key = "<your_thread_id>"
filter_string = f'and(in(metadata_key, ["session_id","thread_id"]), eq(metadata_value, "{group_key}"))'
thread_runs = client.list_runs(
    project_name="<your_project>",
    filter=filter_string,
    is_root=True
)

const groupKey = "<your_thread_id>";
const filterString = `and(in(metadata_key, ["session_id","thread_id"]), eq(metadata_value, "${groupKey}"))`;
const threadRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
  filter: filterString,
  isRoot: true
})) {
  threadRuns.push(run);
};

import com.langchain.smith.models.runs.RunQueryParams;

String groupKey = "<your_thread_id>";

String filterString = String.format(
    "and(in(metadata_key, [\"session_id\",\"thread_id\"]), eq(metadata_value, \"%s\"))",
    groupKey
);

RunQueryParams threadRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter(filterString)
    .build();

列出所有名为 “extractor” 且其追踪根运行被赋予 “user_score” 反馈、分数为 1 的运行

client.list_runs(
    project_name="<your_project>",
    filter='eq(name, "extractor")',
    trace_filter='and(eq(feedback_key, "user_score"), eq(feedback_score, 1))'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'eq(name, "extractor")',
  traceFilter: 'and(eq(feedback_key, "user_score"), eq(feedback_score, 1))'
})

RunQueryParams extractorRuns = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("eq(name, \"extractor\")")
    .traceFilter("and(eq(feedback_key, \"user_score\"), eq(feedback_score, 1))")
    .build();

列出具有 “star_rating” 键且分数大于 4 的运行

client.list_runs(
    project_name="<your_project>",
    filter='and(eq(feedback_key, "star_rating"), gt(feedback_score, 4))'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'and(eq(feedback_key, "star_rating"), gt(feedback_score, 4))'
})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(eq(feedback_key, \"star_rating\"), gt(feedback_score, 4))")
    .build();

列出完成耗时超过 5 秒的运行

client.list_runs(project_name="<your_project>", filter='gt(latency, "5s")')

client.listRuns({projectName: "<your_project>", filter: 'gt(latency, "5s")'})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("gt(latency, \"5s\")")
    .build();

列出所有状态不是 “error” 的运行

client.list_runs(project_name="<your_project>", filter='neq(status, "error")')

client.listRuns({projectName: "<your_project>", filter: 'neq(status, "error")'})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("neq(status, \"error\")")
    .build();

列出所有 `start_time` 大于特定时间戳的运行

client.list_runs(project_name="<your_project>", filter='gt(start_time, "2023-07-15T12:34:56Z")')

client.listRuns({projectName: "<your_project>", filter: 'gt(start_time, "2023-07-15T12:34:56Z")'})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("gt(start_time, \"2023-07-15T12:34:56Z\")")
    .build();

列出包含字符串 “substring” 的所有运行

client.list_runs(project_name="<your_project>", filter='search("substring")')

client.listRuns({projectName: "<your_project>", filter: 'search("substring")'})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("search(\"substring\")")
    .build();

列出所有带有 Git 哈希值 “2aa1cf4” 标签的运行

client.list_runs(project_name="<your_project>", filter='has(tags, "2aa1cf4")')

client.listRuns({projectName: "<your_project>", filter: 'has(tags, "2aa1cf4")'})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("has(tags, \"2aa1cf4\")")
    .build();

列出所有在特定时间戳之后开始，且状态不为错误或 “Correctness” 反馈分数等于 0 的运行

client.list_runs(
  project_name="<your_project>",
  filter='and(gt(start_time, "2023-07-15T12:34:56Z"), or(neq(status, "error"), and(eq(feedback_key, "Correctness"), eq(feedback_score, 0.0))))'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'and(gt(start_time, "2023-07-15T12:34:56Z"), or(neq(status, "error"), and(eq(feedback_key, "Correctness"), eq(feedback_score, 0.0))))'
})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(gt(start_time, \"2023-07-15T12:34:56Z\"), or(neq(status, \"error\"), and(eq(feedback_key, \"Correctness\"), eq(feedback_score, 0.0))))")
    .build();

复杂查询：列出标签包含 “experimental” 或 “beta” 且延迟超过 2 秒的所有运行

client.list_runs(
  project_name="<your_project>",
  filter='and(or(has(tags, "experimental"), has(tags, "beta")), gt(latency, 2))'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'and(or(has(tags, "experimental"), has(tags, "beta")), gt(latency, 2))'
})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(or(has(tags, 'experimental'), has(tags, 'beta')), gt(latency, 2))")
    .build();

通过全文搜索查找追踪树

你可以在不指定具体字段的情况下使用 search() 函数，对运行中的所有字符串字段执行全文搜索。这样可以快速找到匹配搜索词的追踪。

client.list_runs(
  project_name="<your_project>",
  filter='search("image classification")'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'search("image classification")'
})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("search(\"image classification\")")
    .build();

检查元数据是否存在

如果想检查元数据是否存在，可以使用 eq 运算符，也可以选择搭配 and 语句按值进行匹配。当你希望为运行记录更结构化的信息时，这种方式非常有用。

to_search = {
    "user_id": ""
}

# 检查是否存在带有 "user_id" 元数据键的运行
client.list_runs(
  project_name="default",
  filter="eq(metadata_key, 'user_id')"
)
# 检查 user_id=4070f233-f61e-44eb-bff1-da3c163895a3 的运行
client.list_runs(
  project_name="default",
  filter="and(eq(metadata_key, 'user_id'), eq(metadata_value, '4070f233-f61e-44eb-bff1-da3c163895a3'))"
)

// 检查是否存在带有 "user_id" 元数据键的运行
client.listRuns({
  projectName: 'default',
  filter: `eq(metadata_key, 'user_id')`
});
// 检查 user_id=4070f233-f61e-44eb-bff1-da3c163895a3 的运行
client.listRuns({
  projectName: 'default',
  filter: `and(eq(metadata_key, 'user_id'), eq(metadata_value, '4070f233-f61e-44eb-bff1-da3c163895a3'))`
});

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("eq(metadata_key, 'user_id')")
    .build();

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(eq(metadata_key, 'user_id'), eq(metadata_value, '4070f233-f61e-44eb-bff1-da3c163895a3'))")
    .build();

检查元数据中的环境详情

一种常见模式是通过元数据向追踪中添加环境信息。如果想筛选包含环境元数据的运行，可以使用与上面相同的模式：

client.list_runs(
  project_name="default",
  filter="and(eq(metadata_key, 'environment'), eq(metadata_value, 'production'))"
)

client.listRuns({
  projectName: 'default',
  filter: `and(eq(metadata_key, 'environment'), eq(metadata_value, 'production'))`
});

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(eq(metadata_key, 'environment'), eq(metadata_value, 'production'))")
    .build();

检查元数据中的线程 ID

将同一对话中的追踪关联起来的一种常见方式是使用共享线程 ID。如果想根据这种方式存储的线程 ID 筛选运行，可以在元数据中搜索该 ID。

client.list_runs(
  project_name="default",
  filter="and(eq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

client.listRuns({
  projectName: 'default',
  filter: `and(eq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(eq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

对键值对执行否定过滤

你可以对元数据、输入和输出键值对执行否定过滤，从结果中排除特定运行。下面是针对元数据键值对的示例，同样的逻辑也适用于输入和输出键值对。

# 查找所有元数据不包含 "thread_id" 键的运行
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'))"
)

# 查找所有元数据中的 thread_id 不是 "a1b2c3d4-e5f6-7890" 的运行
client.list_runs(
  project_name="default",
  filter="and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

# 查找所有不包含 "thread_id" 元数据键且不存在 "a1b2c3d4-e5f6-7890" 值的运行
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

# 查找所有不存在 thread_id 元数据键但存在 "a1b2c3d4-e5f6-7890" 值的运行
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

// 查找所有元数据不包含 "thread_id" 键的运行
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'))`
});

// 查找所有元数据中的 thread_id 不是 "a1b2c3d4-e5f6-7890" 的运行
client.listRuns({
  projectName: 'default',
  filter: `and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// 查找所有不包含 "thread_id" 元数据键且不存在 "a1b2c3d4-e5f6-7890" 值的运行
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// 查找所有不存在 thread_id 元数据键但存在 "a1b2c3d4-e5f6-7890" 值的运行
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// 查找所有元数据不包含 "thread_id" 键的运行
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(neq(metadata_key, 'thread_id'))")
    .build();

// 查找所有元数据中的 thread_id 不是 "a1b2c3d4-e5f6-7890" 的运行
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

// 查找所有不包含 "thread_id" 元数据键且不存在 "a1b2c3d4-e5f6-7890" 值的运行
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

// 查找所有不存在 thread_id 元数据键但存在 "a1b2c3d4-e5f6-7890" 值的运行
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(neq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

组合多个过滤条件

如果想组合多个条件来细化搜索，可以使用 and 运算符以及其他过滤函数。下面展示如何搜索名称为 “ChatOpenAI” 且元数据中包含特定 thread_id 的运行：

client.list_runs(
  project_name="default",
  filter="and(eq(name, 'ChatOpenAI'), eq(metadata_key, 'thread_id'), eq(metadata_value, '69b12c91-b1e2-46ce-91de-794c077e8151'))"
)

client.listRuns({
  projectName: 'default',
  filter: `and(eq(name, 'ChatOpenAI'), eq(metadata_key, 'thread_id'), eq(metadata_value, '69b12c91-b1e2-46ce-91de-794c077e8151'))`
});

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("and(eq(name, 'ChatOpenAI'), eq(metadata_key, 'thread_id'), eq(metadata_value, '69b12c91-b1e2-46ce-91de-794c077e8151'))")
    .build();

树过滤

列出所有名为 “RetrieveDocs” 的运行，这些运行的根运行具有分数为 1 的 “user_score” 反馈，并且完整追踪中的任意运行名为 “ExpandQuery”。如果希望根据追踪中是否达到各种状态或步骤，有条件地提取特定运行，这类查询非常有用。

client.list_runs(
    project_name="<your_project>",
    filter='eq(name, "RetrieveDocs")',
    trace_filter='and(eq(feedback_key, "user_score"), eq(feedback_score, 1))',
    tree_filter='eq(name, "ExpandQuery")'
)

client.listRuns({
  projectName: "<your_project>",
  filter: 'eq(name, "RetrieveDocs")',
  traceFilter: 'and(eq(feedback_key, "user_score"), eq(feedback_score, 1))',
  treeFilter: 'eq(name, "ExpandQuery")'
})

RunQueryParams runs = RunQueryParams.builder()
    .addSession("<your_project>")
    .filter("eq(name, \"RetrieveDocs\")")
    .traceFilter("and(eq(feedback_key, 'user_score'), eq(feedback_score, 1))")
    .treeFilter("eq(name, 'ExpandQuery')")
    .build();

使用子运行谓词查询追踪树

使用 trace_filter 匹配根运行中的字段，使用 tree_filter 匹配追踪树中任意运行的受支持可搜索字段。对于任意返回的子运行字段（例如嵌套的 inputs、outputs 或 extra 数据）上的谓词，请执行以下步骤：

使用 filter、trace_filter、tree_filter、run_type、元数据过滤器、parent_run_id 以及 ls_run_depth系统元数据键，在服务端缩小候选根追踪的范围。
在 Python 中调用 read_run(..., load_child_runs=True)，或在 TypeScript 中调用 readRun(..., { loadChildRuns: true })，为每个候选根追踪加载子运行。
在本地遍历已加载的 child_runs 树，并对无法作为服务端过滤字段使用的字段应用谓词。

以下示例（Python 0.8 和 JS 0.7）返回包含某个工具运行且该工具运行输出包含特定值的根追踪。服务端的 tree_filter 会将候选范围缩小到包含相关工具运行的追踪，而本地谓词则检查已加载的 outputs 数据。

from datetime import datetime, timedelta

from langsmith import Client

client = Client()
project_name = "<your_project>"


def iter_runs(run):
    yield run
    for child in run.child_runs or []:
        yield from iter_runs(child)


candidate_roots = client.list_runs(
    project_name=project_name,
    is_root=True,
    start_time=datetime.now() - timedelta(days=7),
    tree_filter='and(eq(run_type, "tool"), eq(name, "<tool_name>"))',
    select=["id"],
)

matching_roots = []
for candidate in candidate_roots:
    root = client.read_run(candidate.id, load_child_runs=True)
    has_matching_child = any(
        child.id != root.id
        and child.run_type == "tool"
        and child.name == "<tool_name>"
        and "<expected_value>" in str(child.outputs or {})
        for child in iter_runs(root)
    )
    if has_matching_child:
        matching_roots.append(root)

import { Client, Run } from "langsmith";

const client = new Client();
const projectName = "<your_project>";

function* iterRuns(run: Run): Generator<Run> {
  yield run;
  for (const child of run.child_runs ?? []) {
    yield* iterRuns(child);
  }
}

const candidateRoots: Run[] = [];
for await (const run of client.listRuns({
  projectName,
  isRoot: true,
  startTime: new Date(Date.now() - 1000 * 60 * 60 * 24 * 7),
  treeFilter: 'and(eq(run_type, "tool"), eq(name, "<tool_name>"))',
  select: ["id"],
})) {
  candidateRoots.push(run);
}

const matchingRoots: Run[] = [];
for (const candidate of candidateRoots) {
  const root = await client.readRun(candidate.id, { loadChildRuns: true });
  const hasMatchingChild = [...iterRuns(root)].some(
    (child) =>
      child.id !== root.id &&
      child.run_type === "tool" &&
      child.name === "<tool_name>" &&
      JSON.stringify(child.outputs ?? {}).includes("<expected_value>"),
  );
  if (hasMatchingChild) {
    matchingRoots.push(root);
  }
}

高级用法：导出包含子工具使用情况的扁平化追踪视图

以下 Python 示例演示如何导出扁平化的追踪视图，其中包括代理在每条追踪中使用的工具信息（来自嵌套运行）。你可以使用此方法分析代理在多条追踪中的行为。此示例查询指定天数内的所有工具运行，并根据其父运行（根运行）ID 对它们进行分组。随后，示例会获取每个根运行的相关信息，例如运行名称、输入和输出，并将这些信息与子运行信息合并。为了优化查询，示例：

查询工具运行时只选择必要字段，以减少查询时间。
在并发处理工具运行的同时，分批获取根运行。

from collections import defaultdict
from concurrent.futures import Future, ThreadPoolExecutor
from datetime import datetime, timedelta

from langsmith import Client
from tqdm.auto import tqdm

client = Client()
project_name = "my-project"
num_days = 30

# 列出所有工具运行
tool_runs = client.list_runs(
    project_name=project_name,
    start_time=datetime.now() - timedelta(days=num_days),
    run_type="tool",
    # 不需要获取 inputs、outputs 和其他可能增加查询时间的值
    select=["trace_id", "name", "run_type"],
)

data = []
futures: list[Future] = []
trace_cursor = 0
trace_batch_size = 50

tool_runs_by_parent = defaultdict(lambda: defaultdict(set))
# 不要超过速率限制
with ThreadPoolExecutor(max_workers=2) as executor:
    # 根据父运行 ID 对工具运行进行分组
    for run in tqdm(tool_runs):
        # 收集给定追踪中调用的所有工具
        tool_runs_by_parent[run.trace_id]["tools_involved"].add(run.name)
        # 可以将一批父运行 ID 发送到服务器
        # 这样可以分批查询根运行
        # 同时继续处理工具运行
        if len(tool_runs_by_parent) % trace_batch_size == 0:
            if this_batch := list(tool_runs_by_parent.keys())[
                trace_cursor : trace_cursor + trace_batch_size
            ]:
                trace_cursor += trace_batch_size
                futures.append(
                    executor.submit(
                        client.list_runs,
                        project_name=project_name,
                        run_ids=this_batch,
                        select=["name", "inputs", "outputs", "run_type"],
                    )
                )
    if this_batch := list(tool_runs_by_parent.keys())[trace_cursor:]:
        futures.append(
            executor.submit(
                client.list_runs,
                project_name=project_name,
                run_ids=this_batch,
                select=["name", "inputs", "outputs", "run_type"],
            )
        )

for future in tqdm(futures):
    root_runs = future.result()
    for root_run in root_runs:
        root_data = tool_runs_by_parent[root_run.id]
        data.append(
            {
                "run_id": root_run.id,
                "run_name": root_run.name,
                "run_type": root_run.run_type,
                "inputs": root_run.inputs,
                "outputs": root_run.outputs,
                "tools_involved": list(root_data["tools_involved"]),
            }
        )

# （可选）：转换为 pandas DataFrame
import pandas as pd

df = pd.DataFrame(data)
df.head()

高级用法：导出包含反馈的追踪中的检索器输入输出

如果希望根据检索器行为微调嵌入，或诊断端到端系统性能问题，此查询非常有用。以下 Python 示例演示如何导出具有特定反馈分数的追踪中的检索器输入和输出。

from collections import defaultdict
from concurrent.futures import Future, ThreadPoolExecutor
from datetime import datetime, timedelta

import pandas as pd
from langsmith import Client
from tqdm.auto import tqdm

client = Client()
project_name = "your-project-name"
num_days = 1

# 列出所有工具运行
retriever_runs = client.list_runs(
    project_name=project_name,
    start_time=datetime.now() - timedelta(days=num_days),
    run_type="retriever",
    # 这次需要获取 inputs 和 outputs，因为它们可能会被
    # 查询扩展步骤调整。
    select=["trace_id", "name", "run_type", "inputs", "outputs"],
    trace_filter='eq(feedback_key, "user_score")',
)

data = []
futures: list[Future] = []
trace_cursor = 0
trace_batch_size = 50

retriever_runs_by_parent = defaultdict(lambda: defaultdict(list))
# 不要超过速率限制
with ThreadPoolExecutor(max_workers=2) as executor:
    # 根据父运行 ID 对检索器运行进行分组
    for run in tqdm(retriever_runs):
        # 收集给定追踪中调用的所有检索器
        for k, v in run.inputs.items():
            retriever_runs_by_parent[run.trace_id][f"retriever.inputs.{k}"].append(v)
        for k, v in (run.outputs or {}).items():
            # 扩展文档
            retriever_runs_by_parent[run.trace_id][f"retriever.outputs.{k}"].extend(v)
        # 可以将一批父运行 ID 发送到服务器
        # 这样可以分批查询根运行
        # 同时继续处理检索器运行
        if len(retriever_runs_by_parent) % trace_batch_size == 0:
            if this_batch := list(retriever_runs_by_parent.keys())[
                trace_cursor : trace_cursor + trace_batch_size
            ]:
                trace_cursor += trace_batch_size
                futures.append(
                    executor.submit(
                        client.list_runs,
                        project_name=project_name,
                        run_ids=this_batch,
                        select=[
                            "name",
                            "inputs",
                            "outputs",
                            "run_type",
                            "feedback_stats",
                        ],
                    )
                )
    if this_batch := list(retriever_runs_by_parent.keys())[trace_cursor:]:
        futures.append(
            executor.submit(
                client.list_runs,
                project_name=project_name,
                run_ids=this_batch,
                select=["name", "inputs", "outputs", "run_type"],
            )
        )

for future in tqdm(futures):
    root_runs = future.result()
    for root_run in root_runs:
        root_data = retriever_runs_by_parent[root_run.id]
        feedback = {
            f"feedback.{k}": v.get("avg")
            for k, v in (root_run.feedback_stats or {}).items()
        }
        inputs = {f"inputs.{k}": v for k, v in root_run.inputs.items()}
        outputs = {f"outputs.{k}": v for k, v in (root_run.outputs or {}).items()}
        data.append(
            {
                "run_id": root_run.id,
                "run_name": root_run.name,
                **inputs,
                **outputs,
                **feedback,
                **root_data,
            }
        )

# （可选）：转换为 pandas DataFrame
import pandas as pd
df = pd.DataFrame(data)
df.head()

速率限制

POST /runs/query 端点（Python 中的 list_runs，JavaScript 中的 listRuns）具有每个租户的速率限制，具体限制取决于查询参数：

查询类型	限制	时间窗口
短时间窗口（≤ 7 天）	10 个请求	10 秒
大时间窗口（> 7 天）	3 个请求	10 秒
全文搜索，短时间窗口（≤ 7 天）	3 个请求	10 秒
全文搜索，大时间窗口（> 7 天）	1 个请求	10 秒
选择 `child_run_ids`，短时间窗口（≤ 7 天）	3 个请求	10 秒
选择 `child_run_ids`，大时间窗口（> 7 天）	1 个请求	10 秒

时间窗口由 end_time - start_time 确定。如果未提供 end_time，LangSmith 将使用当前时间。不包含 start_time 的查询会被视为大时间窗口查询。

最佳实践

为了避免触发速率限制并减少查询时间，尤其是对于输入输出较大的运行，请遵循以下建议：

设置 start_time：省略该参数会触发大时间窗口速率限制层级（每 10 秒 3 个请求，而不是 10 个）。如果可能，请使用 7 天或更短的时间窗口。
使用 select：默认情况下会返回所有字段。只指定所需字段（例如 select=["inputs", "outputs"]）可以显著减少响应大小和查询时间，尤其适用于输入输出较大的运行。
设置 limit：如果不需要分页获取全部结果，请限制结果数量。
避免全文搜索：filter='search("...")' 具有最严格的速率限制；请尽可能使用结构化过滤器（例如 eq()、has()）。
避免选择 child_run_ids：选择该字段同样会触发更严格的速率限制层级。

超过这些限制时，API 会返回 429 Too Many Requests 响应。有关速率限制的一般信息，请参阅管理概览。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Viewing & managing traces

Bulk export trace data

Messages view

Data type reference

使用过滤参数

列出项目中的所有运行

列出过去 24 小时内的 LLM 和聊天运行

列出项目中的根运行

列出没有错误的运行

按运行 ID 列出运行

按 ID 获取单个运行

使用过滤查询语言

列出对话线程中的所有根运行

列出所有名为 “extractor” 且其追踪根运行被赋予 “user_score” 反馈、分数为 1 的运行

列出具有 “star_rating” 键且分数大于 4 的运行

列出完成耗时超过 5 秒的运行

列出所有状态不是 “error” 的运行

列出所有 `start_time` 大于特定时间戳的运行

列出包含字符串 “substring” 的所有运行

列出所有带有 Git 哈希值 “2aa1cf4” 标签的运行

列出所有在特定时间戳之后开始，且状态不为错误或 “Correctness” 反馈分数等于 0 的运行

复杂查询：列出标签包含 “experimental” 或 “beta” 且延迟超过 2 秒的所有运行

通过全文搜索查找追踪树

检查元数据是否存在

检查元数据中的环境详情

检查元数据中的线程 ID

对键值对执行否定过滤

组合多个过滤条件

树过滤

使用子运行谓词查询追踪树

高级用法：导出包含子工具使用情况的扁平化追踪视图

高级用法：导出包含反馈的追踪中的检索器输入输出

速率限制

最佳实践

​使用过滤参数

​列出项目中的所有运行

​列出过去 24 小时内的 LLM 和聊天运行

​列出项目中的根运行

​列出没有错误的运行

​按运行 ID 列出运行

​按 ID 获取单个运行

​使用过滤查询语言

​列出对话线程中的所有根运行

​列出所有名为 “extractor” 且其追踪根运行被赋予 “user_score” 反馈、分数为 1 的运行

​列出具有 “star_rating” 键且分数大于 4 的运行

​列出完成耗时超过 5 秒的运行

​列出所有状态不是 “error” 的运行

​列出所有 start_time 大于特定时间戳的运行

​列出包含字符串 “substring” 的所有运行

​列出所有带有 Git 哈希值 “2aa1cf4” 标签的运行

​列出所有在特定时间戳之后开始，且状态不为错误或 “Correctness” 反馈分数等于 0 的运行

​复杂查询：列出标签包含 “experimental” 或 “beta” 且延迟超过 2 秒的所有运行

​通过全文搜索查找追踪树

​检查元数据是否存在

​检查元数据中的环境详情

​检查元数据中的线程 ID

​对键值对执行否定过滤

​组合多个过滤条件

​树过滤

​使用子运行谓词查询追踪树

​高级用法：导出包含子工具使用情况的扁平化追踪视图

​高级用法：导出包含反馈的追踪中的检索器输入输出

​速率限制

​最佳实践

使用过滤参数

列出项目中的所有运行

列出过去 24 小时内的 LLM 和聊天运行

列出项目中的根运行

列出没有错误的运行

按运行 ID 列出运行

按 ID 获取单个运行

使用过滤查询语言

列出对话线程中的所有根运行

列出所有名为 “extractor” 且其追踪根运行被赋予 “user_score” 反馈、分数为 1 的运行

列出具有 “star_rating” 键且分数大于 4 的运行

列出完成耗时超过 5 秒的运行

列出所有状态不是 “error” 的运行

列出所有 `start_time` 大于特定时间戳的运行

列出包含字符串 “substring” 的所有运行

列出所有带有 Git 哈希值 “2aa1cf4” 标签的运行

列出所有在特定时间戳之后开始，且状态不为错误或 “Correctness” 反馈分数等于 0 的运行

复杂查询：列出标签包含 “experimental” 或 “beta” 且延迟超过 2 秒的所有运行

通过全文搜索查找追踪树

检查元数据是否存在

检查元数据中的环境详情

检查元数据中的线程 ID

对键值对执行否定过滤

组合多个过滤条件

树过滤

使用子运行谓词查询追踪树

高级用法：导出包含子工具使用情况的扁平化追踪视图

高级用法：导出包含反馈的追踪中的检索器输入输出

速率限制

最佳实践