refactor(callback): simplify import statements and type annotations for better readability (#2793)

* refactor: add LogFunctionType protocol for type hint

The LogFunctionType protocol is introduced to define the structure of a logging function that takes a message and an optional name parameter. This protocol will be used to ensure consistency and compatibility with other parts of the codebase when logging messages.

* refactor(callback.py): simplify import statements and type annotations for better readability
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-07-18 16:22:41 -03:00 • committed by GitHub
commit 6ca5780908
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 9 additions and 4 deletions

View file

@ -1,16 +1,16 @@
from typing import Any, Callable, Concatenate, Dict, List from typing import Any, Dict, List
from uuid import UUID from uuid import UUID
from langchain.callbacks.base import AsyncCallbackHandler from langchain.callbacks.base import AsyncCallbackHandler
from langchain_core.agents import AgentAction, AgentFinish from langchain_core.agents import AgentAction, AgentFinish
from langflow.schema.log import LoggableType from langflow.schema.log import LogFunctionType
class AgentAsyncHandler(AsyncCallbackHandler): class AgentAsyncHandler(AsyncCallbackHandler):
"""Async callback handler that can be used to handle callbacks from langchain.""" """Async callback handler that can be used to handle callbacks from langchain."""
def __init__(self, log_function: Callable[Concatenate[LoggableType | list[LoggableType], ...], None] | None = None): def __init__(self, log_function: LogFunctionType | None = None):
self.log_function = log_function self.log_function = log_function
async def on_tool_start( async def on_tool_start(

View file

@ -1,5 +1,10 @@
from typing import Union from typing import Optional, Union
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import Protocol
LoggableType = Union[str, dict, list, int, float, bool, None, BaseModel] LoggableType = Union[str, dict, list, int, float, bool, None, BaseModel]
class LogFunctionType(Protocol):
def __call__(self, message: Union[LoggableType, list[LoggableType]], *, name: Optional[str] = None) -> None: ...