若要对利用LangGraph创建的基于工作流的Agent进行轨迹评估前提条件是得提取代表节点流程的执行轨迹。在前文中我们利用在每个节点函数中注入的RunnableConfig提取出PregelScratchpad对象并利用此对象提取当前的Superstep编码在进一步将它与当前节点名称的映射保存下来进而得到完整的节点流转轨迹。这无疑是一种丑陋的编程方式我们不应该将与业务功能无关的评估逻辑定义在节点函数中。这篇文章提供一种更好的方式从Checkpointer持久化的快照中构建执行轨迹。1. 根据持久化历史构建执行轨迹在我们全面优化版本的LangGraph轨迹评估方案中待评估的执行轨迹是一个list[GraphRunTrajectory]其中GraphRunTrajectory定义如下。三个成员分别代表调用Agent的输入、执行结果和描述轨迹的步骤。steps是一个双层列表第一层代表Superstep第二层表示当前Superstep中执行的节点名称。classGraphRunTrajectory(TypedDict):inputs:dict|Noneresults:dict|Nonesteps:list[list[str]]读取持久化的历史状态构建list[GraphRunTrajectory]对象的完整实现体现在如下这个extract_trajectory_async函数中两个参数分别代表作为Agent对象LangChain、DeepAgents和LangGraph创建的Agent对象最终都继承自Pregel这个基类和传入的配置。fromlanggraph.pregelimportPregelasyncdefextract_trajectory_async(agent:Pregel,config:RunnableConfig)-list[GraphRunTrajectory]:trajectories:list[GraphRunTrajectory][]steps:list[list[str]]|NoneNoneresults:dict{}asyncforsnapshotinagent.aget_state_history(configconfig):stepsstepsor[]iflen(snapshot.tasks)0:steps.append([task.namefortaskinsnapshot.tasks])has_interruptany(task.interruptsfortaskinsnapshot.tasks)ifhas_interrupt:steps.append([__interrupt__])results{}ifnotsnapshot.next:resultssnapshot.valuesifsnapshot.metadataisnotNoneandsnapshot.metadata.get(source)input:iflen(steps)0:steps.reverse()trajectory:GraphRunTrajectory{inputs:{task.name:task.resultfortaskinsnapshot.tasks},results:results,steps:steps}trajectories.append(trajectory)stepsNonetrajectories.reverse()returntrajectories我们简单说说读取历史状态构建执行轨迹的大致流程将RunnableConfig配置作为参数调用Pregel的aget_state_history方法读取代表历史的快照遍历每个快照如果名为source的元数据的值为input,意味着Agent从这里开始执行我们在这里创建一个新的GraphRunTrajectory对象快照的tasks一组在当前Superstep中执行的任务任务的名称即为节点名称此时我们创建节点列表并添加到GraphRunTrajectory的steps字段中如果快照的next为None或者某个任务包含中断意味着这是当前Agent执行流程的最后一个快照如果没有发生中断此时的values就是执行的结果。如果发生中断我们约定将执行结果指定为空字典。由于aget_state_history返回的快照列表是逆序的先返回最近Superstep对应的快照所以迭代会有一些特殊的操作还会对构建的轨迹进行翻转。2. 验证构建的执行轨迹我们依然使用前几篇文章给出的例子来验证一下上面这个函数能够正确地将我们希望的执行轨迹构建出来。我们利用LangGraph构建一个具有如下结构的工作流.整个工作流由7个节点组成从起始节点node1根据状态成员shortcut创建了一个条件分支左边被称为捷径的分支通过node1直接抵达完成节点node7。另一个条长程分支先抵达node3然后利用fan-out边向node4和node5广播然后利用fan-in边由node6收口后到达完成节点node7。node2和node6与node7之间并非fan-in边就是条常规的静态边否则流程将永远结束不了。由于不再需要在节点函数中捕捉执行轨迹所以整个工作的构建会简单一下完整的实现如下所示。由于需要从持久化的状态快照提取执行轨迹所以我们在调用StateGraph的compile方法编译生成Agent时指定了一个InMemorySaver作为checkpointer。fromlanggraph.graphimportStateGraphfromtypingimportTypedDict,Callable,Annotated,Required,Iterable,DefaultDictfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph._internal._scratchpadimportPregelScratchpadfromlanggraph._internal._constantsimportCONFIG_KEY_SCRATCHPADimportjson,uuid,asyncio,operatorclassState(TypedDict):shortcut:boolnodes:Required[Annotated[list[str],operator.add]]defcreate_node(name:str)-Callable[[State],dict]:defnode(state:State)-dict: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(checkpointerInMemorySaver()))在如下的演示程序中我们使用同一个RunnableConfig配置分别指定不同的shortcut值调用Agent两次然后基于这个RunnableConfig配置调用extract_trajectory_async方法得到代表两端执行轨迹的list[GraphRunTrajectory]对象。我们将这个列表以JSON的形式输出可以看出两段轨迹的输入、执行结果由于状态字段nodes的reducer返回只进行追加操作我们第二次调用时无法将其清除和执行步骤都是正确的。fromgraphtrajectoryevalsimportextract_trajectory_asyncasyncdefmain():config:RunnableConfig{configurable:{thread_id:uuid.uuid4()}}app.invoke({shortcut:True,nodes:[]},config)app.invoke({shortcut:False,nodes:[]},config)trajectoriesawaitextract_trajectory_async(app,config)print(json.dumps(trajectories,indent2))asyncio.run(main())输出[{inputs:{__start__:{shortcut:true,nodes:[]}},results:{shortcut:true,nodes:[node1,node2,node7]},steps:[[__start__],[node1],[node2],[node7]]},{inputs:{__start__:{shortcut:false,nodes:[]}},results:{shortcut:false,nodes:[node1,node2,node7,node1,node3,node4,node5,node6,node7]},steps:[[__start__],[node1],[node3],[node4,node5],[node6][node7]]}]3. 针对Agent实施轨迹评估有了extract_trajectory_async函数的加持评估工作就简单多了。我们将轨迹评估核心逻辑实现在如下所示的eval函数中两个参数shortcut和reference_outputs分别表示执行Agent时是否走捷径以及提供的作为评估基准的GraphTrajectory对象默认为走长程路径的轨迹。我们利用创建的RunnableConfig配置指定thread_id并连同shortcut参数构建的输入调用Agent。我们最后调用extract_trajectory_async函数构建描述轨迹的list[GraphRunTrajectory]列表并将它和作为评估基准的list[GraphRunReferenceTrajectory]对象传入graph_trajectory_match_async函数实施评估并输出评估结果。fromgraphtrajectoryevalsimport(extract_trajectory_async,ReferenceStep,graph_trajectory_match_async)asyncdefeval(shortcut:bool,reference_outputs:list[GraphRunReferenceTrajectory]|NoneNone)-None:reference_outputsreference_outputsor[{inputs:None,results:None,steps:[[__start__],[node1],[node3],ReferenceStep([node4,node5],unordered),[node6],[node7]]}]input:dict{shortcut:shortcut,nodes:[]}config:RunnableConfig{configurable:{thread_id:uuid.uuid4()}}app.invoke(input,config)outputsawaitextract_trajectory_async(app,config)resultawaitgraph_trajectory_match_async(outputsoutputs,reference_outputsreference_outputs)print(json.dumps(result,indent2))我们按照如下的形式两次调用eval函数并得到不同的评估结果asyncdefmain():awaiteval(shortcutFalse)awaiteval(shortcutTrue)asyncio.run(main())输出结果{key:graph_trajectory_match,score:true,comment:null,metadata:null}{key:graph_trajectory_match,score:false,comment:Trajectory not match. \noutputs:{inputs: {__start__: {shortcut: True, nodes: []}}, results: {shortcut: True, nodes: [node1, node2, node7]}, steps: [[__start__], [node1], [node2], [node7]]}\nreference_outputs:{inputs: None, results: None, steps: [[__start__], [node1], [node3], ReferenceStep(steps[node4, node5], match_modeunordered), [node6], [node7]]}\n,metadata:null}4. AgentEvals自己是如何实现的AgentEvals其实也提供过类似的方式来构建它自己的执行轨迹具体体现在如下这三个方法中底层的实现原理与我上面定义的extract_trajectory_async函数类似。defextract_langgraph_trajectory_from_snapshots(snapshots:Iterable[StateSnapshot],)-ExtractedLangGraphThreadTrajectorydefextract_langgraph_trajectory_from_thread(graph:Pregel,config:RunnableConfig)-ExtractedLangGraphThreadTrajectoryasyncdefaextract_langgraph_trajectory_from_thread(graph:Pregel,config:RunnableConfig)-ExtractedLangGraphThreadTrajectoryclassExtractedLangGraphThreadTrajectory(TypedDict):inputs:listoutputs:GraphTrajectory