Callbacks
Callbacks in the Agent SDK provide hooks into the agent's lifecycle, allowing for custom functionality to be executed at various stages of an agent's run. They enable extensibility by allowing developers to integrate their own logic for tasks such as logging, cost management, and data anonymization.
Usage
You can add preprocessing and postprocessing hooks using callbacks, or write your own by subclassing AsyncCallbackHandler
.
Built-in Callbacks
Built-in callbacks can be used as follows:
from agent.callbacks import (
ImageRetentionCallback,
TrajectorySaverCallback,
BudgetManagerCallback,
LoggingCallback
)
agent = ComputerAgent(
model="anthropic/claude-3-5-sonnet-20241022",
tools=[computer],
callbacks=[
ImageRetentionCallback(only_n_most_recent_images=3),
TrajectorySaverCallback(trajectory_dir="trajectories"),
BudgetManagerCallback(max_budget=10.0, raise_error=True),
LoggingCallback(level=logging.INFO)
]
)
The following built-in callbacks are available:
- BudgetManagerCallback: Stops execution when budget exceeded
- LoggingCallback: Logs agent activities
- ImageRetentionCallback: Limits recent images in context
- TrajectorySaverCallback: Saves conversation trajectories
- PII Anonymization
Custom Callbacks
Create custom callbacks using knowlege of the callback lifecycle as described in Agent Lifecycle.
from agent.callbacks.base import AsyncCallbackHandler
class CustomCallback(AsyncCallbackHandler):
async def on_llm_start(self, messages):
"""Preprocess messages before LLM call"""
# Add custom preprocessing logic
return messages
async def on_llm_end(self, messages):
"""Postprocess messages after LLM call"""
# Add custom postprocessing logic
return messages
async def on_usage(self, usage):
"""Track usage information"""
print(f"Tokens used: {usage.total_tokens}")