How to limit the number of recursion in the graph that contains several loops ？

[Eskiii]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

for the examples of langgraph: https://github.com/langchain-ai/langgraph/blob/main/examples/rag/langgraph_self_rag.ipynb

Description

in the self-rag graph: https://github.com/langchain-ai/langgraph/blob/main/examples/rag/langgraph_self_rag.ipynb，which contain three loops, i wonder how to limit the maximum number of times of one of the loops.

For example, limit the loop: retrive - grade_documents - decide_to_generate - retrieve to 3, and limit the other loop: generate - grade_generation_v_documents_and_question - generate to 5.

Looking forward to reply， thank you!

System Info

langchain-experimental == 0.0.49
langchain-openai == 0.1.1
langchain == 0.1.13
langchain-community == 0.0.29
langchain-core == 0.1.34
langchain-text-splitters == 0.0.1
python == 3.10.12

[dosubot[bot]]

Hey @Eskiii, great to see you diving deeper into the LangChain world! Facing a new challenge, I see? Let's tackle this one together.

To limit the number of iterations for specific loops within your graph, you can create custom chain classes that inherit from SequentialChain or SimpleSequentialChain in the LangChain framework. These custom classes can incorporate logic to track and limit iterations for each loop.

Here's a step-by-step guide to achieve this:

1. Create Custom Chain Classes: Extend SequentialChain or SimpleSequentialChain to include iteration tracking and limiting functionality.

2. Track Iterations: Implement a counter within your custom class to monitor the number of iterations.

3. Limit Iterations: Before proceeding with each iteration, check if the counter has reached the predefined maximum. If so, terminate the loop; otherwise, continue.

Here's an example to illustrate this approach:

from langchain.chains import SequentialChain

class LimitedIterationChain(SequentialChain):
    def __init__(self, components, max_iterations):
        super().__init__(components)
        self.max_iterations = max_iterations
        self.current_iteration = 0

    def __call__(self, *args, **kwargs):
        while self.current_iteration < self.max_iterations:
            super().__call__(*args, **kwargs)
            self.current_iteration += 1
        self.current_iteration = 0  # Reset for potential reuse

# Example usage with your specific loops
sequence_1_chain = LimitedIterationChain(components_sequence_1, max_iterations=3)
sequence_2_chain = LimitedIterationChain(components_sequence_2, max_iterations=5)

Replace components_sequence_1 and components_sequence_2 with the actual components for your loops. This setup ensures that sequence_1_chain and sequence_2_chain will iterate up to their specified limits.

This method requires you to adjust the logic based on your specific implementation details and how components are called within each iteration. For further insights on handling iterations with SequentialChain or SimpleSequentialChain, consider exploring the unit tests in the LangChain repository.

Sources

• libs/langchain/langchain/chains/init.py
• libs/langchain/tests/unit_tests/chains/test_sequential.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[Eskiii]

however, a loop maybe contain within another loop (just like the example: https://github.com/langchain-ai/langgraph/blob/main/examples/rag/langgraph_self_rag.ipynb), which together form a graph, in this case, how to limit a standardalone loop?

[maximeperrindev]

Hi @Eskiii

You can pass {"recursion_limit": 15} for example to limit the graph loop iterations.

chain.invoke(...,{"recursion_limit": 15})

[Eskiii]

It seems that "recursion_limit" works globally, this variable is no more applicable if i want to control the recursion limit for a specific loop in a graph that contains multiple loops.

In the following case, this variable cannot fulfill such a requirement:

1. limit loop1: retrive - grade_documents - decide_to_generate - retrieve to 3
2. limit loop2: retrive - grade_documents - decide_to_generate - generate - grade_generation_v_documents_and_question - retrieve to 3
3. limit loop3: generate - grade_generation_v_documents_and_question - generate to 5.

from langgraph.graph import END, StateGraph

workflow = StateGraph(GraphState)

# Define the nodes
workflow.add_node("retrieve", retrieve) # retrieve
workflow.add_node("grade_documents", grade_documents) # grade documents
workflow.add_node("generate", generate) # generatae
workflow.add_node("transform_query", transform_query) # transform_query

# Build graph
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",   # loop1
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",   #loop2
        "useful": END,
        "not useful": "transform_query",  # loop3
    },
)

# Compile
app = workflow.compile()

[Towhid1]

I am also facing similar issue. stuck in loop.

[electricjimi]

The solution I found is to create a defaultdict { name of step : counter} to keep track of how many times a specific step is performed.
Then I pass it around inside the GraphState: if counter for this specific step is > LIMIT, I skip to the next step.

[vizsatiz]

@electricjimi This looks like a nice way to do it, can you give me example code ?

[isatyamks]

I am using a defaultdict to keep track of the number of times each specific step
Extend the State with a Counter Dictionary:

from collections import defaultdict
from langgraph.graph import GraphState, END, StateGraph

class CustomGraphState(GraphState):
def init(self):
super().init()
self.step_counter = defaultdict(int)

workflow = StateGraph(CustomGraphState)

is it helpful?

[usersina]

How do you use it and increment it?

I tried

class AgentState(TypedDict):
    """
    The agent state is the input to each node in the graph
    """

    messages: Annotated[Sequence[BaseMessage], operator.add]
    """
    The annotation tells the graph that new messages will always be added
    to the current state
    """

    next: str  # [...agents, "FINISH"]
    """
    The next node to execute
    """

    step_counter: dict[str, int]
    """
    The step counter for each node. This is used to keep track of the number of
    times a node has been executed.
    """

workflow.add_conditional_edges(
    "my_entry_node", _decide_next_node, conditional_map
)

def _decide_next_node(state: AgentState) -> str:
    """
    Decide the next node based on the current state and step_counter.
    """
    # Increment the step counter for the current node
    current_node = state["next"]
    if current_node not in state["step_counter"]:
        state["step_counter"][current_node] = 1
    else:
        state["step_counter"][current_node] += 1

    # Check if the loop limit for my_researcher has been reached
    if (
        current_node == "my_researcher"
        and state["step_counter"][current_node] > 3
    ):
        # If the limit is reached, decide the next node to break the loop
        # This could be another node or an end condition
        return "my_final_node"
    else:
        # If the limit is not reached, continue with the current logic
        # This is where you define how to choose the next node based on the current state
        # For example, returning my_researcher to continue the loop
        return "my_researcher"

Throws a KeyError because of "step_counter"

[SrGrace]

You can do something like the below:

have revision_number and max_revision in your graph state

### graph state
from typing import List
from typing_extensions import TypedDict

class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        question: question
        generation: LLM generation
        documents: list of documents
    """
    question: str
    generation: str
    documents: List[str]
    revision_number: int
    max_revisions: int
  

and update your conditional edges for the revisions

 def decide_to_generate(self, state: GraphState):
        """
        Determines whether to generate an answer, or re-generate a question.

        Args:
            state (dict): The current graph state

        Returns:
            str: Binary decision for next node to call
        """

        print("---ASSESS GRADED DOCUMENTS---")
        state["question"]
        filtered_documents = state["documents"]

        if not filtered_documents and state["revision_number"] < state["max_revisions"]:
            # All documents have been filtered check_relevance
            # We will re-generate a new query
            print(
                "---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---"
            )
            return "transform_query"
        else:
            # We have relevant documents, so generate answer
            print("---DECISION: GENERATE---")
            return "generate"

def grade_generation_v_documents_and_question(self, state: GraphState):
        """
        Determines whether the generation is grounded in the document and answers question.

        Args:
            state (dict): The current graph state

        Returns:
            str: Decision for next node to call
        """

        print("---CHECK HALLUCINATIONS---")
        question = state["question"]
        documents = state["documents"]
        generation = state["generation"]

        score = self.hallucination_grader.invoke(
            {
                "documents": documents, "generation": generation
            }
        )
        grade = score["score"]

        # Check hallucination
        if grade == "yes":
            print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
            # Check question-answering
            print("---GRADE GENERATION vs QUESTION---")
            score = self.answer_grader.invoke({
                    "question": question, "generation": generation
                })
            grade = score["score"]
            if grade == "yes":
                print("---DECISION: GENERATION ADDRESSES QUESTION---")
                return "useful"
            else:
                print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
                return "not useful"

        else:
            if state["revision_number"] >= state["max_revisions"]:
                print("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS AND THE REVISION COUNT EXCEEDED THE MAX REVISION, STOP---")
                return "stop"
            else:
                print("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
                return "not supported"

Compile graph like:

# build graph
workflow = StateGraph(GraphState)

# Define the nodes
workflow.add_node("retrieve", self.retrieve)  # retrieve
workflow.add_node("grade_documents", self.grade_documents)  # grade documents
workflow.add_node("generate", self.generate)  # generatae
workflow.add_node("transform_query", self.transform_query)  # transform_query

# Build graph
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    self.decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    self.grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
        "stop": END
    },
)

# Compile
self.app = workflow.compile()

# display
display(Image(self.app.get_graph(xray=True).draw_mermaid_png()))

It will look like this:

Hope this helps!