数据截至 2026-08-17(上游 commit 11e53e8eedb3)

第 4 章 · 跑起来：流水线、pytest、组件级评测、模型抽象

这章讲什么： 前几章讲「单个指标怎么算分」。这章讲把它们组织起来跑的四件事：evaluate() 批量流水线、assert_test() + pytest 做 CI、@observe 做组件级（白盒）评测、以及让裁判模型可换的 DeepEvalBaseLLM 抽象。

---

4.1 两个入口：evaluate() 和 assert_test()

DeepEval 对外主要就两个评测入口（都在 deepeval/__init__.py:_expose_public_api 里暴露）：

• evaluate(test_cases, metrics) —— 批量评测，跑完出一张报告，返回 EvaluationResult。日常迭代用。
• assert_test(test_case, metrics) —— 单用例断言，任一指标没过就 raise AssertionError（标记为 flaky 的用例例外：只发 warning 不阻断）。配合 pytest 用。

assert_test（evaluate/evaluate.py:80）的核心就是「跑一遍 → 收集失败指标 → 拼错误信息抛异常」：

# evaluate/evaluate.py:146 assert_test 末尾（断言逻辑节选）
if not test_result.success:
    failed_metrics_data = [m for m in test_result.metrics_data
                           if m.error is not None or not m.success]
    failed_metrics_str = ", ".join(f"{m.name} (score: {m.score}, threshold: {m.threshold}, ...)"
                                   for m in failed_metrics_data)
    raise AssertionError(f"Metrics: {failed_metrics_str} failed.")

这就是「LLM 输出的 pytest」那句话的字面落地——评测失败 = 测试失败。

---

4.2 evaluate() 流水线做了什么

evaluate()（evaluate/evaluate.py:180）是总控，本身不算分，它负责把一批活儿组织好。主流程：

  校验输入 ──► 重置 test run ──► 打印每个指标说明
      │
      ▼
  按 async_config 选执行器：
    run_async=True  → a_execute_test_cases(...)   并发跑
    run_async=False → execute_test_cases(...)     顺序跑
      │
      ▼
  收集 test_results ──► 渲染控制台报告（可导出 html/md）
      │
      ▼
  保存 test run（本地 + 可选上传 Confident AI）──► 返回 EvaluationResult

它把所有「怎么跑」的旋钮拆成四个配置对象传进去，关注点分离很清楚：

配置	管什么
AsyncConfig	是否异步、最大并发数、限流
DisplayConfig	要不要进度条、详细模式、导出文件格式
CacheConfig	读/写缓存（同样的用例+指标不重复花钱调 LLM）
ErrorConfig	出错是忽略还是抛、缺字段是跳过还是报错

真正的并发执行在 evaluate/execute/（a_execute_test_cases / execute_test_cases）。值得注意的是 assert_test 里写死了 max_concurrent=100、throttle_value=0（evaluate/evaluate.py:97）——单用例断言不需要省并发。

---

4.3 pytest 集成：deepeval test run

DeepEval 注册了一个 pytest 插件（pyproject.toml 里 [tool.poetry.plugins."pytest11"]），代码在 plugins/plugin.py。它让你能写普通 pytest 测试函数，里面调 assert_test，然后用 deepeval test run test_x.py 跑。

插件做两件关键事：

• pytest_sessionstart（plugins/plugin.py:51）： 会话开始时建一个全局 test run，把这次跑的所有断言收集到一起、最后统一汇报/上传。
• pytest_runtest_call（plugins/plugin.py:86）： 用 hookwrapper 把每个测试函数包进一个 deepeval 的「评测 scope」——这样测试里被 @observe 标记的函数产生的 span，能挂到这次 test run 上。这是连接「pytest 测试」和「DeepEval 追踪」的胶水。

---

4.4 组件级（白盒）评测：@observe

前面的评测都是黑盒：只看 input → actual_output。但 agent 内部有很多步（检索、多次 LLM 调用、工具调用），黑盒分数低时你不知道是哪一步坏了。

DeepEval 的解法是 @observe 装饰器（tracing/tracing.py:1406）：给 agent 里的每个函数建一个 span，并能在 span 级别直接挂指标。

# 示意，非源码：组件级评测——给单步挂指标
from deepeval.tracing import observe
from deepeval.metrics import AnswerRelevancyMetric

@observe(type="retriever")                 # 这步是检索
def retrieve(query): ...

@observe(type="llm", metrics=[AnswerRelevancyMetric()])  # 这步是 LLM，单独评
def generate(query, docs): ...

机制要点：

• @observe 包住函数，进入时建 Observer（tracing/tracing.py:1127），__enter__ 时把当前 span 推进上下文，建立父子关系（tracing/tracing.py:1163 起：从 current_span_context 找父，没父就开新 trace）。这样嵌套调用自然形成一棵 span 树。
• 装饰器对同步函数、异步函数、生成器、异步生成器都分别做了包装（tracing.py:1442 起的几个 *_wrapper），连流式输出也能正确建 span。
• span 上挂的 metrics 会在该组件上单独评测（执行在 evaluate/execute/agentic.py）。

还记得 01 章说的吗——BaseMetric.__init_subclass__ 和 DeepEvalBaseLLM.__init_subclass__ 都会自动给方法套 observe。所以指标内部、模型调用都会自动出现在这棵 span 树里，你不用手动埋点。

一句话直觉： @observe 把 agent 的一次运行变成一棵带时间线的调用树；组件级评测就是「给树上的某些节点单独贴一张评分卡」。

---

4.5 模型抽象：让裁判可任意替换

所有指标内部都要调一个「裁判 LLM」。为了让这个裁判能是 OpenAI、Claude、Gemini、Ollama、本地模型……DeepEval 抽了一个统一接口 DeepEvalBaseLLM（models/base_model.py:61）。

子类必须实现的核心方法（都是 @abstractmethod）：

方法	作用
load_model()	加载/构造底层客户端
generate() / a_generate()	同步/异步生成（返回字符串或结构化对象）
get_model_name()	报自己叫什么

还有一层很实用的设计——结构化输出的优雅降级（models/base_model.py:140）：

# models/base_model.py:125 generate_with_schema
def generate_with_schema(self, *args, schema=None, **kwargs):
    if schema is not None:
        try:
            return self.generate(*args, schema=schema, **kwargs)   # 模型原生支持 schema
        except TypeError:
            pass            # 不支持 schema 关键字？退回普通 generate
    return self.generate(*args, **kwargs)

这解释了 02 章里 generate_with_schema_and_extract 为什么有「拿到 schema 对象就用对象、拿到字符串就解析 JSON」两条路——底层模型能力参差，框架在这里统一兜底。

指标怎么挑到具体模型： initialize_model(model)（metrics/utils.py:674）是分发中心。你给指标传 model=：

• 是字符串或 None → 按全局配置（should_use_anthropic_model() 等一串开关）挑一个 provider，多数情况回退到 OpenAIModel。
• 已经是某个原生模型实例 → 直接用，并标记 using_native_model=True。

# metrics/utils.py:674 initialize_model（分发骨架）
if is_native_model(model):
    return model, True
if isinstance(model, DeepEvalBaseLLM):
    return model, False
if should_use_openai_model():   return OpenAIModel(model=model), True
if should_use_gemini_model():   return GeminiModel(model=model), True
# ... litellm / ollama / azure / anthropic / bedrock ...
elif isinstance(model, str) or model is None:
    return OpenAIModel(model=model), True   # 默认 OpenAI

using_native_model 这个布尔很重要：只有原生模型才能精确算成本和 token（generate_with_schema_and_extract 里 if metric.using_native_model: 才累加 cost，metrics/utils.py:525）。你自带的第三方模型，框架不假设能拿到用量。

---

4.6 全章串起来：一次组件级 CI 评测的数据流

  deepeval test run test_agent.py
        │  pytest 插件建全局 test run（plugins/plugin.py）
        ▼
  你的测试函数调你的 agent
        │  agent 里每步 @observe → 建 span 树（tracing/tracing.py）
        ▼
  span 上挂的指标在组件级触发
        │  每个指标 measure() → initialize_model() 选裁判 LLM
        │  裁判 LLM 拆命题/打分（02、03 章）
        ▼
  assert_test / 引擎收集结果 → 没过就 AssertionError → CI 红
        ▼
  保存 test run（本地 + 可选上传 Confident AI）

至此，从「一条用例 + 一个指标」到「挂进 CI 的 agent 组件级评测」，整条链路就通了。

---

代码地图

主题	文件	符号
评测入口	evaluate/evaluate.py	evaluate、assert_test
执行引擎	evaluate/execute/	a_execute_test_cases、execute_test_cases、agentic.py、e2e.py
运行配置	evaluate/configs.py	AsyncConfig、DisplayConfig、CacheConfig、ErrorConfig
pytest 插件	plugins/plugin.py	pytest_sessionstart、pytest_runtest_call
追踪 / 组件级	tracing/tracing.py	observe、Observer、__enter__
模型基类	models/base_model.py	DeepEvalBaseLLM、generate_with_schema、a_generate_with_schema
模型选择	metrics/utils.py	initialize_model、is_native_model、should_use_*_model
公共 API	deepeval/__init__.py	_expose_public_api、instrument