Skip to content

node_logger

Class info

Classes

Name Children Inherits
ChatMessageContainer
llmling_agent.messaging.message_container
Container for tracking and managing chat messages.
    NodeLogger
    llmling_agent.messaging.node_logger
    Handles database logging for node interactions.
      ToolCallInfo
      llmling_agent.models.tools
      Information about an executed tool call.

        🛈 DocStrings

        Logging functionality for node interactions.

        NodeLogger

        Handles database logging for node interactions.

        Source code in src/llmling_agent/messaging/node_logger.py 
        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
        class NodeLogger:
    """Handles database logging for node interactions."""

    def __init__(self, node: MessageEmitter[Any, Any], enable_db_logging: bool = True):
        """Initialize logger.

        Args:
            node: Node to log interactions for
            enable_db_logging: Whether to enable logging
        """
        self.node = node
        self.enable_db_logging = enable_db_logging
        self.conversation_id = str(uuid4())
        self.message_history = ChatMessageContainer()
        self.toolcall_history = EventedList[ToolCallInfo]()

        # Initialize conversation record if enabled
        if enable_db_logging:
            self.init_conversation()
            # Connect to the combined signal to capture all messages
            node.message_received.connect(self.log_message)
            node.message_sent.connect(self.log_message)
            node.tool_used.connect(self.log_tool_call)

    def clear_state(self):
        """Clear node state."""
        self.message_history.clear()
        self.toolcall_history.clear()

    @property
    def last_message(self) -> ChatMessage[Any] | None:
        """Get last message in history."""
        return self.message_history.last_message

    @property
    def last_tool_call(self) -> ToolCallInfo | None:
        """Get last tool call in history."""
        return self.toolcall_history[-1] if self.toolcall_history else None

    def init_conversation(self):
        """Create initial conversation record."""
        self.node.context.storage.log_conversation_sync(
            conversation_id=self.conversation_id,
            node_name=self.node.name,
        )

    def log_message(self, message: ChatMessage):
        """Handle message from chat signal."""
        self.message_history.append(message)

        if not self.enable_db_logging:
            return
        self.node.context.storage.log_message_sync(
            conversation_id=self.conversation_id,
            content=str(message.content),
            role=message.role,
            name=message.name,
            cost_info=message.cost_info,
            model=message.model,
            response_time=message.response_time,
            forwarded_from=message.forwarded_from,
        )

    def log_tool_call(self, tool_call: ToolCallInfo):
        """Handle tool usage signal."""
        self.toolcall_history.append(tool_call)

        if not self.enable_db_logging:
            return
        self.node.context.storage.log_tool_call_sync(
            conversation_id=self.conversation_id,
            message_id=tool_call.message_id or "",
            tool_call=tool_call,
        )

        last_message property 

        last_message: ChatMessage[Any] | None

        Get last message in history.

        last_tool_call property 

        last_tool_call: ToolCallInfo | None

        Get last tool call in history.

        __init__ 

        __init__(node: MessageEmitter[Any, Any], enable_db_logging: bool = True)

        Initialize logger.

        Parameters:

        Name Type Description Default
        node MessageEmitter[Any, Any]

        Node to log interactions for

        		 required
        enable_db_logging bool

        Whether to enable logging

        		 True
        Source code in src/llmling_agent/messaging/node_logger.py 
        26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
        def __init__(self, node: MessageEmitter[Any, Any], enable_db_logging: bool = True):
    """Initialize logger.

    Args:
        node: Node to log interactions for
        enable_db_logging: Whether to enable logging
    """
    self.node = node
    self.enable_db_logging = enable_db_logging
    self.conversation_id = str(uuid4())
    self.message_history = ChatMessageContainer()
    self.toolcall_history = EventedList[ToolCallInfo]()

    # Initialize conversation record if enabled
    if enable_db_logging:
        self.init_conversation()
        # Connect to the combined signal to capture all messages
        node.message_received.connect(self.log_message)
        node.message_sent.connect(self.log_message)
        node.tool_used.connect(self.log_tool_call)

        clear_state 

        clear_state()

        Clear node state.

        Source code in src/llmling_agent/messaging/node_logger.py 
        47
48
49
50
        def clear_state(self):
    """Clear node state."""
    self.message_history.clear()
    self.toolcall_history.clear()

        init_conversation 

        init_conversation()

        Create initial conversation record.

        Source code in src/llmling_agent/messaging/node_logger.py 
        62
63
64
65
66
67
        def init_conversation(self):
    """Create initial conversation record."""
    self.node.context.storage.log_conversation_sync(
        conversation_id=self.conversation_id,
        node_name=self.node.name,
    )

        log_message 

        log_message(message: ChatMessage)

        Handle message from chat signal.

        Source code in src/llmling_agent/messaging/node_logger.py 
        69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
        def log_message(self, message: ChatMessage):
    """Handle message from chat signal."""
    self.message_history.append(message)

    if not self.enable_db_logging:
        return
    self.node.context.storage.log_message_sync(
        conversation_id=self.conversation_id,
        content=str(message.content),
        role=message.role,
        name=message.name,
        cost_info=message.cost_info,
        model=message.model,
        response_time=message.response_time,
        forwarded_from=message.forwarded_from,
    )

        log_tool_call 

        log_tool_call(tool_call: ToolCallInfo)

        Handle tool usage signal.

        Source code in src/llmling_agent/messaging/node_logger.py 
        86
87
88
89
90
91
92
93
94
95
96
        def log_tool_call(self, tool_call: ToolCallInfo):
    """Handle tool usage signal."""
    self.toolcall_history.append(tool_call)

    if not self.enable_db_logging:
        return
    self.node.context.storage.log_tool_call_sync(
        conversation_id=self.conversation_id,
        message_id=tool_call.message_id or "",
        tool_call=tool_call,
    )