How to exit the langgraph and raise exception if any one of the nodes fail?

[kennywongck]

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

# langgraph code
    workflow.set_entry_point("node_a")
    workflow.add_edge("node_a", "node_b")
    workflow.add_edge("node_b", "node_c")
    workflow.add_conditional_edges(
        "node_c",
        decide_to_finish,
        {
            "node_d": "node_d",
            "node_b": "node_b"
        },
    )
    workflow.add_edge("node_d", END)
    app = workflow.compile()
	
	
	
# how i stream it out
async for event in app.astream_events(inputs, version="v1"):
	kind = event["event"]
	tags = event.get("tags", [])
	# on_chat_model_stream / on_chain_stream
	if kind == "on_chat_model_stream":
		if "chunk" in event["data"]:
			content = event["data"]["chunk"].content
			if content:
				yield content

Description

I have some nodes in my langgraph and I will stream out the results of each LLM call in some of the nodes, how can I exit the workflow completely if I encounter some unexpected error in any of the nodes and print out the error message of that node which fail? Do I need to wrap the code in a try catch in all of my nodes and add a conditional edge for each nodes?

System Info

langchain==0.2.1
linux
python 3.11

[dosubot[bot]]

To exit the workflow completely if you encounter an unexpected error in any of the nodes and print out the error message of the node that fails, you can use the ToolException and set the handle_tool_error accordingly. Here is an example of how you can achieve this:

1. Define your tool and raise a ToolException when an error occurs.
2. Set handle_tool_error to a custom function that handles the error and prints the error message.

Here is a code example:

from langchain_core.tools import ToolException
from langchain.tools import StructuredTool

# Define a custom error handler function
def _handle_error(error: ToolException) -> str:
    error_message = f"Error in tool: {error.args[0]}"
    print(error_message)  # Print the error message
    raise SystemExit(error_message)  # Exit the workflow completely

# Define a tool that raises a ToolException
def search_tool1(s: str):
    raise ToolException("The search tool1 is not available.")

# Create the tool with the custom error handler
search = StructuredTool.from_function(
    func=search_tool1,
    name="Search_tool1",
    description="A bad tool",
    handle_tool_error=_handle_error,
)

# Run the tool
search.run("test")

In this example, if search_tool1 encounters an error, it raises a ToolException. The custom error handler _handle_error prints the error message and exits the workflow completely using SystemExit ^[1].

Additionally, you can use conditional edges in LangChain workflows to handle error states and redirect the workflow to an error handling node. Here is an example of how you might add an error handling node:

from langgraph.graph import END, StateGraph

# Define a new graph
workflow = StateGraph(AgentState)

# Define the nodes we will cycle between
workflow.add_node("agent", call_model)  # agent
workflow.add_node("action", call_tool)  # retrieval
workflow.add_node("error_handler", handle_error)  # error handling

# Call agent node to decide to retrieve or not
workflow.set_entry_point("agent")

# Decide whether to retrieve
workflow.add_conditional_edges(
    "agent",
    # Assess agent decision
    should_retrieve,
    {
        # Call tool node
        "continue": "action",
        "end": END,
    },
)

# Edges taken after the `action` node is called.
workflow.add_conditional_edges(
    "action",
    # Assess agent decision
    check_relevance,
    {
        # Call agent node
        "yes": "agent",
        "no": END,  # placeholder
    },
)

# Example of adding an error condition
def check_for_error(state):
    # Implement your error checking logic here
    if "error" in state:
        return "error"
    return "no_error"

workflow.add_conditional_edges(
    "agent",
    check_for_error,
    {
        "error": "error_handler",
        "no_error": "action",
    },
)

# Compile
app = workflow.compile()

In this example, check_for_error is a function that checks the state for errors and redirects to the error_handler node if an error is found ^[2].

To continue talking to Dosu, mention @dosu.

[kennywongck]

@dosu how if many nodes is to call an LLM chain instead of calling a tool? if I encountered some error such as network issue or openai rate limit issue, how can I exit the flow?

[dosubot[bot]]

To handle errors like network issues or rate limits in the LLM chain nodes of your LangChain workflow, exit the workflow gracefully, and print out the error message of the node that fails, you can use the following approach:

1. Set up your LLM with fallbacks.
2. Use a try-except block to catch the specific errors and print the error message.

Here's an example:

from unittest.mock import patch
import httpx
from openai import RateLimitError
from langchain_community.chat_models import ChatAnthropic
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Mocking a RateLimitError from OpenAI
request = httpx.Request("GET", "/")
response = httpx.Response(200, request=request)
error = RateLimitError("rate limit", response=response, body="")

# Set max_retries = 0 to avoid retrying on RateLimits, etc
openai_llm = ChatOpenAI(max_retries=0)
anthropic_llm = ChatAnthropic()
llm = openai_llm.with_fallbacks([anthropic_llm])

# Create a prompt template
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You're a nice assistant who always includes a compliment in your response"),
        ("human", "Why did the {animal} cross the road"),
    ]
)
chain = prompt | llm

# Use the chain and handle errors
with patch("openai.resources.chat.completions.Completions.create", side_effect=error):
    try:
        print(chain.invoke({"animal": "kangaroo"}))
    except RateLimitError as e:
        print(f"Workflow exited gracefully due to error: {e}")

In this example:

• The openai_llm is set up with max_retries=0 to avoid automatic retries.
• The llm is created with fallbacks to anthropic_llm.
• The try-except block catches the RateLimitError and prints the error message, allowing the workflow to exit gracefully ^[1].

Additionally, you can use the try_except_tool function to handle errors more gracefully and provide a helpful message on errors:

from typing import Any
from langchain_core.runnables import Runnable, RunnableConfig

def try_except_tool(tool_args: dict, config: RunnableConfig) -> Runnable:
    try:
        complex_tool.invoke(tool_args, config=config)
    except Exception as e:
        return f"Calling tool with arguments:\n\n{tool_args}\n\nraised the following error:\n\n{type(e)}: {e}"

chain = llm_with_tools | (lambda msg: msg.tool_calls[0]["args"]) | try_except_tool

This approach ensures that any errors during the tool invocation are caught and a helpful error message is returned, allowing the workflow to exit gracefully ^[2].

For more robust error handling, you can also use the retry mechanism provided by LangChain's create_base_retry_decorator function, which is preconfigured to handle various exceptions, including network issues and rate limits:

from langchain_core.language_models.llms import create_base_retry_decorator
from langchain_core.callbacks import CallbackManagerForLLMRun

def _create_retry_decorator(llm, run_manager=None):
    import litellm

    errors = [
        litellm.Timeout,
        litellm.APIError,
        litellm.APIConnectionError,
        litellm.RateLimitError,
    ]
    return create_base_retry_decorator(
        error_types=errors, max_retries=llm.max_retries, run_manager=run_manager
    )

# Example usage
retry_decorator = _create_retry_decorator(openai_llm)
@retry_decorator
def call_llm_with_retry(**kwargs):
    return openai_llm.client.completion(**kwargs)

try:
    response = call_llm_with_retry(messages=message_dicts)
    print(response)
except Exception as e:
    print(f"Workflow exited gracefully due to error: {e}")

This ensures that the LLM calls are retried automatically on specific errors, and if all retries fail, the error is caught and handled gracefully ^[3].

Additionally, you can use the LLMonitorCallbackHandler to track and log errors in your LangChain workflow. This handler can be configured to log detailed error messages, including stack traces, which can help in debugging and gracefully exiting the workflow:

from langchain_community.callbacks.llmonitor_callback import LLMonitorCallbackHandler
import traceback
from uuid import UUID

class CustomLLMonitorCallbackHandler(LLMonitorCallbackHandler):
    def on_llm_error(
        self,
        error: BaseException,
        *,
        run_id: UUID,
        parent_run_id: Union[UUID, None] = None,
        **kwargs: Any,
    ) -> Any:
        if self.__has_valid_config is False:
            return
        try:
            self.__track_event(
                "llm",
                "error",
                run_id=str(run_id),
                parent_run_id=str(parent_run_id) if parent_run_id else None,
                error={"message": str(error), "stack": traceback.format_exc()},
                app_id=self.__app_id,
            )
        except Exception as e:
            logger.error(f"[LLMonitor] An error occurred in on_llm_error: {e}")

# Example usage
callback_handler = CustomLLMonitorCallbackHandler(app_id="your_app_id")
try:
    response = call_llm_with_retry(messages=message_dicts)
    print(response)
except Exception as e:
    callback_handler.on_llm_error(e, run_id=UUID("your_run_id"))
    print(f"Workflow exited gracefully due to error: {e}")

This approach ensures that detailed error information is logged, and the workflow exits gracefully with a clear error message ^[4].

To continue talking to Dosu, mention @dosu.

[kennywongck]

@dosu how if the node which i call LLM is inside a langgraph workflow? how do i exit the whole langgraph workflow execution if any of these node fail?

[dosubot[bot]]

To exit the entire LangGraph workflow if any node that calls an LLM fails and ensure that the workflow exits gracefully while printing out the error message, you can modify the on_llm_error method to raise an exception after logging the error. This will terminate the workflow and print the error message.

Here is the modified on_llm_error method:

async def on_llm_error(
        self,
        error: BaseException,
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> None:
        llm_run = self._errored_llm_run(
            error=error,
            run_id=run_id,
        )
        tasks = [self._on_llm_error(llm_run), self._end_trace(llm_run)]
        await asyncio.gather(*tasks)
        
        # Raise the error to exit the workflow
        raise error

This modification ensures that when an error occurs in any node that calls an LLM, the error is logged, and then the exception is raised, which will terminate the workflow and print the error message ^[1].

Additionally, you can use the LangChainTracer class to handle errors for different types of runs (e.g., LLM, Chain, Tool, Retriever) and update the run status upon error:

def _on_llm_error(self, run: Run) -> None:
    """Process the LLM Run upon error."""
    self._update_run_single(run)

def _on_chain_error(self, run: Run) -> None:
    """Process the Chain Run upon error."""
    self._update_run_single(run)

def _on_tool_error(self, run: Run) -> None:
    """Process the Tool Run upon error."""
    self._update_run_single(run)

def _on_retriever_error(self, run: Run) -> None:
    """Process the Retriever Run upon error."""
    self._update_run_single(run)

These methods ensure that the error is processed and the run status is updated accordingly ^[2][3].

For more detailed examples and documentation on error handling within a LangGraph workflow, you can refer to the LangChain repository, which includes examples of using try/except blocks, fallbacks, and retry mechanisms with custom exceptions ^[4].

To continue talking to Dosu, mention @dosu.