Agent Callbacks

Callback system for monitoring agent execution, collecting metrics, and handling events during agent workflows.

Core Classes

AgentCallbackHandler

Description: Base callback handler for agent monitoring and logging

Parameters:

• agent_id (str, optional): Unique identifier for the agent
• log_level (str): Logging level (default: "INFO")

Returns: AgentCallbackHandler instance

Example:

from recoagent.agents.callbacks import AgentCallbackHandler

# Create callback handler
callback = AgentCallbackHandler(agent_id="agent_001")

# Use with agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=[callback]
)

# Metrics are automatically collected
result = agent.invoke({"query": "What is AI?"})
print(f"Duration: {callback.metrics.duration_ms}ms")

MetricsCallbackHandler

Description: Specialized callback for detailed metrics collection

Parameters:

• agent_id (str, optional): Agent identifier
• collect_detailed_metrics (bool): Enable detailed metrics
• export_format (str): Export format ("json", "prometheus")

Returns: MetricsCallbackHandler instance

Example:

from recoagent.agents.callbacks import MetricsCallbackHandler

# Create metrics handler
metrics_handler = MetricsCallbackHandler(
    agent_id="agent_001",
    collect_detailed_metrics=True,
    export_format="prometheus"
)

# Use with agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=[metrics_handler]
)

# Export metrics
prometheus_metrics = metrics_handler.export_metrics("prometheus")

AgentMetrics

Description: Data class for storing agent execution metrics

Fields:

• start_time (datetime): Execution start time
• end_time (datetime, optional): Execution end time
• duration_ms (float): Total duration in milliseconds
• step_count (int): Number of execution steps
• tool_calls (int): Number of tool calls
• llm_calls (int): Number of LLM calls
• tokens_used (int): Total tokens used
• cost_usd (float): Total cost in USD
• errors (List[str]): List of errors encountered
• metadata (Dict): Additional metadata

Usage Examples

Basic Monitoring

from recoagent.agents.callbacks import AgentCallbackHandler
from recoagent.agents import RAGAgentGraph

# Create callback
callback = AgentCallbackHandler(agent_id="monitoring_agent")

# Create agent with callback
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=[callback]
)

# Execute and monitor
result = agent.invoke({"query": "Explain quantum computing"})

# Access metrics
print(f"Execution time: {callback.metrics.duration_ms}ms")
print(f"Steps taken: {callback.metrics.step_count}")
print(f"Tool calls: {callback.metrics.tool_calls}")
print(f"LLM calls: {callback.metrics.llm_calls}")
print(f"Cost: ${callback.metrics.cost_usd:.4f}")

Detailed Metrics Collection

from recoagent.agents.callbacks import MetricsCallbackHandler

# Create detailed metrics handler
metrics_handler = MetricsCallbackHandler(
    agent_id="detailed_agent",
    collect_detailed_metrics=True
)

# Use with agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=[metrics_handler]
)

# Execute multiple queries
queries = [
    "What is machine learning?",
    "Explain neural networks",
    "How does deep learning work?"
]

for query in queries:
    result = agent.invoke({"query": query})
    
# Get aggregated metrics
total_metrics = metrics_handler.get_aggregated_metrics()
print(f"Total queries: {total_metrics['total_queries']}")
print(f"Average duration: {total_metrics['avg_duration_ms']}ms")
print(f"Total cost: ${total_metrics['total_cost_usd']:.4f}")

Custom Callback Handler

from recoagent.agents.callbacks import AgentCallbackHandler
from typing import Dict, Any, List

class CustomCallbackHandler(AgentCallbackHandler):
    """Custom callback for business-specific monitoring."""
    
    def __init__(self, business_rules: Dict[str, Any]):
        super().__init__()
        self.business_rules = business_rules
        self.business_metrics = {}
    
    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) -> None:
        """Custom LLM start handling."""
        super().on_llm_start(serialized, prompts, **kwargs)
        
        # Track business-specific metrics
        if "pricing" in prompts[0].lower():
            self.business_metrics["pricing_queries"] = self.business_metrics.get("pricing_queries", 0) + 1
    
    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) -> None:
        """Custom tool start handling."""
        super().on_tool_start(serialized, input_str, **kwargs)
        
        # Track tool usage by business category
        tool_name = serialized.get("name", "unknown")
        if tool_name in self.business_rules.get("tracked_tools", []):
            self.business_metrics[f"{tool_name}_usage"] = self.business_metrics.get(f"{tool_name}_usage", 0) + 1
    
    def get_business_metrics(self) -> Dict[str, Any]:
        """Get business-specific metrics."""
        return self.business_metrics

# Create custom callback
business_callback = CustomCallbackHandler({
    "tracked_tools": ["retrieval", "web_search"],
    "business_categories": ["pricing", "support", "technical"]
})

# Use with agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=[business_callback]
)

Multi-Callback System

from recoagent.agents.callbacks import AgentCallbackHandler, MetricsCallbackHandler

# Create multiple callbacks
monitoring_callback = AgentCallbackHandler(agent_id="monitor")
metrics_callback = MetricsCallbackHandler(agent_id="metrics")
business_callback = CustomCallbackHandler(business_rules)

# Combine callbacks
all_callbacks = [monitoring_callback, metrics_callback, business_callback]

# Use with agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    callbacks=all_callbacks
)

# Execute and collect from all callbacks
result = agent.invoke({"query": "Business query"})

# Access metrics from different callbacks
print("Monitoring metrics:", monitoring_callback.metrics)
print("Detailed metrics:", metrics_callback.get_aggregated_metrics())
print("Business metrics:", business_callback.get_business_metrics())

Metrics Export and Analysis

from recoagent.agents.callbacks import MetricsCallbackHandler
import json

# Create metrics handler
metrics_handler = MetricsCallbackHandler(
    collect_detailed_metrics=True,
    export_format="json"
)

# Execute queries
agent = RAGAgentGraph(callbacks=[metrics_handler])
# ... execute queries ...

# Export metrics in different formats
json_metrics = metrics_handler.export_metrics("json")
prometheus_metrics = metrics_handler.export_metrics("prometheus")

# Save to file
with open("agent_metrics.json", "w") as f:
    json.dump(json_metrics, f, indent=2)

# Get performance insights
insights = metrics_handler.get_performance_insights()
print(f"Average response time: {insights['avg_response_time']}ms")
print(f"Cost per query: ${insights['cost_per_query']:.4f}")
print(f"Error rate: {insights['error_rate']:.2%}")

API Reference

AgentCallbackHandler Methods

on_llm_start(serialized: Dict, prompts: List[str]) -> None

Called when LLM execution starts

Parameters:

• serialized (Dict): LLM configuration
• prompts (List[str]): Input prompts

on_llm_end(response: LLMResult) -> None

Called when LLM execution ends

Parameters:

• response (LLMResult): LLM response

on_tool_start(serialized: Dict, input_str: str) -> None

Called when tool execution starts

Parameters:

• serialized (Dict): Tool configuration
• input_str (str): Tool input

on_tool_end(output: str) -> None

Called when tool execution ends

Parameters:

• output (str): Tool output

on_agent_action(action: AgentAction) -> None

Called when agent takes an action

Parameters:

• action (AgentAction): Agent action

on_agent_finish(finish: AgentFinish) -> None

Called when agent finishes

Parameters:

• finish (AgentFinish): Agent finish result

MetricsCallbackHandler Methods

export_metrics(format: str) -> str

Export metrics in specified format

Parameters:

• format (str): Export format ("json", "prometheus")

Returns: Formatted metrics string

get_aggregated_metrics() -> Dict

Get aggregated metrics across all executions

Returns: Dictionary with aggregated metrics

get_performance_insights() -> Dict

Get performance insights and recommendations

Returns: Dictionary with insights

reset_metrics() -> None

Reset all collected metrics

See Also

• Agent Graphs - Agent state machines
• Agent Tools - Available tools
• Agent Policies - Safety and governance
• Agent Middleware - Middleware components