AsyncCallbackManagerForChainGroup Class — langchain Architecture

Architecture documentation for the AsyncCallbackManagerForChainGroup class in manager.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7["AsyncCallbackManagerForChainGroup"]
  351e9b50_c4fb_2046_1e46_9c34747d671c["AsyncCallbackManager"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|extends| 351e9b50_c4fb_2046_1e46_9c34747d671c
  35cf5db6_bcb1_b854_6ebb_5e0368e51b58["manager.py"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|defined in| 35cf5db6_bcb1_b854_6ebb_5e0368e51b58
  98e920f8_cefc_f929_cc32_a91a4cdfb1ce["__init__()"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|method| 98e920f8_cefc_f929_cc32_a91a4cdfb1ce
  2d1469f3_fddc_38c8_3d0b_ce073c9e6949["copy()"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|method| 2d1469f3_fddc_38c8_3d0b_ce073c9e6949
  cde1f342_a223_8f6a_396b_da1ea4aac583["merge()"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|method| cde1f342_a223_8f6a_396b_da1ea4aac583
  142166f9_39f4_3d7f_a74c_ad799fe38437["on_chain_end()"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|method| 142166f9_39f4_3d7f_a74c_ad799fe38437
  2de4e789_0195_c04e_1d11_06317e5459cd["on_chain_error()"]
  0a7c300e_d2d3_3fec_8d89_671ec76d46f7 -->|method| 2de4e789_0195_c04e_1d11_06317e5459cd

Relationship Graph

Source Code

libs/core/langchain_core/callbacks/manager.py lines 2166–2293

class AsyncCallbackManagerForChainGroup(AsyncCallbackManager):
    """Async callback manager for the chain group."""

    def __init__(
        self,
        handlers: list[BaseCallbackHandler],
        inheritable_handlers: list[BaseCallbackHandler] | None = None,
        parent_run_id: UUID | None = None,
        *,
        parent_run_manager: AsyncCallbackManagerForChainRun,
        **kwargs: Any,
    ) -> None:
        """Initialize the async callback manager.

        Args:
            handlers: The list of handlers.
            inheritable_handlers: The list of inheritable handlers.
            parent_run_id: The ID of the parent run.
            parent_run_manager: The parent run manager.
            **kwargs: Additional keyword arguments.
        """
        super().__init__(
            handlers,
            inheritable_handlers,
            parent_run_id,
            **kwargs,
        )
        self.parent_run_manager = parent_run_manager
        self.ended = False

    def copy(self) -> AsyncCallbackManagerForChainGroup:
        """Return a copy the async callback manager."""
        return self.__class__(
            handlers=self.handlers.copy(),
            inheritable_handlers=self.inheritable_handlers.copy(),
            parent_run_id=self.parent_run_id,
            tags=self.tags.copy(),
            inheritable_tags=self.inheritable_tags.copy(),
            metadata=self.metadata.copy(),
            inheritable_metadata=self.inheritable_metadata.copy(),
            parent_run_manager=self.parent_run_manager,
        )

    def merge(
        self: AsyncCallbackManagerForChainGroup, other: BaseCallbackManager
    ) -> AsyncCallbackManagerForChainGroup:
        """Merge the group callback manager with another callback manager.

        Overwrites the merge method in the base class to ensure that the parent run
        manager is preserved. Keeps the `parent_run_manager` from the current object.

        Returns:
            A copy of the current `AsyncCallbackManagerForChainGroup` with the handlers,
                tags, etc. of the other callback manager merged in.

        Example:
            ```python
            # Merging two callback managers
            from langchain_core.callbacks.manager import (
                CallbackManager,
                atrace_as_chain_group,
            )
            from langchain_core.callbacks.stdout import StdOutCallbackHandler

            manager = CallbackManager(handlers=[StdOutCallbackHandler()], tags=["tag2"])
            async with atrace_as_chain_group(
                "My Group Name", tags=["tag1"]
            ) as group_manager:
                merged_manager = group_manager.merge(manager)
                print(type(merged_manager))
                # <class 'langchain_core.callbacks.manager.AsyncCallbackManagerForChainGroup'>

                print(merged_manager.handlers)
                # [
                #    <langchain_core.callbacks.stdout.LangChainTracer object at ...>,
                #    <langchain_core.callbacks.streaming_stdout.StdOutCallbackHandler object at ...>,
                # ]

                print(merged_manager.tags)
                #    ['tag2', 'tag1']
            ```