langchain-ai / langgraph

Build resilient language agents as graphs.
https://langchain-ai.github.io/langgraph/
MIT License
5.61k stars 884 forks source link

[SOLVED] LangGraph plots incorrectly the workflow #1473

Closed Nachoeigu closed 2 weeks ago

Nachoeigu commented 2 weeks ago

Checked other resources

Example Code

import os
from dotenv import load_dotenv
import sys

load_dotenv()
WORKDIR=os.getenv("WORKDIR")
os.chdir(WORKDIR)
sys.path.append(WORKDIR)

from langchain_google_genai.chat_models import ChatGoogleGenerativeAI
from langchain_openai.chat_models import ChatOpenAI
from langgraph.graph import StateGraph
from src.utils import State, GraphInput, GraphOutput, GraphConfig, check_chapter
from src.nodes import *
from src.routers import *
from langgraph.graph import END
from src.utils import State

def should_go_to_brainstorming_writer(state: State):
    if state.get('instructor_documents', '') == '':
        return "human_feedback"
    else:
        return "brainstorming_writer"

def should_continue_with_critique(state: State):
    if state.get('is_plan_approved', None) is None: 
        return "brainstorming_critique"
    elif state['is_plan_approved'] == True:
        return "writer"
    else:
        return "brainstorming_critique"

def has_writer_ended_book(state: State):
    if state['current_chapter'] == len(state['chapters_summaries']):
        return END
    else:
        return "writer"

workflow = StateGraph(State, 
                      input = GraphInput,
                      config_schema = GraphConfig)

workflow.add_node("instructor", get_clear_instructions)
workflow.set_entry_point("instructor")
workflow.add_node("human_feedback", read_human_feedback)
workflow.add_node("brainstorming_writer", making_writer_brainstorming)
workflow.add_node("brainstorming_critique", brainstorming_critique)
workflow.add_node("writer", generate_content)
workflow.add_conditional_edges(
    "instructor",
    should_go_to_brainstorming_writer
)
workflow.add_edge("human_feedback","instructor")
workflow.add_conditional_edges(
    "brainstorming_writer",
    should_continue_with_critique
)

workflow.add_edge("brainstorming_critique","brainstorming_writer")
workflow.add_conditional_edges(
    "writer",
    has_writer_ended_book
)

app = workflow.compile(
    interrupt_before=['human_feedback']
    )

Error Message and Stack Trace (if applicable)

No response

Description

LangGraph plots my workflow in the following way, which is not correct:

image

The workflow should look like:

Screenshot 2024-08-26 at 10 39 09

It created relationships where there are no ones.

System Info

langchain 0.2.14 langchain-community 0.2.12 langchain-core 0.2.34 langchain-google-genai 1.0.10 langchain-groq 0.1.9 langchain-openai 0.1.22 langchain-text-splitters 0.2.2 langgraph 0.2.14 langgraph-checkpoint 1.0.6 langsmith 0.1.104

gbaian10 commented 2 weeks ago
Nachoeigu commented 2 weeks ago

Thank you very much, I couldn´t find them in my research.

I think a good improvement in the framework could be that automaticaly detects the return possible values of the router and, based on it, make the relationships. What do you think?

gbaian10 commented 2 weeks ago

Thank you very much, I couldn´t find them in my research.

I think a good improvement in the framework could be that automaticaly detects the return possible values of the router and, based on it, make the relationships. What do you think?

It seems to be mentioned here. https://langchain-ai.github.io/langgraph/cloud/faq/studio/?h=router#why-are-extra-edges-showing-up-in-my-graph

If it could automatically detect and complete, that would be even better. After all, many people have raised the same issue. At least I'm sure VSCode can automatically detect return types, which proves it's feasible from a programming perspective. I'm just not sure how difficult it would be to implement.

image

image

image

Nachoeigu commented 2 weeks ago

Thank you very much, I couldn´t find them in my research. I think a good improvement in the framework could be that automaticaly detects the return possible values of the router and, based on it, make the relationships. What do you think?

It seems to be mentioned here. https://langchain-ai.github.io/langgraph/cloud/faq/studio/?h=router#why-are-extra-edges-showing-up-in-my-graph

If it could automatically detect and complete, that would be even better. After all, many people have raised the same issue. At least I'm sure VSCode can automatically detect return types, which proves it's feasible from a programming perspective. I'm just not sure how difficult it would be to implement.

image

image

image

I think with ast something like that could be possible.

import ast
import inspect
from langgraph.graph import END
import pandas as pd

class ReturnValueExtractor(ast.NodeVisitor):
    """
    A class to extract all possible return expressions from a Python function.

    Attributes:
        return_expressions (set): A set of unique return expressions found in the function.
    """

    def __init__(self):
        """Initialize the ReturnValueExtractor with an empty set of return expressions."""
        self.return_expressions = set()

    def visit_Return(self, node):
        """
        Visit a return statement in the AST and extract the return expression.

        Args:
            node (ast.Return): The return node in the AST.
        """
        if node.value:
            # Convert the return expression to a string and add to the set
            self.return_expressions.add(ast.dump(node.value, annotate_fields=False))

        # Continue walking the AST
        self.generic_visit(node)

    def extract(self, func):
        """
        Extract all return expressions from the given function.

        Args:
            func (function): The function from which to extract return expressions.

        Returns:
            list: A list of unique return expressions as strings.
        """
        source = inspect.getsource(func)
        tree = ast.parse(source)
        self.visit(tree)

        return list(self.return_expressions)

# Example function to test
def example_function(state):
    if state.get('is_plan_approved', None) is None: 
        return "brainstorming_critique"
    elif state['is_plan_approved'] == True:
        return "writer"
    elif state['is_plan_approved'] == False:
        return "another_agent"
    else:
        return END

def complex_function(state):
    if state.get('is_ready', False):
        return END
    elif 1+1 == 3:
        return pd.DataFrame()
    else:
        return example_function
if __name__ == '__main__':
    extractor = ReturnValueExtractor()
    possible_returns = extractor.extract(example_function)
    print(possible_returns)
    print('--------------')
    possible_returns = extractor.extract(complex_function)
    print(possible_returns)
gbaian10 commented 2 weeks ago

If you include some situations where None is implicitly returned, the problem becomes very complex. It will be much simpler if you only capture all explicitly returned values to check if they are Liteval, leaving the rest to the user.

def add(a: int, b: int):
    if a > b:
        raise ValueError("a should be less than b")
    if a == 1:
        return 1
    if b == 2:
        return "2"
    # return a + b

For example, it might return None, but you cannot easily complete this check just by getting the return value; more complex validation is required.

gbaian10 commented 2 weeks ago
import ast
import inspect
from collections.abc import Callable
from typing import Any

def is_literal_return(func: Callable) -> bool:
    """
    Check if the function only returns constant values.

    Args:
        func (Callable): The function to analyze.

    Returns:
        bool: True if all return statements are constant values, False otherwise.
    """
    source = inspect.getsource(func)
    tree = ast.parse(source)
    return all(
        isinstance(node.value, ast.Constant)
        for node in ast.walk(tree)
        if isinstance(node, ast.Return)
    )

def has_explicit_return(func: Callable) -> bool:
    """
    Check if all code paths in the function have an explicit return statement.

    Args:
        func (Callable): The function to analyze.

    Returns:
        bool: True if all paths have explicit return statements, False otherwise.
    """
    source = inspect.getsource(func)
    tree = ast.parse(source)
    function_defs = [
        node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)
    ]
    if not function_defs:
        return False
    return check_all_paths_return(function_defs[0].body)

def check_all_paths_return(statements: list) -> bool:
    """
    Recursively check if all paths in the given statements have a return statement.

    Args:
        statements (list): A list of AST nodes representing statements.

    Returns:
        bool: True if all paths have return statements, False otherwise.
    """
    has_return = False
    for stmt in statements:
        if isinstance(stmt, ast.Return | ast.Raise):
            has_return = True
        elif isinstance(stmt, ast.If):
            has_return = check_all_paths_return(stmt.body) and check_all_paths_return(
                stmt.orelse
            )
        elif isinstance(stmt, ast.While | ast.For):
            continue
    return has_return

def analyze_function(func: Callable) -> set[Any] | None:
    """
    Analyze the function to determine if it returns a set of constant values.

    Args:
        func (Callable): The function to analyze.

    Returns:
        Optional[Set[Any]]: A set of constant values if the function meets the criteria, None otherwise.
    """
    if not is_literal_return(func) or not has_explicit_return(func):
        return None
    source = inspect.getsource(func)
    tree = ast.parse(source)
    return {node.value.value for node in ast.walk(tree) if isinstance(node, ast.Return)}

# Test examples
def example_func() -> int:
    if True:
        return 42
    return 42

def add(a: int, b: int):
    if a > b:
        raise ValueError("a should be less than b")
    if a == 1:
        return 1
    if b == 2:
        return "2"
    # return a + b

def ok():
    return "ok"

def no_ok():
    while 1 + 1 == 3:
        return 1

def one():
    return 1

# Usage examples
print(analyze_function(example_func))  # {42}
print(analyze_function(add))  # None
print(analyze_function(ok))  # {'ok'}
print(analyze_function(no_ok))  # None
print(analyze_function(one))  # {1}

I used an LLM to generate this code and I'm not sure if it covers all scenarios. Since I rarely use AST, I haven't fully reviewed its logic to ensure it's correct.

I expect the following order of rules:

  1. If path_map is included, use the path provided by path_map as a priority.
  2. If path_map is not included, read the return type hint of the path function.
  3. If neither is included, execute AST to check if the return value always returns a Liteval result in all possible cases. If so, use the set or list of that Liteval result.
  4. If it returns None, or if AST checking is disabled, issue a warning indicating that this point will lead to all points.

Three and four are new, but three would be a breaking change, so it's not suitable for a minor release unless the default is set to False.