requests 库通过 REST API 运行评估,但同样的原理适用于任何语言。
在深入了解本指南内容之前,阅读以下内容可能会有所帮助:
- 评估 LLM 应用。
- LangSmith API 参考:本指南中使用的所有端点的完整 API 文档。
创建数据集
在此示例中,我们使用 Python SDK 快速创建一个数据集。要通过 API 或 UI 创建数据集,请参阅管理数据集。import os
import requests
from datetime import datetime
from langsmith import Client
from openai import OpenAI
from uuid import uuid4
client = Client()
oa_client = OpenAI()
# 创建数据集
examples = [
{
"inputs": {"text": "Shut up, idiot"},
"outputs": {"label": "Toxic"},
},
{
"inputs": {"text": "You're a wonderful person"},
"outputs": {"label": "Not toxic"},
},
{
"inputs": {"text": "This is the worst thing ever"},
"outputs": {"label": "Toxic"},
},
{
"inputs": {"text": "I had a great day today"},
"outputs": {"label": "Not toxic"},
},
{
"inputs": {"text": "Nobody likes you"},
"outputs": {"label": "Toxic"},
},
{
"inputs": {"text": "This is unacceptable. I want to speak to the manager."},
"outputs": {"label": "Not toxic"},
},
]
dataset_name = "Toxic Queries - API Example"
dataset = client.create_dataset(dataset_name=dataset_name)
client.create_examples(dataset_id=dataset.id, examples=examples)
运行单个实验
要通过 API 运行实验,你需要:- 从数据集中获取示例。
- 创建一个实验(在 API 中也称为“会话”)。
- 为每个示例创建引用了该示例和该实验的运行。
- 通过设置
end_time来结束实验。
/examples 端点拉取你想要在实验中使用的所有示例:
# 选择一个数据集 ID。在此示例中,我们使用的是上面创建的数据集。
# API 参考:https://docs.langchain.com/langsmith/smith-api/examples/read-examples
dataset_id = dataset.id
params = { "dataset": dataset_id }
resp = requests.get(
"https://api.smith.langchain.com/api/v1/examples",
params=params,
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
examples = resp.json()
- 通过 POST 请求到
/runs来创建运行对象,并设置reference_example_id和session_id。 - 跟踪运行之间的父子关系(例如,一个父级的 “chain” 运行包含一个子级的 “llm” 运行)。
- 通过 PATCH 请求到
/runs/{run_id}来更新运行的输出。
os.environ["OPENAI_API_KEY"] = "sk-..."
def run_completion_on_example(example, model_name, experiment_id):
"""在示例列表上运行补全。"""
# 我们在此使用 OpenAI API,但你可以使用任何你喜欢的模型
def _post_run(run_id, name, run_type, inputs, parent_id=None):
"""向 API 发布新运行的函数。
API 参考:https://docs.langchain.com/langsmith/smith-api/runs/create-a-run
"""
data = {
"id": run_id.hex,
"name": name,
"run_type": run_type,
"inputs": inputs,
"start_time": datetime.utcnow().isoformat(),
"reference_example_id": example["id"],
"session_id": experiment_id,
}
if parent_id:
data["parent_run_id"] = parent_id.hex
resp = requests.post(
"https://api.smith.langchain.com/api/v1/runs", # 请为自托管安装或区域 SaaS 进行相应更新
json=data,
headers=headers
)
resp.raise_for_status()
def _patch_run(run_id, outputs):
"""用输出修补运行的函数。
API 参考:https://docs.langchain.com/langsmith/smith-api/run/update-run
"""
resp = requests.patch(
f"https://api.smith.langchain.com/api/v1/runs/{run_id}",
json={
"outputs": outputs,
"end_time": datetime.utcnow().isoformat(),
},
headers=headers,
)
resp.raise_for_status()
# 在请求头中发送你的 API 密钥
headers = {"x-api-key": os.environ["LANGSMITH_API_KEY"]}
text = example["inputs"]["text"]
messages = [
{
"role": "system",
"content": "Please review the user query below and determine if it contains any form of toxic behavior, such as insults, threats, or highly negative comments. Respond with 'Toxic' if it does, and 'Not toxic' if it doesn't.",
},
{"role": "user", "content": text},
]
# 创建父运行
parent_run_id = uuid7()
_post_run(parent_run_id, "LLM Pipeline", "chain", {"text": text})
# 创建子运行
child_run_id = uuid7()
_post_run(child_run_id, "OpenAI Call", "llm", {"messages": messages}, parent_run_id)
# 生成补全
chat_completion = oa_client.chat.completions.create(model=model_name, messages=messages)
output_text = chat_completion.choices[0].message.content
# 结束运行
_patch_run(child_run_id, {
"messages": messages,
"output": output_text,
"model": model_name
})
_patch_run(parent_run_id, {"label": output_text})
reference_dataset_id 引用一个数据集。与常规追踪的主要区别在于,实验中的运行必须具有 reference_example_id,将每个运行链接到数据集中的特定示例。
# 使用 /sessions 端点创建新实验
# 实验是一组带有对所用数据集引用的运行集合
# API 参考:https://docs.langchain.com/langsmith/smith-api/tracer-sessions/create-tracer-session
model_names = ("gpt-3.5-turbo", "gpt-5.4-mini")
experiment_ids = []
for model_name in model_names:
resp = requests.post(
"https://api.smith.langchain.com/api/v1/sessions",
json={
"start_time": datetime.utcnow().isoformat(),
"reference_dataset_id": str(dataset_id),
"description": "实验的可选描述",
"name": f"Toxicity detection - API Example - {model_name} - {str(uuid4())[0:8]}", # 实验的名称
"extra": {
"metadata": {"foo": "bar"}, # 可选元数据
},
},
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
experiment = resp.json()
experiment_ids.append(experiment["id"])
# 在所有示例上运行补全
for example in examples:
run_completion_on_example(example, model_name, experiment["id"])
# 发出 PATCH 请求,通过更新 end_time 来“结束”实验
requests.patch(
f"https://api.smith.langchain.com/api/v1/sessions/{experiment['id']}",
json={"end_time": datetime.utcnow().isoformat()},
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
添加评估反馈
运行完你的实验后,你通常会希望通过添加反馈分数来评估结果。这允许你跟踪诸如正确性、准确性或任何自定义评估标准之类的指标。 在此示例中,评估检查每个模型的输出是否与数据集中的预期标签匹配。代码发布一个“正确性”分数(正确为 1.0,不正确为 0.0)来跟踪每个模型分类有毒与无毒文本的准确程度。 以下代码向单个实验示例中的运行添加反馈:# 从其中一个实验中获取运行
# API 参考:https://docs.langchain.com/langsmith/smith-api/run/query-runs
experiment_id = experiment_ids[0] # 评估第一个实验
runs_resp = requests.post(
"https://api.smith.langchain.com/api/v1/runs/query",
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]},
json={
"session": [experiment_id],
"is_root": True, # 仅获取根运行
"select": ["id", "reference_example_id", "outputs", "session_id"],
}
)
runs = runs_resp.json()["runs"]
# 通过比较输出与预期值来评估每个运行
for run in runs:
# 从原始示例获取预期输出
example_id = run["reference_example_id"]
expected_output = next(
ex["outputs"]["label"]
for ex in examples
if ex["id"] == example_id
)
# 将模型输出与预期输出进行比较
actual_output = run["outputs"].get("label", "")
is_correct = expected_output.lower() == actual_output.lower()
# 发布反馈分数
# API 参考:https://docs.langchain.com/langsmith/smith-api/feedback/create-feedback
feedback = {
"run_id": str(run["id"]),
"key": "correctness", # 你的评估指标的名称
"score": 1.0 if is_correct else 0.0,
"session_id": run["session_id"], # 必需:运行的追踪项目 UUID
"comment": f"Expected: {expected_output}, Got: {actual_output}", # 可选
}
resp = requests.post(
"https://api.smith.langchain.com/api/v1/feedback",
json=feedback,
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
resp.raise_for_status()
运行对比实验
接下来,我们将演示如何运行对比实验。在对比实验中,你将两个示例相互比较。 更多信息,请查看如何运行对比评估。# 比较实验允许你对两个或多个实验的输出提供偏好排序
# API 参考:https://docs.langchain.com/langsmith/smith-api/datasets/create-comparative-experiment
resp = requests.post(
"https://api.smith.langchain.com/api/v1/datasets/comparative",
json={
"experiment_ids": experiment_ids,
"name": "Toxicity detection - API Example - Comparative - " + str(uuid4())[0:8],
"description": "对比实验的可选描述",
"extra": {
"metadata": {"foo": "bar"}, # 可选元数据
},
"reference_dataset_id": str(dataset_id),
},
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
comparative_experiment = resp.json()
comparative_experiment_id = comparative_experiment["id"]
# 你可以遍历属于对比实验的实验中的运行,并对输出进行偏好排序
# 获取对比实验
resp = requests.get(
f"https://api.smith.langchain.com/api/v1/datasets/{str(dataset_id)}/comparative",
params={"id": comparative_experiment_id},
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
comparative_experiment = resp.json()[0]
experiment_ids = [info["id"] for info in comparative_experiment["experiments_info"]]
from collections import defaultdict
example_id_to_runs_map = defaultdict(list)
# API 参考:https://docs.langchain.com/langsmith/smith-api/run/query-runs
runs = requests.post(
f"https://api.smith.langchain.com/api/v1/runs/query",
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]},
json={
"session": experiment_ids,
"is_root": True, # 仅获取根运行(跨度),这些运行包含最终输出
"select": ["id", "reference_example_id", "outputs", "session_id"],
}
).json()
runs = runs["runs"]
for run in runs:
example_id = run["reference_example_id"]
example_id_to_runs_map[example_id].append(run)
for example_id, runs in example_id_to_runs_map.items():
print(f"Example ID: {example_id}")
# 对输出进行偏好排序,在本例中我们始终偏好第一个输出
# 实际上,你可以使用 LLM 来对输出进行排序
feedback_group_id = uuid4()
# 为每个运行发布反馈分数,第一个运行是首选的
# API 参考:https://docs.langchain.com/langsmith/smith-api/feedback/create-feedback
# 我们将使用反馈组 ID 将反馈分数关联到同一组
for i, run in enumerate(runs):
print(f"Run ID: {run['id']}")
feedback = {
"score": 1 if i == 0 else 0,
"run_id": str(run["id"]),
"key": "ranked_preference",
"session_id": run["session_id"], # 必需:运行的追踪项目 UUID
"feedback_group_id": str(feedback_group_id),
"comparative_experiment_id": comparative_experiment_id,
}
resp = requests.post(
"https://api.smith.langchain.com/api/v1/feedback",
json=feedback,
headers={"x-api-key": os.environ["LANGSMITH_API_KEY"]}
)
resp.raise_for_status()
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

