feat(skill-evolution): add SkillReviewMiddleware to trigger post-turn review
Implements SkillReviewMiddleware that hooks into after_agent to call maybe_start_review() as a background task after each agent turn, without blocking the HTTP response. Exceptions are caught and logged only. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
1 parent
ef3ef433c1
commit
0e7a1de4db
2 files changed
+126
No files matched your search
+56
@@ -0,0 +1,56 @@
|
||||
"""SkillReviewMiddleware — triggers background skill review after each agent turn.
|
||||
|
||||
Hooks into ``after_agent``: once the lead agent graph has finished a full turn,
|
||||
this middleware counts the tool calls made and, if the threshold is met, hands
|
||||
the conversation messages to the SkillReviewWorker as a background asyncio task.
|
||||
|
||||
No blocking I/O or LLM calls happen in-band — the review runs fully in the
|
||||
background and never delays the HTTP response.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, override
|
||||
|
||||
from langchain.agents import AgentState
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
from deerflow.config.app_config import get_app_config
|
||||
from deerflow.skills.review_worker import maybe_start_review
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SkillReviewMiddlewareState(AgentState):
|
||||
"""Minimal state schema — only needs ``messages``."""
|
||||
pass
|
||||
|
||||
|
||||
class SkillReviewMiddleware(AgentMiddleware[SkillReviewMiddlewareState]):
|
||||
"""Dispatch a background skill review after every substantial agent turn.
|
||||
|
||||
'Substantial' is defined by ``skill_evolution.review_worker.min_tool_calls``
|
||||
(default 3). Turns below the threshold are silently skipped to avoid
|
||||
wasting LLM tokens on trivial exchanges.
|
||||
"""
|
||||
|
||||
state_schema = SkillReviewMiddlewareState
|
||||
|
||||
@override
|
||||
async def after_agent(self, state: Any, runtime: Runtime) -> None:
|
||||
"""Fire background review; never raises — failures are logged only."""
|
||||
try:
|
||||
app_config = get_app_config()
|
||||
messages = state.get("messages", []) if hasattr(state, "get") else []
|
||||
thread_id: str = (
|
||||
(runtime.config or {})
|
||||
.get("configurable", {})
|
||||
.get("thread_id", "")
|
||||
)
|
||||
if not thread_id:
|
||||
return
|
||||
maybe_start_review(thread_id, messages, app_config)
|
||||
except Exception:
|
||||
logger.exception("SkillReviewMiddleware.after_agent: unexpected error (ignored)")
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Tests for SkillReviewMiddleware."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
|
||||
from deerflow.agents.middlewares.skill_review_middleware import SkillReviewMiddleware
|
||||
|
||||
|
||||
def _make_state(messages):
|
||||
state = MagicMock()
|
||||
state.get.side_effect = lambda key, default=None: messages if key == "messages" else default
|
||||
return state
|
||||
|
||||
|
||||
def _make_runtime(thread_id="thread-abc"):
|
||||
runtime = MagicMock()
|
||||
runtime.config = {"configurable": {"thread_id": thread_id}}
|
||||
return runtime
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_after_agent_calls_maybe_start_review():
|
||||
"""When enabled and tool calls exist, maybe_start_review is called with correct thread_id."""
|
||||
middleware = SkillReviewMiddleware()
|
||||
|
||||
ai_msg = MagicMock(spec=AIMessage)
|
||||
tc = MagicMock()
|
||||
tc.name = "bash"
|
||||
ai_msg.tool_calls = [tc]
|
||||
|
||||
state = _make_state([HumanMessage(content="hi"), ai_msg])
|
||||
runtime = _make_runtime("thread-xyz")
|
||||
|
||||
app_cfg = MagicMock()
|
||||
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", return_value=app_cfg):
|
||||
with patch("deerflow.agents.middlewares.skill_review_middleware.maybe_start_review") as mock_review:
|
||||
await middleware.after_agent(state, runtime)
|
||||
|
||||
mock_review.assert_called_once()
|
||||
call_args = mock_review.call_args
|
||||
assert call_args[0][0] == "thread-xyz" # thread_id
|
||||
assert call_args[0][2] is app_cfg # app_config
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_after_agent_no_thread_id_skips_review():
|
||||
"""When thread_id is missing, maybe_start_review is NOT called."""
|
||||
middleware = SkillReviewMiddleware()
|
||||
state = _make_state([HumanMessage(content="hi")])
|
||||
runtime = _make_runtime("")
|
||||
|
||||
app_cfg = MagicMock()
|
||||
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", return_value=app_cfg):
|
||||
with patch("deerflow.agents.middlewares.skill_review_middleware.maybe_start_review") as mock_review:
|
||||
await middleware.after_agent(state, runtime)
|
||||
|
||||
mock_review.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_after_agent_exception_is_swallowed():
|
||||
"""Exceptions in after_agent must not propagate."""
|
||||
middleware = SkillReviewMiddleware()
|
||||
state = _make_state([HumanMessage(content="hi")])
|
||||
runtime = _make_runtime("thread-err")
|
||||
|
||||
with patch("deerflow.agents.middlewares.skill_review_middleware.get_app_config", side_effect=RuntimeError("boom")):
|
||||
# Must not raise
|
||||
await middleware.after_agent(state, runtime)
|
||||
Reference in new issue
Block a user