Source code for langchain.callbacks.flyte_callback
"""FlyteKit callback handler."""
from __future__ import annotations
import logging
from copy import deepcopy
from typing import TYPE_CHECKING, Any, Dict, List, Tuple
from langchain.callbacks.base import BaseCallbackHandler
from langchain.callbacks.utils import (
    BaseMetadataCallbackHandler,
    flatten_dict,
    import_pandas,
    import_spacy,
    import_textstat,
)
from langchain.schema import AgentAction, AgentFinish, LLMResult
if TYPE_CHECKING:
    import flytekit
    from flytekitplugins.deck import renderer
logger = logging.getLogger(__name__)
[docs]def import_flytekit() -> Tuple[flytekit, renderer]:
    """Import flytekit and flytekitplugins-deck-standard."""
    try:
        import flytekit  # noqa: F401
        from flytekitplugins.deck import renderer  # noqa: F401
    except ImportError:
        raise ImportError(
            "To use the flyte callback manager you need"
            "to have the `flytekit` and `flytekitplugins-deck-standard`"
            "packages installed. Please install them with `pip install flytekit`"
            "and `pip install flytekitplugins-deck-standard`."
        )
    return flytekit, renderer
[docs]def analyze_text(
    text: str,
    nlp: Any = None,
    textstat: Any = None,
) -> dict:
    """Analyze text using textstat and spacy.
    Parameters:
        text (str): The text to analyze.
        nlp (spacy.lang): The spacy language model to use for visualization.
    Returns: