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://api.smith.langchain.com/redoc#tag/examples/operation/read_examples_api_v1_examples_get
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。 - 跟踪运行之间的父子关系(例如,包含子“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://api.smith.langchain.com/redoc#tag/run/operation/create_run_api_v1_runs_post
"""
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", # 对于自托管安装或欧盟区域,请相应更新
json=data,
headers=headers
)
resp.raise_for_status()
def _patch_run(run_id, outputs):
"""使用输出更新运行的函数。
API 参考:https://api.smith.langchain.com/redoc#tag/run/operation/update_run_api_v1_runs__run_id__patch
"""
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://api.smith.langchain.com/redoc#tag/tracer-sessions/operation/create_tracer_session_api_v1_sessions_post
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": "An optional description for the experiment",
"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"])
# 发送补丁请求以通过更新 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://api.smith.langchain.com/redoc#tag/run/operation/query_runs_api_v1_runs_query_post
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"],
}
)
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://api.smith.langchain.com/redoc#tag/feedback/operation/create_feedback_api_v1_feedback_post
feedback = {
"run_id": str(run["id"]),
"key": "correctness", # 您的评估指标名称
"score": 1.0 if is_correct else 0.0,
"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://api.smith.langchain.com/redoc#tag/datasets/operation/create_comparative_experiment_api_v1_datasets_comparative_post
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": "An optional description for the comparative experiment",
"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://api.smith.langchain.com/redoc#tag/run/operation/query_runs_api_v1_runs_query_post
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"],
}
).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://api.smith.langchain.com/redoc#tag/feedback/operation/create_feedback_api_v1_feedback_post
# 我们将使用反馈组 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",
"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()
将这些文档连接到 Claude、VSCode 等,通过 MCP 获取实时答案。

