71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from typing import Any
|
|
|
|
from langchain_core.callbacks import BaseCallbackHandler
|
|
|
|
from pers import GLOBALS
|
|
|
|
|
|
class CallbackHandler(BaseCallbackHandler):
|
|
"""Base callback handler that can be used to handle callbacks from langchain."""
|
|
|
|
# def on_llm_start(
|
|
# self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when LLM starts running."""
|
|
#
|
|
# def on_chat_model_start(
|
|
# self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when Chat Model starts running."""
|
|
#
|
|
# def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
|
|
# """Run on new LLM token. Only available when streaming is enabled."""
|
|
#
|
|
# def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
|
|
# """Run when LLM ends running."""
|
|
#
|
|
# def on_llm_error(
|
|
# self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when LLM errors."""
|
|
|
|
# def on_chain_start(
|
|
# self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when chain starts running."""
|
|
|
|
# def on_chain_end(self, *args: Any, **kwargs: Any) -> Any:
|
|
# """Run when chain ends running."""
|
|
# print('on_chain_end', args, kwargs)
|
|
|
|
# def on_chain_error(
|
|
# self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when chain errors."""
|
|
#
|
|
# def on_tool_start(
|
|
# self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when tool starts running."""
|
|
|
|
def on_tool_end(self, output: str, **kwargs: Any) -> Any:
|
|
"""Run when tool ends running."""
|
|
GLOBALS.ChatHistory.add_function_output(
|
|
name=kwargs.get('name', ''),
|
|
output=output
|
|
)
|
|
|
|
# def on_tool_error(
|
|
# self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
|
|
# ) -> Any:
|
|
# """Run when tool errors."""
|
|
#
|
|
# def on_text(self, text: str, **kwargs: Any) -> Any:
|
|
# """Run on arbitrary text."""
|
|
#
|
|
# def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
|
|
# """Run on agent action."""
|
|
#
|
|
# def on_agent_finish(self, *args: Any, **kwargs: Any) -> Any:
|
|
# """Run on agent end."""
|