基于AgentEvals的自动化评估-02:面向LangGraph的轨迹评估介绍了AgentEvals针对LangGraph的轨迹评估，它利用graph_trajectory_strict_match和graph_trajectory_strict_match_async这两个函数完成基于工作流节点流转的轨迹评估。它提供的解决方案具有一个很大的问题：由于将待评估的执行轨迹表示为简单的节点名称列表，导致无法解决基于并发节点执行的场景（并发节点执行顺序不可控，应该在忽略执行顺序的前提下实施评估）。为了我通过重写评估方法提过一种给全新的评估方案。

1. 新评估方案做了哪些改进？

新的评估方案不仅仅是解决上述这个核心问题，还在如下几个方面做了全面的改进和增强：

首先，新的方式提供针对执行轨迹的多种匹配模式。Graph的执行轨迹从单层节点列表list[str]，变成了两层列表list[list[str]]。第一层对应的每个列表对应一个Superstep，第二个层列表对应的是当前Superstep内有序执行的节点。对于每个Superstep对应的节点列表，都可以独立应用如下四种不同的匹配模式。为了更好地说明它们之前的差别，我们将待评估轨迹和基准轨迹的节点列表分别成为outputs列表和reference_outs列表：

• unordered：将outputs列表和reference_outs列表视为顺序无关的集合，集合相等即视为匹配；
• subset：将outputs列表和reference_outs列表视为顺序无关的集合，要求前者是后者子集；
• superset：将outputs列表和reference_outs列表视为顺序无关的集合，要求前者是后者超集；
• exact：两个集合完全相等。

除此之外，AgentEvals默认的轨迹评估会忽略输出结果，该改进的评估方案可以选择包含结果评估。默认评估在评估失败是不提供具体的原因，改进的方法则提供原因以利于Debug。

2. 利用新的评估方案解决多节点并发执行问题

在多个节点在同一个Supestep并发执行的情况下采用unordered模式的匹配规则进行评估，是我们优化评估方案的初衷。我们看看整个问题是如何解决的，我们依然使用基于AgentEvals的自动化评估-02:面向LangGraph的轨迹评估构建的这个具有如下结构的工作流。

整个工作流由7个节点组成，从起始节点node1根据状态成员shortcut创建了一个条件分支，左边被称为捷径的分支通过node1直接抵达完成节点node7。另一个条长程分支先抵达node3，然后利用fan-out边向node4和node5广播，然后利用fan-in边由node6收口后到达完成节点node7。node2和node6与node7之间并非fan-in边，就是条常规的静态边，否则流程将永远结束不了。

用于构建工作流的StateGraph将我们自定义的State作为状态类型，状态字段shortcut决定是否走上述的捷径，nodes字段用于收集整个流程执行过的节点，我们利用指向add_to_set函数的reducer函数实现针对节点的添加。由于新的轨迹是根据Superstep的编号作为顺序构建的，所以我们在节点工厂函数create_node创建的节点函数中注入了RunnableConfig配置，并从中提取PregelScratchpad对象，后者利用step字段提供当前的Superstep编号。我们将当前节点添加到全局变量steps中，这是一个DefaultDict[int,list[str]]类型的字典，Key为Superstep编号。

fromlanggraph.graphimportStateGraphfromtypingimportTypedDict,Callable,Annotated,Required,Iterable,DefaultDict,castfromlangchain_core.runnablesimportRunnableConfigfromlanggraph._internal._scratchpadimportPregelScratchpadfromlanggraph._internal._constantsimportCONFIG_KEY_SCRATCHPADimportjsondefadd_to_set(result:set[str],update:Iterable[str]|str)->set[str]:ifisinstance(update,str):result.add(update)returnresultreturnresult.union(update)classState(TypedDict):shortcut:boolnodes:Required[Annotated[set[str],add_to_set]]steps:DefaultDict[int,list[str]]=DefaultDict(list)defcreate_node(name:str)->Callable[[State,RunnableConfig],dict]:defnode(state:State,config:RunnableConfig)->dict:step=cast(PregelScratchpad,config.get("configurable",{}).get(CONFIG_KEY_SCRATCHPAD)).step steps[step].append(name)return{"nodes":name}returnnode app=(StateGraph(State).add_node("node1",create_node("node1"))# type: ignore.add_node("node2",create_node("node2"))# type: ignore.add_node("node3",create_node("node3"))# type: ignore.add_node("node4",create_node("node4"))# type: ignore.add_node("node5",create_node("node5"))# type: ignore.add_node("node6",create_node("node6"))# type: ignore.add_node("node7",create_node("node7"))# type: ignore.set_entry_point("node1").set_finish_point("node7").add_conditional_edges(source="node1",path=lambdastate:"shortcut=true"ifstate["shortcut"]else"shortcut=false",path_map={"shortcut=true":"node2","shortcut=false":"node3"}).add_edge("node2","node7").add_edge("node3","node4").add_edge("node3","node5").add_edge(["node4","node5"],"node6").add_edge("node2","node7").add_edge("node6","node7").compile())fromgraphtrajectoryevalsimport(graph_trajectory_match,GraphRunReferenceTrajectory,GraphRunTrajectory,ReferenceStep)defgenerate_trajectory(results:dict)->GraphRunTrajectory:step_list=[steps[k]forkinsorted(steps)]step_list.insert(0,["__start__"])return{"inputs":None,"results":results,"steps":step_list}defeval(shortcut:bool,reference_outputs:list[GraphRunReferenceTrajectory]|None=None)->None:steps.clear()input:dict={"shortcut":shortcut,"nodes":set()}reference_outputs=reference_outputsor[{"inputs":None,"results":None,"steps":[["__start__"],["node1"],["node3"],ReferenceStep(["node4","node5"],"unordered"),["node6"],["node7"]]}]response=app.invoke(input)# type: ignoreoutputs=[generate_trajectory(response)]result=graph_trajectory_match(outputs=outputs,reference_outputs=reference_outputs)print(json.dumps(result,ensure_ascii=False,indent=2))eval(shortcut=False)eval(shortcut=True)

Agent的执行和评估都实现在eval方法中，该方法的shortcut参数决定工作流是否走捷径。我们利用它生成执行Agent的输入，并将调用generate_trajectory将得到的响应转换成一个GraphRunTrajectory对象，该类型表示针对单一Agent调用（Run）生成的评估轨迹。由于默认我们不评估执行结果，所以steps字段表示的轨迹是整个对象的核心，整个轨迹来源于用来收集执行节点的steps变量。

传入自定义评估函数graph_trajectory_match中的参数有两个，一个是承载待评估轨迹的GraphRunTrajectory列表，另一个是作为评估基准的GraphRunReferenceTrajectory列表。GraphRunReferenceTrajectory和GraphRunTrajectory的区别在于，它需要为每个Superstep执行的多个节点指定评估模式，所以它的steps字段类型为list[ReferenceStep|list[str]]，如果直接指定节点名称列表，意味着采用默认的unordered匹配模式，如果需要采用其他模式，则需要利用ReferenceStep对象显式指定。

对于我们的工作流来说，node4和node5节点是并发执行的，所以我们只将这一步定义成ReferenceStep(["node4","node5"]，其它的都是一个只包含单一节点的列表。这里仅仅是为了演示如何为每一步指定不同的匹配规则，由于unordered是默认的匹配模式，所以直接将其指定成["node4","node5"]就好。 我们最后指定不同的shortcut参数调用eval函数，得到如下两种不同的评估结果。对于没有通过的评估，我们在comment中提供了待评估的轨迹和基准轨迹。

输出：

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":false,"comment":"Trajectory not match. \noutputs:{'inputs': None, 'results': {'shortcut': True, 'nodes': {'node2', 'node7', 'node1'}}, 'steps': [['__start__'], ['node1'], ['node2'], ['node7']]}\nreference_outputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ['node1'], ['node3'], ReferenceStep(steps=['node4', 'node5'], match_mode='unordered'), ['node6'], ['node7']]}\n","metadata":null}

由于由于单步都采用unordered匹配模式，并发节点node4和node5执行循序的不同不会影响最终的评估结果。

3. 针对不同模式的轨迹匹配

接下来我们通过一个例子来演示上述的四种轨迹匹配模式。在如下这个辅助函数eval中，我们针对一个包含三步的轨迹（["__start__"],[...],["__end__"]）进行评估，中间这一步的节点列表由参数step指定。作为基准的轨迹为[“start”],[“foo”,“bar”,“baz”],[“end”]，中间这一步的匹配规则则由match_mode参数指定。

fromgraphtrajectoryevalsimport(graph_trajectory_match,GraphTrajectoryMatchModel,GraphRunTrajectory,GraphRunReferenceTrajectory,ReferenceStep)importjsondefeval(match_mode:GraphTrajectoryMatchModel,*steps:str):outputs:list[GraphRunTrajectory]=[{"inputs":None,"results":None,"steps":[["__start__"],[*steps],["__end__"]]}]reference_outputs:list[GraphRunReferenceTrajectory]=[{"inputs":None,"results":None,"steps":[["__start__"],ReferenceStep(steps=["foo","bar","baz"],match_mode=match_mode),["__end__"]]}]result=graph_trajectory_match(outputs=outputs,reference_outputs=reference_outputs)print(json.dumps(result,indent=2))

3.1 exact

我们将match_mode参数设置为exact，意味着待评估轨迹必须于评估基准在数量和顺序上完全一致，所以第一次调用针对评估会成功，第二次会失败：

eval("exact","foo","bar","baz")eval("exact","bar","baz","foo")

输出：

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":false,"comment":"Trajectory not match. \noutputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ['bar', 'baz', 'foo'], ['__end__']]}\nreference_outputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ReferenceStep(steps=['foo', 'bar', 'baz'], match_mode='exact'), ['__end__']]}\n","metadata":null}

3.2 unordered

我们将match_mode参数设置为unordered，意味着轨迹包含的节点将视为无序的集合，集合相等则视为评估通过，所以下面三次调用，只有最后一次会失败。

eval("unordered","foo","bar","baz")eval("unordered","bar","baz","foo")eval("unordered","foo","bar")

输出：

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":false,"comment":"Trajectory not match. \noutputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ['foo', 'bar'], ['__end__']]}\nreference_outputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ReferenceStep(steps=['foo', 'bar', 'baz'], match_mode='unordered'), ['__end__']]}\n","metadata":null}

3.3.subset

我们将match_mode参数设置为unordered，意味着轨迹包含的节点将视为无序的集合，待评估的节点集合是评估基准的节点集合的子集则视为评估通过。其实这也是一种常用的匹配模式，如果一个节点具有多个下游节点，并且不要求所有下游节点都执行，此时就是用到这种匹配模式。所以下面三次调用，只有最后一次会失败。

eval("subset","foo","bar")eval("subset","bar","baz")eval("subset","baz","qux")

输出：

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":false,"comment":"Trajectory not match. \noutputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ['baz', 'qux'], ['__end__']]}\nreference_outputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ReferenceStep(steps=['foo', 'bar', 'baz'], match_mode='subset'), ['__end__']]}\n","metadata":null}

3.4.superset

我们将match_mode参数设置为unordered，意味着轨迹包含的节点将视为无序的集合，待评估的节点集合是评估基准的节点集合的子集则视为评估通过。其实这也是一种常用的匹配模式，如果一个节点具有多个下游节点，并且不要求所有下游节点都执行，但要求某协议节点一定要执行，此时就是用到这种匹配模式。所以下面三次调用，只有最后一次会失败。

eval("superset","foo","bar","baz","qux")eval("superset","foo","bar","baz","quux")eval("superset","foo","bar")

输出：

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":true,"comment":null,"metadata":null}

{"key":"graph_trajectory_match","score":false,"comment":"Trajectory not match. \noutputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ['foo', 'bar'], ['__end__']]}\nreference_outputs:{'inputs': None, 'results': None, 'steps': [['__start__'], ReferenceStep(steps=['foo', 'bar', 'baz'], match_mode='superset'), ['__end__']]}\n","metadata":null}