如何创建和控制循环¶

先决条件

本指南假定您熟悉以下内容

当创建带有循环的图时，我们需要一种终止执行的机制。这通常通过添加一个条件边来完成，当达到某个终止条件时，该条件边将路由到 END 节点。

在调用或流式传输图时，您还可以设置图的递归限制。递归限制设置了图在引发错误之前允许执行的超级步骤的数量。在此处阅读更多关于递归限制的概念。

让我们考虑一个带有循环的简单图，以便更好地理解这些机制是如何工作的。

总结¶

创建循环时，可以包含指定终止条件的条件边

const route = async function (state: typeof StateAnnotation.State) {
  if (terminationCondition(state)) {
    return "__END__";
  } else {
    return "a";
  }
}

const graph = StateGraph(State)
  .addNode(a)
  .addNode(b)
  .addConditionalEdges("a", route)
  .addEdge("b", "a")
  .compile();

要控制递归限制，请在配置中指定"recursionLimit"。这将引发GraphRecursionError，您可以捕获并处理此错误。

import { GraphRecursionError } from "@langchain/langgraph";

try {
  await graph.invoke(inputs, { recursionLimit: 3 });
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion Error");
  } else {
    throw error;
  }
}

设置¶

首先，让我们安装所需的包

npm install @langchain/langgraph @langchain/core

为 LangGraph 开发设置 LangSmith

注册 LangSmith，快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据调试、测试和监控您使用 LangGraph 构建的 LLM 应用程序——在此处阅读更多关于如何入门的信息。

定义图¶

让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。

import { StateGraph, Annotation } from "@langchain/langgraph";

// Define the state with a reducer
const StateAnnotation = Annotation.Root({
  aggregate: Annotation<string[]>({
    reducer: (a, b) => a.concat(b),
    default: () => [],
  }),
});

// Define nodes
const a = async function (state: typeof StateAnnotation.State) {
  console.log(`Node A sees ${state.aggregate}`);
  return { aggregate: ["A"] };
}

const b = async function (state: typeof StateAnnotation.State) {
  console.log(`Node B sees ${state.aggregate}`);
  return { aggregate: ["B"] };
}

// Define edges
const route = async function (state: typeof StateAnnotation.State) {
  if (state.aggregate.length < 7) {
    return "b";
  } else {
    return "__end__";
  }
}


// Define the graph
const graph = new StateGraph(StateAnnotation)
  .addNode("a", a)
  .addNode("b", b)
  .addEdge("__start__", "a")
  .addConditionalEdges("a", route)
  .addEdge("b", "a")
  .compile();

import * as tslab from "tslab";

const drawableGraph = graph.getGraph();
const image = await drawableGraph.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();

await tslab.display.png(new Uint8Array(arrayBuffer));

这种架构类似于一个ReAct 智能体，其中节点"a"是工具调用模型，节点"b"表示工具。

在我们的route条件边中，我们指定当状态中的"aggregate"列表长度超过阈值时，我们应该结束。

调用图后，我们看到在达到终止条件之前，我们会在节点"a"和"b"之间交替执行，然后终止。

await graph.invoke({ aggregate: [] });

Node A sees 
Node B sees A
Node A sees A,B
Node B sees A,B,A
Node A sees A,B,A,B
Node B sees A,B,A,B,A
Node A sees A,B,A,B,A,B
{
  aggregate: [
    'A', 'B', 'A',
    'B', 'A', 'B',
    'A'
  ]
}

施加递归限制¶

在某些应用程序中，我们可能无法保证达到给定的终止条件。在这种情况下，我们可以设置图的递归限制。这将在给定数量的超级步骤后引发GraphRecursionError。然后我们可以捕获并处理此异常。

import { GraphRecursionError } from "@langchain/langgraph";

try {
  await graph.invoke({ aggregate: [] }, { recursionLimit: 4 });
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion Error");
  } else {
    throw error;
  }
}

Node A sees 
Node B sees A
Node A sees A,B
Node B sees A,B,A
Recursion Error

请注意，这次我们在第四步后终止。

带分支的循环¶

为了更好地理解递归限制的工作原理，让我们考虑一个更复杂的示例。下面我们实现一个循环，但其中一步会分散到两个节点。

import { StateGraph, Annotation } from "@langchain/langgraph";

// Define the state with a reducer
const StateAnnotationWithLoops = Annotation.Root({
  aggregate: Annotation<string[]>({
    reducer: (a, b) => a.concat(b),
    default: () => [],
  }),
});

// Define nodes
const nodeA = async function (state: typeof StateAnnotationWithLoops.State) {
  console.log(`Node A sees ${state.aggregate}`);
  return { aggregate: ["A"] };
}

const nodeB = async function (state: typeof StateAnnotationWithLoops.State) {
  console.log(`Node B sees ${state.aggregate}`);
  return { aggregate: ["B"] };
}

const nodeC = async function (state: typeof StateAnnotationWithLoops.State) {
  console.log(`Node C sees ${state.aggregate}`);
  return { aggregate: ["C"] };
}

const nodeD = async function (state: typeof StateAnnotationWithLoops.State) {
  console.log(`Node D sees ${state.aggregate}`);
  return { aggregate: ["D"] };
}

// Define edges
const loopRouter = async function (state: typeof StateAnnotationWithLoops.State) {
  if (state.aggregate.length < 7) {
    return "b";
  } else {
    return "__end__";
  }
}

// Define the graph
const graphWithLoops = new StateGraph(StateAnnotationWithLoops)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addNode("c", nodeC)
  .addNode("d", nodeD)
  .addEdge("__start__", "a")
  .addConditionalEdges("a", loopRouter)
  .addEdge("b", "c")
  .addEdge("b", "d")
  .addEdge(["c", "d"], "a")
  .compile();

import * as tslab from "tslab";

const drawableGraphWithLoops = graphWithLoops.getGraph();
const imageWithLoops = await drawableGraphWithLoops.drawMermaidPng();
const arrayBufferWithLoops = await imageWithLoops.arrayBuffer();

await tslab.display.png(new Uint8Array(arrayBufferWithLoops));

这个图看起来很复杂，但可以被概念化为超级步骤的循环。

节点 A
节点 B
节点 C 和 D
节点 A
...

我们有一个由四个超级步骤组成的循环，其中节点 C 和 D 并发执行。

像之前一样调用图，我们看到在达到终止条件之前，我们完成了两个完整的“循环”。

await graphWithLoops.invoke({ aggregate: [] })

Node A sees 
Node B sees A
Node C sees A,B
Node D sees A,B
Node A sees A,B,C,D
Node B sees A,B,C,D,A
Node C sees A,B,C,D,A,B
Node D sees A,B,C,D,A,B
Node A sees A,B,C,D,A,B,C,D
{
  aggregate: [
    'A', 'B', 'C',
    'D', 'A', 'B',
    'C', 'D', 'A'
  ]
}

然而，如果我们将递归限制设置为四，我们只完成一个循环，因为每个循环是四个超级步骤。

import { GraphRecursionError } from "@langchain/langgraph";

try {
  await graphWithLoops.invoke({ aggregate: [] }, { recursionLimit: 4 });
} catch (error) {
  if (error instanceof GraphRecursionError) {
    console.log("Recursion Error");
  } else {
    throw error;
  }
}

Node A sees 
Node B sees A
Node C sees A,B
Node D sees A,B
Node A sees A,B,C,D
Recursion Error