Skip to content

Agent Event Monitoring

Learn how to track and monitor your QuantaLogic agent's lifecycle and performance using the built-in event monitoring system.

Overview

Event monitoring provides real-time insights into agent operations, enabling: - Debugging and troubleshooting - Performance tracking - Detailed operational transparency

Prerequisites

  • Python 3.12 or later
  • QuantaLogic library installed
  • DeepSeek API key (or alternative LLM provider)

Setting Up Event Monitoring

Python
1
2
3
4
5
6
7
8
import os
from quantalogic import Agent
from quantalogic.console_print_events import console_print_events
from quantalogic.console_print_token import console_print_token
from quantalogic.tools import LLMTool

# Set API key
os.environ["DEEPSEEK_API_KEY"] = "your-deepseek-key"

Creating an Agent with Event Listeners

Python
 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
# Initialize agent with event monitoring
agent = Agent(
    model_name="deepseek/deepseek-chat",
    tools=[LLMTool(
        model_name="deepseek/deepseek-chat", 
        name="deepseek_llm_tool", 
        on_token=console_print_token
    )]
)

# Configure event listeners
agent.event_emitter.on(
    event=[
        "task_complete",
        "task_think_start",
        "task_think_end",
        "tool_execution_start",
        "tool_execution_end",
        "error_max_iterations_reached",
        "memory_full",
        "memory_compacted",
        "memory_summary"
    ],
    listener=console_print_events
)

Supported Event Types

Event Type Description
task_complete Triggered when a task is successfully finished
task_think_start Marks the beginning of task analysis
task_think_end Indicates completion of task analysis
tool_execution_start Signals the start of a tool's execution
tool_execution_end Marks the end of a tool's execution
error_max_iterations_reached Warns about reaching maximum iteration limit
memory_full Indicates memory capacity has been reached
memory_compacted Shows memory has been optimized
memory_summary Provides an overview of memory usage

Best Practices

  • Use event monitoring for debugging complex tasks
  • Monitor performance and identify bottlenecks
  • Customize event listeners for specific tracking needs
  • Avoid adding too many event listeners to prevent performance overhead

Flexibility

Event monitoring can be customized to suit your specific observability requirements.

Example Task Execution

Python
1
2
3
# Execute a task with event monitoring
result = agent.solve_task("Create a complex data processing script")
print(result)

Customizing Event Listeners

You can create custom event listeners to: - Log events to a file - Send notifications - Perform advanced analytics - Integrate with monitoring systems

Python
1
2
3
4
5
6
7
8
def custom_event_listener(event_data):
    # Implement your custom event handling logic
    print(f"Custom Event: {event_data}")

agent.event_emitter.on(
    event=["task_complete"], 
    listener=custom_event_listener
)

Performance Consideration

Custom event listeners should be lightweight to avoid impacting agent performance.

Complete Code

Here's the full example:

Python
 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
import os
from quantalogic import Agent
from quantalogic.console_print_events import console_print_events
from quantalogic.console_print_token import console_print_token
from quantalogic.tools import LLMTool

# Set API key
os.environ["DEEPSEEK_API_KEY"] = "your-deepseek-key"

# Initialize agent with event monitoring
agent = Agent(
    model_name="deepseek/deepseek-chat",
    tools=[LLMTool(
        model_name="deepseek/deepseek-chat", 
        name="deepseek_llm_tool", 
        on_token=console_print_token
    )]
)

# Configure event listeners
agent.event_emitter.on(
    event=[
        "task_complete",
        "task_think_start",
        "task_think_end",
        "tool_execution_start",
        "tool_execution_end",
        "error_max_iterations_reached",
        "memory_full",
        "memory_compacted",
        "memory_summary"
    ],
    listener=console_print_events
)

# Execute a task with event monitoring
result = agent.solve_task("Create a complex data processing script")
print(result)

Next Steps