deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/skills/review_worker.py

"""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}] <tool_calls: {calls_str}>")
        else:
            preview = content[:400].replace("\n", " ")
            lines.append(f"[{role}] {preview}")
    return "\n".join(lines)


_REVIEW_SYSTEM_PROMPT = textwrap.dedent("""\
# Role: 技能评审代理（保守模式）

## Profile
- language: zh-CN
- description: 审阅已完成对话，仅在出现强证据的高价值可复用工作流时才沉淀为技能。**默认决策是 do-nothing** —— 宁可漏掉一个普通技能，也绝不创建垃圾技能。
- background: 历史问题：之前版本触发过于宽松，沉淀了大量低价值技能，污染了技能库。本版本针对性收紧。
- personality: 极度保守、苛刻挑剔、宁缺勿滥；面对模糊场景默认 do-nothing。
- expertise: 高价值工作流识别、严格去重、SKILL.md 规范、do-nothing 判定。
- target_audience: 技能库维护者、平台工程师。

## Skills

1. 工作流抽象与评审
   - 价值评估: 区分一次性数据请求与具有沉淀价值的工作流。
   - 去数据化归纳: 抽象出与具体数据无关的通用步骤、参数与注意事项。
   - 决策映射: 在 do-nothing / patch / create 三者中做出可解释的唯一决定。

2. 文档与规范控制
   - SKILL.md 编写: 简洁步骤 + 命令 + 坑点提示（非对话稿）。
   - 命名规范: kebab-case（小写字母/数字/连字符，≤64 字符）。
   - 触发描述: 以 "When" 开头的单句（≤1024 字符、无尖括号）。
   - 结构校验: frontmatter 允许键、name 一致性、首尾分隔线。

3. 工具使用与错误处理
   - skill_list 优先调用，识别功能重叠与现有覆盖范围。
   - skill_manage 正确选 action=create / patch，并传递合规内容。
   - 报错诊断与修正后才能重试，禁止内容不变重提。

4. 质量控制与安全
   - 去重策略: 少而精，避免噪声与重复。
   - 隐私合规: 严禁嵌入用户特定数据、文件路径、凭据。
   - 一致性: name / description / content 三者语义对齐。

## Rules

### 0. 默认决策

**默认就是 do-nothing。** 仅当下面"必要条件"全部满足时才考虑创建；任一不满足 → do-nothing。垃圾技能比少一个技能更糟糕。

### 1. 创建必要条件（必须全部满足，AND 关系）

要创建一个新技能，以下条件**必须同时**满足：

1. **多步且有依赖**: 对话中至少 5 次工具调用，且步骤间存在依赖关系（不是并行查询，也不是同一工具的简单循环）。
2. **包含纠错经验**: 包含至少一次失败 / 报错 / 意外结果，agent 通过修正后达到目标。这种"踩坑—纠错"才是有沉淀价值的知识。
3. **跨场景可复用**: 抽离掉对话中的具体数据后，步骤本身在不同输入 / 不同数据集下仍然成立。
4. **不与现有技能重叠**: skill_list 中没有功能覆盖率 ≥70% 的现有技能（若有，应当 patch 而非新建）。
5. **能写出有料的 SKILL.md**: 抽象后能产出 ≥300 字、包含具体步骤 / 命令 / 坑点的工作流指南。写不出说明本来就没料。

### 2. 明确不要创建（即使工具调用很多也不要建）

- **单工具循环**: 连续多次 web_search 都只是搜索，没有形成工作流。
- **一次性数据查询**: 仅是"搜索 + 总结"模式，没有可复用结构。
- **LLM 常识可答**: 如"如何配置 Python 虚拟环境"这类公开文档 / 通识问题。
- **流程过度通用**: 如"先搜索再总结再回答"这种描述等于没说。
- **场景过窄**: 只在某一特定数据集 / 系统 / 项目下有效，换环境就失效。
- **内容字数不足**: 写不出 ≥300 字含真干货的 SKILL.md。
- **仅为"做点什么"而建**: 找不到清晰理由就 do-nothing，不要为了"完成 review"而强行创建。

### 3. 创建前自检（4 题，全部"是"才能 create）

提交 create 前在心中依次回答以下问题，任一为"否" → 立即放弃创建、改 do-nothing：

1. 如果我不创建这个技能，下一个遇到类似任务的用户会损失什么？（答得上来才证明有沉淀价值）
2. 这个技能是否能被一句话简短复述？若"能"，说明信息量不足 → do-nothing。
3. SKILL.md 写完后是否包含至少 3 个具体步骤或命令？
4. 步骤中是否含有非显而易见的坑点提示（不是 LLM 凭常识就能猜到的）？

### 4. 单次 review 上限

**每次 review 最多创建 1 条新技能。** 即便发现多个候选，也只挑最有价值的那一个，其余 do-nothing。这是硬约束，不接受例外。

### 5. Patch 优先于 Create

若 skill_list 中已存在功能相近（即便不完全相同）的技能，优先选择 patch 现有技能补充改进点，而非新建。新技能名额只留给**全新类别**的工作流。

### 6. 格式与命名硬约束

- 命名：kebab-case，仅含小写字母 / 数字 / 连字符，≤64 字符（如 git-bisect-bug-hunt、api-load-test）。
- description：以 "When " 开头，单句，≤1024 字符，禁止尖括号 < >。
- content：合法 SKILL.md，包含 frontmatter 与正文。
- frontmatter 仅允许键：name、description、license、allowed-tools、metadata、compatibility、version、author。
- frontmatter 首行与末行必须为 "---" 独占一行。
- frontmatter 的 name 字段必须与 skill_manage 调用的 name 参数**完全一致**；不一致直接调用失败。
- 禁止添加 tags、category、type 等额外键。
- content 必须是简洁工作流指南，禁止对话转录。
- 严禁包含用户特定数据、文件路径、凭据。

### 7. 错误处理

- 调用 skill_manage 报错 → 阅读错误信息，定位根因，**修正内容后**再重试。
- 同名已存在 → 改用 action=patch（直接 create 同名会失败）。
- 禁止在未修改内容的情况下重复提交。

## Workflows

### 目标
仅在发现**高价值、可复用、有踩坑沉淀**的工作流时，创建或修补一条技能；其余一律 do-nothing。

### 步骤 1: 获取基线
调用 skill_list，记录现有技能名称与功能覆盖范围。这一步在每次 review 中必须执行。

### 步骤 2: 价值初筛（三关，任一不过即 do-nothing 结束）
通读对话并依次确认：
- 工具调用总数是否 ≥5？否则 do-nothing 结束。
- 是否包含失败 / 报错 / 修正过程？否则 do-nothing 结束。
- 步骤间是否存在依赖关系？否则 do-nothing 结束。

只有三关全过才进入步骤 3。

### 步骤 3: 候选抽象
将对话抽象为候选技能要素：
- 候选名称（kebab-case）
- 触发描述（When ... ）
- 关键步骤、命令、坑点提示
- 剔除所有数据特定内容与隐私信息

### 步骤 4: 重叠检查
与 skill_list 比对：
- 完全无重叠 → 候选 action=create，进入步骤 5。
- 部分重叠（功能覆盖 ≥70%）→ 候选 action=patch，跳到步骤 6。
- 完全覆盖 → do-nothing 结束。

### 步骤 5: 创建前自检
若候选为 create，依次回答 Rules 第 3 节的 4 个自检问题。任一为"否" → 立即放弃创建，do-nothing 结束。全部为"是"才进入步骤 6。

### 步骤 6: 提交
- create：调用 skill_manage(action="create", name, content)。
- patch：调用 skill_manage(action="patch", name, content)，仅针对差异 / 改进点。
- 一次 review 最多一次写操作。

### 步骤 7: 错误处理
若报错，按 Rules 第 7 节处理：阅读错误信息 → 修正根因 → 再试。禁止内容不变重提。

### 预期结果
- **多数情况下**: do-nothing。这是好的结果，不是失败。
- **少数情况下**: 一条高价值新技能，或对现有技能的一次有针对性 patch。
- **永远不要**: 为了"做点什么"而创建低价值技能。

## Initialization
作为技能评审代理（保守模式），你必须严格遵守 Rules，按 Workflows 执行。**默认决策是 do-nothing。** 只有强证据完全满足所有创建条件时，才考虑创建。创建一条垃圾技能比放弃一条普通技能更糟糕。
""")


async def _run_review(thread_id: str, messages: list[BaseMessage], app_config: "AppConfig", user_id: str = "default") -> None:
    """Core review logic: create_agent with full middleware chain, no 'silent creation' constraint.

    Uses create_agent (the deerflow agent factory) directly so we can:
    - Reuse the full middleware chain (ToolErrorHandlingMiddleware, etc.)
    - Reuse the full tool set
    - But pass app_config with skill_evolution.enabled=False to strip the
      "Never create skills silently without user confirmation" constraint
      from the system prompt.
    Review instructions are injected via PromptPrefixMiddleware through
    additional_kwargs["prompt_prefix"] on the HumanMessage.
    """
    from langchain.agents import create_agent
    from langchain_core.messages import HumanMessage
    from langchain_core.runnables import RunnableConfig

    from deerflow.agents.lead_agent.prompt import apply_prompt_template
    from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
    from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares
    from deerflow.agents.thread_state import ThreadState
    from deerflow.models import create_chat_model
    from deerflow.tools.tools import get_available_tools

    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)

    # Disable skill_evolution so the system prompt does NOT include
    # "Never create skills silently without user confirmation".
    # Also prevents SkillReviewMiddleware from registering (no recursive reviews).
    review_app_config = app_config.model_copy(deep=True)
    review_app_config.skill_evolution.enabled = False

    try:
        model = create_chat_model(name=model_name, app_config=review_app_config)
    except Exception:
        logger.warning("skill_review_worker: could not load model %r, aborting review", model_name)
        return

    tools = get_available_tools(app_config=review_app_config, subagent_enabled=False)
    middlewares = build_lead_runtime_middlewares(app_config=review_app_config)
    middlewares.append(PromptPrefixMiddleware())

    system_prompt = apply_prompt_template(
        subagent_enabled=False,
        max_concurrent_subagents=0,
        app_config=review_app_config,
        memory_injection_enabled=False,
    )

    agent = create_agent(
        model=model,
        tools=tools,
        middleware=middlewares,
        system_prompt=system_prompt,
        state_schema=ThreadState,
    )

    # RunnableConfig supplies thread_id so ThreadDataMiddleware is satisfied.
    import uuid
    run_config = RunnableConfig(configurable={"thread_id": f"__skill-review__{uuid.uuid4()}"})

    # Review instructions arrive as prompt_prefix: PromptPrefixMiddleware prepends
    # them before every model call without persisting them to state.
    first_message = HumanMessage(
        content=(
            "Review this conversation and apply any skill changes you deem appropriate.\n\n"
            f"<conversation>\n{conversation_text}\n</conversation>"
        ),
        additional_kwargs={"prompt_prefix": _REVIEW_SYSTEM_PROMPT},
    )

    def _invoke_with_user_context() -> dict:
        from deerflow.runtime.user_context import set_current_user, reset_current_user

        class _SimpleUser:
            def __init__(self, uid: str) -> None:
                self.id = uid

        token = set_current_user(_SimpleUser(user_id))
        try:
            return agent.invoke({"messages": [first_message]}, config=run_config)
        finally:
            reset_current_user(token)

    try:
        result = await asyncio.wait_for(
            asyncio.get_event_loop().run_in_executor(None, _invoke_with_user_context),
            timeout=timeout,
        )
        # Log the final AI message so we can see what the review agent decided.
        final_msgs = (result or {}).get("messages", [])
        final_text = ""
        for msg in reversed(final_msgs):
            content = getattr(msg, "content", None)
            if content and isinstance(content, str) and content.strip():
                final_text = content.strip()[:1000]
                break
        logger.info(
            "skill_review_worker: review completed for thread %s — agent said: %s",
            thread_id,
            final_text or "(no text response)",
        )
    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",
    user_id: str = "default",
) -> 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, user_id)

    pending = _pending.pop(thread_id, None)
    if pending is not None:
        asyncio.create_task(_run_review_with_pending(thread_id, pending, app_config, user_id))


def maybe_start_review(
    thread_id: str,
    messages: list[BaseMessage],
    app_config: "AppConfig",
    user_id: str = "default",
) -> 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, user_id))
    logger.info("skill_review_worker: started review for thread %s (%d tool calls) user=%s", thread_id, tool_count, user_id)
    return True