使用 SDK 查询追踪 - 以编程方式使用文档

推荐的查询运行（LangSmith 追踪中的 span 数据）的方式是使用 list_runs 中的 SDK or /runs/query 端点。LangSmith 以简单格式存储追踪，具体格式在 API中定义。运行（span）数据格式.

本页面涵盖：

- 使用过滤参数：使用 SDK 参数进行基于关键字的过滤。
- 使用过滤查询语言：使用 LangSmith 的过滤语法进行复杂查询。
- 使用子运行谓词查询追踪树：将服务端筛选与本地子运行遍历相结合。
- 速率限制：每个租户的限流限制以及保持在限制范围内的最佳实践。

使用过滤参数

对于简单查询，您无需依赖我们的查询语法。您可以使用过滤参数参考.

from langsmith import Client

client = Client()

const client = new Client();

LangsmithClient client = LangsmithOkHttpClient.fromEnv();

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

列出项目中的所有运行

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

// Download runs in a project
const projectRuns: Run[] = [];
for await (const run of client.listRuns({
  projectName: "<your_project>",
})) {
  projectRuns.push(run);
};

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();

列出项目中的根运行

根运行是没有父级的运行。这些会被分配一个值 True 作为 is_root。您可以用它来过滤根运行。

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);
};

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);
};

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

按运行 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);
};

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)

# Access run data
print(run.inputs)
print(run.outputs)
print(run.name)

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

// Access run data
console.log(run.inputs);
console.log(run.outputs);
console.log(run.name);

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

使用筛选查询语言

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

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

这是获取对话线程中运行的方法。有关设置线程的更多信息，请参阅我们的设置线程的操作指南. 线程通过设置共享线程ID进行分组。LangSmith UI允许您使用以下任一元数据键： session_id or 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);
};

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();

列出所有开始_时间大于特定时间戳的运行

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": ""
}

# Check for any run with the "user_id" metadata key
client.list_runs(
  project_name="default",
  filter="eq(metadata_key, 'user_id')"
)
# Check for runs with 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'))"
)

// Check for any run with the "user_id" metadata key
client.listRuns({
  projectName: 'default',
  filter: `eq(metadata_key, 'user_id')`
});
// Check for runs with 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();

键值对上的负向筛选

您可以对元数据、输入和输出键值对使用负向筛选，从结果中排除特定运行。以下是元数据键值对的一些示例，但相同的逻辑也适用于输入和输出键值对。

# Find all runs where the metadata does not contain a "thread_id" key
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'))"
)

# Find all runs where the thread_id in metadata is not "a1b2c3d4-e5f6-7890"
client.list_runs(
  project_name="default",
  filter="and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

# Find all runs where there is no "thread_id" metadata key and the "a1b2c3d4-e5f6-7890" value is not present
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

# Find all runs where the thread_id metadata key is not present but the "a1b2c3d4-e5f6-7890" value is present
client.list_runs(
  project_name="default",
  filter="and(neq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))"
)

// Find all runs where the metadata does not contain a "thread_id" key
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'))`
});

// Find all runs where the thread_id in metadata is not "a1b2c3d4-e5f6-7890"
client.listRuns({
  projectName: 'default',
  filter: `and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// Find all runs where there is no "thread_id" metadata key and the "a1b2c3d4-e5f6-7890" value is not present
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// Find all runs where the thread_id metadata key is not present but the "a1b2c3d4-e5f6-7890" value is present
client.listRuns({
  projectName: 'default',
  filter: `and(neq(metadata_key, 'thread_id'), eq(metadata_value, 'a1b2c3d4-e5f6-7890'))`
});

// Find all runs where the metadata does not contain a "thread_id" key
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(neq(metadata_key, 'thread_id'))")
    .build();

// Find all runs where the thread_id in metadata is not "a1b2c3d4-e5f6-7890"
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(eq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

// Find all runs where there is no "thread_id" metadata key and the "a1b2c3d4-e5f6-7890" value is not present
RunQueryParams runs = RunQueryParams.builder()
    .addSession("default")
    .filter("and(neq(metadata_key, 'thread_id'), neq(metadata_value, 'a1b2c3d4-e5f6-7890'))")
    .build();

// Find all runs where the thread_id metadata key is not present but the "a1b2c3d4-e5f6-7890" value is present
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" 且其根运行具有 "user_score" 反馈为 1 且完整跟踪中任何运行名为 "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, or extra 负载，请使用以下步骤：

使用 filter, trace_filter, tree_filter, run_type、元数据过滤器、 parent_run_id和 ls_run_depth 系统元数据键在服务端缩小候选根跟踪范围.
通过调用 read_run(..., load_child_runs=True) 在 Python 中或 readRun(..., { loadChildRuns: true }) 在 TypeScript 中用子运行填充每个候选根跟踪
在本地遍历已填充的 child_runs 树，并将您的谓词应用到不可作为服务端过滤字段的字段上。

以下示例（Python 0.8 和 JS 0.7）返回包含其输出包含特定值的工具运行的根跟踪。服务端 tree_filter 将候选项缩小到包含相关工具运行的跟踪，本地谓词检查已填充的 outputs payload.

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)

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

function* iterRuns(run: Run): Generator {
  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

# List all tool runs
tool_runs = client.list_runs(
    project_name=project_name,
    start_time=datetime.now() - timedelta(days=num_days),
    run_type="tool",
    # We don't need to fetch inputs, outputs, and other values that # may increase the query time
    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))
# Do not exceed rate limit
with ThreadPoolExecutor(max_workers=2) as executor:
    # Group tool runs by parent run ID
    for run in tqdm(tool_runs):
        # Collect all tools invoked within a given trace
        tool_runs_by_parent[run.trace_id]["tools_involved"].add(run.name)
        # maybe send a batch of parent run IDs to the server
        # this lets us query for the root runs in batches
        # while still processing the tool runs
        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"]),
            }
        )

# (Optional): Convert to a pandas DataFrame


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

高级：为有反馈的跟踪导出检索器 IO

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

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 = "your-project-name"
num_days = 1

# List all tool runs
retriever_runs = client.list_runs(
    project_name=project_name,
    start_time=datetime.now() - timedelta(days=num_days),
    run_type="retriever",
    # This time we do want to fetch the inputs and outputs, since they
    # may be adjusted by query expansion steps.
    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))
# Do not exceed rate limit
with ThreadPoolExecutor(max_workers=2) as executor:
    # Group retriever runs by parent run ID
    for run in tqdm(retriever_runs):
        # Collect all retriever calls invoked within a given trace
        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():
            # Extend the docs
            retriever_runs_by_parent[run.trace_id][f"retriever.outputs.{k}"].extend(v)
        # maybe send a batch of parent run IDs to the server
        # this lets us query for the root runs in batches
        # while still processing the retriever runs
        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,
            }
        )

# (Optional): Convert to a pandas DataFrame

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

速率限制

POST /runs/query 端点（list_runs 在 Python 中， listRuns 在 JavaScript 中）具有基于查询参数而变化的多租户速率限制：

查询类型	限制	窗口
短时间窗口（≤ 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. If end_time 未提供时，LangSmith 将使用当前时间。未指定 start_time 的查询被视为长时间窗口查询。

最佳实践

To avoid hitting rate limits and reduce query time, especially for runs with large inputs/outputs:

- **设置 start_time**：省略此参数会触发长时间窗口速率限制层级（每 10 秒 3 个请求，而非 10 个）。请尽可能使用 7 天或更短的时间窗口。
- **使用 select**：默认情况下返回所有字段。仅指定您需要的字段（例如 select=["inputs", "outputs"]) substantially reduces response size and query time, especially for runs with large inputs/outputs.
- **设置 limit**：如果您不需要分页遍历所有结果，请限制返回数量。
避免全文搜索: filter='search("...")' 具有最严格的速率限制；请尽可能使用结构化过滤器（例如 eq(), has()）。
- **避免选择 child_run_ids**：这也会触发更严格的速率限制层级。

当您超过这些限制时，API 会返回 429 Too Many Requests 响应。如需常规速率限制信息，请参阅管理概述.