Pydantic Evals 基于 Span 的评估实战：用 OpenTelemetry 追踪断言 AI 系统“如何执行“

Pydantic Evals 基于 Span 的评估实战用 OpenTelemetry 追踪断言 AI 系统如何执行【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai在 Pydantic Evals 中HasMatchingSpan评估器与SpanQuery查询语言让你不只看 AI 系统输出了什么还能验证它是怎么执行的——调用了哪些工具、走了哪些代码路径、耗时是否达标、哪里发生了重试与回退。本文以 docs/evals/evaluators/span-based.md 为主体结合 pydantic_evals 仓库源码与测试完整讲解 SpanTree 的采集原理、SpanQuery 全部查询条件与多场景实战写法。为什么需要基于 Span 的评估传统的评估器只检查任务的输入与输出。对简单任务这或许够用——输出正确即任务成功。但对复杂的多步 Agent过程与结果同样重要错误地得到正确答案Agent 可能靠猜测、使用本应搜索的缓存数据、误调工具却碰巧正确最终输出是对的行为却是错的验证必需行为需要确认特定工具被调用、特定代码分支被执行、特定模式被遵循性能与效率Agent 应高效地达成答案不应有无谓的工具调用、死循环或过度重试安全与合规必须确认危险操作未被尝试、敏感数据未被不当访问、护栏未被绕过。基于 span 的评估能验证**行为契约behavioral contracts**而非简单的输入-输出关系适合这些场景RAG 系统确认生成前确实完成了检索与重排而不只是答案里带了引用多智能体协同确认编排器按正确顺序把任务委派给了正确的专家 Agent工具调用型 Agent确认特定工具被使用或被规避且调用顺序符合预期调试与回归测试捕获输出仍正确但内部逻辑劣化的行为回归生产对齐让评估断言作用在生产环境采集到的同一份遥测数据上评估洞察可直接迁移到生产监控。工作原理从 logfire.configure() 到 SpanTree当你调用logfire.configure()后Pydantic Evals 会在每次任务执行期间采集所有 OpenTelemetry span并在评估时把它们组织成一棵SpanTreelogfire.configure()设置全局 TracerProviderLogfire 使用ProxyTracerProvider包装真实的 provider任务运行时会进入context_subtree()上下文管理器见 pydantic_evals/pydantic_evals/otel/_context_in_memory_span_exporter.py通过_ContextInMemorySpanExporter把上下文期间结束的 span 收集到内存并按上下文 ID 隔离因此并发任务互不串扰上下文退出后SpanTree.add_readable_spans()把ReadableSpan转换为SpanNode并重建父子层级见 pydantic_evals/pydantic_evals/otel/span_tree.py评估器通过EvaluatorContext.span_tree拿到这棵树见 pydantic_evals/pydantic_evals/evaluators/context.py。随后你可以用HasMatchingSpan断言各种执行事实调用了哪些工具HasMatchingSpan(query{name_contains: search_tool})执行了哪些代码路径验证特定函数运行或特定分支被走到时序特征检查操作是否在 SLA 时限内完成错误条件检测重试、回退或特定失败模式执行结构验证父子关系、委派模式或执行顺序注意Span 采集依赖 Logfire或兼容的 OpenTelemetry 环境。未安装或未配置时ctx.span_tree会抛出SpanTreeRecordingError错误信息由 pydantic_evals/pydantic_evals/otel/_context_subtree.py 中的 fallback 实现产生。若使用不带add_span_processor的自定义 TracerProvider如 ddtrace评估仍可运行但span_tree不会被填充。环境准备基于 span 的评估需要安装并配置 Logfirepip install pydantic-evals[logfire]安装后在代码入口处调用logfire.configure()即可开始采集 span。基本用法import logfire from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import HasMatchingSpan # Configure logfire to capture spans logfire.configure(send_to_logfireif-token-present) dataset Dataset( namespan_basic, cases[Case(inputstest)], evaluators[ # Check that database was queried HasMatchingSpan( query{name_contains: database_query}, evaluation_nameused_database, ), ], )send_to_logfireif-token-present表示有 token 才上传遥测即便不上传span 依然会被内存 exporter 采集用于本地评估。evaluation_name用于在报告中标识这条断言。HasMatchingSpan 评估器HasMatchingSpan检查 span 树中是否存在任意一个匹配查询的 spanfrom pydantic_evals.evaluators import HasMatchingSpan HasMatchingSpan( query{name_contains: test}, evaluation_namespan_check, )返回bool——只要任一 span 匹配查询即返回True。其底层实现非常简洁见 pydantic_evals/pydantic_evals/evaluators/common.pydef evaluate(self, ctx: EvaluatorContext) - bool: return ctx.span_tree.any(self.query)SpanTree.any()从每个根节点出发按 DFS 顺序扫描整棵树遇到第一个匹配节点即返回True见 pydantic_evals/pydantic_evals/otel/span_tree.py。这正是后续错误检测章节中需要把查询锚定在根 span 上的原因——HasMatchingSpan是存在性断言只要树里任何span 满足条件就通过。在评估器体系中HasMatchingSpan属于确定性评估器参见 docs/evals/evaluators/overview.md用于行为验证工具调用、代码路径特点是执行快微秒到毫秒级、结果确定、零成本、易调试。SpanQuery 完整参考SpanQuery是一个TypedDict字段全部可选默认按 AND 逻辑组合。其字段顺序个体条件 → 逻辑组合 → 相关 span 条件与SpanNode._matches_query的求值顺序一致先评估代价最低的个体条件最后评估需要递归遍历的相关 span 条件见 pydantic_evals/pydantic_evals/otel/span_tree.py。名称条件# 精确匹配名称 {name_equals: search_database} # 包含子串 {name_contains: tool_call} # 正则模式使用 re.match从头匹配 {name_matches_regex: rllm_call_\d}注意源码中name_matches_regex用re.match而非re.search即模式必须从 span 名称开头匹配测试中的r^grand.*\d$见 tests/evals/test_otel.py 的test_span_query_basics同样印证了这一点。属性条件# 具有指定属性值 {has_attributes: {operation: search, status: success}} # 具有指定属性键值任意 {has_attribute_keys: [user_id, request_id]}属性值按相等性比较但有一个重要的源码级细节OTel 属性无法存放嵌套对象因此 Logfire 等插桩库会把 dict/list 值序列化为 JSON 字符串存储。_attribute_matches见 pydantic_evals/pydantic_evals/otel/span_tree.py做了兼容处理查询值是 dict/list 且存储值是字符串时尝试json.loads反序列化后比较测试见test_span_node_matches_json_serialized_attributes查询值是 list 且存储值是 tuple 时OTel SDK 原生把序列属性存为 tuple转为 list 比较测试见test_span_node_matches_native_sequence_attributes存储的字符串不会被反序列化去匹配原始类型字符串42不会匹配整数42true不会匹配布尔True。状态条件# 记录了错误的 span {has_status: error} # 显式标记为 OK 的 span注意成功 span 通常是 unset 而非 ok {has_status: ok}SpanStatus是Literal[unset, ok, error]镜像 OpenTelemetry 的StatusCode见 pydantic_evals/pydantic_evals/otel/span_tree.py。当 span 所代表的操作抛出异常时其状态为error见SpanNode.status字段注释同文件 L102-L103。从ReadableSpan转换时ERROR→error、OK→ok、其余 →unset见from_readable_span同文件 L140-L155。时长条件from datetime import timedelta # 最小时长 {min_duration: 1.0} # 秒 {min_duration: timedelta(seconds1)} # 最大时长 {max_duration: 5.0} # 秒 {max_duration: timedelta(seconds5)} # 范围 {min_duration: 0.5, max_duration: 2.0}min_duration/max_duration同时接受 float 秒数和timedelta源码中非timedelta值会被timedelta(seconds...)包装后与node.durationend_timestamp - start_timestamp比较见 pydantic_evals/pydantic_evals/otel/span_tree.py。逻辑操作符# NOT {not_: {name_contains: error}} # AND全部满足 {and_: [ {name_contains: tool}, {max_duration: 1.0}, ]} # OR任一满足 {or_: [ {name_equals: search}, {name_equals: query}, ]}源码级注意点or_不能与同一层级的其他条件组合——_matches_query中若or_存在且查询还包含其他字段会直接抛出ValueError(Cannot combine or_ conditions with other conditions at the same level)见 pydantic_evals/pydantic_evals/otel/span_tree.py。需要组合时应把or_放进and_的子查询里正如test_span_query_logical_combinations中的写法{and_: [{has_attributes: {level: 1}}, {or_: [...]}]}。子节点 / 后代条件# 直接子节点数量 {min_child_count: 1} {max_child_count: 5} # 存在子节点匹配查询 {some_child_has: {name_contains: retry}} # 所有子节点匹配查询 {all_children_have: {max_duration: 0.5}} # 没有子节点匹配查询 {no_child_has: {has_status: error}} # 后代查询递归 {min_descendant_count: 5} {some_descendant_has: {name_contains: api_call}}此外源码还支持未在示例中展示的max_descendant_count、all_descendants_have、no_descendant_has见 pydantic_evals/pydantic_evals/otel/span_tree.py。后代按 DFS 顺序遍历实现中用cache装饰的局部函数缓存descendants与剪枝后的pruned_descendants避免多个后代条件对同一棵树重复求值。祖先 / 深度条件# 深度根 span 深度为 0 {min_depth: 1} # 非根 span {max_depth: 2} # 至多 2 层深 # 祖先查询 {some_ancestor_has: {name_equals: agent_run}} {all_ancestors_have: {max_duration: 10.0}} {no_ancestor_has: {has_status: error}}深度等价于祖先数量源码中min_depth/max_depth通过len(ancestors())计算见 pydantic_evals/pydantic_evals/otel/span_tree.py因此根 span 深度为 0。tests/evals/test_otel.py的test_span_tree_ancestors_methods对 5 层嵌套 span 验证了min_depth/max_depth/some_ancestor_has/all_ancestors_have/no_ancestor_has及stop_recursing_when的组合行为。停止递归stop_recursing_when{ some_descendant_has: {name_contains: expensive}, stop_recursing_when: {name_equals: boundary}, } # 只搜索到名为 boundary 的 span 为止不再向下深入stop_recursing_when会同时作用于祖先与后代方向的递归查询遍历时一旦遇到匹配该条件的节点即停止继续深入后代方向continue祖先方向break见 pydantic_evals/pydantic_evals/otel/span_tree.py。剪枝结果会以列表缓存并被同一查询中的每个递归条件复用针对 issue #7484 的回归测试test_span_query_stop_recursing_when_with_multiple_conditions专门验证了这一点防止生成器被消费耗尽导致后续条件空转。实战示例验证工具使用from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import HasMatchingSpan dataset Dataset( nametool_verification, cases[Case(inputstest)], evaluators[ # Must call search tool HasMatchingSpan( query{name_contains: search_tool}, evaluation_nameused_search, ), # Must NOT call dangerous tool HasMatchingSpan( query{not_: {name_contains: delete_database}}, evaluation_namesafe_execution, ), ], )检查多个工具from pydantic_evals.evaluators import HasMatchingSpan evaluators [ HasMatchingSpan( query{name_contains: retrieve_context}, evaluation_nameretrieved_context, ), HasMatchingSpan( query{name_contains: generate_response}, evaluation_namegenerated_response, ), HasMatchingSpan( query{and_: [ {name_contains: cite}, {has_attribute_keys: [source_id]}, ]}, evaluation_nameadded_citations, ), ]性能断言from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # Database queries should be fast HasMatchingSpan( query{and_: [ {name_contains: database}, {max_duration: 0.1}, # 100ms max ]}, evaluation_namefast_db_queries, ), # Overall should complete quickly HasMatchingSpan( query{and_: [ {name_equals: task_execution}, {max_duration: 2.0}, ]}, evaluation_namewithin_sla, ), ]错误检测from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # An error occurred somewhere in the trace HasMatchingSpan( query{has_status: error}, evaluation_namehad_errors, ), # No errors occurred: since HasMatchingSpan passes if *any* span matches, # anchor the query on the root span and check it and all its descendants HasMatchingSpan( query{ name_equals: task_execution, not_: {has_status: error}, no_descendant_has: {has_status: error}, }, evaluation_nameno_errors, ), # Retries happened HasMatchingSpan( query{name_contains: retry}, evaluation_namehad_retries, ), # Fallback was used HasMatchingSpan( query{name_contains: fallback_model}, evaluation_nameused_fallback, ), ]no_errors 是理解HasMatchingSpan语义的关键示例因为它是任一匹配即通过的存在性断言要表达无错误必须锚定到根 spanname_equals: task_execution并同时断言自身无错误、且所有后代都无错误。复杂行为检查from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # Agent delegated to sub-agent HasMatchingSpan( query{and_: [ {name_contains: agent}, {some_child_has: {name_contains: delegate}}, ]}, evaluation_nameused_delegation, ), # Made multiple LLM calls with retries HasMatchingSpan( query{and_: [ {name_contains: llm_call}, {some_descendant_has: {name_contains: retry}}, {min_descendant_count: 3}, ]}, evaluation_nameretry_pattern, ), ]这类嵌套查询在测试中有对应验证test_span_query_complex_hierarchical_conditions构建了app → request(methodGET/POST) → db_query / cache_lookup / notification的树用{name_equals: app, some_child_has: {name_equals: request, has_attributes: {method: POST}, some_child_has: {name_equals: notification}}}精确定位包含 POST 请求且有 notification 子 span 的 app span证明条件可以任意深度递归嵌套。自定义评估器与 SpanTree API对于更复杂的 span 分析可以编写自定义评估器通过ctx.span_tree直接访问整棵树from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class CustomSpanCheck(Evaluator): def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | int]: span_tree ctx.span_tree # Find specific spans llm_spans span_tree.find(lambda node: llm in node.name) tool_spans span_tree.find(lambda node: tool in node.name) # Calculate metrics total_llm_time sum( span.duration.total_seconds() for span in llm_spans ) return { used_llm: len(llm_spans) 0, used_tools: len(tool_spans) 0, tool_count: len(tool_spans), llm_fast: total_llm_time 2.0, }SpanTree APISpanTree提供以下查询方法find/first/any同时接受SpanQuery字典或SpanPredicate可调用对象见同文件 L470from pydantic_evals.otel import SpanTree # Example API (requires span_tree from context) def example_api(span_tree: SpanTree) - None: span_tree.find(lambda n: True) # Find all matching nodes span_tree.any({name_contains: test}) # Check if any span matches span_tree.all({name_contains: test}) # Check if all spans match span_tree.count({name_contains: test}) # Count matching spans # Iteration for node in span_tree: print(node.name, node.duration, node.attributes)注all与count的示例为文档演示写法源码中SpanTree直接提供的是find/first/any见 pydantic_evals/pydantic_evals/otel/span_tree.py以及__iter__按 start_timestamp 排序的扁平迭代SpanNode侧还提供find_children/first_child/any_child/find_descendants/first_descendant/any_descendant/find_ancestors/first_ancestor/any_ancestor等方法同文件 L169-L247测试test_span_tree_find_all、test_span_node_find_children等均有覆盖。SpanTree还支持repr_xml()输出 XML 风格的可读树形结构并导出SPAN_TREE_ADAPTER TypeAdapter(SpanTree)用于 JSON 序列化/反序列化同文件 L584-L585。SpanNode 属性每个SpanNode包含from pydantic_evals.otel import SpanNode # Example properties (requires node from context) def example_properties(node: SpanNode) - None: _ node.name # Span name _ node.duration # timedelta _ node.attributes # dict[str, AttributeValue] _ node.start_timestamp # datetime _ node.end_timestamp # datetime _ node.status # unset | ok | error _ node.children # list[SpanNode] _ node.descendants # list[SpanNode] (recursive) _ node.ancestors # list[SpanNode] _ node.parent # SpanNode | Noneduration是计算属性end_timestamp - start_timestampchildren/descendants/ancestors分别返回直接子节点、DFS 顺序的全部后代、全部祖先见 pydantic_evals/pydantic_evals/otel/span_tree.py。attributes的值类型AttributeValue对齐 OpenTelemetrystr | bool | int | float | Sequence[str] | ...同文件 L20。调试 Span 查询在 Logfire 中查看 Span如果数据发送到了 Logfire可以在其 Web UI 中查看全部 span直观理解 trace 结构父子关系、耗时、状态、属性这通常是编写查询前最有效的第一步。打印 Span 树编写一个简单的调试评估器把整棵树按缩进打印出来from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class DebugSpans(Evaluator): def evaluate(self, ctx: EvaluatorContext) - bool: for node in ctx.span_tree: print(f{ * len(node.ancestors)}{node.name} ({node.duration})) return True缩进深度由len(node.ancestors)决定能直观还原 span 的层级结构。增量测试查询从最简单的条件开始逐步叠加from pydantic_evals.evaluators import HasMatchingSpan # Start simple query {name_contains: tool} # Add conditions gradually query {and_: [ {name_contains: tool}, {max_duration: 1.0}, ]} # Test in evaluator HasMatchingSpan(queryquery, evaluation_nametest)典型场景用例RAG 系统验证from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # Retrieved documents HasMatchingSpan( query{name_contains: vector_search}, evaluation_nameretrieved_docs, ), # Reranked results HasMatchingSpan( query{name_contains: rerank}, evaluation_namereranked_results, ), # Generated with context HasMatchingSpan( query{and_: [ {name_contains: generate}, {has_attribute_keys: [context_ids]}, ]}, evaluation_nameused_context, ), ]多智能体系统from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # Master agent ran HasMatchingSpan( query{name_equals: master_agent}, evaluation_namemaster_ran, ), # Delegated to specialist HasMatchingSpan( query{and_: [ {name_contains: specialist_agent}, {some_ancestor_has: {name_equals: master_agent}}, ]}, evaluation_namedelegated_correctly, ), # No circular delegation HasMatchingSpan( query{not_: {and_: [ {name_contains: agent}, {some_descendant_has: {name_contains: agent}}, {some_ancestor_has: {name_contains: agent}}, ]}}, evaluation_nameno_circular_delegation, ), ]工具使用模式from pydantic_evals.evaluators import HasMatchingSpan evaluators [ # Used search before answering HasMatchingSpan( query{and_: [ {name_contains: search}, {some_ancestor_has: {name_contains: answer}}, ]}, evaluation_namesearched_before_answering, ), # Limited tool calls (no loops) HasMatchingSpan( query{and_: [ {name_contains: tool}, {max_child_count: 5}, ]}, evaluation_namereasonable_tool_usage, ), ]最佳实践从简开始先用基本的名称查询需要时再增加复杂度or_不能与同层其他条件混用复杂逻辑请包进and_使用描述性命名在应用代码中为 span 起好名字——span 名称就是你的断言语言良好的命名让查询一目了然先测查询在跑完整评估前用调试评估器或 Logfire UI 验证查询确实能命中预期的 span与其他评估器组合span 检查应配合输出验证如Contains、LLMJudge等一起使用行为断言与结果断言互补记录预期在注释中说明为什么某些 span 应当/不应当存在这既是文档也是回归防护。下一步Logfire 集成配置 —— 配置 Logfire 以采集 span自定义评估器 —— 编写更高级的 span 分析逻辑内置评估器 —— 其他类型的评估器【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考