Plugin

autogen.beta.agent.Plugin #

Plugin(*, prompt=(), hitl_hook=None, tools=(), middleware=(), observers=(), dependencies=None, variables=None)

Source code in autogen/beta/agent.py

def __init__(
    self,
    *,
    prompt: PromptType | Iterable[PromptType] = (),
    hitl_hook: HumanHook | None = None,
    tools: Iterable[Callable[..., Any] | Tool] = (),
    middleware: Iterable[MiddlewareFactory] = (),
    observers: Iterable[Observer] = (),
    dependencies: dict[Any, Any] | None = None,
    variables: dict[Any, Any] | None = None,
) -> None:
    self._tools = list(tools)
    self._middleware = list(middleware)
    self._observers = list(observers)
    self._dependencies = dependencies or {}
    self._variables = variables or {}
    self._hitl_hook = hitl_hook

    self._system_prompt: list[str] = []
    self._dynamic_prompt: list[Callable[[ModelRequest, Context], Awaitable[str]]] = []

    if isinstance(prompt, str) or callable(prompt):
        prompt = [prompt]
    for p in prompt:
        if isinstance(p, str):
            self._system_prompt.append(p)
        else:
            self._dynamic_prompt.append(_wrap_prompt_hook(p))

register #

register(agent)

Apply this plugin's contributions to an Agent instance.

Source code in autogen/beta/agent.py

def register(self, agent: "Agent[Any]") -> None:
    """Apply this plugin's contributions to an Agent instance."""
    for t in self._tools:
        agent.add_tool(t)

    for m in self._middleware:
        agent.add_middleware(m)

    if self._hitl_hook is not None:
        if agent._hitl_hook is not None:
            warnings.warn(
                f"Agent '{agent.name}' already has a HITL hook; the plugin's hook will be ignored.",
                stacklevel=2,
            )
        else:
            agent._hitl_hook = wrap_hitl(self._hitl_hook)

    agent._agent_dependencies = self._dependencies | agent._agent_dependencies
    agent._agent_variables.update(self._variables)

    agent._observers.extend(self._observers)
    agent._system_prompt.extend(self._system_prompt)
    agent._dynamic_prompt.extend(self._dynamic_prompt)

hitl_hook #

hitl_hook(func)

Source code in autogen/beta/agent.py

def hitl_hook(self, func: HumanHook) -> HumanHook:
    if self._hitl_hook is not None:
        warnings.warn(
            "You already set HITL hook, provided value overrides it",
            category=RuntimeWarning,
            stacklevel=2,
        )
    self._hitl_hook = func
    return func

prompt #

prompt(func: PromptHook) -> PromptHook

prompt(func: None = None) -> Callable[[PromptHook], PromptHook]

prompt(func=None)

Source code in autogen/beta/agent.py

def prompt(
    self,
    func: PromptHook | None = None,
) -> PromptHook | Callable[[PromptHook], PromptHook]:
    def wrapper(f: PromptHook) -> PromptHook:
        self._dynamic_prompt.append(_wrap_prompt_hook(f))
        return f

    if func:
        return wrapper(func)
    return wrapper

tool #

tool(function: Callable[..., Any], *, name: str | None = None, description: str | None = None, schema: FunctionParameters | None = None, sync_to_thread: bool = True, middleware: Iterable[ToolMiddleware] = ()) -> FunctionTool

tool(function: None = None, *, name: str | None = None, description: str | None = None, schema: FunctionParameters | None = None, sync_to_thread: bool = True, middleware: Iterable[ToolMiddleware] = ()) -> Callable[[Callable[..., Any]], FunctionTool]

tool(function=None, *, name=None, description=None, schema=None, sync_to_thread=True, middleware=())

Source code in autogen/beta/agent.py

def tool(
    self,
    function: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    schema: FunctionParameters | None = None,
    sync_to_thread: bool = True,
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    def make_tool(f: Callable[..., Any]) -> FunctionTool:
        t = tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self._tools.append(t)
        return t

    if function:
        return make_tool(function)
    return make_tool

observer #

observer(condition: ClassInfo | Condition, callback: Callable[..., Any]) -> Callable[..., Any]

observer(condition: ClassInfo | Condition) -> Callable[[Callable[..., Any]], Callable[..., Any]]

observer(condition, callback=None)

Source code in autogen/beta/agent.py

def observer(
    self,
    condition: ClassInfo | Condition,
    callback: Callable[..., Any] | None = None,
) -> Callable[..., Any] | Callable[[Callable[..., Any]], Callable[..., Any]]:
    def wrapper(func: Callable[..., Any]) -> Callable[..., Any]:
        obs = observer_factory(condition, func)
        self._observers.append(obs)
        return func

    if callback is not None:
        return wrapper(callback)
    return wrapper