Added Python logging tracer (#14190)

This PR creates a logging handler and adds a simple unit test of it

Supercedes https://github.com/langchain-ai/langchain/pull/12862

---------

Co-authored-by: Harrison Chase <hw.chase.17@gmail.com>
This commit is contained in:
James Braza 2023-12-03 12:36:30 -05:00 committed by GitHub
parent 1ea48a31da
commit 052e23be3e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 87 additions and 2 deletions

View File

@ -7,12 +7,14 @@ from langchain_core.tracers.stdout import (
FunctionCallbackHandler, FunctionCallbackHandler,
) )
from langchain.callbacks.tracers.logging import LoggingCallbackHandler
from langchain.callbacks.tracers.wandb import WandbTracer from langchain.callbacks.tracers.wandb import WandbTracer
__all__ = [ __all__ = [
"ConsoleCallbackHandler",
"FunctionCallbackHandler",
"LoggingCallbackHandler",
"LangChainTracer", "LangChainTracer",
"LangChainTracerV1", "LangChainTracerV1",
"FunctionCallbackHandler",
"ConsoleCallbackHandler",
"WandbTracer", "WandbTracer",
] ]

View File

@ -0,0 +1,46 @@
__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, # noqa: ARG002
**kwargs: Any, # noqa: ARG002
) -> 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}'
)

View File

@ -0,0 +1,37 @@
import logging
import sys
import uuid
import pytest
from langchain.callbacks.tracers import LoggingCallbackHandler
def test_logging(
caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str]
) -> None:
# Set up a Logger and a handler so we can check the Logger's handlers work too
logger = logging.getLogger("test_logging")
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(sys.stdout))
handler = LoggingCallbackHandler(logger, extra={"test": "test_extra"})
handler.on_text("test", run_id=uuid.uuid4())
# Assert logging actually took place
assert len(caplog.record_tuples) == 1
record = caplog.records[0]
assert record.name == logger.name
assert record.levelno == logging.INFO
assert (
record.msg == "\x1b[36;1m\x1b[1;3m[text]\x1b[0m \x1b[1mNew text:\x1b[0m\ntest"
)
# Check the extra shows up
assert record.test == "test_extra" # type: ignore[attr-defined]
# Assert log handlers worked
cap_result = capsys.readouterr()
assert (
cap_result.out
== "\x1b[36;1m\x1b[1;3m[text]\x1b[0m \x1b[1mNew text:\x1b[0m\ntest\n"
)