Plugins

from definable.agent import Agent
from definable.agent.plugin.builtin import LoggingPlugin, MetricsPlugin, CachingPlugin

metrics = MetricsPlugin()

agent = Agent(
    model="gpt-4o",
    plugins=[LoggingPlugin(), metrics, CachingPlugin()],
)

output = await agent.arun("Hello")
print(metrics.last.total_duration_ms)  # e.g., 250.5

from definable.agent.plugin.builtin import LoggingPlugin

plugin = LoggingPlugin(
    verbose=False,  # True for full state details
    log_fn=None,    # Custom log function (default: log_info)
)

from definable.agent.plugin.builtin import MetricsPlugin

metrics = MetricsPlugin(max_history=100)

agent = Agent(model="gpt-4o", plugins=[metrics])
await agent.arun("Hello")

print(metrics.last.total_duration_ms)     # Latest run
print(metrics.last.phase_durations)       # {"prepare": 5.2, "invoke_loop": 200.1, ...}
print(metrics.last.tool_call_count)       # 3
print(metrics.average_duration_ms)        # Average across all runs
print(len(metrics.history))               # Run count

from definable.agent.plugin.builtin import CachingPlugin

cache = CachingPlugin(
    max_size=256,       # Max cached responses
    ttl_seconds=0,      # 0 = no expiry
)

agent = Agent(model="gpt-4o", plugins=[cache])

r1 = await agent.arun("What is 2+2?")  # Hits model
r2 = await agent.arun("What is 2+2?")  # Returns cached response

print(cache.hit_count)   # 1
print(cache.miss_count)  # 1
print(cache.size)        # 1
cache.clear()            # Reset cache and counters

from definable.agent.plugin import Plugin

class RateLimitPlugin(Plugin):
    def __init__(self, max_calls_per_minute: int = 60):
        self._limit = max_calls_per_minute
        self._calls: list[float] = []

    @property
    def name(self) -> str:
        return "rate-limit"

    @property
    def description(self) -> str:
        return f"Rate limits to {self._limit} calls/minute."

    @property
    def modifies(self):
        return frozenset({"invoke_loop"})

    async def on_load(self, agent):
        agent.pipeline.hook("before:invoke_loop", self._check_rate, priority=100)

    async def on_unload(self, agent):
        agent.pipeline.remove_hook("before:invoke_loop", self._check_rate)

    async def _check_rate(self, state, agent):
        import time
        now = time.time()
        self._calls = [t for t in self._calls if now - t < 60]
        if len(self._calls) >= self._limit:
            raise RuntimeError("Rate limit exceeded")
        self._calls.append(now)

agent = Agent(model="gpt-4o")

# Add plugins after construction
agent.plugin_registry.add(LoggingPlugin())
agent.plugin_registry.add(MetricsPlugin())

# Check state
print(agent.plugin_registry.plugin_names)   # ["logging", "metrics"]
print(agent.plugin_registry.loaded_names)   # [] (not loaded yet)

# Plugins load on first arun()
await agent.arun("Hello")
print(agent.plugin_registry.loaded_names)   # ["logging", "metrics"]

# Disable without unloading
agent.plugin_registry.disable("logging")

# Unload a specific plugin
await agent.plugin_registry.unload_one("metrics", agent)

class AnalyticsPlugin(Plugin):
    @property
    def name(self):
        return "analytics"

    @property
    def requires(self):
        return frozenset({"metrics"})  # MetricsPlugin must load first

    @property
    def conflicts(self):
        return frozenset({"legacy-analytics"})  # Cannot coexist

    async def on_load(self, agent):
        metrics = agent.plugin_registry.get("metrics")
        # Use metrics plugin...

from definable.agent import Plugin, PluginRegistry
from definable.agent.plugin.builtin import LoggingPlugin, MetricsPlugin, CachingPlugin