base.py — langchain Source File

Source Code

"""Base callback handler for LangChain."""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from collections.abc import Sequence
    from uuid import UUID

    from tenacity import RetryCallState
    from typing_extensions import Self

    from langchain_core.agents import AgentAction, AgentFinish
    from langchain_core.documents import Document
    from langchain_core.messages import BaseMessage
    from langchain_core.outputs import ChatGenerationChunk, GenerationChunk, LLMResult

_LOGGER = logging.getLogger(__name__)


class RetrieverManagerMixin:
    """Mixin for `Retriever` callbacks."""

    def on_retriever_error(
        self,
        error: BaseException,
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> Any:
        """Run when `Retriever` errors.

        Args:
            error: The error that occurred.
            run_id: The ID of the current run.
            parent_run_id: The ID of the parent run.
            **kwargs: Additional keyword arguments.
        """

    def on_retriever_end(
        self,
        documents: Sequence[Document],
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> Any:
        """Run when `Retriever` ends running.

        Args:
            documents: The documents retrieved.
            run_id: The ID of the current run.
            parent_run_id: The ID of the parent run.
            **kwargs: Additional keyword arguments.
        """


// ... (1058 more lines)