diff --git a/apps/server/api/llm.py b/apps/server/api/llm.py new file mode 100644 index 0000000..b534e40 --- /dev/null +++ b/apps/server/api/llm.py @@ -0,0 +1,186 @@ +"""LLM 状态/成本/控制 API (MVP-3 / D34) + +提供给前端的 RESTful 接口: + +- ``GET /llm/status`` 返回 LLM 是否启用、provider、熔断状态、降级原因 +- ``GET /llm/cost`` 返回今日费用 / 预算 / token 用量 +- ``GET /llm/cost/history`` 返回最近 N 天的日级统计 +- ``GET /llm/cost/records`` 返回最近 N 次调用明细 +- ``POST /llm/disable`` 手动禁用 LLM 调用 (运维) +- ``POST /llm/enable`` 手动重新启用 +- ``POST /llm/reset`` 重置成本统计 + 熔断状态 + +设计要点: + +- ``llm_service`` 与 ``cost_tracker`` 都通过 ``main.py`` 在 lifespan 中注入， + 此处仅暴露读取接口 +- 前端的"规则配置 / LLM 状态面板"通过本接口拿到实时数据 +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel, Field + +from core.settings import get_settings +from services.llm_analysis_service import LLMAnalysisService +from services.llm_cost_tracker import LLMCostTracker + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/llm", tags=["LLM 二次判断"]) + + +# --------------------------------------------------------------------------- +# 全局注入 (由 main.py 调用) +# --------------------------------------------------------------------------- + + +_llm_service: Optional[LLMAnalysisService] = None +_cost_tracker: Optional[LLMCostTracker] = None + + +def init_llm_api( + llm_service: Optional[LLMAnalysisService], + cost_tracker: Optional[LLMCostTracker], +) -> None: + """由 main.py 在启动阶段注入服务实例。""" + + global _llm_service, _cost_tracker + _llm_service = llm_service + _cost_tracker = cost_tracker + logger.info( + "LLM API 已初始化: service=%s tracker=%s", + bool(llm_service), + bool(cost_tracker), + ) + + +def _require_tracker() -> LLMCostTracker: + if _cost_tracker is None: + raise HTTPException( + status_code=503, detail="LLMCostTracker 未初始化" + ) + return _cost_tracker + + +# --------------------------------------------------------------------------- +# 响应模型 +# --------------------------------------------------------------------------- + + +class LLMStatusResponse(BaseModel): + enabled: bool + provider: str + model: str + api_base: Optional[str] = None + triggers: Dict[str, Any] = Field(default_factory=dict) + fusion: Dict[str, Any] = Field(default_factory=dict) + cost: Dict[str, Any] = Field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# API 路由 +# --------------------------------------------------------------------------- + + +@router.get("/status", response_model=LLMStatusResponse) +async def get_llm_status() -> LLMStatusResponse: + """获取 LLM 当前运行状态 (供前端 LLM 状态面板)。""" + + settings = get_settings() + llm_cfg = settings.llm + trigger_cfg = settings.llm_trigger + fusion_cfg = settings.fusion + + cost_data: Dict[str, Any] + if _cost_tracker is not None: + cost_data = _cost_tracker.stats() + else: + cost_data = {"enabled": llm_cfg.enabled, "circuit_state": "n/a"} + + provider_name = llm_cfg.provider + if _llm_service is not None: + provider_name = getattr(_llm_service.provider, "name", provider_name) + + return LLMStatusResponse( + enabled=llm_cfg.enabled, + provider=provider_name, + model=llm_cfg.model, + api_base=llm_cfg.api_base, + triggers={ + "enabled": trigger_cfg.enabled, + "window_seconds": trigger_cfg.window_seconds, + "min_consecutive_hits": trigger_cfg.min_consecutive_hits, + "min_avg_confidence": trigger_cfg.min_avg_confidence, + "cooldown_seconds": trigger_cfg.cooldown_seconds, + "severity_bypass": trigger_cfg.severity_bypass, + }, + fusion={ + "strategy": fusion_cfg.strategy, + "yolo_weight": fusion_cfg.yolo_weight, + "llm_weight": fusion_cfg.llm_weight, + "suppress_on_llm_negative": fusion_cfg.suppress_on_llm_negative, + "fallback_to_yolo": fusion_cfg.fallback_to_yolo, + }, + cost=cost_data, + ) + + +@router.get("/cost") +async def get_llm_cost() -> Dict[str, Any]: + """获取今日 LLM 成本概览。""" + + return _require_tracker().stats() + + +@router.get("/cost/history") +async def get_llm_cost_history( + days: int = Query(default=7, ge=1, le=30), +) -> Dict[str, List[Dict[str, Any]]]: + """最近 N 天的日级成本/调用统计。""" + + summary = _require_tracker().daily_summary() + return {"days": summary[-days:]} + + +@router.get("/cost/records") +async def get_llm_cost_records( + limit: int = Query(default=20, ge=1, le=200), +) -> Dict[str, List[Dict[str, Any]]]: + """最近 N 次 LLM 调用明细。""" + + return {"records": _require_tracker().recent_records(limit=limit)} + + +@router.post("/disable") +async def disable_llm() -> Dict[str, Any]: + """手动禁用 LLM 调用。""" + + tracker = _require_tracker() + tracker.disable() + return {"success": True, "enabled": tracker.is_enabled} + + +@router.post("/enable") +async def enable_llm() -> Dict[str, Any]: + """手动重新启用 LLM 调用 (重置熔断)。""" + + tracker = _require_tracker() + tracker.enable() + return {"success": True, "enabled": tracker.is_enabled} + + +@router.post("/reset") +async def reset_llm() -> Dict[str, Any]: + """重置成本统计与熔断状态。""" + + tracker = _require_tracker() + tracker.reset() + return {"success": True, "stats": tracker.stats()} + + +__all__ = ["router", "init_llm_api"] diff --git a/apps/server/api/rules.py b/apps/server/api/rules.py new file mode 100644 index 0000000..690cb33 --- /dev/null +++ b/apps/server/api/rules.py @@ -0,0 +1,281 @@ +"""规则配置 API (MVP-3 / D33 配套后端) + +提供 YAML 规则文件的增删改查接口，供前端 ``RuleConfiguration.vue`` 调用。 + +接口列表: + +- ``GET /rules`` 列出所有规则 (按文件分组) +- ``GET /rules/{name}`` 获取单条规则详情 +- ``POST /rules`` 新增规则 +- ``PUT /rules/{name}`` 更新规则 +- ``DELETE /rules/{name}`` 删除规则 +- ``POST /rules/reload`` 热重载规则目录 +- ``GET /rules/_stats`` 规则引擎运行时状态 + +规则文件落盘策略: + +- 默认所有自定义规则写入 ``config/rules/custom.yaml`` +- 不修改 MVP-1 内置的事件类型规则文件 (fire.yaml / smoking.yaml 等) 的"内置规则"， + 但可以通过本接口在其上追加新规则或编辑已有规则 +- 安全保护: 规则名作为唯一键，禁止同名覆盖；删除时仅允许删除 "custom.yaml" 内的规则 + +设计要点: + +- 规则编辑后会刷新 ``DetectionService.rule_engine`` 实例，确保实时生效 +- 写入 YAML 时使用 ``yaml.safe_dump``，保留 schema 兼容性 +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any, Dict, List, Optional + +import yaml +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, ConfigDict, Field + +from core.settings import get_settings +from models.event_schemas import EventType, SeverityLevel +from services.event.rule_engine import AlertRule, AlertRuleEngine + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/rules", tags=["规则配置"]) + + +# --------------------------------------------------------------------------- +# 全局引擎注入 +# --------------------------------------------------------------------------- + + +_rule_engine_provider = None + + +def init_rules_api(rule_engine_provider) -> None: + """注入规则引擎获取函数。 + + 使用 callable 而非实例，便于多个模块 (DetectionService 等) 共享同一引擎， + 并在调用时拿到最新引用。 + """ + + global _rule_engine_provider + _rule_engine_provider = rule_engine_provider + + +def _get_engine() -> AlertRuleEngine: + if _rule_engine_provider is None: + raise HTTPException(status_code=503, detail="规则引擎未初始化") + engine = _rule_engine_provider() + if engine is None: + raise HTTPException(status_code=503, detail="规则引擎未就绪") + return engine + + +def _custom_rules_path() -> Path: + settings = get_settings() + rules_dir = settings.paths.server_dir / settings.event_engine.rules_dir + rules_dir.mkdir(parents=True, exist_ok=True) + return rules_dir / "custom.yaml" + + +# --------------------------------------------------------------------------- +# 请求 / 响应模型 +# --------------------------------------------------------------------------- + + +class RulePayload(BaseModel): + """规则的可序列化表示。""" + + model_config = ConfigDict(extra="ignore") + + name: str = Field(..., min_length=1, max_length=64) + event_type: EventType + enabled: bool = True + min_confidence: float = Field(default=0.0, ge=0.0, le=1.0) + severity: Optional[SeverityLevel] = None + allowed_sources: Optional[List[str]] = None + required_labels: Optional[List[str]] = None + min_bbox_area: int = Field(default=0, ge=0) + description: str = "" + + @classmethod + def from_rule(cls, rule: AlertRule) -> "RulePayload": + return cls( + name=rule.name, + event_type=rule.event_type, + enabled=rule.enabled, + min_confidence=rule.min_confidence, + severity=rule.severity, + allowed_sources=rule.allowed_sources, + required_labels=rule.required_labels, + min_bbox_area=rule.min_bbox_area, + description=rule.description, + ) + + def to_yaml_dict(self) -> Dict[str, Any]: + data: Dict[str, Any] = { + "name": self.name, + "event_type": self.event_type.value, + "enabled": self.enabled, + "min_confidence": self.min_confidence, + "min_bbox_area": self.min_bbox_area, + "description": self.description, + } + if self.severity is not None: + data["severity"] = self.severity.value + if self.allowed_sources: + data["allowed_sources"] = list(self.allowed_sources) + if self.required_labels: + data["required_labels"] = list(self.required_labels) + return data + + +# --------------------------------------------------------------------------- +# YAML 持久化 +# --------------------------------------------------------------------------- + + +def _load_custom_rules() -> List[Dict[str, Any]]: + path = _custom_rules_path() + if not path.exists(): + return [] + with path.open("r", encoding="utf-8") as fp: + data = yaml.safe_load(fp) or {} + if isinstance(data, dict) and "rules" in data: + return list(data["rules"]) + if isinstance(data, list): + return list(data) + return [] + + +def _save_custom_rules(rules: List[Dict[str, Any]]) -> None: + path = _custom_rules_path() + payload = { + "# generated by /api/rules MVP-3 D33": None, + "rules": rules, + } + payload.pop("# generated by /api/rules MVP-3 D33") # 仅作注释占位 + with path.open("w", encoding="utf-8") as fp: + fp.write("# 由前端规则配置页面写入，请勿手工同时编辑\n") + yaml.safe_dump( + {"rules": rules}, + fp, + allow_unicode=True, + sort_keys=False, + indent=2, + ) + + +def _reload_engine() -> AlertRuleEngine: + """重新从规则目录加载，并替换全局引擎的规则集合。""" + + settings = get_settings() + rules_dir = settings.paths.server_dir / settings.event_engine.rules_dir + new_engine = AlertRuleEngine.from_directory(rules_dir) + + engine = _get_engine() + engine.rules = new_engine.rules + engine._stats = new_engine._stats # noqa: SLF001 + logger.info("规则引擎已热重载: 共 %d 条", len(engine.rules)) + return engine + + +# --------------------------------------------------------------------------- +# API +# --------------------------------------------------------------------------- + + +@router.get("") +async def list_rules() -> Dict[str, Any]: + """列出所有规则 (含来源文件)。""" + + engine = _get_engine() + return { + "rules": [RulePayload.from_rule(r).model_dump(mode="json") for r in engine.rules], + "stats": engine.stats, + } + + +@router.get("/_stats") +async def rule_stats() -> Dict[str, Any]: + """规则引擎运行状态。""" + + engine = _get_engine() + return engine.stats + + +@router.get("/{name}") +async def get_rule(name: str) -> RulePayload: + """获取单条规则。""" + + engine = _get_engine() + for rule in engine.rules: + if rule.name == name: + return RulePayload.from_rule(rule) + raise HTTPException(status_code=404, detail=f"规则 {name} 不存在") + + +@router.post("") +async def create_rule(payload: RulePayload) -> Dict[str, Any]: + """新增规则 (写入 custom.yaml)。""" + + engine = _get_engine() + if any(r.name == payload.name for r in engine.rules): + raise HTTPException(status_code=409, detail=f"规则 {payload.name} 已存在") + + rules = _load_custom_rules() + rules.append(payload.to_yaml_dict()) + _save_custom_rules(rules) + + _reload_engine() + return {"success": True, "rule": payload.model_dump(mode="json")} + + +@router.put("/{name}") +async def update_rule(name: str, payload: RulePayload) -> Dict[str, Any]: + """更新规则 (仅支持 custom.yaml 中的规则)。""" + + if payload.name != name: + raise HTTPException(status_code=400, detail="路径与请求体中的规则名不一致") + + rules = _load_custom_rules() + for idx, item in enumerate(rules): + if item.get("name") == name: + rules[idx] = payload.to_yaml_dict() + _save_custom_rules(rules) + _reload_engine() + return {"success": True, "rule": payload.model_dump(mode="json")} + + raise HTTPException( + status_code=404, + detail=f"规则 {name} 不在 custom.yaml 中，无法编辑 (内置规则只读)", + ) + + +@router.delete("/{name}") +async def delete_rule(name: str) -> Dict[str, Any]: + """删除规则 (仅 custom.yaml)。""" + + rules = _load_custom_rules() + new_rules = [r for r in rules if r.get("name") != name] + if len(new_rules) == len(rules): + raise HTTPException( + status_code=404, + detail=f"规则 {name} 不在 custom.yaml 中，无法删除 (内置规则只读)", + ) + + _save_custom_rules(new_rules) + _reload_engine() + return {"success": True, "deleted": name} + + +@router.post("/reload") +async def reload_rules() -> Dict[str, Any]: + """热重载规则目录 (供运维直接编辑 YAML 后调用)。""" + + engine = _reload_engine() + return {"success": True, "stats": engine.stats} + + +__all__ = ["router", "init_rules_api"]