What is the use of raise_error and run_inline in BaseCallbackHandler?

[wuodar]

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

""

Description

BaseCallbackHandler contains two attributes

• raise_error
• run_inline

Question 1
I see that run_inline is being used somewhere in the code, but I don't understand what it actually does?

Question 2
Regarding raise_error - I don't see it is being used anywhere in the code. I also use langgraph with langchain StructuredTool and when the tool raise an Error, it is not being populated (raised). It is catched somewhere in the code. So my question is, to propagate the Error that was raised in the tool, should I set raise_error to True, and then in my callback handler implement sth like:

MyCustomCallback(BaseCallbackHandler):
    raise_error = True

    def on_tool_error(self, error: BaseException, **kwargs: Any) -> Any:
        logger.error("Tool error", extra={"error": error})
        if self.raise_error:
            raise error

Or I'm missing the point and it can be done easier (e.g. by setting some flag or class attribute)?

System Info

""

[dosubot[bot]]

Hey there, @wuodar! 👋 I'm here to help you with any bugs, questions, or contributions you have in mind. Let's work together to solve the issue you're facing.

The raise_error and run_inline attributes in the BaseCallbackHandler class control the behavior of the callback handler when an exception occurs and whether the callback should be executed inline, respectively.

1. run_inline: This boolean attribute determines whether the callback should be executed immediately in the current execution flow. If set to True, the callback will run inline; if set to False, it may be executed in a different context or thread.

2. raise_error: This boolean attribute determines whether the callback handler should raise an error if an exception occurs during the execution of a callback. If set to True, the handler will raise the exception; if set to False, it will suppress the exception.

Here is how these attributes are used in the _ahandle_event_for_handler function:

async def _ahandle_event_for_handler(
    handler: BaseCallbackHandler,
    event_name: str,
    ignore_condition_name: Optional[str],
    *args: Any,
    **kwargs: Any,
) -> None:
    try:
        if ignore_condition_name is None or not getattr(handler, ignore_condition_name):
            event = getattr(handler, event_name)
            if asyncio.iscoroutinefunction(event):
                await event(*args, **kwargs)
            else:
                if handler.run_inline:
                    event(*args, **kwargs)
                else:
                    await asyncio.get_event_loop().run_in_executor(
                        None,
                        cast(
                            Callable,
                            functools.partial(
                                copy_context().run, event, *args, **kwargs
                            ),
                        ),
                    )
    except Exception as e:
        if handler.raise_error:
            raise e

In your case, to propagate the error raised in the tool, you should set raise_error to True and implement the on_tool_error method in your custom callback handler. Your implementation is correct:

class MyCustomCallback(BaseCallbackHandler):
    raise_error = True

    def on_tool_error(self, error: BaseException, **kwargs: Any) -> Any:
        logger.error("Tool error", extra={"error": error})
        if self.raise_error:
            raise error

This setup will ensure that the error is logged and then raised if raise_error is set to True ^[1][2].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

Setting raise_error to True in BaseCallbackHandler does have an effect. It determines whether an error should be raised if an exception occurs during the execution of a callback. By default, raise_error is set to False, meaning exceptions will not be raised. If you set it to True, exceptions will be raised when they occur, affecting the flow of the program ^[1][2].

To continue talking to Dosu, mention @dosu.

[wuodar]

Thank you.

Could you describe more in depth what is the use of run_inline? What does it mean that by setting it to False may results in either executing the callback in

1. Different context
2. Different thread?

[dosubot[bot]]

The run_inline attribute in BaseCallbackHandler determines whether the callback should be executed inline within the current context or in a different context or thread.

1. Different Context: When run_inline is set to False, the callback may be executed in a different context. This means that the callback is not executed immediately within the current coroutine or function. Instead, it is scheduled to run in a separate context, which can help in managing long-running or blocking operations without affecting the main execution flow.

2. Different Thread: Setting run_inline to False can also result in the callback being executed in a different thread. This is achieved using asyncio.get_event_loop().run_in_executor, which offloads the execution of the callback to a thread pool executor. This is particularly useful for I/O-bound or CPU-bound tasks that would otherwise block the main event loop.

Here is the relevant code snippet:

async def _ahandle_event_for_handler(
    handler: BaseCallbackHandler,
    event_name: str,
    ignore_condition_name: Optional[str],
    *args: Any,
    **kwargs: Any,
) -> None:
    try:
        if ignore_condition_name is None or not getattr(handler, ignore_condition_name):
            event = getattr(handler, event_name)
            if asyncio.iscoroutinefunction(event):
                await event(*args, **kwargs)
            else:
                if handler.run_inline:
                    event(*args, **kwargs)
                else:
                    await asyncio.get_event_loop().run_in_executor(
                        None,
                        cast(
                            Callable,
                            functools.partial(
                                copy_context().run, event, *args, **kwargs
                            ),
                        ),
                    )
    except NotImplementedError as e:
        # Handle NotImplementedError
    except Exception as e:
        # Handle other exceptions

In summary, setting run_inline to False allows the callback to be executed in a different context or thread, which can help in avoiding blocking the main event loop, especially for long-running or blocking operations ^[1].

To continue talking to Dosu, mention @dosu.

[wuodar]

Can I start a thread using threading module in my Callback that inherits from BaseCallbackHandler? E.g. to send periodic notifications after the on_llm_start is called and before on_llm_new_token is called for the first time? Should I then set run_inline to some specific value?

[dosubot[bot]]

Yes, you can start a thread using the threading module in your callback that inherits from BaseCallbackHandler to send periodic notifications after on_llm_start is called and before on_llm_new_token is called for the first time. You should set run_inline to False to ensure that the callback does not block the main execution flow.

Here's an example of how you might implement this:

import threading
import time
from uuid import UUID
from typing import Any, Optional, Union, List, Dict
from langchain_core.callbacks.base import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> None:
        def send_periodic_notifications():
            while True:
                # Send your notification here
                print("Sending periodic notification...")
                time.sleep(5)  # Adjust the sleep time as needed

        self.notification_thread = threading.Thread(target=send_periodic_notifications)
        self.notification_thread.start()

    def on_llm_new_token(
        self,
        token: str,
        *,
        chunk: Optional[Union[GenerationChunk, ChatGenerationChunk]] = None,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        **kwargs: Any,
    ) -> Any:
        # Stop the notification thread when the first token is received
        if hasattr(self, 'notification_thread'):
            self.notification_thread.join(timeout=0)

    run_inline = False

In this example, the send_periodic_notifications function runs in a separate thread, sending notifications every 5 seconds. The thread is started in the on_llm_start method and stopped in the on_llm_new_token method. The run_inline attribute is set to False to ensure that the callback runs asynchronously ^[1][2].

To continue talking to Dosu, mention @dosu.