LangSmith源码学习路线：第 20 篇 Trace、Evaluation 与 Dataset 闭环源码解剖#

1
from langsmith import Client, traceable
2
from langchain_core.runnables import RunnableConfig
3

4
client = Client()
5

6
@traceable(name="answer_question", run_type="chain")
7
def answer_question(inputs: dict) -> dict:
8
    ...

1
# 简化伪代码，不是源码逐字复制
2
def prepare_langsmith_tracing(user_config: dict | None):
3
    env = {
4
        "api_url": read_env("LANGSMITH_ENDPOINT"),
5
        "api_key": read_env("LANGSMITH_API_KEY"),
6
        "project": read_env("LANGSMITH_PROJECT"),
7
        "tracing_enabled": read_env("LANGSMITH_TRACING"),
8
    }
9

10
    client = Client(
11
        api_url=user_config.get("api_url", env["api_url"]),
12
        api_key=user_config.get("api_key", env["api_key"]),
13
    )
14

15
    callback_handler = LangSmithCallbackHandler(
16
        client=client,
17
        project_name=user_config.get("project_name", env["project"]),
18
    )
19

20
    runnable_config = RunnableConfig(
21
        callbacks=[callback_handler, *user_config.get("callbacks", [])],
22
        tags=user_config.get("tags", []),
23
        metadata=user_config.get("metadata", {}),
24
        run_name=user_config.get("run_name"),
25
    )
26

27
    return client, runnable_config

1
client.create_dataset(dataset_name="qa-regression")
2
client.create_examples(
3
    inputs=[{"question": "..."}],
4
    outputs=[{"answer": "..."}],
5
    dataset_id=dataset.id,
6
)

1
# 简化伪代码，不是源码逐字复制
2
def build_regression_dataset(client: Client, dataset_name: str, rows: list[dict]):
3
    dataset = client.create_dataset(
4
        dataset_name=dataset_name,
5
        description="Questions promoted from production traces",
6
    )
7

8
    example_inputs = []
9
    reference_outputs = []
10
    example_metadata = []
11

12
    for row in rows:
13
        example_inputs.append({"question": row["question"]})
14
        reference_outputs.append({"answer": row["expected_answer"]})
15
        example_metadata.append({
16
            "source_run_id": row.get("run_id"),
17
            "failure_type": row.get("failure_type"),
18
            "added_by": row.get("reviewer", "system"),
19
        })
20

21
    examples = client.create_examples(
22
        dataset_id=dataset.id,
23
        inputs=example_inputs,
24
        outputs=reference_outputs,
25
        metadata=example_metadata,
26
    )
27

28
    return dataset, examples

1
def correctness_evaluator(run, example) -> dict:
2
    return {"key": "correctness", "score": 1, "comment": "..."}

1
# 简化伪代码，不是源码逐字复制
2
def build_correctness_evaluator(model):
3
    def correctness(run, example):
4
        prediction = run.outputs.get("answer")
5
        reference = example.outputs.get("answer")
6

7
        if prediction is None:
8
            return {
9
                "key": "correctness",
10
                "score": 0,
11
                "comment": "target did not return answer",
12
            }
13

14
        judge_result = model.invoke({
15
            "prediction": prediction,
16
            "reference": reference,
17
            "question": example.inputs.get("question"),
18
        })
19

20
        return {
21
            "key": "correctness",
22
            "score": judge_result["score"],
23
            "comment": judge_result.get("reasoning"),
24
            "metadata": {"judge_model": model.name},
25
        }
26

27
    return correctness

1
with tracing_context(project_name="qa-prod"):
2
    result = app.invoke(inputs, config={"tags": ["prod"]})

1
# 简化伪代码，不是源码逐字复制
2
def run_traced_runnable(runnable, inputs, config):
3
    callback_manager = configure_callbacks(config.callbacks)
4
    root_run_id = new_uuid()
5

6
    callback_manager.on_chain_start(
7
        serialized={"name": config.run_name or runnable.get_name()},
8
        inputs=inputs,
9
        run_id=root_run_id,
10
        parent_run_id=config.get("parent_run_id"),
11
        tags=config.tags,
12
        metadata=config.metadata,
13
    )
14

15
    try:
16
        outputs = runnable._invoke_with_config(
17
            inputs,
18
            config={**config, "parent_run_id": root_run_id},
19
        )
20
    except Exception as exc:
21
        callback_manager.on_chain_error(exc, run_id=root_run_id)
22
        raise
23

24
    callback_manager.on_chain_end(outputs, run_id=root_run_id)
25
    return outputs

1
client.create_feedback(
2
    run_id=run_id,
3
    key="correctness",
4
    score=0,
5
    comment="回答引用了不存在的政策",
6
)

1
# 简化伪代码，不是源码逐字复制
2
def attach_feedback_to_run(client, run_id, user_vote, review_note):
3
    run = client.read_run(run_id)
4

5
    if user_vote == "thumbs_up":
6
        score = 1
7
        key = "user_helpfulness"
8
    elif user_vote == "thumbs_down":
9
        score = 0
10
        key = "user_helpfulness"
11
    else:
12
        score = None
13
        key = "needs_review"
14

15
    feedback = client.create_feedback(
16
        run_id=run.id,
17
        key=key,
18
        score=score,
19
        comment=review_note,
20
        metadata={
21
            "input_hash": stable_hash(run.inputs),
22
            "model_version": run.extra.get("metadata", {}).get("revision"),
23
        },
24
    )
25

26
    if score == 0:
27
        enqueue_for_dataset_curation(run.id, feedback.id)
28

29
    return feedback

1
client.evaluate(
2
    target=answer_question,
3
    data="qa-regression",
4
    evaluators=[correctness_evaluator],
5
    experiment_prefix="candidate-v2",
6
)

1
# 简化伪代码，不是源码逐字复制
2
def run_experiment(client, target, dataset, evaluators, experiment_prefix):
3
    experiment = client.create_project(
4
        project_name=f"{experiment_prefix}-{timestamp()}",
5
        reference_dataset_id=dataset.id,
6
    )
7

8
    results = []
9
    for example in client.list_examples(dataset_id=dataset.id):
10
        run = trace_target_call(
11
            project_name=experiment.name,
12
            target=target,
13
            inputs=example.inputs,
14
            reference_example_id=example.id,
15
        )
16

17
        feedback_items = []
18
        for evaluator in evaluators:
19
            try:
20
                feedback_payload = evaluator(run, example)
21
            except Exception as exc:
22
                feedback_payload = {
23
                    "key": "evaluator_error",
24
                    "score": 0,
25
                    "comment": repr(exc),
26
                }
27

28
            feedback_items.append(
29
                client.create_feedback(run_id=run.id, **feedback_payload)
30
            )
31

32
        results.append({
33
            "example_id": example.id,
34
            "run_id": run.id,
35
            "feedback": feedback_items,
36
        })
37

38
    return experiment, results

1
if feedback.score == 0:
2
    promote_run_to_dataset(run_id, dataset_name="qa-regression")

1
# 简化伪代码，不是源码逐字复制
2
def promote_bad_run_to_dataset(client, run_id, feedback_id, dataset_name):
3
    run = client.read_run(run_id)
4
    feedback = client.read_feedback(feedback_id)
5

6
    if feedback.score is None or feedback.score > 0:
7
        return {"action": "skip", "reason": "not a confirmed failure"}
8

9
    if contains_private_data(run.inputs, run.outputs):
10
        return {"action": "send_to_redaction", "run_id": run.id}
11

12
    dataset = get_or_create_dataset(client, dataset_name)
13

14
    example = client.create_example(
15
        dataset_id=dataset.id,
16
        inputs=extract_replayable_inputs(run.inputs),
17
        outputs=derive_reference_output(run, feedback),
18
        metadata={
19
            "source_run_id": run.id,
20
            "source_feedback_id": feedback.id,
21
            "failure_comment": feedback.comment,
22
            "trace_url": run.url,
23
        },
24
    )
25

26
    return {"action": "created_example", "example_id": example.id}

1
if judge_confidence_low:
2
    client.create_feedback(run_id=run.id, key="needs_human_review", score=None)

1
# 简化伪代码，不是源码逐字复制
2
def evaluate_with_uncertainty(run, example, judge_model):
3
    judge = judge_model.invoke({
4
        "question": example.inputs["question"],
5
        "prediction": run.outputs.get("answer"),
6
        "reference": example.outputs.get("answer"),
7
    })
8

9
    if judge.get("confidence", 1.0) < 0.7:
10
        return {
11
            "key": "needs_human_review",
12
            "score": None,
13
            "comment": judge.get("reasoning"),
14
            "metadata": {
15
                "raw_judge": judge,
16
                "route": "human_queue",
17
            },
18
        }
19

20
    return {
21
        "key": "correctness",
22
        "score": judge["score"],
23
        "comment": judge.get("reasoning"),
24
    }

1
if candidate_score < baseline_score - tolerance:
2
    block_release(experiment_id)

1
# 简化伪代码，不是源码逐字复制
2
def compare_experiment_to_baseline(client, candidate_project, baseline_project):
3
    candidate = load_experiment_scores(client, candidate_project)
4
    baseline = load_experiment_scores(client, baseline_project)
5

6
    failing_examples = []
7
    for example_id, candidate_row in candidate.by_example.items():
8
        baseline_row = baseline.by_example.get(example_id)
9
        if baseline_row is None:
10
            continue
11

12
        delta = candidate_row.score("correctness") - baseline_row.score("correctness")
13
        if delta < -0.05:
14
            failing_examples.append({
15
                "example_id": example_id,
16
                "candidate_run_id": candidate_row.run_id,
17
                "baseline_run_id": baseline_row.run_id,
18
                "delta": delta,
19
            })
20

21
    if failing_examples:
22
        return {
23
            "decision": "block_release",
24
            "reason": "regression_detected",
25
            "failures": failing_examples,
26
        }
27

28
    return {"decision": "allow_release"}