105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
from typing import Any, Generic
|
|
|
|
from .agent import Agent
|
|
from .run_context import RunContextWrapper, TContext
|
|
from .tool import Tool
|
|
|
|
|
|
class RunHooks(Generic[TContext]):
|
|
"""A class that receives callbacks on various lifecycle events in an agent run. Subclass and
|
|
override the methods you need.
|
|
"""
|
|
|
|
async def on_agent_start(
|
|
self, context: RunContextWrapper[TContext], agent: Agent[TContext]
|
|
) -> None:
|
|
"""Called before the agent is invoked. Called each time the current agent changes."""
|
|
pass
|
|
|
|
async def on_agent_end(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
output: Any,
|
|
) -> None:
|
|
"""Called when the agent produces a final output."""
|
|
pass
|
|
|
|
async def on_handoff(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
from_agent: Agent[TContext],
|
|
to_agent: Agent[TContext],
|
|
) -> None:
|
|
"""Called when a handoff occurs."""
|
|
pass
|
|
|
|
async def on_tool_start(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
tool: Tool,
|
|
) -> None:
|
|
"""Called before a tool is invoked."""
|
|
pass
|
|
|
|
async def on_tool_end(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
tool: Tool,
|
|
result: str,
|
|
) -> None:
|
|
"""Called after a tool is invoked."""
|
|
pass
|
|
|
|
|
|
class AgentHooks(Generic[TContext]):
|
|
"""A class that receives callbacks on various lifecycle events for a specific agent. You can
|
|
set this on `agent.hooks` to receive events for that specific agent.
|
|
|
|
Subclass and override the methods you need.
|
|
"""
|
|
|
|
async def on_start(self, context: RunContextWrapper[TContext], agent: Agent[TContext]) -> None:
|
|
"""Called before the agent is invoked. Called each time the running agent is changed to this
|
|
agent."""
|
|
pass
|
|
|
|
async def on_end(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
output: Any,
|
|
) -> None:
|
|
"""Called when the agent produces a final output."""
|
|
pass
|
|
|
|
async def on_handoff(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
source: Agent[TContext],
|
|
) -> None:
|
|
"""Called when the agent is being handed off to. The `source` is the agent that is handing
|
|
off to this agent."""
|
|
pass
|
|
|
|
async def on_tool_start(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
tool: Tool,
|
|
) -> None:
|
|
"""Called before a tool is invoked."""
|
|
pass
|
|
|
|
async def on_tool_end(
|
|
self,
|
|
context: RunContextWrapper[TContext],
|
|
agent: Agent[TContext],
|
|
tool: Tool,
|
|
result: str,
|
|
) -> None:
|
|
"""Called after a tool is invoked."""
|
|
pass
|