343 lines
16 KiB
Python
343 lines
16 KiB
Python
"""Hermes 自然語言閘道 — ADR-094
|
||
|
||
Layer 1 意圖路由（關鍵字正則）→ Ollama 本地模型（111）→ Telegram 格式化輸出。
|
||
|
||
2026-04-24 Claude Sonnet 4.6 (WS4 Hermes NL)
|
||
2026-04-24 Claude Sonnet 4.6 (WS4 Hermes NL T1+T2+T3): hermes_dispatch_log DB 寫入 /
|
||
Redis per-chat_id 速率限制 / Multi-turn session (Redis Hash TTL=300s)
|
||
2026-04-25 Claude Sonnet 4.6: 改用 Ollama 本地模型（111），按 agent 類型選模型，零費用
|
||
debugger/vuln → deepseek-r1:14b（推理）; code agents → qwen2.5-coder:7b; 其他 → qwen2.5:7b-instruct
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
import time
|
||
|
||
import httpx
|
||
import structlog
|
||
from sqlalchemy import text
|
||
|
||
from src.core.redis_client import get_redis
|
||
from src.db.base import get_db_context
|
||
from src.hermes.agent_loader import get_agent_system_prompt
|
||
from src.hermes.display_names import DEFAULT_AGENT, format_response_header
|
||
from src.hermes.safety_hooks import is_dangerous_input, is_mutate_intent
|
||
from src.services.ollama_endpoint_resolver import resolve_ollama_order
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Layer 1 意圖路由（關鍵字正則，<10ms）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_ROUTING_RULES: list[tuple[re.Pattern, str]] = [
|
||
(re.compile(r"(資料庫|postgres|sql|index|query|migration|schema)", re.IGNORECASE), "db-expert"),
|
||
(re.compile(r"(漏洞|CVE|injection|XSS|CSRF|security|安全)", re.IGNORECASE), "vuln-verifier"),
|
||
(re.compile(r"(bug|crash|error|exception|fail|失敗|崩潰|為什麼壞|不通)", re.IGNORECASE), "debugger"),
|
||
(re.compile(r"(重構|refactor|clean|重寫)", re.IGNORECASE), "refactor-specialist"),
|
||
(re.compile(r"(升級|upgrade|migration|migrate|版本)", re.IGNORECASE), "migration-engineer"),
|
||
(re.compile(r"(設計|UI|frontend|頁面|按鈕|樣式)", re.IGNORECASE), "frontend-designer"),
|
||
(re.compile(r"(工具|tool|hook|MCP|plugin)", re.IGNORECASE), "tool-expert"),
|
||
(re.compile(r"(文件|document|官方|API spec|how to)", re.IGNORECASE), "web-researcher"),
|
||
(re.compile(r"(導覽|介紹|架構|codebase|overview)", re.IGNORECASE), "onboarder"),
|
||
(re.compile(r"(拆解|任務|plan|規劃)", re.IGNORECASE), "planner"),
|
||
(re.compile(r"(審查|review|code review|找問題)", re.IGNORECASE), "critic"),
|
||
(re.compile(r"(實作|implement|develop|功能|feature)", re.IGNORECASE), "fullstack-engineer"),
|
||
]
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# T2：速率限制常數（ADR-094）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_RATE_LIMIT_MAX = 20
|
||
_RATE_LIMIT_WINDOW_SEC = 60
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Ollama 模型路由（按 agent 專業選最適模型，111 主機）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_MODEL_BY_AGENT: dict[str, str] = {
|
||
# 推理型（找根因 / 安全分析）→ deepseek-r1:14b（CoT 推理）
|
||
"debugger": "deepseek-r1:14b",
|
||
"vuln-verifier": "deepseek-r1:14b",
|
||
# 程式碼 + 通用（review / 實作 / 重構 / DB / 前端 / 工具 / 規劃 / 文件）→ qwen3:8b
|
||
# 2026-04-25 ogt + Claude Sonnet 4.6: qwen2.5-coder:7b + qwen2.5:7b-instruct → qwen3:8b
|
||
# qwen3:8b Hybrid Thinking 同時勝任程式碼與指令；gemma4 尚未在 Ollama 釋出
|
||
"critic": "qwen3:8b",
|
||
"db-expert": "qwen3:8b",
|
||
"fullstack-engineer": "qwen3:8b",
|
||
"refactor-specialist":"qwen3:8b",
|
||
"migration-engineer": "qwen3:8b",
|
||
"frontend-designer": "qwen3:8b",
|
||
"tool-expert": "qwen3:8b",
|
||
"planner": "qwen3:8b",
|
||
"onboarder": "qwen3:8b",
|
||
"web-researcher": "qwen3:8b",
|
||
}
|
||
_DEFAULT_MODEL = "deepseek-r1:14b"
|
||
_OLLAMA_TIMEOUT = 90.0 # deepseek-r1:14b 推理較慢，給 90s
|
||
|
||
|
||
def _pick_model(agent_name: str) -> str:
|
||
return _MODEL_BY_AGENT.get(agent_name, _DEFAULT_MODEL)
|
||
|
||
|
||
def _strip_think_tags(text: str) -> str:
|
||
"""移除 deepseek-r1 的 <think>...</think> 內部推理塊，只留最終回答。"""
|
||
return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
|
||
|
||
|
||
def _route_intent_layer1(msg: str) -> str:
|
||
"""Layer 1: 關鍵字正則路由，回傳 agent 名稱"""
|
||
for pattern, agent in _ROUTING_RULES:
|
||
if pattern.search(msg):
|
||
return agent
|
||
return DEFAULT_AGENT
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# T1：hermes_dispatch_log DB 寫入（ADR-094，非阻擋）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def _write_dispatch_log(
|
||
*,
|
||
chat_id: str,
|
||
user_id: int,
|
||
username: str,
|
||
agent_name: str,
|
||
input_preview: str,
|
||
latency_ms: int,
|
||
success: bool,
|
||
error_type: str | None,
|
||
) -> None:
|
||
"""寫入派發審計日誌；失敗只 warning，不影響主流程。"""
|
||
try:
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
text("""
|
||
INSERT INTO hermes_dispatch_log
|
||
(chat_id, user_id, username, agent_name, input_preview,
|
||
latency_ms, success, error_type)
|
||
VALUES
|
||
(:chat_id, :user_id, :username, :agent_name, :input_preview,
|
||
:latency_ms, :success, :error_type)
|
||
"""),
|
||
{
|
||
"chat_id": chat_id,
|
||
"user_id": user_id,
|
||
"username": username,
|
||
"agent_name": agent_name,
|
||
"input_preview": input_preview,
|
||
"latency_ms": latency_ms,
|
||
"success": success,
|
||
"error_type": error_type,
|
||
},
|
||
)
|
||
await db.commit()
|
||
except Exception as exc:
|
||
logger.warning("hermes_dispatch_log_write_failed", error=str(exc))
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# T2：per-chat_id 速率限制（ADR-094，fail-open）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def _check_rate_limit(chat_id: str, project_id: str = "awoooi") -> bool:
|
||
"""True = 允許；False = 超過限制（20 req/min per chat_id）。Redis 不可用時放行。"""
|
||
try:
|
||
redis = get_redis()
|
||
key = f"{project_id}:hermes:rl:{chat_id}"
|
||
count = await redis.incr(key)
|
||
if count == 1:
|
||
await redis.expire(key, _RATE_LIMIT_WINDOW_SEC)
|
||
return count <= _RATE_LIMIT_MAX
|
||
except Exception:
|
||
return True # Redis 不可用 → fail open
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# T3：Multi-turn session（Redis Hash TTL=300s，ADR-094）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def _load_session_context(chat_id: str, user_id: int, project_id: str = "awoooi") -> str:
|
||
"""載入最近 3 輪對話歷史（最多 600 字），組成 context prefix。Redis 不可用時回空字串。"""
|
||
try:
|
||
redis = get_redis()
|
||
key = f"{project_id}:hermes:session:{chat_id}:{user_id}"
|
||
data = await redis.hgetall(key)
|
||
if not data:
|
||
# Phase A: fallback 到舊 key（滾動部署相容）
|
||
data = await redis.hgetall(f"hermes:session:{chat_id}:{user_id}")
|
||
if not data:
|
||
return ""
|
||
turns = sorted(
|
||
[(k, v) for k, v in data.items() if (k if isinstance(k, str) else k.decode()).startswith("turn_")],
|
||
key=lambda x: x[0],
|
||
)[-3:]
|
||
parts = [v.decode() if isinstance(v, bytes) else v for _, v in turns]
|
||
return "【近期對話記錄】\n" + "\n".join(parts) + "\n\n"
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
async def _save_session_turn(
|
||
chat_id: str, user_id: int, user_msg: str, assistant_reply: str, project_id: str = "awoooi"
|
||
) -> None:
|
||
"""將本輪對話存入 Redis Hash，並重置 TTL=300s。Redis 不可用時靜默忽略。"""
|
||
try:
|
||
redis = get_redis()
|
||
key = f"{project_id}:hermes:session:{chat_id}:{user_id}"
|
||
legacy_key = f"hermes:session:{chat_id}:{user_id}" # Phase A dual-write
|
||
turn_key = f"turn_{int(time.time())}"
|
||
value = f"用戶：{user_msg[:100]}\nHermes：{assistant_reply[:200]}"
|
||
await redis.hset(key, turn_key, value)
|
||
await redis.expire(key, 300)
|
||
await redis.hset(legacy_key, turn_key, value)
|
||
await redis.expire(legacy_key, 300)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 主入口
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def process_nl_message(
|
||
user_message: str,
|
||
*,
|
||
chat_id: str,
|
||
user_id: int,
|
||
username: str = "",
|
||
project_id: str = "awoooi",
|
||
) -> str:
|
||
"""
|
||
處理 NL 訊息，回傳 Telegram 格式的回覆文字。
|
||
|
||
流程：
|
||
1. 安全守門（DENY + MUTATE）
|
||
2. T2 速率限制（20 req/min per chat_id）
|
||
3. Layer 1 關鍵字路由 → agent_name
|
||
4. 讀取 agent system prompt（.claude/agents/*.md）
|
||
5. T3 載入 session context（最近 3 輪）
|
||
6. 呼叫 Claude Agent SDK query()
|
||
7. T3 儲存本輪對話
|
||
8. 格式化為 Telegram MarkdownV2 訊息
|
||
9. T1 非阻擋寫入 hermes_dispatch_log
|
||
"""
|
||
# 安全守門
|
||
if is_dangerous_input(user_message):
|
||
logger.warning(
|
||
"hermes_nl_dangerous_input",
|
||
user_id=user_id,
|
||
chat_id=chat_id,
|
||
preview=user_message[:80],
|
||
)
|
||
return "⛔ 偵測到危險指令，拒絕處理。"
|
||
|
||
if is_mutate_intent(user_message):
|
||
return (
|
||
"⚠️ 此操作涉及變更，需透過正式審批流程執行。\n"
|
||
"請在 Telegram 告警卡片上操作，或聯繫值班 SRE。"
|
||
)
|
||
|
||
# T2：速率限制
|
||
if not await _check_rate_limit(chat_id, project_id):
|
||
return "⚠️ 請求太頻繁，請稍後再試（每分鐘上限 20 次）。"
|
||
|
||
# Layer 1 意圖路由
|
||
agent_name = _route_intent_layer1(user_message)
|
||
|
||
# 確認 agent 存在，否則 fallback
|
||
system_prompt = get_agent_system_prompt(agent_name)
|
||
if system_prompt is None:
|
||
logger.warning(
|
||
"hermes_nl_agent_not_found",
|
||
agent=agent_name,
|
||
fallback=DEFAULT_AGENT,
|
||
)
|
||
agent_name = DEFAULT_AGENT
|
||
system_prompt = get_agent_system_prompt(agent_name) or ""
|
||
|
||
# T3：載入 session context（最近 3 輪）
|
||
session_ctx = await _load_session_context(chat_id, user_id, project_id)
|
||
prompt_with_ctx = f"{session_ctx}{user_message}" if session_ctx else user_message
|
||
|
||
t0 = time.monotonic()
|
||
|
||
# 呼叫 Ollama 模型（GCP-A → GCP-B → 111，零費用，按 agent 選模型）
|
||
model = _pick_model(agent_name)
|
||
success = False
|
||
error_type: str | None = None
|
||
result_text = ""
|
||
async with httpx.AsyncClient(timeout=_OLLAMA_TIMEOUT) as _hc:
|
||
for endpoint in resolve_ollama_order("hermes"):
|
||
if not endpoint.url:
|
||
continue
|
||
try:
|
||
resp = await _hc.post(
|
||
f"{endpoint.url}/api/chat",
|
||
json={
|
||
"model": model,
|
||
# Keep Hermes responses in message.content across Ollama 0.24+.
|
||
"think": False,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": prompt_with_ctx},
|
||
],
|
||
"stream": False,
|
||
"options": {"num_predict": 1500, "temperature": 0.3},
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
result_text = resp.json().get("message", {}).get("content", "")
|
||
result_text = _strip_think_tags(result_text)
|
||
if not result_text:
|
||
result_text = "_Agent 回應為空，請稍後再試。_"
|
||
success = True
|
||
break
|
||
except Exception as exc:
|
||
error_type = type(exc).__name__
|
||
logger.error(
|
||
"hermes_nl_ollama_error",
|
||
error=str(exc),
|
||
agent=agent_name,
|
||
model=model,
|
||
provider=endpoint.provider_name,
|
||
exc_type=error_type,
|
||
)
|
||
if not success:
|
||
result_text = f"_Hermes 暫時無法連線（{error_type}），請稍後再試。_"
|
||
|
||
latency_ms = int((time.monotonic() - t0) * 1000)
|
||
logger.info(
|
||
"hermes_nl_dispatch",
|
||
agent=agent_name,
|
||
model=model,
|
||
user_id=user_id,
|
||
chat_id=chat_id,
|
||
username=username,
|
||
latency_ms=latency_ms,
|
||
success=success,
|
||
)
|
||
|
||
# T3：儲存本輪對話（只在成功時存）
|
||
if success:
|
||
await _save_session_turn(chat_id, user_id, user_message, result_text, project_id)
|
||
|
||
# T1：非阻擋寫入 hermes_dispatch_log（失敗不影響回覆）
|
||
asyncio.create_task(
|
||
_write_dispatch_log(
|
||
chat_id=chat_id,
|
||
user_id=user_id,
|
||
username=username,
|
||
agent_name=agent_name,
|
||
input_preview=user_message[:200],
|
||
latency_ms=latency_ms,
|
||
success=success,
|
||
error_type=error_type,
|
||
)
|
||
)
|
||
|
||
header = format_response_header(agent_name)
|
||
# Telegram 訊息上限 4096 字元，超過截斷
|
||
body = result_text[:3800]
|
||
return f"{header}{body}"
|