由于LangChain和DeepAgents最终都是利用create_agent函数来创建Agent这是一个状态类型为AgentState的CompiledStateGraph对象承载对话历史的messages字段是AgentState的核心成员。加上LangChain采用类似OpenAI的消息结构所以正好可以利用OpenEvals针对Agent执行轨迹的评估AgentEvals将用于创建对应评估器的工厂函数照搬了进来我们在基于AgentEvals的自动化评估-01:面向LangChain的轨迹评估对此进行了详细介绍。LangChain体系还有另一个重要的分支那就是LangGraph而且它是前两者的基础。LangGraph可以构建任意结构的工作流所谓的轨迹就成了工作流中各节点的流程这是OpenEvals默认没有的我们来看看如何利用AgentEvals来完成这种类型的轨迹评估背后的原理有时怎样针对LangGraph这种基于自由工作流节点流程的轨迹评估AgentEvals同样提供了两种方式一种是不需要LLM参数直接比较真实轨迹和参考轨迹另一种则是采用LLM-as-a-Judge评估模式。前者可以指定调用如下这两个函数来完成。defgraph_trajectory_strict_match(*,outputs:GraphTrajectory,reference_outputs:GraphTrajectory,**kwargs:Any,)-EvaluatorResultasyncdefgraph_trajectory_strict_match_async(*,outputs:GraphTrajectory,reference_outputs:GraphTrajectory,**kwargs:Any,)-EvaluatorResult从graph_trajectory_strict_match和graph_trajectory_strict_match_async函数的定义可以看出不论是作为待评估的轨迹对应outputs参数还是作为评估基准的参考轨迹对应reference_outputs参数对应的类型都是GraphTrajectory。1. 轨迹的表示AgentEvals中表示评估轨迹的GraphTrajectory类型是具有如下定义的TypedDict。可以看出它的三个成员均为列表这是因为GraphTrajectory表示的并不是针对单一Agent调用的执行轨迹而是针对同一个Thread中针对Agent多次调用的执行轨迹。列表中的每个元素对应着每次调用所以三个列表成员的长度应该是一样的。classGraphTrajectory(TypedDict):inputs:Optional[list[dict]]results:list[dict]steps:list[list[str]]三个字段成员说明如下inputs每次调用的输入其中作为单次输入的字典只包含一个KVValue才是真正传入的输入Key为接收输入的节点名称除非我们直接采用最底层的Pregel编程控制每个节点的名称否则LangGraph总是会自动创建接收输入的初始节点__start__所以我们看到的基本都是这个值results每次调用完成后的输出。如果发生中断意味着调用尚未结束此时输出为一个空字典steps针对每次调用依次执行的节点名称如果发生中断最后一个节点名称会被设置为__interrupt_,恢复调用后会此节点会消息。这里你会发现一个问题节点流转轨迹体现在steps字段中的每个节点列表它无法表达哪些相邻节点其实是在同一个Superstep中执行的。由于大部分情况下同一Superstep中并发执行的多个节点的顺序是不受控制的评估时也不应该考虑顺序但是目前的结构是做不到的。这可能也是为什么会将上面两个评估方法以strict_match作为后缀命名的原因了。但即使如此作为一个描述轨迹的基础数据类型采用单层的扁平结构来定义依然是严重的设计缺陷steps字段的类型应该是list[list[list[str]]]才对中间一层表示Superstep。这样可以采用不同的模式来实施评估比如Unordered、Exactly、Subset和Superset。2. 针对工作流节点流转的评估接下来我们通过一个实例演示如是对一个采用LangGraph编程模式创建的Agent实施轨迹评估。这个待评估的Agent对应的工作流程如下整个工作流由7个节点组成从起始节点node1根据状态成员shortcut创建了一个条件分支左边被称为捷径的分支通过node1直接抵达完成节点node7。另一个条长程分支先抵达node3然后利用fan-out边向node4和node5广播然后利用fan-in边由node6收口后到达完成节点node7。node2和node6与node7之间并非fan-in边就是条常规的静态边否则流程将永远结束不了。2.1 工作流的构建我们为工作流定义了如下的状态类型State字段shortcut作为输入决定是否走捷径nodes以集合的方式收集整个流程执行的节点名称。我们注册了reducer函数实现针对集合的添加操作。defadd_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。辅助函数create_node根据指定的节点名称创建对应的节点函数具体是一个Callable[[State],dict]对象。create_node返回的节点函数中处理将状态作为参数外。函数执行后我们将当前节点名称添加到steps中并通过返回的{nodes:name}对象往状态的nodes字段添加节点名称。steps:list[str][__start__]defcreate_node(name:str)-Callable[[State],dict]:defnode(state:State)-dict:if(len(steps)0):steps.append(__start__)steps.append(name)return{nodes:name}returnnode整个基于上图所示工作流的Agent通过如下的程序构建而成app(StateGraph(State).add_node(node1,create_node(node1)).add_node(node2,create_node(node2)).add_node(node3,create_node(node3)).add_node(node4,create_node(node4)).add_node(node5,create_node(node5)).add_node(node6,create_node(node6)).add_node(node7,create_node(node7)).set_entry_point(node1).set_finish_point(node7).add_conditional_edges(sourcenode1,pathlambdastate:shortcuttrueifstate[shortcut]elseshortcutfalse,path_map{shortcuttrue:node2,shortcutfalse: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())2.2 轨迹评估具体的评估实现在如下程序中。我们定义了真正用来实施评估的eval函数参数shortcut具体执行Agent时是走捷径分支。如果没有提供作为评估基准的reference_outputs参数我们会使用走长程路径的执行结果和节点流转轨迹构建的GraphTrajectory对象。在该函数中我们在完成Agent的之后后利用返回的结果和steps收集到的步骤创建待评估的GraphTrajectory对象然后调用graph_trajectory_strict_match函数实施评估并将评估结果以JSON形式输出。我们先后指定的不同shortcut参数调用eval函数输出结果表明前者通过评估后者评估失败。ffrom agentevals.graph_trajectory.strictimportgraph_trajectory_strict_matchfromagentevals.typesimportGraphTrajectorydefeval(shortcut:bool,reference_outputs:GraphTrajectory|NoneNone)-None:steps.clear()input:dict{shortcut:shortcut,nodes:set()}reference_outputsreference_outputsor{inputs:[{__start__:input}],results:[{nodes:{node1,node3,node4,node5,node6,node7}}],steps:[[__start__,node1,node3,node4,node5,node6,node7]]}responseapp.invoke(input)# type: ignoreoutputs:GraphTrajectory{inputs:[{__start__:input}],results:[{nodes:response[nodes]}],steps:[steps]}resultgraph_trajectory_strict_match(outputsoutputs,reference_outputsreference_outputs)print(json.dumps(result,ensure_asciiFalse,indent2))eval(shortcutFalse)eval(shortcutTrue)reference_outputs:GraphTrajectory{inputs:[{__start__:{shortcut:False,nodes:set()}}],results:[{nodes:{node1,node3,node4,node5,node6,node7}}],steps:[[__start__,node1,node3,node5,node4,node6,node7]]}eval(shortcutFalse,reference_outputsreference_outputs)输出{key:graph_trajectory_strict_match,score:true,comment:null,metadata:null}{key:graph_trajectory_strict_match,score:false,comment:null,metadata:null}2.3 并发节点的评估我们在上面已经说过由于表达执行轨迹的GraphTrajectory类型在设计上的局限导致无法表示在同一Superstep并发执行多个节点的场景自然也无法完成对应的评估。以我们构建的工作流为例如果不走捷径考虑到源自node3的两个fan-out节点node4和node5如下两种执行轨迹都是可以的“start”,“node1”,“node3”,“node4”,“node5”,“node6”,node7“start”,“node1”,“node3”,“node5”,“node4”,“node6”,node7但是它们只有一个能通过如下的评估eval(shortcutFalse)reference_outputs:GraphTrajectory{inputs:[{__start__:{shortcut:False,nodes:set()}}],results:[{nodes:{node1,node3,node4,node5,node6,node7}}],steps:[[__start__,node1,node3,node5,node4,node6,node7]]}eval(shortcutFalse,reference_outputsreference_outputs)输出{key:graph_trajectory_strict_match,score:true,comment:null,metadata:null}{key:graph_trajectory_strict_match,score:false,comment:null,metadata:null}附上完整的评估代码fromlanggraph.graphimportStateGraphfromtypingimportTypedDict,Callable,Annotated,Required,Iterablefromlangchain_core.runnablesimportRunnableConfigimportjsondefadd_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:list[str][__start__]defcreate_node(name:str)-Callable[[State],dict]:defnode(state:State)-dict:if(len(steps)0):steps.append(__start__)steps.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(sourcenode1,pathlambdastate:shortcuttrueifstate[shortcut]elseshortcutfalse,path_map{shortcuttrue:node2,shortcutfalse: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())fromagentevals.graph_trajectory.strictimportgraph_trajectory_strict_matchfromagentevals.typesimportGraphTrajectorydefeval(shortcut:bool,reference_outputs:GraphTrajectory|NoneNone)-None:steps.clear()input:dict{shortcut:shortcut,nodes:set()}reference_outputsreference_outputsor{inputs:[{__start__:input}],results:[{nodes:{node1,node3,node4,node5,node6,node7}}],steps:[[__start__,node1,node3,node4,node5,node6,node7]]}responseapp.invoke(input)# type: ignoreoutputs:GraphTrajectory{inputs:[{__start__:input}],results:[{nodes:response[nodes]}],steps:[steps]}resultgraph_trajectory_strict_match(outputsoutputs,reference_outputsreference_outputs)print(json.dumps(result,ensure_asciiFalse,indent2))eval(shortcutFalse)eval(shortcutTrue)reference_outputs:GraphTrajectory{inputs:[{__start__:{shortcut:False,nodes:set()}}],results:[{nodes:{node1,node3,node4,node5,node6,node7}}],steps:[[__start__,node1,node3,node5,node4,node6,node7]]}eval(shortcutFalse,reference_outputsreference_outputs)