[基于AgentEvals的自动化评估-04]全面优化面向LangGraph的轨迹评估[下篇]

为了彻底解决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"],["bar"，"baz"]]
• [["foo","bar"]，["baz"]]

2. 将匹配模式纳入评估基准

由于同一个Superstep涉及多个节点的并发执行，由此引入了四种针对单步的节点匹配模式：

• Exact： 虽然LangGraph无法控制并发节点的执行顺序，但是如果并发节点针对具有依赖关系，也可以通过编程的手段来实现。这种情况下，相当于采用AgentEvals默认的评估模式；
• Unordered：这是默认的匹配模式，将节点列表视为无序集合（set），元素一样即可；
• Subset: 将节点列表视为无序集合（set），但要求执行的节点在指定的节点内。如果某个节点具有m个下游节点，只要执行n个节点（n<=m）就可以了，可以采用此模式；
• 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:GraphTrajectoryMatchModel="unordered"classGraphRunReferenceTrajectory(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:ReferenceStep=raw_ref_stepifisinstance(raw_ref_step,ReferenceStep)elseReferenceStep(raw_ref_step,"exact")match_mode=ref_step.match_modeifmatch_mode=="exact"andout_step!=ref_step.steps:returnFalseout_step_set=set(out_step)ref_step_set=set(ref_step.steps)matchmatch_mode:case"unordered":ifout_step_set!=ref_step_set:returnFalsecase"subset":ifnotout_step_set.issubset(ref_step_set):returnFalsecase"superset":ifnotout_step_set.issuperset(ref_step_set):returnFalsereturnTruedefgraph_trajectory_match(*,outputs:list[GraphRunTrajectory],reference_outputs:list[GraphRunReferenceTrajectory],eval_results:bool=False)->EvaluatorResult:ifoutputsisNoneorreference_outputsisNoneorlen(outputs)!=len(reference_outputs):raiseValueError("Strict trajectory match requires both outputs and reference_outputs")matric_name="graph_trajectory_match"result:EvaluatorResultforoutput,referenceinzip(outputs,reference_outputs):result=cast(EvaluatorResult,_run_evaluator(run_name=matric_name,scorer=functools.partial(_scorer,eval_results=eval_results),feedback_key=matric_name,outputs=output,reference_outputs=reference,))ifnotcast(bool,result.get("score",False)):comment=f"""\ Trajectory not match. outputs:{output}reference_outputs:{reference}"""return{**result,"comment":comment}# type: ignorereturnresult# type: ignore

graph_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:bool=False)->EvaluatorResult:asyncdefasync_wrapper(**kwargs:Any):return_scorer(eval_results=eval_results,**kwargs)ifoutputsisNoneorreference_outputsisNoneorlen(outputs)!=len(reference_outputs)orlen(outputs)==0:raiseValueError("Strict trajectory match requires both outputs and reference_outputs")matric_name=f"graph_trajectory_match"result:EvaluatorResultforoutput,referenceinzip(outputs,reference_outputs):result=cast(EvaluatorResult,await_arun_evaluator(run_name=matric_name,scorer=async_wrapper,feedback_key=matric_name,outputs=output,reference_outputs=reference,))ifnotcast(bool,result.get("score",False)):comment=f"""\ Trajectory not match. outputs:{output}reference_outputs:{reference}"""return{**result,"comment":comment}# type: ignorereturnresult# type: ignore