| __all__ = ["LoggingCallbackHandler"] |
|
|
| import logging |
| from typing import Any, Optional |
| from uuid import UUID |
|
|
| from langchain_core.exceptions import TracerException |
| from langchain_core.tracers.stdout import FunctionCallbackHandler |
| from langchain_core.utils.input import get_bolded_text, get_colored_text |
|
|
|
|
| class LoggingCallbackHandler(FunctionCallbackHandler): |
| """Tracer that logs via the input Logger.""" |
|
|
| name: str = "logging_callback_handler" |
|
|
| def __init__( |
| self, |
| logger: logging.Logger, |
| log_level: int = logging.INFO, |
| extra: Optional[dict] = None, |
| **kwargs: Any, |
| ) -> None: |
| log_method = getattr(logger, logging.getLevelName(level=log_level).lower()) |
|
|
| def callback(text: str) -> None: |
| log_method(text, extra=extra) |
|
|
| super().__init__(function=callback, **kwargs) |
|
|
| def on_text( |
| self, |
| text: str, |
| *, |
| run_id: UUID, |
| parent_run_id: Optional[UUID] = None, |
| **kwargs: Any, |
| ) -> None: |
| try: |
| crumbs_str = f"[{self.get_breadcrumbs(run=self._get_run(run_id=run_id))}] " |
| except TracerException: |
| crumbs_str = "" |
| self.function_callback( |
| f'{get_colored_text("[text]", color="blue")}' |
| f' {get_bolded_text(f"{crumbs_str}New text:")}\n{text}' |
| ) |
|
|