feat: add event logging api and more tracing (microsoft#2478)

* feat: add event logging api and more tracing

* code fmt shenanigans

* fixup

* Update test_agent_logging.py

* Update test_agent_logging.py

* Update test_agent_logging.py

* Update sqlite_logger.py

* Update test_agent_logging.py

* Update sqlite_logger.py

---------

Co-authored-by: Chi Wang <wang.chi@microsoft.com>

autogen/agentchat/conversable_agent.py

@@ -32,7 +32,7 @@
 from ..function_utils import get_function_schema, load_basemodels_if_needed, serialize_to_str
 from ..io.base import IOStream
 from ..oai.client import ModelClient, OpenAIWrapper
-from ..runtime_logging import log_new_agent, logging_enabled
+from ..runtime_logging import log_event, log_new_agent, logging_enabled
 from .agent import Agent, LLMAgent
 from .chat import ChatResult, a_initiate_chats, initiate_chats
 from .utils import consolidate_chat_info, gather_usage_summary
@@ -757,6 +757,9 @@ def _print_received_message(self, message: Union[Dict, str], sender: Agent):
     def _process_received_message(self, message: Union[Dict, str], sender: Agent, silent: bool):
         # When the agent receives a message, the role of the message is "user". (If 'role' exists and is 'function', it will remain unchanged.)
         valid = self._append_oai_message(message, "user", sender)
+        if logging_enabled():
+            log_event(self, "received_message", message=message, sender=sender.name, valid=valid)
+
         if not valid:
             raise ValueError(
                 "Received message can't be converted into a valid ChatCompletion message. Either content or function_call must be provided."
@@ -1939,6 +1942,15 @@ def generate_reply(
                 continue
             if self._match_trigger(reply_func_tuple["trigger"], sender):
                 final, reply = reply_func(self, messages=messages, sender=sender, config=reply_func_tuple["config"])
+                if logging_enabled():
+                    log_event(
+                        self,
+                        "reply_func_executed",
+                        reply_func_module=reply_func.__module__,
+                        reply_func_name=reply_func.__name__,
+                        final=final,
+                        reply=reply,
+                    )
                 if final:
                     return reply
         return self._default_auto_reply

autogen/logger/base_logger.py

@@ -9,7 +9,7 @@
 from openai.types.chat import ChatCompletion
 
 if TYPE_CHECKING:
-    from autogen import ConversableAgent, OpenAIWrapper
+    from autogen import Agent, ConversableAgent, OpenAIWrapper
 
 ConfigItem = Dict[str, Union[str, List[str]]]
 LLMConfig = Dict[str, Union[None, float, int, ConfigItem, List[ConfigItem]]]
@@ -68,6 +68,18 @@ def log_new_agent(self, agent: ConversableAgent, init_args: Dict[str, Any]) -> N
         """
         ...
 
+    @abstractmethod
+    def log_event(self, source: Union[str, Agent], name: str, **kwargs: Dict[str, Any]) -> None:
+        """
+        Log an event for an agent.
+
+        Args:
+            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
+        """
+        ...
+
     @abstractmethod
     def log_new_wrapper(self, wrapper: OpenAIWrapper, init_args: Dict[str, Union[LLMConfig, List[LLMConfig]]]) -> None:
         """

autogen/logger/sqlite_logger.py

@@ -17,7 +17,7 @@
 from .base_logger import LLMConfig
 
 if TYPE_CHECKING:
-    from autogen import ConversableAgent, OpenAIWrapper
+    from autogen import Agent, ConversableAgent, OpenAIWrapper
 
 logger = logging.getLogger(__name__)
 lock = threading.Lock()
@@ -103,6 +103,20 @@ class TEXT,                                         -- type or class name of cli
             """
             self._run_query(query=query)
 
+            query = """
+            CREATE TABLE IF NOT EXISTS events (
+                event_name TEXT,
+                source_id INTEGER,
+                source_name TEXT,
+                agent_module TEXT DEFAULT NULL,
+                agent_class_name TEXT DEFAULT NULL,
+                id INTEGER PRIMARY KEY,
+                json_state TEXT,
+                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
+            );
+            """
+            self._run_query(query=query)
+
             current_verion = self._get_current_db_version()
             if current_verion is None:
                 self._run_query(
@@ -246,6 +260,41 @@ class = excluded.class,
         )
         self._run_query(query=query, args=args)
 
+    def log_event(self, source: Union[str, Agent], name: str, **kwargs: Dict[str, Any]) -> None:
+        from autogen import Agent
+
+        if self.con is None:
+            return
+
+        json_args = json.dumps(kwargs, default=lambda o: f"<<non-serializable: {type(o).__qualname__}>>")
+
+        if isinstance(source, Agent):
+            query = """
+            INSERT INTO events (source_id, source_name, event_name, agent_module, agent_class_name, json_state, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?)
+            """
+            args = (
+                id(source),
+                source.name if hasattr(source, "name") else source,
+                name,
+                source.__module__,
+                source.__class__.__name__,
+                json_args,
+                get_current_ts(),
+            )
+            self._run_query(query=query, args=args)
+        else:
+            query = """
+            INSERT INTO events (source_id, source_name, event_name, json_state, timestamp) VALUES (?, ?, ?, ?, ?)
+            """
+            args_str_based = (
+                id(source),
+                source.name if hasattr(source, "name") else source,
+                name,
+                json_args,
+                get_current_ts(),
+            )
+            self._run_query(query=query, args=args_str_based)
+
     def log_new_wrapper(self, wrapper: OpenAIWrapper, init_args: Dict[str, Union[LLMConfig, List[LLMConfig]]]) -> None:
         if self.con is None:
             return

autogen/runtime_logging.py

@@ -12,7 +12,7 @@
 from autogen.logger.logger_factory import LoggerFactory
 
 if TYPE_CHECKING:
-    from autogen import ConversableAgent, OpenAIWrapper
+    from autogen import Agent, ConversableAgent, OpenAIWrapper
 
 logger = logging.getLogger(__name__)
 
@@ -62,6 +62,14 @@ def log_new_agent(agent: ConversableAgent, init_args: Dict[str, Any]) -> None:
     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")

samples/tools/autogenbench/autogenbench/template/testbed_utils.py

@@ -70,7 +70,7 @@ def init():
 
     # Start logging
     if LOGGING_ENABLED:
-        autogen.runtime_logging.start(config={"dbname": "telemetry.db"})
+        autogen.runtime_logging.start(config={"dbname": "telemetry.sqlite"})
 
 
 def finalize(agents):

test/agentchat/test_agent_logging.py

@@ -34,6 +34,9 @@
 
 OAI_WRAPPERS_QUERY = "SELECT id, wrapper_id, session_id, init_args, timestamp FROM oai_wrappers"
 
+EVENTS_QUERY = (
+    "SELECT source_id, source_name, event_name, agent_module, agent_class_name, json_state, timestamp FROM events"
+)
 
 if not skip_openai:
     config_list = autogen.config_list_from_json(
@@ -242,6 +245,14 @@ def test_groupchat_logging(db_connection):
     rows = cur.fetchall()
     assert len(rows) == 3
 
+    # Verify events
+    cur.execute(EVENTS_QUERY)
+    rows = cur.fetchall()
+    json_state = json.loads(rows[0]["json_state"])
+    assert rows[0]["event_name"] == "received_message"
+    assert json_state["message"] == "Can you explain the difference between eigenvalues and singular values again?"
+    assert len(rows) == 15
+
     # Verify schema version
     version_query = "SELECT id, version_number from version"
     cur.execute(version_query)