-
Notifications
You must be signed in to change notification settings - Fork 4.5k
/
runtime_logging.py
105 lines (73 loc) · 2.95 KB
/
runtime_logging.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
from __future__ import annotations
import logging
import sqlite3
import uuid
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
from openai import AzureOpenAI, OpenAI
from openai.types.chat import ChatCompletion
from autogen.logger.base_logger import LLMConfig
from autogen.logger.logger_factory import LoggerFactory
if TYPE_CHECKING:
from autogen import Agent, ConversableAgent, OpenAIWrapper
logger = logging.getLogger(__name__)
autogen_logger = None
is_logging = False
def start(logger_type: str = "sqlite", config: Optional[Dict[str, Any]] = None) -> str:
global autogen_logger
global is_logging
autogen_logger = LoggerFactory.get_logger(logger_type=logger_type, config=config)
try:
session_id = autogen_logger.start()
is_logging = True
except Exception as e:
logger.error(f"[runtime logging] Failed to start logging: {e}")
finally:
return session_id
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:
if autogen_logger is None:
logger.error("[runtime logging] log_chat_completion: autogen logger is None")
return
autogen_logger.log_chat_completion(
invocation_id, client_id, wrapper_id, request, response, is_cached, cost, start_time
)
def log_new_agent(agent: ConversableAgent, init_args: Dict[str, Any]) -> None:
if autogen_logger is None:
logger.error("[runtime logging] log_new_agent: autogen logger is None")
return
autogen_logger.log_new_agent(agent, init_args)
def log_event(source: Union[str, Agent], name: str, **kwargs: Dict[str, Any]) -> None:
if autogen_logger is None:
logger.error("[runtime logging] log_event: autogen logger is None")
return
autogen_logger.log_event(source, name, **kwargs)
def log_new_wrapper(wrapper: OpenAIWrapper, init_args: Dict[str, Union[LLMConfig, List[LLMConfig]]]) -> None:
if autogen_logger is None:
logger.error("[runtime logging] log_new_wrapper: autogen logger is None")
return
autogen_logger.log_new_wrapper(wrapper, init_args)
def log_new_client(client: Union[AzureOpenAI, OpenAI], wrapper: OpenAIWrapper, init_args: Dict[str, Any]) -> None:
if autogen_logger is None:
logger.error("[runtime logging] log_new_client: autogen logger is None")
return
autogen_logger.log_new_client(client, wrapper, init_args)
def stop() -> None:
global is_logging
if autogen_logger:
autogen_logger.stop()
is_logging = False
def get_connection() -> Union[None, sqlite3.Connection]:
if autogen_logger is None:
logger.error("[runtime logging] get_connection: autogen logger is None")
return None
return autogen_logger.get_connection()
def logging_enabled() -> bool:
return is_logging