为了彻底解决AgentEvals针对LagnGraph轨迹评估无法解决同一Superstep内多个节点并发执行的问题我通过定义一个全新的graph_trajectory_match和graph_trajectory_match_async函数提供一种更加灵活的评估方案。上篇提供了针对这种方案的编程体验本篇介绍这这两个函数涉及的轨迹评估究竟是如何实现的。1. 针对Agent执行轨迹的表达我们先来回顾一下定义在AgentEvals中的如下这个GraphTrajectory类型它用来表示作为工作流的Agent的执行轨迹。由于它表示的并非针对Agent的某一次单一调用的执行轨迹而是站在Thread的角度将同于Thread中的多次调用轨迹合并在一起所以它的输入、执行结果和执行步骤都是一个列表而且每个列表的长度是相同的。问题就出在steps字段使用节点名称列表来表示每次Agent调用的执行轨迹无法确定执行的节点于Superstep之间的关系。classGraphTrajectory(TypedDict):inputs:Optional[list[dict]]results:list[dict]steps:list[list[str]]比如[“foo”,“bar”“baz”]具有如下几种执行场景三个节点在三个不同的Superstep中执行“foo和bar”baz在两个相邻Superstep中执行“foo”,bar和baz在两个相邻Superstep中执行。为此我定义了如下这个类似的GraphRunTrajectory它仅仅表示针对单一Agent调用的执行轨迹。所以输入和执行结果都是一个字典steps字段则是一个两层列表第一个对应Superstep第二个对应具体某个Superstep中执行的节点。classGraphRunTrajectory(TypedDict):inputs:dict|Noneresults:dict|Nonesteps:list[list[str]]针对前面列举的三种执行场景steps字段可以通过如下的两层列表来表示[[foo],[bar][baz]][[foo],[barbaz]][[foo,bar][baz]]2. 将匹配模式纳入评估基准由于同一个Superstep涉及多个节点的并发执行由此引入了四种针对单步的节点匹配模式Exact 虽然LangGraph无法控制并发节点的执行顺序但是如果并发节点针对具有依赖关系也可以通过编程的手段来实现。这种情况下相当于采用AgentEvals默认的评估模式Unordered这是默认的匹配模式将节点列表视为无序集合set元素一样即可Subset: 将节点列表视为无序集合set但要求执行的节点在指定的节点内。如果某个节点具有m个下游节点只要执行n个节点nm就可以了可以采用此模式Superset将节点列表视为无序集合set但要求执行的节点包含指定的节点。如果某个节点具有m个下游节点严格要求执行指定的一个或者多个下游节点可以采用此模式。我们将匹配模式应用到作为评估基准的GraphRunReferenceTrajectory类型上。GraphRunReferenceTrajectory是针对GraphTrajectory的评估基准它们之间的不同之处在于其steps字段的定义这里的集合元素不仅仅可以是表示节点名称列表的list[str]对象还可以是一个ReferenceStep对象。classGraphRunTrajectory(TypedDict):inputs:dict|Noneresults:dict|Nonesteps:list[list[str]]classReferenceStep(NamedTuple):steps:list[str]match_mode:GraphTrajectoryMatchModelunorderedclassGraphRunReferenceTrajectory(TypedDict):inputs:dict|Noneresults:dict|Nonesteps:list[ReferenceStep|list[str]]ReferenceStep表示针对一个Superstep的评估轨迹基准steps字段表示节点列表match_mode表示评估时采用的匹配规则。如果直接使用list[str]对象意味着使用默认的Unordered匹配模式。3. 同步轨迹评估同步版本的轨迹评估实现在如下这个graph_trajectory_match函数中它具有如下三个参数outputs 待评估的执行轨迹是一个GraphRunTrajectory对象的列表reference_outputs执行轨迹评估基准是一个GraphRunReferenceTrajectory对象的列表eval_results是否在评估轨迹的时候验证执行结果是否相同默认为False。针对单次Agent调用的轨迹评估实现在_scorer函数中逻辑其实很简单先验证表示轨迹的列表长度是否相同在遍历针对每一步的节点列表根据对应评估基准指定的匹配模式进行对比就可以了。如果开启了eval_results开关则先比较执行结果是否一致。def_scorer(*,outputs:GraphRunTrajectory,reference_outputs:GraphRunReferenceTrajectory,eval_results:bool)-bool:ifeval_resultsandoutputs[results]!reference_outputs[results]:returnFalseforout_step,raw_ref_stepinzip(outputs[steps],reference_outputs[steps]):ref_step:ReferenceStepraw_ref_stepifisinstance(raw_ref_step,ReferenceStep)elseReferenceStep(raw_ref_step,exact)match_moderef_step.match_modeifmatch_modeexactandout_step!ref_step.steps:returnFalseout_step_setset(out_step)ref_step_setset(ref_step.steps)matchmatch_mode:caseunordered:ifout_step_set!ref_step_set:returnFalsecasesubset:ifnotout_step_set.issubset(ref_step_set):returnFalsecasesuperset:ifnotout_step_set.issuperset(ref_step_set):returnFalsereturnTruedefgraph_trajectory_match(*,outputs:list[GraphRunTrajectory],reference_outputs:list[GraphRunReferenceTrajectory],eval_results:boolFalse)-EvaluatorResult:ifoutputsisNoneorreference_outputsisNoneorlen(outputs)!len(reference_outputs):raiseValueError(Strict trajectory match requires both outputs and reference_outputs)matric_namegraph_trajectory_matchresult:EvaluatorResultforoutput,referenceinzip(outputs,reference_outputs):resultcast(EvaluatorResult,_run_evaluator(run_namematric_name,scorerfunctools.partial(_scorer,eval_resultseval_results),feedback_keymatric_name,outputsoutput,reference_outputsreference,))ifnotcast(bool,result.get(score,False)):commentf\ Trajectory not match. outputs:{output}reference_outputs:{reference}return{**result,comment:comment}# type: ignorereturnresult# type: ignoregraph_trajectory_match函数会先验证outputs和reference_outputs在长度上是否匹配在从中依次提取出GraphRunTrajectory和GraphRunReferenceTrajectory通过调用_run_evaluator函数借助_scorer函数实施评估。如果评估没有通过直接返回生成的EvaluatorResult对象。为了利于Debug我们会将待评估轨迹和基准轨迹放到comment字段中。4. 异步轨迹评估同步版本的轨迹评估实现在如下这个graph_trajectory_match_async函数中其实现的本质就是利用async_wrapper函数将针对同步函数_scorer的调用转换成异步形式然后在每次迭代中通过调用_arun_evaluator函数利用async_wrapper函数实施评估。asyncdefgraph_trajectory_match_async(*,outputs:list[GraphRunTrajectory],reference_outputs:list[GraphRunReferenceTrajectory],eval_results:boolFalse)-EvaluatorResult:asyncdefasync_wrapper(**kwargs:Any):return_scorer(eval_resultseval_results,**kwargs)ifoutputsisNoneorreference_outputsisNoneorlen(outputs)!len(reference_outputs)orlen(outputs)0:raiseValueError(Strict trajectory match requires both outputs and reference_outputs)matric_namefgraph_trajectory_matchresult:EvaluatorResultforoutput,referenceinzip(outputs,reference_outputs):resultcast(EvaluatorResult,await_arun_evaluator(run_namematric_name,scorerasync_wrapper,feedback_keymatric_name,outputsoutput,reference_outputsreference,))ifnotcast(bool,result.get(score,False)):commentf\ Trajectory not match. outputs:{output}reference_outputs:{reference}return{**result,comment:comment}# type: ignorereturnresult# type: ignore