SqliteLogger

SqliteLogger(config: dict[str, Any])

Sqlite logger class.
Initialize the SqliteLogger.

Parameters:
NameDescription
configType: dict[str, Any]

Class Attributes

schema_version



Instance Methods

get_connection

get_connection(self) -> None | Connection

Get connection.


log_chat_completion

log_chat_completion(
    self,
    invocation_id: UUID,
    client_id: int,
    wrapper_id: int,
    source: str | Agent,
    request: dict[str, float | str | list[dict[str, str]]],
    response: str | ChatCompletion,
    is_cached: int,
    cost: float,
    start_time: str
) -> None

Log chat completion.

Parameters:
NameDescription
invocation_idInvocation ID.

Type: uuid.UUID
client_idClient ID.

Type: int
wrapper_idWrapper ID.

Type: int
sourceSource of the chat completion.

Type: str | Agent
requestRequest for the chat completion.

Type: dict[str, float | str | list[dict[str, str]]]
responseResponse for the chat completion.

Type: str | ChatCompletion
is_cachedWhether the response is cached.

Type: int
costCost of the chat completion.

Type: float
start_timeStart time of the chat completion.

Type: str

log_event

log_event(
    self,
    source: str | Agent,
    name: str,
    **kwargs: dict[str, Any]
) -> None

Log event.

Parameters:
NameDescription
sourceSource of the event.

Type: str | Agent
nameName of the event.

Type: str
**kwargsAdditional arguments for the event.

Type: dict[str, Any]

log_function_use

log_function_use(
    self,
    source: str | Agent,
    function: F,
    args: dict[str, Any],
    returns: Any
) -> None

Log function use.

Parameters:
NameDescription
sourceSource of the function use.

Type: str | Agent
functionFunction to log.

Type: F
argsArguments of the function.

Type: dict[str, Any]
returnsReturns of the function.

Type: Any

log_new_agent

log_new_agent(
    self,
    agent: ConversableAgent,
    init_args: dict[str, Any]
) -> None

Log new agent.

Parameters:
NameDescription
agentAgent to log.

Type: ConversableAgent
init_argsInitialization arguments of the agent

Type: dict[str, Any]

log_new_client

log_new_client(
    self,
    client: AzureOpenAI | OpenAI | CerebrasClient | GeminiClient | AnthropicClient | MistralAIClient | TogetherClient | GroqClient | CohereClient | OllamaClient | BedrockClient,
    wrapper: OpenAIWrapper,
    init_args: dict[str, Any]
) -> None

Log new client.

Parameters:
NameDescription
clientClient to log.

Type: AzureOpenAI | OpenAI | CerebrasClient | GeminiClient | AnthropicClient | MistralAIClient | TogetherClient | GroqClient | CohereClient | OllamaClient | BedrockClient
wrapperWrapper of the client.

Type: OpenAIWrapper
init_argsInitialization arguments of the client.

Type: dict[str, Any]

log_new_wrapper

log_new_wrapper(
    self,
    wrapper: OpenAIWrapper,
    init_args: dict[str, LLMConfig | list[LLMConfig]]
) -> None

Log new wrapper.

Parameters:
NameDescription
wrapperWrapper to log.

Type: OpenAIWrapper
init_argsInitialization arguments of the wrapper

Type: dict[str, LLMConfig | list[LLMConfig]]

start

start(self) -> str

Open a connection to the logging database, and start recording.
Returns:
session_id (str): a unique id for the logging session

Returns:
TypeDescription
strsession_id (str): a unique id for the logging session

stop

stop(self) -> None

Stop the logger