FunctionCallbackHandler Class — langchain Architecture

Architecture documentation for the FunctionCallbackHandler class in stdout.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  56d9fac7_db71_9c71_7cac_1d680d392fa0["FunctionCallbackHandler"]
  0f6b3261_31fa_c34e_ca33_cb141bdf78ff["BaseTracer"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|extends| 0f6b3261_31fa_c34e_ca33_cb141bdf78ff
  7fb50eed_e591_4b0f_cce8_91a5fb920a0a["stdout.py"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|defined in| 7fb50eed_e591_4b0f_cce8_91a5fb920a0a
  0da0ad2a_0827_9618_158e_ceb0da119317["__init__()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 0da0ad2a_0827_9618_158e_ceb0da119317
  b242d9f1_5ecd_074d_1971_a8b97464311a["_persist_run()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| b242d9f1_5ecd_074d_1971_a8b97464311a
  ce337549_19ba_f194_9b7a_4c9f15e4f177["get_parents()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| ce337549_19ba_f194_9b7a_4c9f15e4f177
  1bda6774_d332_dec5_ed8f_8d6eef57961e["get_breadcrumbs()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 1bda6774_d332_dec5_ed8f_8d6eef57961e
  42164529_c052_e374_4090_851b8fe251db["_on_chain_start()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 42164529_c052_e374_4090_851b8fe251db
  52082883_bf43_56da_0542_07f3c24da80e["_on_chain_end()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 52082883_bf43_56da_0542_07f3c24da80e
  a241034e_3f27_e335_bb6f_691ec2e801f8["_on_chain_error()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| a241034e_3f27_e335_bb6f_691ec2e801f8
  3847ea36_9090_74f7_44f0_b0626355ef3b["_on_llm_start()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 3847ea36_9090_74f7_44f0_b0626355ef3b
  431be517_3f37_ea62_947a_c527994dac16["_on_llm_end()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 431be517_3f37_ea62_947a_c527994dac16
  b0ee5d69_7295_f562_0d37_2fd471178028["_on_llm_error()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| b0ee5d69_7295_f562_0d37_2fd471178028
  442b4b55_0d0b_ec70_5d85_b110cbb0d4af["_on_tool_start()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 442b4b55_0d0b_ec70_5d85_b110cbb0d4af
  2ef5d58a_7e99_0446_20d8_b2a7f5e0e437["_on_tool_end()"]
  56d9fac7_db71_9c71_7cac_1d680d392fa0 -->|method| 2ef5d58a_7e99_0446_20d8_b2a7f5e0e437

Relationship Graph

Source Code

libs/core/langchain_core/tracers/stdout.py lines 48–195

class FunctionCallbackHandler(BaseTracer):
    """Tracer that calls a function with a single str parameter."""

    name: str = "function_callback_handler"
    """The name of the tracer.

    This is used to identify the tracer in the logs.
    """

    def __init__(self, function: Callable[[str], None], **kwargs: Any) -> None:
        """Create a `FunctionCallbackHandler`.

        Args:
            function: The callback function to call.
        """
        super().__init__(**kwargs)
        self.function_callback = function

    def _persist_run(self, run: Run) -> None:
        pass

    def get_parents(self, run: Run) -> list[Run]:
        """Get the parents of a run.

        Args:
            run: The run to get the parents of.

        Returns:
            A list of parent runs.
        """
        parents = []
        current_run = run
        while current_run.parent_run_id:
            parent = self.run_map.get(str(current_run.parent_run_id))
            if parent:
                parents.append(parent)
                current_run = parent
            else:
                break
        return parents

    def get_breadcrumbs(self, run: Run) -> str:
        """Get the breadcrumbs of a run.

        Args:
            run: The run to get the breadcrumbs of.

        Returns:
            A string with the breadcrumbs of the run.
        """
        parents = self.get_parents(run)[::-1]
        return " > ".join(
            f"{parent.run_type}:{parent.name}"
            for i, parent in enumerate([*parents, run])
        )

    # logging methods
    def _on_chain_start(self, run: Run) -> None:
        crumbs = self.get_breadcrumbs(run)
        run_type = run.run_type.capitalize()
        self.function_callback(
            f"{get_colored_text('[chain/start]', color='green')} "
            + get_bolded_text(f"[{crumbs}] Entering {run_type} run with input:\n")
            + f"{try_json_stringify(run.inputs, '[inputs]')}"
        )

    def _on_chain_end(self, run: Run) -> None:
        crumbs = self.get_breadcrumbs(run)
        run_type = run.run_type.capitalize()
        self.function_callback(
            f"{get_colored_text('[chain/end]', color='blue')} "
            + get_bolded_text(
                f"[{crumbs}] [{elapsed(run)}] Exiting {run_type} run with output:\n"
            )
            + f"{try_json_stringify(run.outputs, '[outputs]')}"
        )

    def _on_chain_error(self, run: Run) -> None:
        crumbs = self.get_breadcrumbs(run)
        run_type = run.run_type.capitalize()
        self.function_callback(