Skip to main content

logger.base_logger

BaseLogger​

class BaseLogger(ABC)

start​

@abstractmethod
def start() -> str

Open a connection to the logging database, and start recording.

Returns:

  • session_id str - a unique id for the logging session

log_chat_completion​

@abstractmethod
def log_chat_completion(invocation_id: uuid.UUID, client_id: int,
                        wrapper_id: int, request: Dict[str,
                                                       Union[float, str,
                                                             List[Dict[str,
                                                                       str]]]],
                        response: Union[str, ChatCompletion], is_cached: int,
                        cost: float, start_time: str) -> None

Log a chat completion to database.

In AutoGen, chat completions are somewhat complicated because they are handled by the autogen.oai.OpenAIWrapper class. One invocation to create can lead to multiple underlying OpenAI calls, depending on the llm_config list used, and any errors or retries.

Arguments:

  • invocation_id uuid - A unique identifier for the invocation to the OpenAIWrapper.create method call
  • client_id int - A unique identifier for the underlying OpenAI client instance
  • wrapper_id int - A unique identifier for the OpenAIWrapper instance
  • request dict - A dictionary representing the the request or call to the OpenAI client endpoint
  • response str or ChatCompletion - The response from OpenAI
  • is_chached int - 1 if the response was a cache hit, 0 otherwise
  • cost(float) - The cost for OpenAI response
  • start_time str - A string representing the moment the request was initiated

log_new_agent​

@abstractmethod
def log_new_agent(agent: ConversableAgent, init_args: Dict[str, Any]) -> None

Log the birth of a new agent.

Arguments:

  • agent ConversableAgent - The agent to log.
  • init_args dict - The arguments passed to the construct the conversable agent

log_event​

@abstractmethod
def log_event(source: Union[str, Agent], name: str,
              **kwargs: Dict[str, Any]) -> None

Log an event for an agent.

Arguments:

  • source str or Agent - The source/creator of the event as a string name or an Agent instance
  • name str - The name of the event
  • kwargs dict - The event information to log

log_new_wrapper​

@abstractmethod
def log_new_wrapper(
        wrapper: OpenAIWrapper,
        init_args: Dict[str, Union[LLMConfig, List[LLMConfig]]]) -> None

Log the birth of a new OpenAIWrapper.

Arguments:

  • wrapper OpenAIWrapper - The wrapper to log.
  • init_args dict - The arguments passed to the construct the wrapper

log_new_client​

@abstractmethod
def log_new_client(client: Union[AzureOpenAI, OpenAI], wrapper: OpenAIWrapper,
                   init_args: Dict[str, Any]) -> None

Log the birth of a new OpenAIWrapper.

Arguments:

  • wrapper OpenAI - The OpenAI client to log.
  • init_args dict - The arguments passed to the construct the client

stop​

@abstractmethod
def stop() -> None

Close the connection to the logging database, and stop logging.

get_connection​

@abstractmethod
def get_connection() -> Union[None, sqlite3.Connection]

Return a connection to the logging database.