Langgraph条件END状态无法终止执行问题排查求助

问题根源 #

你代码里的核心问题是给eval节点同时配置了条件分支边和硬编码的普通边：

graph.add_conditional_edges(
    "eval",
    self.should_continue,
    {
        "END": END,
        "CONTINUE": "generate",
    },
)
graph.add_edge("eval", "generate")  # 这行是多余且冲突的

Langgraph的规则是：如果一个节点同时存在条件边和普通边，两种边的逻辑会同时执行。也就是说，不管should_continue返回END还是CONTINUE，这条普通边都会强制把流程导向generate节点，直接忽略了END终止的逻辑，最终导致死循环触发GraphRecursionError。

修复方法 #

直接删掉graph.add_edge("eval", "generate")这行代码。条件边已经定义了CONTINUE时走向generate的逻辑，不需要重复添加普通边。

修复后的完整代码 #

import operator
import logging
import random
from typing import TypedDict, Annotated, Sequence, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage

logging.basicConfig(level=logging.INFO)


class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]


class Agent:
    def __init__(self):
        # graph init
        graph = StateGraph(AgentState)
        graph.add_node("init", self.init_node)
        graph.add_node("generate", self.generate_node)
        graph.add_node("eval", self.eval_node)
        graph.add_node("search", self.search_node)

        graph.set_entry_point("init")
        graph.add_conditional_edges(
            "eval",
            self.should_continue,
            {
                "END": END,
                "CONTINUE": "generate",
            },
        )
        graph.add_edge("init", "search")
        graph.add_edge("search", "eval")
        # 删掉了冲突的graph.add_edge("eval", "generate")
        graph.add_edge("generate", "search")

        self.graph = graph.compile()

    def init_node(self, state: AgentState):
        content = "some keywords"
        logging.info(content)
        return {"messages": [AIMessage(content=content)]}

    def search_node(self, state: AgentState):
        content = "some results...."
        logging.info(content)
        return {"messages": [AIMessage(content=content)]}

    def eval_node(self, state: AgentState):
        content = "some evaluation...."
        logging.info(content)
        return {"messages": [AIMessage(content=content)]}

    def generate_node(self, state: AgentState):
        content = "some new keywords...."
        logging.info(content)
        return {"messages": [AIMessage(content=content)]}

    def should_continue(self, state: AgentState):
        choice = random.choice(["END", "CONTINUE"])
        logging.info(choice)
        return choice

验证效果 #

修复后，当should_continue返回END时，流程会从eval节点直接终止，不会再走到generate；返回CONTINUE时，流程才会进入generate->search->eval的循环，完全符合预期。

你之前尝试让should_continue直接返回END的思路是对的，但因为没删掉那条冲突的普通边，所以逻辑一直不生效，现在去掉后就能正常工作了。

内容的提问来源于stack exchange，提问作者FcoZ