用 SDK 查询追踪

查询运行（LangSmith 追踪中的 span 数据）的推荐方式，是用 SDK 中的 list_runs 方法或 API 中的 /runs/query 端点。LangSmith 用一种简单格式存储追踪，该格式在运行（span）数据格式中规定。

本页涵盖：

使用过滤参数：用 SDK 参数做基于关键字的过滤。
使用过滤查询语言：用 LangSmith 的过滤语法做复杂查询。
用子运行谓词查询追踪树：把服务端收窄与本地子运行遍历结合。
速率限制：按租户的限制以及待在其内的最佳实践。

NOTE

如果你想要导出大量追踪，我们推荐你用批量数据导出功能，因为它能更好地处理大数据量，并会支持自动重试和跨分区并行。

使用过滤参数

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

WARNING

前置条件

在运行下面的代码片段之前初始化客户端。

PythonTypeScriptJava

python

from langsmith import Client

client = Client()

typescript

import { Client, Run } from "langsmith";

const client = new Client();

java

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

LangsmithClient client = LangsmithOkHttpClient.fromEnv();

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

列出项目中所有运行

PythonTypeScriptJava

python

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

typescript

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

java

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

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

列出最近 24 小时的 LLM 和 chat 运行

PythonTypeScriptJava

python

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

typescript

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

java

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 值。你可以用它来过滤根运行。

PythonTypeScriptJava

python

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

typescript

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

java

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

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

列出无错误的运行

PythonTypeScriptJava

python

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

typescript

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

java

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

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

按运行 ID 列出运行

WARNING

忽略其他参数

如果你按上面描述的方式提供一个运行 ID 列表，它会忽略所有其他过滤参数，如 project_name、run_type 等，而直接返回匹配所给 ID 的运行。

如果你有一个运行 ID 列表，你可以直接列出它们：

PythonTypeScriptJava

python

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

typescript

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

java

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

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

按 ID 获取单个运行

要按 ID 获取单个运行（追踪），用 read_run 方法。当你有一个特定的追踪 ID（例如从一个形如 https://smith.langchain.com/public/<trace-id>/r 的 LangSmith 分享链接）并想取回它的完整数据时，这很有用。

PythonTypeScriptJava

python

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)

typescript

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

java

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

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

TIP

用 LangGraph 在本地回放追踪

如果你在使用带检查点的 LangGraph，你可以从 LangSmith 取回一条追踪并在本地回放以调试。有关从检查点恢复执行的详情，见 LangGraph 的时间旅行与回放文档。

使用过滤查询语言

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

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

这是获取会话线程中运行的方式。有关设置线程的更多信息，参阅我们的设置线程 how-to 指南。线程通过设置一个共享的线程 ID 来分组。LangSmith UI 让你用以下两个元数据键之一：session_id 或 thread_id。会话 ID 也被称为追踪项目 ID。下面的查询匹配其中任一个。

PythonTypeScriptJava

python

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
)

typescript

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

java

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 的运行

PythonTypeScriptJava

python

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

typescript

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

java

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 的运行

PythonTypeScriptJava

python

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

typescript

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

java

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

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

PythonTypeScriptJava

python

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

typescript

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

java

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

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

PythonTypeScriptJava

python

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

typescript

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

java

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

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

PythonTypeScriptJava

python

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

typescript

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

java

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

列出含字符串 "substring" 的所有运行

PythonTypeScriptJava

python

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

typescript

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

java

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

列出被标记为 git hash "2aa1cf4" 的所有运行

PythonTypeScriptJava

python

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

typescript

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

java

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

列出在一个特定时间戳之后开始、且要么有非错误状态要么有一个 "Correctness" 反馈分数等于 0 的所有运行

PythonTypeScriptJava

python

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))))'
)

typescript

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

java

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

复杂查询：列出 tags 含 "experimental" 或 "beta" 且延迟大于 2 秒的所有运行

PythonTypeScriptJava

python

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

typescript

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

java

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

用全文搜索追踪树

你可以用 search() 函数而不带任何特定字段，以在一次运行的所有字符串字段上做全文搜索。这让你能快速找到匹配一个搜索词的追踪。

PythonTypeScriptJava

python

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

typescript

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

java

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

检查元数据的存在

如果你想检查元数据的存在，你可以用 eq 运算符，可选地带一个 and 语句按值匹配。当你想记录关于你运行的更结构化信息时这很有用。

PythonTypeScriptJava

python

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'))"
)

typescript

// 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'))`
});

java

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

检查元数据中的环境详情

一个常见模式是通过元数据把你的环境信息添加到你的追踪。如果你想过滤含环境元数据的运行，你可以用与上面相同的模式：

PythonTypeScriptJava

python

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

typescript

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

java

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

检查元数据中的线程 ID

把同一会话中的追踪关联起来的一个常见方式是使用一个共享的线程 ID。如果你想以这种方式基于一个线程 ID 过滤运行，你可以在元数据中搜索那个 ID。

PythonTypeScriptJava

python

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

typescript

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

java

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

对键值对做负向过滤

你可以对元数据、输入和输出的键值对使用负向过滤，以从你的结果中排除特定运行。这里有一些针对元数据键值对的例子，同样的逻辑适用于输入和输出的键值对。

PythonTypeScriptJava

python

# 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'))"
)

typescript

// 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'))`
});

java

// 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 的运行：

PythonTypeScriptJava

python

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

typescript

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

java

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" 的运行。

如果你想基于追踪内达到的各种状态或步骤来提取一个特定运行，这类查询很有用。

PythonTypeScriptJava

python

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")'
)

typescript

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

java

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 载荷。

PythonTypeScript

python

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)

typescript

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 例子演示了如何导出追踪的扁平视图，包括关于每条追踪内 agent 所用工具（来自嵌套运行）的信息。这可用于跨多条追踪分析你 agent 的行为。

这个例子查询指定天数内的所有工具运行，并按其父（根）运行 ID 分组。然后它为每个根运行获取相关信息，例如运行名、输入、输出，并把那信息与子运行信息结合。

为优化查询，这个例子：

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

Python

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 = "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
import pandas as pd

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

高级：为含反馈的追踪导出检索器 IO

如果你想基于检索器行为微调嵌入，或诊断端到端系统性能问题，这个查询很有用。下面的 Python 例子演示了如何在含一个特定反馈分数的追踪内导出检索器的输入和输出。

Python

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

# 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
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 响应。一般速率限制信息，参阅管理概览。

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织