diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/review_worker.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/review_worker.py new file mode 100644 index 00000000..b73895f2 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/review_worker.py @@ -0,0 +1,183 @@ +"""Post-turn background skill review worker. + +After a conversation turn completes, SkillReviewMiddleware calls +``maybe_start_review()``. This module runs a lightweight LLM React agent +(with skill_manage + skill_list tools) that reads the conversation history +and decides whether to create or patch skills. + +Concurrency model: per-thread asyncio.Lock + a single _pending slot. +If a review is already running for the thread, the newer messages overwrite +the pending slot; once the current review finishes, it drains that slot. +""" + +from __future__ import annotations + +import asyncio +import logging +import textwrap +from typing import TYPE_CHECKING, Any + +from langchain_core.messages import BaseMessage + +if TYPE_CHECKING: + from deerflow.config.app_config import AppConfig + +logger = logging.getLogger(__name__) + +_locks: dict[str, asyncio.Lock] = {} +_pending: dict[str, list[BaseMessage]] = {} + + +def _get_lock(thread_id: str) -> asyncio.Lock: + if thread_id not in _locks: + _locks[thread_id] = asyncio.Lock() + return _locks[thread_id] + + +def extract_tool_call_count(messages: list[BaseMessage]) -> int: + """Count total tool calls across all AI messages in the conversation.""" + count = 0 + for msg in messages: + tool_calls: Any = None + if hasattr(msg, "tool_calls"): + tool_calls = msg.tool_calls + elif isinstance(msg, dict): + tool_calls = msg.get("tool_calls") + if tool_calls: + count += len(tool_calls) + return count + + +def _format_messages_for_review(messages: list[BaseMessage], max_messages: int) -> str: + """Serialize recent messages to a readable text block for the LLM reviewer.""" + recent = messages[-max_messages:] + lines: list[str] = [] + for msg in recent: + role = type(msg).__name__.replace("Message", "") + content = msg.content if isinstance(msg.content, str) else str(msg.content) + tool_calls = getattr(msg, "tool_calls", None) + if tool_calls: + calls_str = ", ".join( + f"{tc.name if hasattr(tc, 'name') else tc.get('name', '?')}()" + for tc in tool_calls + ) + lines.append(f"[{role}] ") + else: + preview = content[:400].replace("\n", " ") + lines.append(f"[{role}] {preview}") + return "\n".join(lines) + + +_REVIEW_SYSTEM_PROMPT = textwrap.dedent("""\ + You are a skill review agent. Your job is to analyze a completed conversation + and decide whether any reusable multi-step workflow should be saved as a skill. + + ## Decision rules + 1. Call skill_list first to see existing skills — avoid creating duplicates. + 2. CREATE a new skill if: the conversation demonstrates a non-trivial, repeatable + workflow (3+ distinct tool operations) that a future agent could reuse verbatim. + 3. PATCH an existing skill if: the conversation revealed a gap, error, or improvement + to a skill that was already invoked. + 4. DO NOTHING if: the task was one-off, data-specific, trivial, or already covered. + + ## Skill quality bar + - Name: kebab-case, describes WHAT the skill does (e.g. "git-bisect-bug-hunt"). + - Description: one sentence starting with "When" that states the trigger condition. + - Content: concise workflow guide (steps, commands, pitfalls) — NOT a transcript. + - Never embed user-specific data, file paths, or credentials. + + Be conservative. Fewer high-quality skills beat many noisy ones. +""") + + +async def _run_review(thread_id: str, messages: list[BaseMessage], app_config: "AppConfig") -> None: + """Core review logic: spin up a minimal React agent and invoke it.""" + from langgraph.prebuilt import create_react_agent + + from deerflow.models.factory import create_chat_model + from deerflow.tools.builtins.skill_tools import skill_list_tool + from deerflow.tools.skill_manage_tool import skill_manage_tool + + evolution_cfg = app_config.skill_evolution + timeout = evolution_cfg.review_worker.timeout_seconds + max_messages = evolution_cfg.review_worker.max_messages + model_name = evolution_cfg.review_worker.model_name + + conversation_text = _format_messages_for_review(messages, max_messages) + + try: + model = create_chat_model(model_name) + except Exception: + logger.warning("skill_review_worker: could not load model %r, aborting review", model_name) + return + + tools = [skill_list_tool, skill_manage_tool] + agent = create_react_agent(model, tools, prompt=_REVIEW_SYSTEM_PROMPT) + + user_message = ( + "Review this conversation and apply any skill changes you deem appropriate.\n\n" + f"\n{conversation_text}\n" + ) + + try: + await asyncio.wait_for( + asyncio.get_event_loop().run_in_executor( + None, + lambda: agent.invoke({"messages": [("user", user_message)]}), + ), + timeout=timeout, + ) + logger.info("skill_review_worker: review completed for thread %s", thread_id) + except TimeoutError: + logger.warning("skill_review_worker: review timed out after %ds (thread %s)", timeout, thread_id) + except Exception: + logger.exception("skill_review_worker: review failed for thread %s", thread_id) + + +async def _run_review_with_pending( + thread_id: str, + messages: list[BaseMessage], + app_config: "AppConfig", +) -> None: + """Run one review; if a newer snapshot arrived while we ran, drain it.""" + lock = _get_lock(thread_id) + async with lock: + await _run_review(thread_id, messages, app_config) + + pending = _pending.pop(thread_id, None) + if pending is not None: + asyncio.create_task(_run_review_with_pending(thread_id, pending, app_config)) + + +def maybe_start_review( + thread_id: str, + messages: list[BaseMessage], + app_config: "AppConfig", +) -> bool: + """Schedule a background review for ``thread_id``. + + Returns True if a new task was started, False if queued or skipped. + """ + evolution_cfg = app_config.skill_evolution + if not evolution_cfg.enabled or not evolution_cfg.review_worker.enabled: + return False + + tool_count = extract_tool_call_count(messages) + if tool_count < evolution_cfg.review_worker.min_tool_calls: + logger.debug( + "skill_review_worker: skipping review for thread %s (%d tool calls < min %d)", + thread_id, + tool_count, + evolution_cfg.review_worker.min_tool_calls, + ) + return False + + lock = _get_lock(thread_id) + if lock.locked(): + _pending[thread_id] = messages + logger.debug("skill_review_worker: queued review for thread %s (review in progress)", thread_id) + return False + + asyncio.create_task(_run_review_with_pending(thread_id, messages, app_config)) + logger.info("skill_review_worker: started review for thread %s (%d tool calls)", thread_id, tool_count) + return True diff --git a/offline-backend-20260512/backend/tests/test_skill_review_worker.py b/offline-backend-20260512/backend/tests/test_skill_review_worker.py index 19ebecef..549db4b0 100644 --- a/offline-backend-20260512/backend/tests/test_skill_review_worker.py +++ b/offline-backend-20260512/backend/tests/test_skill_review_worker.py @@ -18,3 +18,85 @@ def test_review_worker_config_from_dict(): assert cfg.review_worker.enabled is True assert cfg.review_worker.min_tool_calls == 5 assert cfg.review_worker.model_name == "my-model" + + +from unittest.mock import AsyncMock, MagicMock, patch +import asyncio +import pytest +from langchain_core.messages import HumanMessage + + +def _make_ai_message(tool_calls_data: list[dict]): + msg = MagicMock() + msg.__class__.__name__ = "AIMessage" + tcs = [] + for tc in tool_calls_data: + m = MagicMock() + m.name = tc["name"] + m.args = tc.get("args", {}) + tcs.append(m) + msg.tool_calls = tcs + return msg + + +def test_extract_tool_call_count_counts_all(): + from deerflow.skills.review_worker import extract_tool_call_count + msgs = [ + HumanMessage(content="do something"), + _make_ai_message([{"name": "bash"}, {"name": "read_file"}]), + _make_ai_message([{"name": "write_file"}]), + ] + assert extract_tool_call_count(msgs) == 3 + + +def test_extract_tool_call_count_empty(): + from deerflow.skills.review_worker import extract_tool_call_count + assert extract_tool_call_count([]) == 0 + + +def test_maybe_start_review_disabled_returns_false(): + from deerflow.skills.review_worker import maybe_start_review + cfg = MagicMock() + cfg.skill_evolution.enabled = False + cfg.skill_evolution.review_worker.enabled = True + cfg.skill_evolution.review_worker.min_tool_calls = 2 + msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])] + assert maybe_start_review("thread-1", msgs, cfg) is False + + +def test_maybe_start_review_too_few_tool_calls_returns_false(): + from deerflow.skills.review_worker import maybe_start_review + cfg = MagicMock() + cfg.skill_evolution.enabled = True + cfg.skill_evolution.review_worker.enabled = True + cfg.skill_evolution.review_worker.min_tool_calls = 5 + msgs = [_make_ai_message([{"name": "bash"}])] + assert maybe_start_review("thread-4", msgs, cfg) is False + + +def test_maybe_start_review_starts_task(): + from deerflow.skills import review_worker as rw + # Reset module state so thread-3 has no lock + rw._locks.pop("thread-3", None) + rw._pending.pop("thread-3", None) + + cfg = MagicMock() + cfg.skill_evolution.enabled = True + cfg.skill_evolution.review_worker.enabled = True + cfg.skill_evolution.review_worker.min_tool_calls = 2 + cfg.skill_evolution.review_worker.timeout_seconds = 30 + cfg.skill_evolution.review_worker.max_messages = 20 + cfg.skill_evolution.review_worker.model_name = None + + msgs = [_make_ai_message([{"name": "bash"}, {"name": "read_file"}, {"name": "write_file"}])] + + with patch("deerflow.skills.review_worker.asyncio") as mock_asyncio: + fake_lock = MagicMock() + fake_lock.locked.return_value = False + mock_asyncio.Lock.return_value = fake_lock + mock_asyncio.create_task = MagicMock() + + result = rw.maybe_start_review("thread-3", msgs, cfg) + + assert result is True + mock_asyncio.create_task.assert_called_once()