feat(core/agent): 添加 ToolDispatcher Protocol 和 AgentLoopSpec hookspec
ToolDispatcher async + context 参数。AgentLoopSpec 四个 async 生命周期 hook。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,45 @@
|
|||||||
|
"""Agent 专属 Protocol 端口。"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any, Protocol, runtime_checkable
|
||||||
|
|
||||||
|
import pluggy
|
||||||
|
|
||||||
|
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: ...
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
"""core/agent/protocols.py 单元测试。"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import pluggy
|
||||||
|
|
||||||
|
from core.agent.protocols import AgentLoopSpec, ToolDispatcher
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeDispatcher:
|
||||||
|
async def dispatch(
|
||||||
|
self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
|
||||||
|
) -> str:
|
||||||
|
return f"executed {tool_name}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_fake_dispatcher_satisfies_protocol() -> None:
|
||||||
|
assert isinstance(_FakeDispatcher(), ToolDispatcher)
|
||||||
|
|
||||||
|
def test_plain_object_not_dispatcher() -> None:
|
||||||
|
assert not isinstance(object(), ToolDispatcher)
|
||||||
|
|
||||||
|
def test_hookspec_can_register() -> None:
|
||||||
|
pm = pluggy.PluginManager("agent_loop")
|
||||||
|
pm.add_hookspecs(AgentLoopSpec)
|
||||||
|
assert pm.hook.before_step is not None
|
||||||
|
assert pm.hook.after_tool is not None
|
||||||
|
assert pm.hook.after_step is not None
|
||||||
|
assert pm.hook.on_finish is not None
|
||||||
Reference in New Issue
Block a user