42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
"""Agent 专属 Protocol 端口。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
|
|
|
|
import pluggy
|
|
|
|
if TYPE_CHECKING:
|
|
from core.agent.types import LoopResult, Step
|
|
|
|
hookspec = pluggy.HookspecMarker("agent_loop")
|
|
hookimpl = pluggy.HookimplMarker("agent_loop")
|
|
|
|
|
|
@runtime_checkable
|
|
class ToolDispatcher(Protocol):
|
|
"""Agent 工具调度端口。无效工具名抛 ValueError。"""
|
|
|
|
async def dispatch(
|
|
self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
|
|
) -> str: ...
|
|
|
|
|
|
class AgentLoopSpec:
|
|
"""AgentLoop 生命周期扩展点。
|
|
|
|
每个 hookimpl 可选择观察（返回 None）或变换（返回值）。
|
|
"""
|
|
|
|
@hookspec
|
|
async def before_step(self, iteration: int, messages: list[dict[str, Any]]) -> None: ...
|
|
|
|
@hookspec
|
|
async def after_tool(self, iteration: int, step: Step) -> str | None: ...
|
|
|
|
@hookspec
|
|
async def after_step(self, iteration: int, messages: list[dict[str, Any]]) -> None: ...
|
|
|
|
@hookspec
|
|
async def on_finish(self, result: LoopResult) -> None: ...
|