269 lines
10 KiB
Python
269 lines
10 KiB
Python
"""Request preprocessor — minimal preprocessing layer for unified REACT agent loop.
|
||
|
||
Replaces the 4-layer CostAwareRouter with a simple approach:
|
||
1. @skill:xxx prefix → explicit skill selection
|
||
2. Greeting/chitchat regex → DIRECT_CHAT (fast path)
|
||
3. Everything else → REACT (LLM decides tool usage in agent loop)
|
||
|
||
This follows the Hermes/Trae/Codex pattern: no intent prediction layer,
|
||
LLM sees full tool descriptions and decides autonomously.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from typing import TYPE_CHECKING
|
||
|
||
from agentkit.chat.skill_routing import (
|
||
ExecutionMode,
|
||
SkillRoutingResult,
|
||
build_skill_system_prompt,
|
||
parse_skill_prefix,
|
||
_resolve_execution_mode,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from agentkit.skills.registry import SkillRegistry
|
||
from agentkit.tools.base import Tool
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Regex patterns for zero-cost direct chat (no LLM call needed)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_GREETING_RE = re.compile(
|
||
r"^(你好|hi|hello|hey|嗨|哈喽|早上好|下午好|晚上好|good morning|good afternoon|good evening)"
|
||
r"\s*[!！.。?？]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
_CHAT_MODE_RE = re.compile(
|
||
r"^(谢谢|感谢|thanks|thank you|ok|好的|嗯|对|是|不是|没关系|再见|bye|goodbye)"
|
||
r"\s*[!！.。?？]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
_IDENTITY_RE = re.compile(
|
||
r"^(你是谁|你叫什么|你是什么|你是哪个|who are you|what are you|what's your name"
|
||
r"|介绍一下你自己|自我介绍|你叫啥|你叫什么名字|你的名字)"
|
||
r"\s*[?？!！.。]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# 中文知识问答：什么是X/解释X/定义X — 中文不需要空格分隔
|
||
# 仅匹配纯知识性问句，排除需要实时数据的请求（由 _TOOL_CONTEXT_RE 过滤）
|
||
# 支持尾部标点（？/！/。等），与 _GREETING_RE/_IDENTITY_RE 保持一致
|
||
_FACTUAL_CN_RE = re.compile(
|
||
r"^(什么是|解释一下|解释下|定义一下|定义|说说什么是|介绍下什么是)"
|
||
r"[\u4e00-\u9fa5a-zA-Z0-9 \t]+[?？!！.。]*$"
|
||
)
|
||
|
||
# English factual questions — requires whitespace separator
|
||
_FACTUAL_EN_RE = re.compile(
|
||
r"^(what\s+is|what's|define|explain)\s+[\u4e00-\u9fa5a-zA-Z0-9 \t]+[?？!！.。]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# 需要工具/实时数据的上下文关键词 — 出现这些词时不走 DIRECT_CHAT
|
||
# 包含中英文关键词，覆盖服务器/数据库/系统状态/配置文件等场景
|
||
_TOOL_CONTEXT_RE = re.compile(
|
||
r"(当前|现在|服务器|数据库|系统|状态|最新|实时|今天|昨天|本机|本地|线上|"
|
||
r"线上环境|生产环境|测试环境|配置文件|日志|进程|端口|IP|CPU|内存|磁盘|"
|
||
r"current|server|database|system\s+status|latest|realtime|today|yesterday|"
|
||
r"local|process|port|log|config\s+file)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# 纯算术：计算 1+2+3 / 算一下 15*23 — 仅匹配数字和运算符
|
||
# 不匹配含中文/字母的复杂表达式（如"计算斐波那契数列"）
|
||
_MATH_RE = re.compile(
|
||
r"^(计算|算一下|算下|calculate|compute)\s+[\d +\-*/().\t]+[?？!！.。]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# 纯翻译：翻译 X / translate X — 需要空格分隔，排除"翻译X为Y"格式
|
||
# 排除含工具上下文关键词的请求（如"翻译 这个配置文件"）
|
||
_TRANSLATION_RE = re.compile(
|
||
r"^(翻译|translate)\s+.+$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
class RequestPreprocessor:
|
||
"""Minimal preprocessing layer: regex fast-path + default REACT.
|
||
|
||
Design rationale:
|
||
- No HeuristicClassifier: keyword enumeration can never cover all colloquial expressions
|
||
- No IntentRouter: LLM blind-classification without tool context is unreliable
|
||
- No SemanticRouter: embedding similarity is not intent recognition
|
||
- LLM in the REACT agent loop sees full tool descriptions and decides autonomously
|
||
- This matches Codex/Trae/Hermes architecture: unified agent loop, no preprocessing layer
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
skill_registry: SkillRegistry | None = None,
|
||
default_tools: list[Tool] | None = None,
|
||
default_system_prompt: str | None = None,
|
||
default_model: str = "default",
|
||
default_agent_name: str = "default",
|
||
) -> None:
|
||
self._skill_registry = skill_registry
|
||
self._default_tools = default_tools or []
|
||
self._default_system_prompt = default_system_prompt
|
||
self._default_model = default_model
|
||
self._default_agent_name = default_agent_name
|
||
|
||
async def preprocess(
|
||
self,
|
||
content: str,
|
||
*,
|
||
skill_registry: SkillRegistry | None = None,
|
||
default_tools: list[Tool] | None = None,
|
||
default_system_prompt: str | None = None,
|
||
default_model: str | None = None,
|
||
default_agent_name: str | None = None,
|
||
session_id: str = "",
|
||
transparency: str = "SILENT",
|
||
) -> SkillRoutingResult:
|
||
"""Preprocess user input to determine the appropriate execution path.
|
||
|
||
Decision tree:
|
||
1. @skill:xxx prefix → explicit skill (SKILL_REACT or skill's execution_mode)
|
||
2. Greeting/chitchat/identity → DIRECT_CHAT (zero-cost fast path)
|
||
3. Everything else → REACT (LLM decides tool usage in agent loop)
|
||
"""
|
||
registry = skill_registry or self._skill_registry
|
||
tools = default_tools if default_tools is not None else self._default_tools
|
||
sys_prompt = (
|
||
default_system_prompt
|
||
if default_system_prompt is not None
|
||
else self._default_system_prompt
|
||
)
|
||
model = default_model or self._default_model
|
||
agent_name = default_agent_name or self._default_agent_name
|
||
|
||
# --- Layer 0: @skill:xxx prefix ---
|
||
explicit_skill, clean_content = parse_skill_prefix(content)
|
||
if explicit_skill and registry is not None:
|
||
result = self._resolve_explicit_skill(
|
||
explicit_skill, clean_content, registry, model, agent_name
|
||
)
|
||
return result
|
||
|
||
# --- Layer 1: Greeting/chitchat/identity regex (<1ms, zero tokens) ---
|
||
stripped = content.strip()
|
||
if self._is_trivial_input(stripped):
|
||
result = SkillRoutingResult(
|
||
clean_content=stripped,
|
||
matched=False,
|
||
match_method="regex_direct",
|
||
match_confidence=1.0,
|
||
agent_name=agent_name,
|
||
model=model,
|
||
system_prompt=sys_prompt,
|
||
tools=[],
|
||
execution_mode=ExecutionMode.DIRECT_CHAT,
|
||
)
|
||
return result
|
||
|
||
# --- Default: REACT (LLM decides tool usage) ---
|
||
result = SkillRoutingResult(
|
||
clean_content=stripped,
|
||
matched=False,
|
||
match_method="default_react",
|
||
match_confidence=0.8,
|
||
agent_name=agent_name,
|
||
model=model,
|
||
system_prompt=sys_prompt,
|
||
tools=tools,
|
||
execution_mode=ExecutionMode.REACT,
|
||
)
|
||
return result
|
||
|
||
def _resolve_explicit_skill(
|
||
self,
|
||
skill_name: str,
|
||
clean_content: str,
|
||
registry: SkillRegistry,
|
||
model: str,
|
||
agent_name: str,
|
||
) -> SkillRoutingResult:
|
||
"""Resolve an explicitly specified skill via @skill:xxx prefix."""
|
||
try:
|
||
skill = registry.get(skill_name)
|
||
except Exception:
|
||
logger.warning(f"Skill '{skill_name}' not found, falling back to REACT")
|
||
return SkillRoutingResult(
|
||
clean_content=clean_content,
|
||
matched=False,
|
||
match_method="skill_not_found_fallback",
|
||
match_confidence=0.5,
|
||
agent_name=agent_name,
|
||
model=model,
|
||
execution_mode=ExecutionMode.REACT,
|
||
)
|
||
|
||
skill_tools = getattr(skill, "tools", []) or []
|
||
skill_config = getattr(skill, "config", skill) # Skill wraps SkillConfig
|
||
skill_prompt = build_skill_system_prompt(skill_config)
|
||
execution_mode = _resolve_execution_mode(skill_config)
|
||
|
||
# U5: 渐进式技能加载 — disclosure_level == 0 时注入 skill_detail 工具
|
||
merged_tools = list(skill_tools)
|
||
if getattr(skill_config, "disclosure_level", 1) == 0:
|
||
from agentkit.skills.skill_detail import SkillDetailTool
|
||
|
||
if not any(t.name == "skill_detail" for t in merged_tools):
|
||
merged_tools.append(SkillDetailTool(skill_registry=registry))
|
||
|
||
return SkillRoutingResult(
|
||
clean_content=clean_content,
|
||
matched=True,
|
||
match_method="skill_prefix",
|
||
match_confidence=1.0,
|
||
skill_name=skill_name,
|
||
skill_config=skill,
|
||
skill_tools=skill_tools,
|
||
agent_name=skill_name,
|
||
model=model,
|
||
system_prompt=skill_prompt,
|
||
tools=merged_tools,
|
||
execution_mode=execution_mode,
|
||
)
|
||
|
||
@staticmethod
|
||
def _is_trivial_input(text: str) -> bool:
|
||
"""Check if the input is a greeting, chitchat, identity question, or pure knowledge/math/translation.
|
||
|
||
These are zero-cost direct chat: no tool usage, no ReAct loop needed.
|
||
Factual/translation patterns are conservative — they exclude requests
|
||
that contain tool-context keywords (当前/服务器/数据库/config etc.) to avoid
|
||
misrouting tool-requiring queries to DIRECT_CHAT.
|
||
"""
|
||
# Multi-line inputs always go to REACT (avoid bypassing tools via newline)
|
||
if "\n" in text or "\r" in text:
|
||
return False
|
||
|
||
# Greeting / chitchat / identity — always safe
|
||
if _GREETING_RE.match(text) or _CHAT_MODE_RE.match(text) or _IDENTITY_RE.match(text):
|
||
return True
|
||
|
||
# Factual questions (CN/EN) — only if no tool-context keywords present
|
||
if (
|
||
_FACTUAL_CN_RE.match(text) or _FACTUAL_EN_RE.match(text)
|
||
) and not _TOOL_CONTEXT_RE.search(text):
|
||
return True
|
||
|
||
# Pure arithmetic — only digits and operators, no tool context possible
|
||
if _MATH_RE.match(text):
|
||
return True
|
||
|
||
# Pure translation — exclude tool-context (e.g. "翻译 这个配置文件")
|
||
if _TRANSLATION_RE.match(text) and not _TOOL_CONTEXT_RE.search(text):
|
||
return True
|
||
|
||
return False
|