Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3898530585 | ||
|
|
fcc601fcf3 | ||
|
|
c6cde2500b | ||
|
|
0006e91e67 | ||
|
|
866febec21 | ||
|
|
9b50b8f0ce | ||
|
|
eb940e6590 | ||
|
|
e37ac7b0a4 | ||
|
|
574b113827 | ||
|
|
1132a4cece | ||
|
|
aedb1c1267 | ||
|
|
fc4b7b9495 | ||
|
|
83782f1d0a |
@@ -118,7 +118,7 @@ cd frontend
|
||||
pnpm test
|
||||
```
|
||||
|
||||
当前回归基线为后端 81 项测试、前端 27 项测试,且生产构建通过。测试数量会随功能增长,以本地实际输出和 CI 为准。
|
||||
当前回归基线为后端 157 项测试、前端 29 项测试,且生产构建通过。测试数量会随功能增长,以本地实际输出和 CI 为准。
|
||||
|
||||
构建产物位于 `frontend/dist`,该目录不提交到 Git。
|
||||
|
||||
@@ -132,6 +132,7 @@ pnpm test
|
||||
| [后端接口契约](docs/contracts/后端接口契约-开发版.md) | HTTP/SSE 接口、错误和当前实现状态 |
|
||||
| [第二阶段接口契约](docs/contracts/第二阶段接口契约-开发版.md) | 第二阶段公共 DTO、计划接口、SSE、错误码与联调顺序 |
|
||||
| [AI Core 与 Agent Core](docs/development/AI-Core与Agent-Core开发说明.md) | Provider、Agent、Tool、Permission 与 Extension Core |
|
||||
| [MCP Bridge 与 Plugin Host](docs/development/MCP-Bridge与Plugin-Host开发说明.md) | stdio MCP、隔离进程、Tool 映射、状态与错误边界 |
|
||||
| [Git 使用细则](docs/guides/Git使用细则-团队开发版.md) | 分支、提交、PR、Review 与合并流程 |
|
||||
| [CI/CD 细则](docs/guides/CI-CD细则-团队开发版.md) | Gitea 流水线、质量门禁、产物、发布与回滚规则 |
|
||||
| [Agent Trace 复盘](docs/retrospectives/Agent-Core第二阶段问题与修复复盘.md) | Agent 持久化、SSE 恢复、事件契约与脱敏问题复盘 |
|
||||
|
||||
+1
-1
@@ -23,7 +23,7 @@ uv run uvicorn app.main:app --reload --host 127.0.0.1 --port 8000
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
当前基线为 81 项测试通过。Provider API Key 可通过前端设置页写入,也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入;不要把真实密钥写入仓库。
|
||||
当前基线为 92 项测试通过。Provider API Key 可通过前端设置页写入,也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入;不要把真实密钥写入仓库。
|
||||
|
||||
团队接口清单见 `../docs/contracts/后端接口契约-开发版.md`,机器可读契约以运行时的 `/openapi.json` 为准。
|
||||
|
||||
|
||||
@@ -491,7 +491,13 @@ class AgentRuntime:
|
||||
async def _invoke_tool(self, record: RunRecord, call: ToolCall) -> ToolResult:
|
||||
try:
|
||||
return await asyncio.wait_for(
|
||||
self.tools.execute(call, ToolExecutionContext(run_id=record.run.run_id)),
|
||||
self.tools.execute(
|
||||
call,
|
||||
ToolExecutionContext(
|
||||
run_id=record.run.run_id,
|
||||
tool_call_id=call.tool_call_id,
|
||||
),
|
||||
),
|
||||
timeout=record.request.tool_timeout_seconds,
|
||||
)
|
||||
except TimeoutError:
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Agent 工具注册与执行边界。"""
|
||||
|
||||
import inspect
|
||||
import threading
|
||||
from dataclasses import dataclass
|
||||
from time import perf_counter
|
||||
from typing import Any, Awaitable, Callable
|
||||
@@ -17,6 +18,7 @@ ToolExecutor = Callable[[BaseModel, "ToolExecutionContext"], Any | Awaitable[Any
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ToolExecutionContext:
|
||||
run_id: str
|
||||
tool_call_id: str | None = None
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
@@ -30,11 +32,21 @@ class ToolNotFoundError(LookupError):
|
||||
pass
|
||||
|
||||
|
||||
class ToolExecutionError(RuntimeError):
|
||||
"""Executor 可预期失败,保留领域错误码而不是折叠成通用异常。"""
|
||||
|
||||
def __init__(self, code: str, message: str) -> None:
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.message = message
|
||||
|
||||
|
||||
class ToolRegistry:
|
||||
"""统一校验工具入参并隔离执行异常,避免单个工具击穿 Agent 主循环。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._tools: dict[str, RegisteredTool] = {}
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def register(
|
||||
self,
|
||||
@@ -42,6 +54,7 @@ class ToolRegistry:
|
||||
arguments_model: type[BaseModel],
|
||||
executor: ToolExecutor,
|
||||
) -> None:
|
||||
with self._lock:
|
||||
if definition.name in self._tools:
|
||||
raise ValueError(f"Tool already registered: {definition.name}")
|
||||
self._tools[definition.name] = RegisteredTool(
|
||||
@@ -51,12 +64,15 @@ class ToolRegistry:
|
||||
)
|
||||
|
||||
def unregister(self, name: str) -> None:
|
||||
with self._lock:
|
||||
self._tools.pop(name, None)
|
||||
|
||||
def contains(self, name: str) -> bool:
|
||||
with self._lock:
|
||||
return name in self._tools
|
||||
|
||||
def get(self, name: str) -> RegisteredTool:
|
||||
with self._lock:
|
||||
try:
|
||||
return self._tools[name]
|
||||
except KeyError as exc:
|
||||
@@ -64,6 +80,7 @@ class ToolRegistry:
|
||||
|
||||
def definitions(self, allowed: list[str] | None = None) -> list[ToolDefinition]:
|
||||
names = set(allowed) if allowed is not None else None
|
||||
with self._lock:
|
||||
return [
|
||||
item.definition.model_copy(deep=True)
|
||||
for name, item in self._tools.items()
|
||||
@@ -108,6 +125,15 @@ class ToolRegistry:
|
||||
output=output,
|
||||
duration_ms=round((perf_counter() - started) * 1000),
|
||||
)
|
||||
except ToolExecutionError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=call.tool_call_id,
|
||||
name=call.name,
|
||||
success=False,
|
||||
error_code=exc.code,
|
||||
error_message=exc.message,
|
||||
duration_ms=round((perf_counter() - started) * 1000),
|
||||
)
|
||||
except Exception as exc: # 工具失败转换成结构化结果,由模型决定是否降级或重试。
|
||||
return ToolResult(
|
||||
tool_call_id=call.tool_call_id,
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
"""Benchmark 服务:RAG / Agent 数据集注册、指标计算与运行管理。
|
||||
|
||||
模块划分:
|
||||
- metrics.py 纯函数指标(Hit@K / Recall@K / MRR / CitationHit / 分位数)
|
||||
- datasets.py 受控目录的 Dataset 注册与校验
|
||||
- rag.py RAG Benchmark Runner(调用 retrieval.engine.search)
|
||||
- service.py 运行注册表、配置快照与报告组装
|
||||
"""
|
||||
@@ -0,0 +1,198 @@
|
||||
"""Benchmark Dataset 注册:从受控目录加载 JSON 数据集并校验。
|
||||
|
||||
Dataset 只能来自配置目录(settings.benchmark_datasets_path),API 不接受调用方提交
|
||||
任意文件路径。目录不存在或为空时按「无数据集」处理,不报错。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import BaseModel, Field, ValidationError
|
||||
|
||||
from app.config import get_settings
|
||||
from app.contracts import (
|
||||
BenchmarkDatasetInfo,
|
||||
BenchmarkKind,
|
||||
RAGDatasetCase,
|
||||
)
|
||||
from app.errors import ApiError
|
||||
|
||||
|
||||
@dataclass
|
||||
class RAGDataset:
|
||||
"""内存中的 RAG 数据集:元信息 + 已校验的 Case 列表 + 内容哈希。"""
|
||||
|
||||
dataset_id: str
|
||||
kind: BenchmarkKind
|
||||
version: str
|
||||
description: str
|
||||
cases: list[RAGDatasetCase] = field(default_factory=list)
|
||||
content_hash: str = ""
|
||||
|
||||
|
||||
class _DatasetMeta(BaseModel):
|
||||
"""Dataset 元数据的最小校验模型。
|
||||
|
||||
list_datasets 用它逐文件校验元信息字段结构,把「合法 JSON 但字段类型错误」
|
||||
(如 cases: 42)这类损坏文件隔离掉,而不是让 len() 抛 TypeError 拖垮整个列表。
|
||||
"""
|
||||
|
||||
dataset_id: str = Field(min_length=1)
|
||||
kind: str = ""
|
||||
version: str = ""
|
||||
description: str = ""
|
||||
cases: list = Field(default_factory=list)
|
||||
|
||||
|
||||
def _datasets_dir() -> Path:
|
||||
return get_settings().benchmark_datasets_path
|
||||
|
||||
|
||||
def _dataset_files() -> list[Path]:
|
||||
directory = _datasets_dir()
|
||||
if not directory.is_dir():
|
||||
return []
|
||||
return sorted(directory.glob("*.json"))
|
||||
|
||||
|
||||
def _content_hash(raw: bytes) -> str:
|
||||
return "sha256:" + hashlib.sha256(raw).hexdigest()
|
||||
|
||||
|
||||
def _read_json(path: Path) -> tuple[dict, bytes]:
|
||||
"""读取并解析 JSON 文件,返回 (dict, 原始字节);非法 JSON 抛 BENCHMARK_DATASET_INVALID。"""
|
||||
try:
|
||||
raw_bytes = path.read_bytes()
|
||||
return json.loads(raw_bytes.decode("utf-8")), raw_bytes
|
||||
except (json.JSONDecodeError, OSError, UnicodeDecodeError) as exc:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
f"Dataset file is not valid JSON: {path.name}",
|
||||
{"path": str(path)},
|
||||
) from exc
|
||||
|
||||
|
||||
def _dataset_from_raw(raw: dict, raw_bytes: bytes, kind: BenchmarkKind) -> RAGDataset:
|
||||
"""把单个数据集 JSON 解析为 RAGDataset,非法结构抛 BENCHMARK_DATASET_INVALID。"""
|
||||
dataset_id = raw.get("dataset_id")
|
||||
if not isinstance(dataset_id, str) or not dataset_id:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
"Dataset must declare a non-empty string 'dataset_id'.",
|
||||
{},
|
||||
)
|
||||
file_kind = raw.get("kind", kind.value)
|
||||
if file_kind != kind.value:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
f"Dataset kind mismatch: expected '{kind.value}', got '{file_kind}'.",
|
||||
{"dataset_id": dataset_id},
|
||||
)
|
||||
raw_cases = raw.get("cases")
|
||||
if not isinstance(raw_cases, list) or not raw_cases:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
"Dataset 'cases' must be a non-empty list.",
|
||||
{"dataset_id": dataset_id},
|
||||
)
|
||||
|
||||
cases: list[RAGDatasetCase] = []
|
||||
for index, case in enumerate(raw_cases):
|
||||
try:
|
||||
parsed = RAGDatasetCase.model_validate(case)
|
||||
except ValidationError as exc:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
f"Dataset case #{index} is invalid.",
|
||||
{"dataset_id": dataset_id, "case_index": index, "errors": exc.errors()},
|
||||
) from exc
|
||||
# 每个 Case 至少要声明一个期望 id,否则无法计算命中/召回
|
||||
if not parsed.expected_note_ids and not parsed.expected_block_ids:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
f"Dataset case '{parsed.case_id}' must declare expected_note_ids or expected_block_ids.",
|
||||
{"dataset_id": dataset_id, "case_id": parsed.case_id},
|
||||
)
|
||||
# citation_required=true 时必须声明 expected_block_ids,否则无法计算 Citation Hit Rate
|
||||
if parsed.citation_required and not parsed.expected_block_ids:
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
f"Dataset case '{parsed.case_id}' requires expected_block_ids when citation_required is true.",
|
||||
{"dataset_id": dataset_id, "case_id": parsed.case_id},
|
||||
)
|
||||
cases.append(parsed)
|
||||
|
||||
return RAGDataset(
|
||||
dataset_id=dataset_id,
|
||||
kind=kind,
|
||||
version=str(raw.get("version", "")),
|
||||
description=str(raw.get("description", "")),
|
||||
cases=cases,
|
||||
content_hash=_content_hash(raw_bytes),
|
||||
)
|
||||
|
||||
|
||||
def list_datasets(kind: BenchmarkKind) -> list[BenchmarkDatasetInfo]:
|
||||
"""枚举受控目录下指定 kind 的数据集元信息(不含 Case 内容)。
|
||||
|
||||
逐文件用 _DatasetMeta 校验元信息字段结构,单个损坏文件隔离跳过而非整体失败,
|
||||
保证列表接口健壮;损坏细节由 load_dataset 抛出。
|
||||
"""
|
||||
infos: list[BenchmarkDatasetInfo] = []
|
||||
for path in _dataset_files():
|
||||
try:
|
||||
raw, raw_bytes = _read_json(path)
|
||||
meta = _DatasetMeta.model_validate(raw)
|
||||
except (ApiError, ValidationError):
|
||||
continue
|
||||
if meta.kind not in ("", kind.value):
|
||||
continue
|
||||
infos.append(
|
||||
BenchmarkDatasetInfo(
|
||||
dataset_id=meta.dataset_id,
|
||||
kind=kind,
|
||||
version=meta.version,
|
||||
description=meta.description,
|
||||
case_count=len(meta.cases),
|
||||
content_hash=_content_hash(raw_bytes),
|
||||
)
|
||||
)
|
||||
return infos
|
||||
|
||||
|
||||
def load_dataset(dataset_id: str, kind: BenchmarkKind) -> RAGDataset:
|
||||
"""按文件名加载并校验数据集;找不到抛 BENCHMARK_DATASET_NOT_FOUND。
|
||||
|
||||
只读取与请求 dataset_id 同名的文件({dataset_id}.json),无关文件的损坏(JSON 语法
|
||||
错误、UTF-8 解码错误、顶层非对象)不会阻断目标数据集加载;只有目标文件本身损坏
|
||||
才抛 BENCHMARK_DATASET_INVALID。按现有文件 stem 精确匹配,不拼接调用方传入的路径。
|
||||
"""
|
||||
for path in _dataset_files():
|
||||
if path.stem != dataset_id:
|
||||
continue
|
||||
raw, raw_bytes = _read_json(path)
|
||||
if not isinstance(raw, dict):
|
||||
raise ApiError(
|
||||
422,
|
||||
"BENCHMARK_DATASET_INVALID",
|
||||
"Dataset top-level must be a JSON object.",
|
||||
{"dataset_id": dataset_id, "path": path.name},
|
||||
)
|
||||
return _dataset_from_raw(raw, raw_bytes, kind)
|
||||
raise ApiError(
|
||||
404,
|
||||
"BENCHMARK_DATASET_NOT_FOUND",
|
||||
f"Benchmark dataset does not exist: {dataset_id}",
|
||||
{"dataset_id": dataset_id, "kind": kind.value},
|
||||
)
|
||||
@@ -0,0 +1,58 @@
|
||||
"""Benchmark 指标纯函数。
|
||||
|
||||
所有指标只依赖「按相关性降序的 retrieved id 列表」和「期望 id 集合」,不接触任何
|
||||
外部状态,便于单元测试与未来 Agent Benchmark 复用。retrieved 顺序越靠前越相关。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def hit_at_k(retrieved: list[str], expected: set[str], k: int) -> bool:
|
||||
"""前 k 个结果里是否命中任意期望 id(用于 Hit@1 / Hit@5)。"""
|
||||
return any(item in expected for item in retrieved[:k])
|
||||
|
||||
|
||||
def recall_at_k(retrieved: list[str], expected: set[str], k: int) -> float:
|
||||
"""前 k 个结果召回的期望 id 占比;期望为空时视为 0。
|
||||
|
||||
结果先去重:检索结果是 Block 级,同一 Note 可能经多个 Block 重复出现,
|
||||
直接逐项计数会把同一 Note 算多次、导致 Recall 超过 1。
|
||||
"""
|
||||
if not expected:
|
||||
return 0.0
|
||||
return len(set(retrieved[:k]) & expected) / len(expected)
|
||||
|
||||
|
||||
def reciprocal_rank(retrieved: list[str], expected: set[str]) -> float:
|
||||
"""首个命中的倒数排名;未命中返回 0。rank 从 1 开始。"""
|
||||
for rank, item in enumerate(retrieved, start=1):
|
||||
if item in expected:
|
||||
return 1.0 / rank
|
||||
return 0.0
|
||||
|
||||
|
||||
def citation_hit(retrieved_block_ids: list[str], expected: set[str]) -> bool:
|
||||
"""首条结果的 block_id 是否为期望引用块(Citation Hit Rate 的逐 Case 判据)。"""
|
||||
if not retrieved_block_ids or not expected:
|
||||
return False
|
||||
return retrieved_block_ids[0] in expected
|
||||
|
||||
|
||||
def mean(values: list[float]) -> float:
|
||||
return sum(values) / len(values) if values else 0.0
|
||||
|
||||
|
||||
def percentile(values: list[float], p: float) -> float:
|
||||
"""线性插值分位数(p ∈ [0, 100]),用于 P50 / P95 延迟。空列表返回 0。"""
|
||||
if not values:
|
||||
return 0.0
|
||||
ordered = sorted(values)
|
||||
if len(ordered) == 1:
|
||||
return ordered[0]
|
||||
rank = (len(ordered) - 1) * (p / 100.0)
|
||||
lo = int(rank)
|
||||
hi = lo + 1
|
||||
if hi >= len(ordered):
|
||||
return ordered[-1]
|
||||
frac = rank - lo
|
||||
return ordered[lo] + (ordered[hi] - ordered[lo]) * frac
|
||||
@@ -0,0 +1,143 @@
|
||||
"""RAG Benchmark Runner:调用检索引擎对数据集逐 Case 求值并聚合指标。
|
||||
|
||||
只读操作,直接复用 app.retrieval.engine 的 search(),不旁路检索链路。指标按
|
||||
(mode, case, repeat) 逐样本计算,再按 mode 聚合;失败样本按零分计入质量指标分母,
|
||||
避免把执行失败误判为检索质量(同时保留 total/successful/failed/failure_rate)。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
|
||||
from app.benchmarks import metrics as m
|
||||
from app.benchmarks.datasets import RAGDataset
|
||||
from app.contracts import (
|
||||
RAGCaseResult,
|
||||
RAGDatasetCase,
|
||||
RAGMetrics,
|
||||
RAGRunRequest,
|
||||
SearchMode,
|
||||
SearchRequest,
|
||||
)
|
||||
from app.retrieval.engine import engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BenchmarkCancelled(Exception):
|
||||
"""运行在 Case 之间被取消时抛出,用于中断后台执行并标记 cancelled。"""
|
||||
|
||||
|
||||
async def run_rag(
|
||||
dataset: RAGDataset,
|
||||
request: RAGRunRequest,
|
||||
on_case: Callable[[RAGCaseResult, int, int], None] | None = None,
|
||||
should_cancel: Callable[[], bool] | None = None,
|
||||
) -> tuple[dict[str, RAGMetrics], list[RAGCaseResult]]:
|
||||
"""执行 RAG Benchmark,返回 (按 mode 聚合的指标, 全部逐样本结果)。
|
||||
|
||||
on_case 在每个样本求值完成后回调 (result, done, total),供上层更新进度与事件。
|
||||
should_cancel 在每个样本开始前被检查;返回 True 时抛出 BenchmarkCancelled 中断运行。
|
||||
"""
|
||||
total = len(request.modes) * len(dataset.cases) * request.repeat
|
||||
done = 0
|
||||
results: list[RAGCaseResult] = []
|
||||
|
||||
for mode in request.modes:
|
||||
for case in dataset.cases:
|
||||
for repeat in range(request.repeat):
|
||||
if should_cancel is not None and should_cancel():
|
||||
raise BenchmarkCancelled()
|
||||
result = await _evaluate_one(case, mode, request, repeat)
|
||||
results.append(result)
|
||||
done += 1
|
||||
if on_case is not None:
|
||||
on_case(result, done, total)
|
||||
|
||||
metrics_by_mode = {mode.value: _aggregate(results, mode) for mode in request.modes}
|
||||
return metrics_by_mode, results
|
||||
|
||||
|
||||
async def _evaluate_one(
|
||||
case: RAGDatasetCase, mode: SearchMode, request: RAGRunRequest, repeat: int
|
||||
) -> RAGCaseResult:
|
||||
search_request = SearchRequest(
|
||||
query=case.query,
|
||||
mode=mode,
|
||||
limit=request.retrieval.top_k,
|
||||
include_snippet=False,
|
||||
rrf_k=request.retrieval.rrf_k,
|
||||
rerank=request.retrieval.rerank,
|
||||
rerank_candidates=request.retrieval.rerank_candidates,
|
||||
score_threshold=request.retrieval.score_threshold,
|
||||
)
|
||||
start = time.perf_counter()
|
||||
try:
|
||||
response = await engine.search(search_request)
|
||||
latency_ms = (time.perf_counter() - start) * 1000.0
|
||||
except Exception as exc: # 单个样本失败不中断整个 Benchmark
|
||||
# 详细异常只进日志,公开响应只带项目错误码与安全消息,避免泄露路径/SQL 等敏感信息
|
||||
logger.warning(
|
||||
"RAG case evaluation failed: case=%s mode=%s", case.case_id, mode.value,
|
||||
exc_info=exc,
|
||||
)
|
||||
return RAGCaseResult(
|
||||
case_id=case.case_id,
|
||||
mode=mode,
|
||||
repeat=repeat,
|
||||
latency_ms=(time.perf_counter() - start) * 1000.0,
|
||||
citation_applicable=case.citation_required,
|
||||
error="RAG case evaluation failed.",
|
||||
error_code="BENCHMARK_CASE_EVALUATION_FAILED",
|
||||
)
|
||||
|
||||
retrieved_note_ids = [item.note_id for item in response.items]
|
||||
retrieved_block_ids = [item.block_id for item in response.items]
|
||||
expected_notes = set(case.expected_note_ids)
|
||||
expected_blocks = set(case.expected_block_ids)
|
||||
k = request.retrieval.top_k
|
||||
|
||||
return RAGCaseResult(
|
||||
case_id=case.case_id,
|
||||
mode=mode,
|
||||
repeat=repeat,
|
||||
latency_ms=latency_ms,
|
||||
retrieved_note_ids=retrieved_note_ids,
|
||||
retrieved_block_ids=retrieved_block_ids,
|
||||
hit_at_1=m.hit_at_k(retrieved_note_ids, expected_notes, 1),
|
||||
hit_at_5=m.hit_at_k(retrieved_note_ids, expected_notes, 5),
|
||||
recall=m.recall_at_k(retrieved_note_ids, expected_notes, k),
|
||||
reciprocal_rank=m.reciprocal_rank(retrieved_note_ids, expected_notes),
|
||||
citation_hit=m.citation_hit(retrieved_block_ids, expected_blocks),
|
||||
citation_applicable=case.citation_required,
|
||||
)
|
||||
|
||||
|
||||
def _aggregate(cases: list[RAGCaseResult], mode: SearchMode) -> RAGMetrics:
|
||||
samples = [c for c in cases if c.mode == mode]
|
||||
total = len(samples)
|
||||
failed = sum(1 for c in samples if c.error is not None)
|
||||
successful = total - failed
|
||||
if total == 0:
|
||||
return RAGMetrics()
|
||||
|
||||
# 延迟只统计成功样本;失败样本按零分计入质量指标分母,避免汇总虚高
|
||||
latencies = [c.latency_ms for c in samples if c.error is None]
|
||||
citation_samples = [c for c in samples if c.citation_applicable]
|
||||
return RAGMetrics(
|
||||
hit_at_1=m.mean([1.0 if (c.error is None and c.hit_at_1) else 0.0 for c in samples]),
|
||||
hit_at_5=m.mean([1.0 if (c.error is None and c.hit_at_5) else 0.0 for c in samples]),
|
||||
recall_at_k=m.mean([c.recall if c.error is None else 0.0 for c in samples]),
|
||||
mrr=m.mean([c.reciprocal_rank if c.error is None else 0.0 for c in samples]),
|
||||
citation_hit_rate=m.mean(
|
||||
[1.0 if (c.error is None and c.citation_hit) else 0.0 for c in citation_samples]
|
||||
),
|
||||
p50_latency_ms=m.percentile(latencies, 50.0),
|
||||
p95_latency_ms=m.percentile(latencies, 95.0),
|
||||
total_cases=total,
|
||||
successful_cases=successful,
|
||||
failed_cases=failed,
|
||||
failure_rate=failed / total,
|
||||
)
|
||||
@@ -0,0 +1,348 @@
|
||||
"""Benchmark 服务:运行注册表、配置快照与报告组装。
|
||||
|
||||
RAG Benchmark 采用「创建即返回 queued、后台 Task 异步执行」的模式(与 index_service
|
||||
的 rebuild 一致):POST 创建后立即返回 202 queued 的 BenchmarkRun,由受管 asyncio.Task
|
||||
在后台逐 Case 求值,进度与事件实时写入内存注册表,供 SSE 订阅。运行记录、事件与报告
|
||||
暂存内存(_runs/_events/_reports),不持久化到 SQLite;后续接入异步任务队列时再落库。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from uuid import uuid4
|
||||
|
||||
from app import repository
|
||||
from app.benchmarks import datasets
|
||||
from app.benchmarks.datasets import RAGDataset
|
||||
from app.benchmarks.rag import BenchmarkCancelled, run_rag
|
||||
from app.config import get_settings
|
||||
from app.contracts import (
|
||||
BenchmarkEvent,
|
||||
BenchmarkEventType,
|
||||
BenchmarkKind,
|
||||
BenchmarkReport,
|
||||
BenchmarkRun,
|
||||
BenchmarkStatus,
|
||||
RAGCaseResult,
|
||||
RAGMetrics,
|
||||
RAGRunRequest,
|
||||
SearchMode,
|
||||
)
|
||||
from app.errors import ApiError
|
||||
from app.retrieval.engine import engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_runs: dict[str, BenchmarkRun] = {}
|
||||
_events: dict[str, list[BenchmarkEvent]] = {}
|
||||
_reports: dict[str, BenchmarkReport] = {}
|
||||
_tasks: dict[str, asyncio.Task] = {}
|
||||
_subscribers: dict[str, list[asyncio.Queue[BenchmarkEvent]]] = {}
|
||||
_cancel_flags: dict[str, asyncio.Event] = {}
|
||||
MAX_RUNS = 100
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _forget(run_id: str) -> None:
|
||||
"""移除一条 run 的全部内存态;仅在 run 处于终态时调用,避免打断活动任务。"""
|
||||
_runs.pop(run_id, None)
|
||||
_events.pop(run_id, None)
|
||||
_reports.pop(run_id, None)
|
||||
_tasks.pop(run_id, None)
|
||||
_subscribers.pop(run_id, None)
|
||||
_cancel_flags.pop(run_id, None)
|
||||
|
||||
|
||||
def _evict_terminal() -> bool:
|
||||
"""超过容量时淘汰最旧的终态 run;全部为活动 run 无法淘汰时返回 False。
|
||||
|
||||
绝不能删除仍在运行(queued/running)的 run:那会连带移除其 _cancel_flags 与
|
||||
_subscribers,使后台 Task 访问时抛出 KeyError。
|
||||
"""
|
||||
terminal = (BenchmarkStatus.completed, BenchmarkStatus.failed, BenchmarkStatus.cancelled)
|
||||
while len(_runs) >= MAX_RUNS:
|
||||
victim = next(
|
||||
(rid for rid, run in _runs.items() if run.status in terminal), None
|
||||
)
|
||||
if victim is None:
|
||||
return False
|
||||
_forget(victim)
|
||||
return True
|
||||
|
||||
|
||||
def _config_snapshot(request: RAGRunRequest, dataset: RAGDataset) -> dict:
|
||||
"""记录运行时的模型 / 索引 / 环境信息,保证报告可解释、可复现。"""
|
||||
settings = get_settings()
|
||||
return {
|
||||
"dataset_id": dataset.dataset_id,
|
||||
"dataset_hash": dataset.content_hash,
|
||||
"dataset_version": dataset.version,
|
||||
"modes": [m.value for m in request.modes],
|
||||
"retrieval": request.retrieval.model_dump(),
|
||||
"repeat": request.repeat,
|
||||
"embedding": {
|
||||
"model_id": engine.embedding.model_id,
|
||||
"version": engine.embedding.version,
|
||||
"dim": engine.embedding.dim,
|
||||
},
|
||||
"reranker": {
|
||||
"model_id": engine.reranker.model_id,
|
||||
"version": engine.reranker.version,
|
||||
},
|
||||
"index_meta": repository.get_index_meta(),
|
||||
"app": {"version": settings.version, "environment": settings.environment},
|
||||
"python": sys.version.split()[0],
|
||||
"metadata": request.metadata,
|
||||
}
|
||||
|
||||
|
||||
async def _validate_index_compatibility(request: RAGRunRequest) -> None:
|
||||
"""创建 RAG Run 前校验索引已建立且与当前 Embedding 模型/维度兼容。
|
||||
|
||||
空索引或不兼容索引会让所有模式得到全 0 指标,把环境/索引错误误判为检索质量差,
|
||||
故在创建时即拒绝,返回 BENCHMARK_INDEX_INCOMPATIBLE。
|
||||
"""
|
||||
stats = repository.stats()
|
||||
meta = repository.get_index_meta()
|
||||
needs_vector = any(m in (SearchMode.vector, SearchMode.hybrid) for m in request.modes)
|
||||
|
||||
reasons: list[str] = []
|
||||
if stats["blocks"] == 0:
|
||||
reasons.append("index is empty (no indexed blocks; run /api/index/rebuild first)")
|
||||
if needs_vector:
|
||||
if meta.get("embedding_model") != engine.embedding.model_id:
|
||||
reasons.append(
|
||||
f"embedding model mismatch: index={meta.get('embedding_model')!r}, "
|
||||
f"engine={engine.embedding.model_id!r}"
|
||||
)
|
||||
if meta.get("embedding_dim") != str(engine.embedding.dim):
|
||||
reasons.append(
|
||||
f"embedding dimension mismatch: index={meta.get('embedding_dim')!r}, "
|
||||
f"engine={engine.embedding.dim}"
|
||||
)
|
||||
if await engine.vector_store.count() == 0:
|
||||
reasons.append("vector index is empty")
|
||||
if reasons:
|
||||
raise ApiError(
|
||||
409,
|
||||
"BENCHMARK_INDEX_INCOMPATIBLE",
|
||||
"Benchmark index is not built or is incompatible with the current retrieval engine.",
|
||||
{"reasons": reasons},
|
||||
)
|
||||
|
||||
|
||||
async def create_rag_run(request: RAGRunRequest) -> BenchmarkRun:
|
||||
"""创建一次 RAG Benchmark,立即返回 queued 的 BenchmarkRun,由后台 Task 执行。"""
|
||||
dataset = datasets.load_dataset(request.dataset_id, BenchmarkKind.rag)
|
||||
await _validate_index_compatibility(request)
|
||||
|
||||
# 容量检查:先淘汰终态 run 腾空间;满容量且全为活动 run 时拒绝创建
|
||||
if not _evict_terminal():
|
||||
raise ApiError(
|
||||
429,
|
||||
"BENCHMARK_CAPACITY_EXCEEDED",
|
||||
"Benchmark run capacity exceeded; wait for active runs to finish.",
|
||||
{},
|
||||
)
|
||||
|
||||
run_id = "benchmark_" + uuid4().hex[:12]
|
||||
snapshot = _config_snapshot(request, dataset)
|
||||
run = BenchmarkRun(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
dataset_id=dataset.dataset_id,
|
||||
dataset_hash=dataset.content_hash,
|
||||
status=BenchmarkStatus.queued,
|
||||
progress=0.0,
|
||||
config_snapshot=snapshot,
|
||||
created_at=_now(),
|
||||
)
|
||||
_runs[run_id] = run
|
||||
_events[run_id] = []
|
||||
_subscribers[run_id] = []
|
||||
_cancel_flags[run_id] = asyncio.Event()
|
||||
_tasks[run_id] = asyncio.create_task(_execute_rag(run_id, request, dataset, snapshot))
|
||||
return run
|
||||
|
||||
|
||||
async def _execute_rag(
|
||||
run_id: str, request: RAGRunRequest, dataset: RAGDataset, snapshot: dict
|
||||
) -> None:
|
||||
"""后台执行 RAG Benchmark,实时更新进度/事件,结束后写入报告并关闭订阅。"""
|
||||
cancel_event = _cancel_flags[run_id]
|
||||
|
||||
def emit(event_type: BenchmarkEventType, data: dict) -> None:
|
||||
sequence = len(_events[run_id])
|
||||
event = BenchmarkEvent(
|
||||
event=event_type, run_id=run_id, sequence=sequence, data=data, timestamp=_now()
|
||||
)
|
||||
_events[run_id].append(event)
|
||||
for queue in _subscribers.get(run_id, []):
|
||||
queue.put_nowait(event)
|
||||
|
||||
def finish() -> None:
|
||||
_subscribers.pop(run_id, None)
|
||||
_cancel_flags.pop(run_id, None)
|
||||
|
||||
_runs[run_id] = _runs[run_id].model_copy(
|
||||
update={"status": BenchmarkStatus.running, "started_at": _now()}
|
||||
)
|
||||
emit(
|
||||
BenchmarkEventType.run_started,
|
||||
{"dataset_id": dataset.dataset_id, "modes": [m.value for m in request.modes]},
|
||||
)
|
||||
total = len(request.modes) * len(dataset.cases) * request.repeat
|
||||
|
||||
def on_case(result: RAGCaseResult, done: int, _total: int) -> None:
|
||||
progress = done / total if total else 1.0
|
||||
_runs[run_id] = _runs[run_id].model_copy(update={"progress": progress})
|
||||
emit(BenchmarkEventType.case_completed, result.model_dump(mode="json"))
|
||||
|
||||
try:
|
||||
metrics_by_mode, results = await run_rag(
|
||||
dataset,
|
||||
request,
|
||||
on_case=on_case,
|
||||
should_cancel=cancel_event.is_set,
|
||||
)
|
||||
except BenchmarkCancelled:
|
||||
_runs[run_id] = _runs[run_id].model_copy(
|
||||
update={
|
||||
"status": BenchmarkStatus.cancelled,
|
||||
"progress": 1.0,
|
||||
"completed_at": _now(),
|
||||
}
|
||||
)
|
||||
emit(BenchmarkEventType.run_cancelled, {"status": BenchmarkStatus.cancelled.value})
|
||||
_reports[run_id] = BenchmarkReport(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
dataset_id=dataset.dataset_id,
|
||||
dataset_hash=dataset.content_hash,
|
||||
status=BenchmarkStatus.cancelled,
|
||||
config_snapshot=snapshot,
|
||||
)
|
||||
finish()
|
||||
return
|
||||
except Exception as exc: # 单次运行失败不拖垮服务,记录错误后结束
|
||||
# 详细异常只进日志,公开响应仅带项目错误码与安全消息,避免泄露路径/SQL 等敏感信息
|
||||
logger.exception("Benchmark run failed: run_id=%s", run_id)
|
||||
_runs[run_id] = _runs[run_id].model_copy(
|
||||
update={
|
||||
"status": BenchmarkStatus.failed,
|
||||
"progress": 1.0,
|
||||
"error": "Benchmark run failed.",
|
||||
"error_code": "BENCHMARK_RUN_FAILED",
|
||||
"completed_at": _now(),
|
||||
}
|
||||
)
|
||||
emit(
|
||||
BenchmarkEventType.run_failed,
|
||||
{"error": "Benchmark run failed.", "error_code": "BENCHMARK_RUN_FAILED"},
|
||||
)
|
||||
_reports[run_id] = BenchmarkReport(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
dataset_id=dataset.dataset_id,
|
||||
dataset_hash=dataset.content_hash,
|
||||
status=BenchmarkStatus.failed,
|
||||
config_snapshot=snapshot,
|
||||
error="Benchmark run failed.",
|
||||
error_code="BENCHMARK_RUN_FAILED",
|
||||
)
|
||||
finish()
|
||||
return
|
||||
|
||||
metrics = {mode: m.model_dump() for mode, m in metrics_by_mode.items()}
|
||||
_runs[run_id] = _runs[run_id].model_copy(
|
||||
update={
|
||||
"status": BenchmarkStatus.completed,
|
||||
"progress": 1.0,
|
||||
"metrics": metrics,
|
||||
"completed_at": _now(),
|
||||
}
|
||||
)
|
||||
emit(BenchmarkEventType.run_completed, {"metrics": metrics})
|
||||
_reports[run_id] = BenchmarkReport(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
dataset_id=dataset.dataset_id,
|
||||
dataset_hash=dataset.content_hash,
|
||||
status=BenchmarkStatus.completed,
|
||||
config_snapshot=snapshot,
|
||||
metrics=metrics,
|
||||
cases=results,
|
||||
)
|
||||
finish()
|
||||
|
||||
|
||||
def list_runs(
|
||||
kind: BenchmarkKind | None = None,
|
||||
status: BenchmarkStatus | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[BenchmarkRun], int]:
|
||||
runs = list(_runs.values())
|
||||
if kind is not None:
|
||||
runs = [r for r in runs if r.kind == kind]
|
||||
if status is not None:
|
||||
runs = [r for r in runs if r.status == status]
|
||||
runs.sort(key=lambda r: r.created_at, reverse=True)
|
||||
total = len(runs)
|
||||
return runs[offset : offset + limit], total
|
||||
|
||||
|
||||
def get_run(run_id: str) -> BenchmarkRun | None:
|
||||
return _runs.get(run_id)
|
||||
|
||||
|
||||
def get_report(run_id: str) -> BenchmarkReport | None:
|
||||
return _reports.get(run_id)
|
||||
|
||||
|
||||
def get_events(run_id: str) -> list[BenchmarkEvent]:
|
||||
return _events.get(run_id, [])
|
||||
|
||||
|
||||
def cancel_run(run_id: str) -> BenchmarkRun | None:
|
||||
"""取消运行:对 queued/running 设置取消标志,后台 Task 在 Case 边界检查后置为 cancelled。"""
|
||||
run = _runs.get(run_id)
|
||||
if run is None:
|
||||
return None
|
||||
if run.status in (BenchmarkStatus.queued, BenchmarkStatus.running):
|
||||
_cancel_flags[run_id].set()
|
||||
return run
|
||||
|
||||
|
||||
def subscribe(run_id: str) -> asyncio.Queue[BenchmarkEvent] | None:
|
||||
"""订阅运行事件流;运行已结束(completed/failed/cancelled)时返回 None。"""
|
||||
run = _runs.get(run_id)
|
||||
if run is None or run.status in (
|
||||
BenchmarkStatus.completed,
|
||||
BenchmarkStatus.failed,
|
||||
BenchmarkStatus.cancelled,
|
||||
):
|
||||
return None
|
||||
queue: asyncio.Queue[BenchmarkEvent] = asyncio.Queue()
|
||||
_subscribers.setdefault(run_id, []).append(queue)
|
||||
return queue
|
||||
|
||||
|
||||
def unsubscribe(run_id: str, queue: asyncio.Queue[BenchmarkEvent]) -> None:
|
||||
subscribers = _subscribers.get(run_id)
|
||||
if subscribers and queue in subscribers:
|
||||
subscribers.remove(queue)
|
||||
|
||||
|
||||
async def wait_for_run(run_id: str) -> BenchmarkRun:
|
||||
"""等待后台任务结束(测试/轮询用);无任务时直接返回当前状态。"""
|
||||
task = _tasks.get(run_id)
|
||||
if task is not None:
|
||||
await task
|
||||
return _runs.get(run_id)
|
||||
@@ -24,6 +24,7 @@ class Settings:
|
||||
db_path: Path
|
||||
vault_path: Path
|
||||
attachments_path: Path
|
||||
benchmark_datasets_path: Path
|
||||
|
||||
|
||||
@lru_cache
|
||||
@@ -41,4 +42,7 @@ def get_settings() -> Settings:
|
||||
attachments_path=Path(
|
||||
os.getenv("APP_ATTACHMENTS_PATH", str(data_dir / "attachments"))
|
||||
),
|
||||
benchmark_datasets_path=Path(
|
||||
os.getenv("APP_BENCHMARK_DATASETS_PATH", str(data_dir / "benchmarks"))
|
||||
),
|
||||
)
|
||||
|
||||
@@ -3,7 +3,7 @@ from dataclasses import dataclass
|
||||
from app.agent import AgentRuntime, PermissionManager, PermissionPolicy, ToolRegistry
|
||||
from app.agent.builtin_tools import register_builtin_tools
|
||||
from app.contracts import ModelCapability, ProviderConfig, ProviderType
|
||||
from app.config import BACKEND_DIR
|
||||
from app.config import BACKEND_DIR, get_settings
|
||||
from app.extensions import PluginRuntime, SkillRuntime
|
||||
from app.providers import MockProvider, ProviderFactory, ProviderRegistry
|
||||
from app.providers.credentials import (
|
||||
@@ -26,6 +26,7 @@ class ApplicationContainer:
|
||||
|
||||
|
||||
def build_container() -> ApplicationContainer:
|
||||
settings = get_settings()
|
||||
credentials = EncryptedCredentialStore()
|
||||
provider_factory = ProviderFactory(
|
||||
ChainedCredentialResolver(credentials, EnvironmentCredentialResolver())
|
||||
@@ -50,7 +51,12 @@ def build_container() -> ApplicationContainer:
|
||||
tools = ToolRegistry()
|
||||
register_builtin_tools(tools)
|
||||
|
||||
plugins = PluginRuntime(tools)
|
||||
plugins = PluginRuntime(
|
||||
tools,
|
||||
# 当前 Python Host 尚无 OS 沙箱。生产构建必须保持关闭,直到
|
||||
# Tauri/Rust Host 能签发绑定命令摘要的可信启动许可。
|
||||
allow_unsandboxed_mcp=settings.environment == "development",
|
||||
)
|
||||
plugins.install(BACKEND_DIR / "extensions" / "plugins" / "text-tools")
|
||||
plugins.enable("text-tools")
|
||||
|
||||
|
||||
+181
-1
@@ -2,7 +2,7 @@ from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, SecretStr
|
||||
from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator
|
||||
|
||||
|
||||
class Contract(BaseModel):
|
||||
@@ -144,6 +144,12 @@ class SearchRequest(Contract):
|
||||
limit: int = Field(default=20, ge=1, le=100)
|
||||
offset: int = Field(default=0, ge=0)
|
||||
include_snippet: bool = True
|
||||
# 检索调优参数(Benchmark 与 Skill 共用):控制 RRF / 精排 / 候选池 / 分数阈值。
|
||||
# rerank_candidates=None 表示对全部候选精排(保留原有行为),Benchmark 传显式值。
|
||||
rrf_k: int = Field(default=60, ge=1)
|
||||
rerank: bool = True
|
||||
rerank_candidates: int | None = Field(default=None, ge=1)
|
||||
score_threshold: float = Field(default=0.0, ge=0.0)
|
||||
|
||||
|
||||
class Citation(Contract):
|
||||
@@ -417,6 +423,10 @@ class ExtensionInstallRequest(Contract):
|
||||
class PluginBackend(Contract):
|
||||
type: Literal["mcp", "internal_rpc", "none"] = "none"
|
||||
transport: Literal["stdio", "http", "none"] = "none"
|
||||
command: str | None = None
|
||||
args: list[str] = Field(default_factory=list)
|
||||
startup_timeout_seconds: int = Field(default=10, ge=1, le=60)
|
||||
tool_timeout_seconds: int = Field(default=30, ge=1, le=600)
|
||||
|
||||
|
||||
class PluginContribution(Contract):
|
||||
@@ -460,6 +470,28 @@ class PluginListResponse(Contract):
|
||||
items: list[Plugin] = Field(default_factory=list)
|
||||
|
||||
|
||||
class PluginHostState(str, Enum):
|
||||
stopped = "stopped"
|
||||
starting = "starting"
|
||||
ready = "ready"
|
||||
unhealthy = "unhealthy"
|
||||
error = "error"
|
||||
|
||||
|
||||
class PluginHostStatus(Contract):
|
||||
plugin_id: str
|
||||
backend_type: Literal["mcp", "internal_rpc", "none"]
|
||||
transport: Literal["stdio", "http", "none"]
|
||||
status: PluginHostState
|
||||
tools_count: int = 0
|
||||
started_at: datetime | None = None
|
||||
last_seen_at: datetime | None = None
|
||||
protocol_version: str | None = None
|
||||
server_name: str | None = None
|
||||
server_version: str | None = None
|
||||
error: str | None = None
|
||||
|
||||
|
||||
class PluginPermissionGrantRequest(Contract):
|
||||
permissions: list[str] = Field(default_factory=list)
|
||||
|
||||
@@ -626,3 +658,151 @@ class IndexJob(Contract):
|
||||
status: Literal["queued", "running", "completed", "failed"]
|
||||
scope: Literal["all", "notes", "vectors"]
|
||||
created_at: datetime
|
||||
|
||||
|
||||
# Benchmark
|
||||
class BenchmarkKind(str, Enum):
|
||||
rag = "rag"
|
||||
agent = "agent"
|
||||
|
||||
|
||||
class BenchmarkStatus(str, Enum):
|
||||
queued = "queued"
|
||||
running = "running"
|
||||
completed = "completed"
|
||||
failed = "failed"
|
||||
cancelled = "cancelled"
|
||||
|
||||
|
||||
class RAGDatasetCase(Contract):
|
||||
case_id: str
|
||||
query: str = Field(min_length=1)
|
||||
expected_note_ids: list[str] = Field(default_factory=list)
|
||||
expected_block_ids: list[str] = Field(default_factory=list)
|
||||
citation_required: bool = False
|
||||
tags: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class RAGRetrievalConfig(Contract):
|
||||
"""RAG Benchmark 的检索参数。top_k 映射到 SearchRequest.limit,
|
||||
其余参数透传到 SearchRequest,由检索引擎实际执行。"""
|
||||
|
||||
top_k: int = Field(default=10, ge=1, le=100)
|
||||
rrf_k: int = Field(default=60, ge=1)
|
||||
rerank: bool = True
|
||||
rerank_candidates: int = Field(default=20, ge=1)
|
||||
score_threshold: float = Field(default=0.0, ge=0.0)
|
||||
|
||||
|
||||
class RAGRunRequest(Contract):
|
||||
dataset_id: str = Field(min_length=1)
|
||||
modes: list[SearchMode] = Field(
|
||||
default_factory=lambda: [SearchMode.fts, SearchMode.vector, SearchMode.hybrid],
|
||||
min_length=1,
|
||||
)
|
||||
retrieval: RAGRetrievalConfig = Field(default_factory=RAGRetrievalConfig)
|
||||
repeat: int = Field(default=1, ge=1, le=10)
|
||||
metadata: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
@field_validator("modes")
|
||||
@classmethod
|
||||
def _no_duplicate_modes(cls, value: list[SearchMode]) -> list[SearchMode]:
|
||||
if len(value) != len(set(value)):
|
||||
raise ValueError("modes must not contain duplicates")
|
||||
return value
|
||||
|
||||
|
||||
class RAGMetrics(Contract):
|
||||
hit_at_1: float = 0.0
|
||||
hit_at_5: float = 0.0
|
||||
recall_at_k: float = 0.0
|
||||
mrr: float = 0.0
|
||||
citation_hit_rate: float = 0.0
|
||||
p50_latency_ms: float = 0.0
|
||||
p95_latency_ms: float = 0.0
|
||||
# 样本构成:失败样本按零分计入质量指标,汇总不虚高;报告据此可知实际分母
|
||||
total_cases: int = 0
|
||||
successful_cases: int = 0
|
||||
failed_cases: int = 0
|
||||
failure_rate: float = 0.0
|
||||
|
||||
|
||||
class BenchmarkDatasetInfo(Contract):
|
||||
dataset_id: str
|
||||
kind: BenchmarkKind
|
||||
version: str
|
||||
description: str = ""
|
||||
case_count: int
|
||||
content_hash: str
|
||||
|
||||
|
||||
class BenchmarkDatasetListResponse(Contract):
|
||||
items: list[BenchmarkDatasetInfo] = Field(default_factory=list)
|
||||
|
||||
|
||||
class BenchmarkRun(Contract):
|
||||
run_id: str
|
||||
kind: BenchmarkKind
|
||||
dataset_id: str
|
||||
dataset_hash: str
|
||||
status: BenchmarkStatus
|
||||
progress: float | None = None
|
||||
metrics: dict[str, Any] | None = None
|
||||
config_snapshot: dict[str, Any] = Field(default_factory=dict)
|
||||
error: str | None = None
|
||||
error_code: str | None = None
|
||||
created_at: datetime
|
||||
started_at: datetime | None = None
|
||||
completed_at: datetime | None = None
|
||||
|
||||
|
||||
class BenchmarkRunListResponse(Contract):
|
||||
items: list[BenchmarkRun] = Field(default_factory=list)
|
||||
page: PageMeta = Field(default_factory=PageMeta)
|
||||
|
||||
|
||||
class BenchmarkEventType(str, Enum):
|
||||
run_started = "RunStarted"
|
||||
case_completed = "CaseCompleted"
|
||||
run_completed = "RunCompleted"
|
||||
run_failed = "RunFailed"
|
||||
run_cancelled = "RunCancelled"
|
||||
|
||||
|
||||
class BenchmarkEvent(Contract):
|
||||
event: BenchmarkEventType
|
||||
run_id: str
|
||||
sequence: int
|
||||
data: dict[str, Any] = Field(default_factory=dict)
|
||||
timestamp: datetime
|
||||
|
||||
|
||||
class RAGCaseResult(Contract):
|
||||
case_id: str
|
||||
mode: SearchMode
|
||||
repeat: int
|
||||
latency_ms: float
|
||||
retrieved_note_ids: list[str] = Field(default_factory=list)
|
||||
retrieved_block_ids: list[str] = Field(default_factory=list)
|
||||
hit_at_1: bool = False
|
||||
hit_at_5: bool = False
|
||||
recall: float = 0.0
|
||||
reciprocal_rank: float = 0.0
|
||||
citation_hit: bool = False
|
||||
# 该 Case 是否声明了 expected_block_ids(决定是否计入 citation_hit_rate 分母)
|
||||
citation_applicable: bool = False
|
||||
error: str | None = None
|
||||
error_code: str | None = None
|
||||
|
||||
|
||||
class BenchmarkReport(Contract):
|
||||
run_id: str
|
||||
kind: BenchmarkKind
|
||||
dataset_id: str
|
||||
dataset_hash: str
|
||||
status: BenchmarkStatus
|
||||
config_snapshot: dict[str, Any] = Field(default_factory=dict)
|
||||
metrics: dict[str, Any] = Field(default_factory=dict)
|
||||
cases: list[RAGCaseResult] = Field(default_factory=list)
|
||||
error: str | None = None
|
||||
error_code: str | None = None
|
||||
|
||||
@@ -4,5 +4,13 @@ from app.extensions.runtime import (
|
||||
PluginRuntime,
|
||||
SkillRuntime,
|
||||
)
|
||||
from app.extensions.mcp import McpBridge, McpBridgeError
|
||||
|
||||
__all__ = ["AgentConfiguration", "ExtensionError", "PluginRuntime", "SkillRuntime"]
|
||||
__all__ = [
|
||||
"AgentConfiguration",
|
||||
"ExtensionError",
|
||||
"McpBridge",
|
||||
"McpBridgeError",
|
||||
"PluginRuntime",
|
||||
"SkillRuntime",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,785 @@
|
||||
"""本地 stdio MCP Bridge。
|
||||
|
||||
第三方 Server 始终运行在子进程中。Bridge 只把通过校验的 MCP Tool 转换为项目内部
|
||||
ToolDefinition/ToolResult,不把 MCP 原始协议泄露给 Agent Runtime 或前端。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import queue
|
||||
import subprocess
|
||||
import threading
|
||||
from collections import deque
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from jsonschema import Draft202012Validator
|
||||
from jsonschema.exceptions import SchemaError
|
||||
|
||||
from app.agent.permissions import KNOWN_PERMISSIONS
|
||||
from app.agent.tools import ToolExecutionError
|
||||
from app.contracts import (
|
||||
PluginBackend,
|
||||
PluginHostState,
|
||||
PluginHostStatus,
|
||||
ToolDefinition,
|
||||
)
|
||||
|
||||
MCP_PROTOCOL_VERSION = "2025-11-25"
|
||||
SUPPORTED_PROTOCOL_VERSIONS = {
|
||||
MCP_PROTOCOL_VERSION,
|
||||
"2025-06-18",
|
||||
"2025-03-26",
|
||||
"2024-11-05",
|
||||
}
|
||||
MAX_MCP_MESSAGE_BYTES = 2 * 1024 * 1024
|
||||
MAX_MCP_TOOL_RESULT_BYTES = 256 * 1024
|
||||
MAX_MCP_TOOLS = 500
|
||||
MAX_MCP_LIST_PAGES = 100
|
||||
|
||||
|
||||
class McpBridgeError(RuntimeError):
|
||||
def __init__(self, code: str, message: str, *, status_code: int = 502) -> None:
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.message = message
|
||||
self.status_code = status_code
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class McpDiscoveredTool:
|
||||
remote_name: str
|
||||
definition: ToolDefinition
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _PendingRequest:
|
||||
response: queue.Queue[dict[str, Any] | BaseException]
|
||||
|
||||
|
||||
class McpStdioClient:
|
||||
"""线程驱动的换行分隔 JSON-RPC 客户端,避免阻塞 FastAPI 事件循环。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
command: list[str],
|
||||
*,
|
||||
cwd: Path,
|
||||
on_seen: Callable[[], None],
|
||||
on_broken: Callable[[str], None],
|
||||
on_tools_changed: Callable[[], None],
|
||||
) -> None:
|
||||
self.command = command
|
||||
self.cwd = cwd
|
||||
self.on_seen = on_seen
|
||||
self.on_broken = on_broken
|
||||
self.on_tools_changed = on_tools_changed
|
||||
self.process: subprocess.Popen[str] | None = None
|
||||
self._write_lock = threading.Lock()
|
||||
self._pending_lock = threading.Lock()
|
||||
self._pending: dict[int, _PendingRequest] = {}
|
||||
self._next_id = 1
|
||||
self._stopping = False
|
||||
# stderr 只在 Host 内部保留有限尾部,不进入 API、Trace 或普通日志。
|
||||
self._stderr_tail: deque[str] = deque(maxlen=50)
|
||||
|
||||
def start(self) -> None:
|
||||
if self.process is not None and self.process.poll() is None:
|
||||
return
|
||||
# TODO(extension-security): 社区 Plugin 开放前迁移到 Tauri/Rust Host 的
|
||||
# 平台级沙箱启动器;uvx 只隔离 Python 依赖,不能替代系统权限限制。
|
||||
creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0
|
||||
environment = _subprocess_environment()
|
||||
environment.setdefault("PYTHONUNBUFFERED", "1")
|
||||
try:
|
||||
self.process = subprocess.Popen(
|
||||
self.command,
|
||||
cwd=self.cwd,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
bufsize=1,
|
||||
shell=False,
|
||||
env=environment,
|
||||
creationflags=creation_flags,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_START_FAILED",
|
||||
f"Cannot start MCP server process: {exc}",
|
||||
status_code=503,
|
||||
) from exc
|
||||
threading.Thread(target=self._stdout_loop, daemon=True).start()
|
||||
threading.Thread(target=self._stderr_loop, daemon=True).start()
|
||||
|
||||
def request(
|
||||
self,
|
||||
method: str,
|
||||
params: dict[str, Any],
|
||||
*,
|
||||
timeout: float,
|
||||
timeout_code: str,
|
||||
response_error_code: str = "MCP_TOOL_CALL_FAILED",
|
||||
) -> dict[str, Any]:
|
||||
request_id, pending = self.begin_request(method, params)
|
||||
return self.wait_response(
|
||||
request_id,
|
||||
pending,
|
||||
timeout=timeout,
|
||||
timeout_code=timeout_code,
|
||||
response_error_code=response_error_code,
|
||||
)
|
||||
|
||||
def begin_request(
|
||||
self, method: str, params: dict[str, Any]
|
||||
) -> tuple[int, _PendingRequest]:
|
||||
self._ensure_running()
|
||||
with self._pending_lock:
|
||||
request_id = self._next_id
|
||||
self._next_id += 1
|
||||
pending = _PendingRequest(response=queue.Queue(maxsize=1))
|
||||
self._pending[request_id] = pending
|
||||
try:
|
||||
self._send(
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": request_id,
|
||||
"method": method,
|
||||
"params": params,
|
||||
}
|
||||
)
|
||||
except BaseException:
|
||||
with self._pending_lock:
|
||||
self._pending.pop(request_id, None)
|
||||
raise
|
||||
return request_id, pending
|
||||
|
||||
def wait_response(
|
||||
self,
|
||||
request_id: int,
|
||||
pending: _PendingRequest,
|
||||
*,
|
||||
timeout: float,
|
||||
timeout_code: str,
|
||||
response_error_code: str = "MCP_TOOL_CALL_FAILED",
|
||||
) -> dict[str, Any]:
|
||||
try:
|
||||
response = pending.response.get(timeout=timeout)
|
||||
except queue.Empty as exc:
|
||||
self.cancel(request_id, "Request timed out.")
|
||||
self.abandon(request_id)
|
||||
raise McpBridgeError(timeout_code, "MCP request timed out.", status_code=504) from exc
|
||||
if isinstance(response, BaseException):
|
||||
raise response
|
||||
if "error" in response:
|
||||
error = response.get("error")
|
||||
message = (
|
||||
str(error.get("message", "MCP JSON-RPC error."))
|
||||
if isinstance(error, dict)
|
||||
else "MCP JSON-RPC error."
|
||||
)
|
||||
raise McpBridgeError(response_error_code, message)
|
||||
result = response.get("result")
|
||||
if not isinstance(result, dict):
|
||||
raise McpBridgeError(
|
||||
response_error_code, "MCP response result must be an object."
|
||||
)
|
||||
return result
|
||||
|
||||
def notify(self, method: str, params: dict[str, Any] | None = None) -> None:
|
||||
payload: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
|
||||
if params is not None:
|
||||
payload["params"] = params
|
||||
self._send(payload)
|
||||
|
||||
def cancel(self, request_id: int, reason: str = "Cancelled by host.") -> None:
|
||||
try:
|
||||
self.notify(
|
||||
"notifications/cancelled",
|
||||
{"requestId": request_id, "reason": reason},
|
||||
)
|
||||
except McpBridgeError:
|
||||
pass
|
||||
|
||||
def abandon(
|
||||
self, request_id: int, wake_error: BaseException | None = None
|
||||
) -> None:
|
||||
with self._pending_lock:
|
||||
pending = self._pending.pop(request_id, None)
|
||||
# asyncio.to_thread 被取消时不会停止底层线程;主动唤醒 Queue,避免线程
|
||||
# 一直占用默认线程池直至远端超时。
|
||||
if pending is not None and wake_error is not None:
|
||||
try:
|
||||
pending.response.put_nowait(wake_error)
|
||||
except queue.Full:
|
||||
pass
|
||||
|
||||
def stop(self) -> None:
|
||||
process = self.process
|
||||
if process is None:
|
||||
return
|
||||
self._stopping = True
|
||||
try:
|
||||
if process.stdin:
|
||||
try:
|
||||
process.stdin.close()
|
||||
except (BrokenPipeError, OSError, ValueError):
|
||||
pass
|
||||
try:
|
||||
process.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
process.wait(timeout=2)
|
||||
finally:
|
||||
self._fail_pending(
|
||||
McpBridgeError("PLUGIN_HOST_UNAVAILABLE", "MCP host stopped.", status_code=503)
|
||||
)
|
||||
self.process = None
|
||||
|
||||
def _send(self, message: dict[str, Any]) -> None:
|
||||
self._ensure_running()
|
||||
encoded = json.dumps(message, ensure_ascii=False, separators=(",", ":"))
|
||||
if len(encoded.encode("utf-8")) > MAX_MCP_MESSAGE_BYTES:
|
||||
raise McpBridgeError("MCP_TOOL_CALL_FAILED", "MCP request is too large.")
|
||||
process = self.process
|
||||
assert process is not None and process.stdin is not None
|
||||
try:
|
||||
with self._write_lock:
|
||||
process.stdin.write(encoded + "\n")
|
||||
process.stdin.flush()
|
||||
except (BrokenPipeError, OSError, ValueError) as exc:
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_UNAVAILABLE", "MCP host input is closed.", status_code=503
|
||||
) from exc
|
||||
|
||||
def _stdout_loop(self) -> None:
|
||||
process = self.process
|
||||
assert process is not None and process.stdout is not None
|
||||
failure: str | None = None
|
||||
try:
|
||||
while True:
|
||||
# readline(size) 在换行缺失时仍有硬上限,不能先把任意大的
|
||||
# 第三方 stdout 行完整读入宿主内存再检查。
|
||||
raw_line = process.stdout.readline(MAX_MCP_MESSAGE_BYTES + 1)
|
||||
if raw_line == "":
|
||||
break
|
||||
if not raw_line.endswith("\n"):
|
||||
failure = "MCP server emitted an oversized or unterminated message."
|
||||
break
|
||||
if len(raw_line.encode("utf-8")) > MAX_MCP_MESSAGE_BYTES:
|
||||
failure = "MCP server emitted an oversized protocol message."
|
||||
break
|
||||
try:
|
||||
message = json.loads(raw_line)
|
||||
except json.JSONDecodeError:
|
||||
failure = "MCP server emitted invalid JSON on stdout."
|
||||
break
|
||||
if not isinstance(message, dict) or message.get("jsonrpc") != "2.0":
|
||||
failure = "MCP server emitted an invalid JSON-RPC message."
|
||||
break
|
||||
self.on_seen()
|
||||
if "id" in message and ("result" in message or "error" in message):
|
||||
request_id = message.get("id")
|
||||
if isinstance(request_id, int):
|
||||
with self._pending_lock:
|
||||
pending = self._pending.pop(request_id, None)
|
||||
if pending:
|
||||
pending.response.put(message)
|
||||
continue
|
||||
method = message.get("method")
|
||||
if method == "notifications/tools/list_changed":
|
||||
self.on_tools_changed()
|
||||
elif isinstance(method, str) and "id" in message:
|
||||
self._send(
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": message["id"],
|
||||
"error": {"code": -32601, "message": "Method not supported."},
|
||||
}
|
||||
)
|
||||
except (McpBridgeError, OSError, ValueError) as exc:
|
||||
failure = f"MCP stdout closed unexpectedly: {type(exc).__name__}."
|
||||
finally:
|
||||
if failure and process.poll() is None:
|
||||
process.terminate()
|
||||
exit_code = process.poll()
|
||||
if exit_code is None:
|
||||
try:
|
||||
exit_code = process.wait(timeout=1)
|
||||
except subprocess.TimeoutExpired:
|
||||
exit_code = None
|
||||
if not self._stopping:
|
||||
message = failure or f"MCP host exited unexpectedly with code {exit_code}."
|
||||
error = McpBridgeError(
|
||||
"PLUGIN_HOST_UNAVAILABLE", message, status_code=503
|
||||
)
|
||||
self._fail_pending(error)
|
||||
self.on_broken(message)
|
||||
|
||||
def _stderr_loop(self) -> None:
|
||||
process = self.process
|
||||
assert process is not None and process.stderr is not None
|
||||
try:
|
||||
while True:
|
||||
# stderr 不是协议通道,但同样按块读取,避免无换行日志造成
|
||||
# 宿主侧的无界字符串分配。
|
||||
line = process.stderr.readline(1025)
|
||||
if line == "":
|
||||
break
|
||||
self._stderr_tail.append(line.rstrip()[:1024])
|
||||
except (OSError, ValueError):
|
||||
return
|
||||
|
||||
def _ensure_running(self) -> None:
|
||||
if self.process is None or self.process.poll() is not None:
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_UNAVAILABLE", "MCP host is not running.", status_code=503
|
||||
)
|
||||
|
||||
def _fail_pending(self, error: BaseException) -> None:
|
||||
with self._pending_lock:
|
||||
pending = list(self._pending.values())
|
||||
self._pending.clear()
|
||||
for item in pending:
|
||||
item.response.put(error)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _McpHost:
|
||||
backend: PluginBackend
|
||||
client: McpStdioClient
|
||||
status: PluginHostStatus
|
||||
|
||||
|
||||
class McpBridge:
|
||||
"""管理每个 Plugin 的独立 MCP Client,并执行 Contract 转换。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._hosts: dict[str, _McpHost] = {}
|
||||
self._statuses: dict[str, PluginHostStatus] = {}
|
||||
self._calls: dict[tuple[str, str], int] = {}
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def start(
|
||||
self,
|
||||
plugin_id: str,
|
||||
backend: PluginBackend,
|
||||
package_path: Path,
|
||||
declared_permissions: list[str],
|
||||
on_unavailable: Callable[[str, str], None],
|
||||
) -> list[McpDiscoveredTool]:
|
||||
if backend.transport != "stdio":
|
||||
raise McpBridgeError(
|
||||
"MCP_CAPABILITY_UNSUPPORTED",
|
||||
"Phase C only supports the MCP stdio transport.",
|
||||
status_code=501,
|
||||
)
|
||||
command = self._resolve_command(package_path, backend)
|
||||
now = datetime.now(timezone.utc)
|
||||
status = PluginHostStatus(
|
||||
plugin_id=plugin_id,
|
||||
backend_type="mcp",
|
||||
transport="stdio",
|
||||
status=PluginHostState.starting,
|
||||
started_at=now,
|
||||
last_seen_at=now,
|
||||
)
|
||||
host_ref: dict[str, _McpHost] = {}
|
||||
|
||||
def seen() -> None:
|
||||
host = host_ref.get("host")
|
||||
if host:
|
||||
host.status.last_seen_at = datetime.now(timezone.utc)
|
||||
|
||||
def broken(message: str) -> None:
|
||||
host = host_ref.get("host")
|
||||
if host:
|
||||
host.status.status = PluginHostState.unhealthy
|
||||
host.status.error = message
|
||||
on_unavailable(plugin_id, message)
|
||||
|
||||
def tools_changed() -> None:
|
||||
broken("MCP tool list changed; restart the Plugin Host to revalidate tools.")
|
||||
|
||||
client = McpStdioClient(
|
||||
command,
|
||||
cwd=package_path,
|
||||
on_seen=seen,
|
||||
on_broken=broken,
|
||||
on_tools_changed=tools_changed,
|
||||
)
|
||||
host = _McpHost(backend=backend, client=client, status=status)
|
||||
host_ref["host"] = host
|
||||
with self._lock:
|
||||
if plugin_id in self._hosts:
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_START_FAILED",
|
||||
f"MCP host is already running: {plugin_id}",
|
||||
status_code=409,
|
||||
)
|
||||
self._hosts[plugin_id] = host
|
||||
self._statuses[plugin_id] = status
|
||||
try:
|
||||
client.start()
|
||||
initialize = client.request(
|
||||
"initialize",
|
||||
{
|
||||
"protocolVersion": MCP_PROTOCOL_VERSION,
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "NotesAgent", "version": "0.1.0"},
|
||||
},
|
||||
timeout=backend.startup_timeout_seconds,
|
||||
timeout_code="MCP_INITIALIZE_FAILED",
|
||||
response_error_code="MCP_INITIALIZE_FAILED",
|
||||
)
|
||||
version = initialize.get("protocolVersion")
|
||||
if version not in SUPPORTED_PROTOCOL_VERSIONS:
|
||||
raise McpBridgeError(
|
||||
"MCP_INITIALIZE_FAILED",
|
||||
f"Unsupported MCP protocol version: {version}",
|
||||
)
|
||||
capabilities = initialize.get("capabilities")
|
||||
if not isinstance(capabilities, dict) or not isinstance(
|
||||
capabilities.get("tools"), dict
|
||||
):
|
||||
raise McpBridgeError(
|
||||
"MCP_CAPABILITY_UNSUPPORTED",
|
||||
"MCP server does not declare the tools capability.",
|
||||
)
|
||||
server_info = initialize.get("serverInfo")
|
||||
if not isinstance(server_info, dict):
|
||||
server_info = {}
|
||||
status.protocol_version = str(version)
|
||||
status.server_name = _optional_string(server_info.get("name"))
|
||||
status.server_version = _optional_string(server_info.get("version"))
|
||||
client.notify("notifications/initialized")
|
||||
discovered = self._discover_tools(
|
||||
plugin_id, client, backend, declared_permissions
|
||||
)
|
||||
status.status = PluginHostState.ready
|
||||
status.tools_count = len(discovered)
|
||||
status.last_seen_at = datetime.now(timezone.utc)
|
||||
status.error = None
|
||||
return discovered
|
||||
except McpBridgeError as exc:
|
||||
status.status = PluginHostState.error
|
||||
status.error = exc.message
|
||||
client.stop()
|
||||
with self._lock:
|
||||
self._hosts.pop(plugin_id, None)
|
||||
raise
|
||||
except Exception as exc:
|
||||
status.status = PluginHostState.error
|
||||
status.error = f"MCP initialization failed: {type(exc).__name__}."
|
||||
client.stop()
|
||||
with self._lock:
|
||||
self._hosts.pop(plugin_id, None)
|
||||
raise McpBridgeError("MCP_INITIALIZE_FAILED", status.error) from exc
|
||||
|
||||
async def call_tool(
|
||||
self,
|
||||
plugin_id: str,
|
||||
remote_name: str,
|
||||
arguments: dict[str, Any],
|
||||
*,
|
||||
request_id: str,
|
||||
) -> Any:
|
||||
host = self._host(plugin_id)
|
||||
rpc_id, pending = host.client.begin_request(
|
||||
"tools/call", {"name": remote_name, "arguments": arguments}
|
||||
)
|
||||
call_key = (plugin_id, request_id)
|
||||
with self._lock:
|
||||
self._calls[call_key] = rpc_id
|
||||
try:
|
||||
result = await asyncio.to_thread(
|
||||
host.client.wait_response,
|
||||
rpc_id,
|
||||
pending,
|
||||
timeout=host.backend.tool_timeout_seconds,
|
||||
timeout_code="MCP_TOOL_CALL_FAILED",
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
host.client.cancel(rpc_id)
|
||||
host.client.abandon(
|
||||
rpc_id,
|
||||
McpBridgeError(
|
||||
"MCP_TOOL_CALL_FAILED", "MCP request was cancelled."
|
||||
),
|
||||
)
|
||||
raise
|
||||
except McpBridgeError as exc:
|
||||
raise ToolExecutionError(exc.code, exc.message) from exc
|
||||
finally:
|
||||
with self._lock:
|
||||
self._calls.pop(call_key, None)
|
||||
|
||||
encoded_size = len(
|
||||
json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||||
)
|
||||
if encoded_size > MAX_MCP_TOOL_RESULT_BYTES:
|
||||
raise ToolExecutionError(
|
||||
"MCP_TOOL_RESULT_TOO_LARGE",
|
||||
"MCP tool result exceeds the configured size limit.",
|
||||
)
|
||||
if result.get("isError") is True:
|
||||
raise ToolExecutionError(
|
||||
"MCP_TOOL_CALL_FAILED", _mcp_error_message(result.get("content"))
|
||||
)
|
||||
structured = result.get("structuredContent")
|
||||
if structured is not None:
|
||||
if not isinstance(structured, dict):
|
||||
raise ToolExecutionError(
|
||||
"MCP_TOOL_CALL_FAILED",
|
||||
"MCP structuredContent must be an object.",
|
||||
)
|
||||
return structured
|
||||
content = result.get("content", [])
|
||||
if not isinstance(content, list):
|
||||
raise ToolExecutionError(
|
||||
"MCP_TOOL_CALL_FAILED", "MCP tool content must be an array."
|
||||
)
|
||||
return {"content": content}
|
||||
|
||||
def cancel(self, plugin_id: str, request_id: str) -> None:
|
||||
with self._lock:
|
||||
rpc_id = self._calls.get((plugin_id, request_id))
|
||||
host = self._hosts.get(plugin_id)
|
||||
if rpc_id is not None and host is not None:
|
||||
host.client.cancel(rpc_id)
|
||||
|
||||
def stop(self, plugin_id: str) -> None:
|
||||
with self._lock:
|
||||
host = self._hosts.pop(plugin_id, None)
|
||||
if host:
|
||||
host.client.stop()
|
||||
host.status.status = PluginHostState.stopped
|
||||
host.status.tools_count = 0
|
||||
host.status.error = None
|
||||
|
||||
def remove(self, plugin_id: str) -> None:
|
||||
"""停止 Host,并清除卸载后不应跨安装保留的状态与调用索引。"""
|
||||
|
||||
self.stop(plugin_id)
|
||||
with self._lock:
|
||||
self._statuses.pop(plugin_id, None)
|
||||
stale_calls = [key for key in self._calls if key[0] == plugin_id]
|
||||
for key in stale_calls:
|
||||
self._calls.pop(key, None)
|
||||
|
||||
def status(self, plugin_id: str, backend: PluginBackend) -> PluginHostStatus:
|
||||
with self._lock:
|
||||
status = self._statuses.get(plugin_id)
|
||||
if status:
|
||||
return status.model_copy(deep=True)
|
||||
return PluginHostStatus(
|
||||
plugin_id=plugin_id,
|
||||
backend_type=backend.type,
|
||||
transport=backend.transport,
|
||||
status=PluginHostState.stopped,
|
||||
)
|
||||
|
||||
def _discover_tools(
|
||||
self,
|
||||
plugin_id: str,
|
||||
client: McpStdioClient,
|
||||
backend: PluginBackend,
|
||||
declared_permissions: list[str],
|
||||
) -> list[McpDiscoveredTool]:
|
||||
discovered: list[McpDiscoveredTool] = []
|
||||
cursor: str | None = None
|
||||
for _ in range(MAX_MCP_LIST_PAGES):
|
||||
params = {"cursor": cursor} if cursor else {}
|
||||
result = client.request(
|
||||
"tools/list",
|
||||
params,
|
||||
timeout=backend.startup_timeout_seconds,
|
||||
timeout_code="MCP_INITIALIZE_FAILED",
|
||||
response_error_code="MCP_INITIALIZE_FAILED",
|
||||
)
|
||||
raw_tools = result.get("tools")
|
||||
if not isinstance(raw_tools, list):
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP tools/list must return a tools array."
|
||||
)
|
||||
for raw in raw_tools:
|
||||
discovered.append(
|
||||
self._map_tool(plugin_id, raw, declared_permissions)
|
||||
)
|
||||
if len(discovered) > MAX_MCP_TOOLS:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"MCP server exposes more than {MAX_MCP_TOOLS} tools.",
|
||||
)
|
||||
next_cursor = result.get("nextCursor")
|
||||
if next_cursor is None:
|
||||
break
|
||||
if not isinstance(next_cursor, str) or not next_cursor:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP nextCursor must be a non-empty string."
|
||||
)
|
||||
cursor = next_cursor
|
||||
else:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP tools/list exceeded the page limit."
|
||||
)
|
||||
names = [item.definition.name for item in discovered]
|
||||
if len(names) != len(set(names)):
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP server returned duplicate tool names."
|
||||
)
|
||||
return discovered
|
||||
|
||||
@staticmethod
|
||||
def _map_tool(
|
||||
plugin_id: str, raw: Any, declared_permissions: list[str]
|
||||
) -> McpDiscoveredTool:
|
||||
if not isinstance(raw, dict):
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP tool definition must be an object."
|
||||
)
|
||||
remote_name = raw.get("name")
|
||||
if not isinstance(remote_name, str) or not remote_name:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID", "MCP tool name must be a non-empty string."
|
||||
)
|
||||
if (
|
||||
len(remote_name) > 128
|
||||
or not remote_name[0].isalnum()
|
||||
or not all(
|
||||
character.islower()
|
||||
or character.isdigit()
|
||||
or character in "._-"
|
||||
for character in remote_name
|
||||
)
|
||||
):
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"MCP tool name is not a valid NotesAgent id: {remote_name}",
|
||||
)
|
||||
schema = raw.get("inputSchema", {"type": "object", "properties": {}})
|
||||
if not isinstance(schema, dict) or schema.get("type", "object") != "object":
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"MCP tool inputSchema must be an object schema: {remote_name}",
|
||||
)
|
||||
try:
|
||||
Draft202012Validator.check_schema(schema)
|
||||
except SchemaError as exc:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"Invalid MCP tool schema for {remote_name}: {exc.message}",
|
||||
) from exc
|
||||
metadata = raw.get("_meta")
|
||||
permission = (
|
||||
metadata.get("notesagent/permission") if isinstance(metadata, dict) else None
|
||||
)
|
||||
if permission is not None and (
|
||||
not isinstance(permission, str) or permission not in KNOWN_PERMISSIONS
|
||||
):
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"MCP tool declares an unknown permission: {remote_name}",
|
||||
)
|
||||
if permission and permission not in declared_permissions:
|
||||
raise McpBridgeError(
|
||||
"MCP_TOOL_SCHEMA_INVALID",
|
||||
f"MCP tool permission is missing from Plugin manifest: {permission}",
|
||||
)
|
||||
description = raw.get("description")
|
||||
return McpDiscoveredTool(
|
||||
remote_name=remote_name,
|
||||
definition=ToolDefinition(
|
||||
name=f"{plugin_id}.{remote_name}",
|
||||
description=description if isinstance(description, str) else remote_name,
|
||||
parameters=schema,
|
||||
permission=permission,
|
||||
source="plugin",
|
||||
),
|
||||
)
|
||||
|
||||
def _host(self, plugin_id: str) -> _McpHost:
|
||||
with self._lock:
|
||||
host = self._hosts.get(plugin_id)
|
||||
if host is None or host.status.status != PluginHostState.ready:
|
||||
raise ToolExecutionError(
|
||||
"PLUGIN_HOST_UNAVAILABLE", f"MCP Plugin Host is not ready: {plugin_id}"
|
||||
)
|
||||
return host
|
||||
|
||||
@staticmethod
|
||||
def _resolve_command(root: Path, backend: PluginBackend) -> list[str]:
|
||||
if not backend.command or not backend.command.strip():
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_START_FAILED", "MCP stdio backend requires a command."
|
||||
)
|
||||
command = backend.command.strip()
|
||||
if Path(command).is_absolute() or "/" in command or "\\" in command:
|
||||
executable = (
|
||||
(root / command).resolve()
|
||||
if not Path(command).is_absolute()
|
||||
else Path(command).resolve()
|
||||
)
|
||||
try:
|
||||
executable.relative_to(root)
|
||||
except ValueError as exc:
|
||||
raise McpBridgeError(
|
||||
"PLUGIN_HOST_START_FAILED",
|
||||
"MCP executable path must stay inside the Plugin package.",
|
||||
) from exc
|
||||
command = str(executable)
|
||||
return [command, *backend.args]
|
||||
|
||||
|
||||
def _mcp_error_message(content: Any) -> str:
|
||||
if isinstance(content, list):
|
||||
texts = [
|
||||
item.get("text")
|
||||
for item in content
|
||||
if isinstance(item, dict)
|
||||
and item.get("type") == "text"
|
||||
and isinstance(item.get("text"), str)
|
||||
]
|
||||
if texts:
|
||||
return "\n".join(texts)[:4096]
|
||||
return "MCP tool returned an error result."
|
||||
|
||||
|
||||
def _optional_string(value: Any) -> str | None:
|
||||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
def _subprocess_environment() -> dict[str, str]:
|
||||
"""只传递启动进程所需的系统变量,隔离 Provider Key、Vault 路径等宿主状态。"""
|
||||
|
||||
allowed = {
|
||||
"PATH",
|
||||
"PATHEXT",
|
||||
"SYSTEMROOT",
|
||||
"WINDIR",
|
||||
"COMSPEC",
|
||||
"TEMP",
|
||||
"TMP",
|
||||
"TMPDIR",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"VIRTUAL_ENV",
|
||||
}
|
||||
environment = {
|
||||
key: value for key, value in os.environ.items() if key.upper() in allowed
|
||||
}
|
||||
environment["PYTHONUNBUFFERED"] = "1"
|
||||
environment["PYTHONIOENCODING"] = "utf-8"
|
||||
return environment
|
||||
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import threading
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
@@ -16,6 +17,7 @@ from app.contracts import (
|
||||
ModelCapability,
|
||||
Plugin,
|
||||
PluginManifest,
|
||||
PluginHostStatus,
|
||||
PluginStatus,
|
||||
RetrievalConfig,
|
||||
Skill,
|
||||
@@ -23,6 +25,7 @@ from app.contracts import (
|
||||
SkillStatus,
|
||||
ToolDefinition,
|
||||
)
|
||||
from app.extensions.mcp import McpBridge, McpBridgeError, McpDiscoveredTool
|
||||
|
||||
_EXTENSION_ID = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
|
||||
|
||||
@@ -242,18 +245,29 @@ class _PluginRecord:
|
||||
tools: list[DeclarativeToolSpec]
|
||||
package_path: Path
|
||||
registered_tools: list[str]
|
||||
mcp_remote_names: dict[str, str]
|
||||
|
||||
|
||||
class PluginRuntime:
|
||||
"""Plugin Manifest、生命周期及 Tool Contribution 注册。"""
|
||||
|
||||
def __init__(self, tools: ToolRegistry, host: DeclarativePluginHost | None = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
tools: ToolRegistry,
|
||||
host: DeclarativePluginHost | None = None,
|
||||
mcp_bridge: McpBridge | None = None,
|
||||
*,
|
||||
allow_unsandboxed_mcp: bool = False,
|
||||
) -> None:
|
||||
self.registry = tools
|
||||
self.host = host or DeclarativePluginHost()
|
||||
self.mcp = mcp_bridge or McpBridge()
|
||||
self.allow_unsandboxed_mcp = allow_unsandboxed_mcp
|
||||
self._records: dict[str, _PluginRecord] = {}
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def install(self, package_path: str | Path) -> Plugin:
|
||||
# 当前只加载声明式清单,不导入或执行插件包中的任意 Python 代码。
|
||||
# 安装阶段只读取清单;MCP 子进程必须在权限授予后的 enable 阶段启动。
|
||||
root = _package_dir(package_path)
|
||||
raw = _read_yaml(root / "plugin.yaml")
|
||||
if "id" in raw and "plugin_id" not in raw:
|
||||
@@ -271,7 +285,9 @@ class PluginRuntime:
|
||||
status_code=409,
|
||||
)
|
||||
|
||||
specs = self._load_tools(root)
|
||||
_validate_backend(manifest)
|
||||
specs = [] if manifest.backend.type == "mcp" else self._load_tools(root)
|
||||
if manifest.backend.type != "mcp":
|
||||
declared = set(manifest.contributes.tools)
|
||||
actual = {spec.name for spec in specs}
|
||||
if declared != actual:
|
||||
@@ -302,6 +318,7 @@ class PluginRuntime:
|
||||
tools=specs,
|
||||
package_path=root,
|
||||
registered_tools=[],
|
||||
mcp_remote_names={},
|
||||
)
|
||||
self._records[manifest.plugin_id] = record
|
||||
return record.plugin.model_copy(deep=True)
|
||||
@@ -313,18 +330,14 @@ class PluginRuntime:
|
||||
return self._record(plugin_id).plugin.model_copy(deep=True)
|
||||
|
||||
def enable(self, plugin_id: str) -> Plugin:
|
||||
# Host 启动和 Tool 批量注册必须串行,避免并发 enable 产生重复进程或半注册状态。
|
||||
with self._lock:
|
||||
return self._enable(plugin_id)
|
||||
|
||||
def _enable(self, plugin_id: str) -> Plugin:
|
||||
record = self._record(plugin_id)
|
||||
if record.plugin.enabled:
|
||||
return record.plugin.model_copy(deep=True)
|
||||
if record.plugin.manifest.backend.type == "mcp":
|
||||
# TODO(extension): 第二阶段以隔离进程实现 MCP Host,并补充签名与来源校验。
|
||||
record.plugin.status = PluginStatus.dependency_missing
|
||||
raise ExtensionError(
|
||||
"PLUGIN_HOST_UNAVAILABLE",
|
||||
"MCP Plugin Host is reserved for the second development phase.",
|
||||
status_code=501,
|
||||
details={"plugin_id": plugin_id, "backend": "mcp"},
|
||||
)
|
||||
missing_grants = sorted(
|
||||
set(record.plugin.manifest.permissions) - set(record.plugin.granted_permissions)
|
||||
)
|
||||
@@ -336,7 +349,18 @@ class PluginRuntime:
|
||||
status_code=409,
|
||||
details={"plugin_id": plugin_id, "permissions": missing_grants},
|
||||
)
|
||||
conflicts = [spec.name for spec in record.tools if self.registry.contains(spec.name)]
|
||||
if (
|
||||
record.plugin.manifest.backend.type == "mcp"
|
||||
and not self.allow_unsandboxed_mcp
|
||||
):
|
||||
raise ExtensionError(
|
||||
"MCP_TRUST_APPROVAL_REQUIRED",
|
||||
"Unsandboxed MCP Hosts are disabled outside development mode.",
|
||||
status_code=403,
|
||||
details={"plugin_id": plugin_id},
|
||||
)
|
||||
declared_tools = list(record.plugin.manifest.contributes.tools)
|
||||
conflicts = [name for name in declared_tools if self.registry.contains(name)]
|
||||
if conflicts:
|
||||
raise ExtensionError(
|
||||
"PLUGIN_TOOL_CONFLICT",
|
||||
@@ -346,6 +370,19 @@ class PluginRuntime:
|
||||
)
|
||||
record.plugin.status = PluginStatus.starting
|
||||
try:
|
||||
if record.plugin.manifest.backend.type == "mcp":
|
||||
discovered = self._start_mcp(record)
|
||||
actual = {item.definition.name for item in discovered}
|
||||
declared = set(declared_tools)
|
||||
if actual != declared:
|
||||
raise ExtensionError(
|
||||
"PLUGIN_CONTRIBUTION_INVALID",
|
||||
"Discovered MCP tools must exactly match Plugin contributions.",
|
||||
details={"declared": sorted(declared), "actual": sorted(actual)},
|
||||
)
|
||||
for item in discovered:
|
||||
self._register_mcp_tool(record, item)
|
||||
else:
|
||||
for spec in record.tools:
|
||||
arguments_model = _arguments_model(spec)
|
||||
|
||||
@@ -373,15 +410,35 @@ class PluginRuntime:
|
||||
for name in record.registered_tools:
|
||||
self.registry.unregister(name)
|
||||
record.registered_tools.clear()
|
||||
record.mcp_remote_names.clear()
|
||||
self.mcp.stop(plugin_id)
|
||||
record.plugin.status = PluginStatus.error
|
||||
record.plugin.error_message = str(exc)
|
||||
record.plugin.error_message = _safe_extension_message(exc)
|
||||
if isinstance(exc, ExtensionError):
|
||||
raise
|
||||
if isinstance(exc, McpBridgeError):
|
||||
raise ExtensionError(
|
||||
exc.code,
|
||||
exc.message,
|
||||
status_code=exc.status_code,
|
||||
details={"plugin_id": plugin_id},
|
||||
) from exc
|
||||
raise ExtensionError(
|
||||
"PLUGIN_HOST_START_FAILED",
|
||||
record.plugin.error_message,
|
||||
status_code=503,
|
||||
details={"plugin_id": plugin_id},
|
||||
) from exc
|
||||
record.plugin.enabled = True
|
||||
record.plugin.status = PluginStatus.ready
|
||||
record.plugin.error_message = None
|
||||
return record.plugin.model_copy(deep=True)
|
||||
|
||||
def set_permissions(self, plugin_id: str, permissions: list[str]) -> Plugin:
|
||||
with self._lock:
|
||||
return self._set_permissions(plugin_id, permissions)
|
||||
|
||||
def _set_permissions(self, plugin_id: str, permissions: list[str]) -> Plugin:
|
||||
record = self._record(plugin_id)
|
||||
requested = set(permissions)
|
||||
declared = set(record.plugin.manifest.permissions)
|
||||
@@ -403,15 +460,125 @@ class PluginRuntime:
|
||||
return record.plugin.model_copy(deep=True)
|
||||
|
||||
def disable(self, plugin_id: str) -> Plugin:
|
||||
with self._lock:
|
||||
return self._disable(plugin_id)
|
||||
|
||||
def _disable(self, plugin_id: str) -> Plugin:
|
||||
record = self._record(plugin_id)
|
||||
for name in record.registered_tools:
|
||||
self.registry.unregister(name)
|
||||
record.registered_tools.clear()
|
||||
record.mcp_remote_names.clear()
|
||||
if record.plugin.manifest.backend.type == "mcp":
|
||||
self.mcp.stop(plugin_id)
|
||||
record.plugin.enabled = False
|
||||
record.plugin.status = PluginStatus.disabled
|
||||
return record.plugin.model_copy(deep=True)
|
||||
|
||||
def get_host_status(self, plugin_id: str) -> PluginHostStatus:
|
||||
record = self._record(plugin_id)
|
||||
return self.mcp.status(plugin_id, record.plugin.manifest.backend)
|
||||
|
||||
def restart_host(self, plugin_id: str) -> PluginHostStatus:
|
||||
with self._lock:
|
||||
return self._restart_host(plugin_id)
|
||||
|
||||
def _restart_host(self, plugin_id: str) -> PluginHostStatus:
|
||||
record = self._record(plugin_id)
|
||||
if record.plugin.manifest.backend.type != "mcp":
|
||||
raise ExtensionError(
|
||||
"PLUGIN_HOST_UNAVAILABLE",
|
||||
"Plugin does not use an MCP Host.",
|
||||
status_code=409,
|
||||
details={"plugin_id": plugin_id},
|
||||
)
|
||||
if record.plugin.status in {
|
||||
PluginStatus.installed,
|
||||
PluginStatus.disabled,
|
||||
PluginStatus.permission_required,
|
||||
}:
|
||||
raise ExtensionError(
|
||||
"PLUGIN_HOST_UNAVAILABLE",
|
||||
"Disabled or inactive MCP Plugins must be started with Enable.",
|
||||
status_code=409,
|
||||
details={"plugin_id": plugin_id, "status": record.plugin.status.value},
|
||||
)
|
||||
for name in record.registered_tools:
|
||||
self.registry.unregister(name)
|
||||
record.registered_tools.clear()
|
||||
record.mcp_remote_names.clear()
|
||||
self.mcp.stop(plugin_id)
|
||||
record.plugin.enabled = False
|
||||
record.plugin.status = PluginStatus.installed
|
||||
record.plugin.error_message = None
|
||||
self.enable(plugin_id)
|
||||
return self.get_host_status(plugin_id)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""关闭所有隔离 Host;用于 FastAPI lifespan 和测试清理。"""
|
||||
|
||||
with self._lock:
|
||||
for plugin_id, record in list(self._records.items()):
|
||||
if record.plugin.manifest.backend.type == "mcp":
|
||||
self.mcp.stop(plugin_id)
|
||||
|
||||
def _start_mcp(self, record: _PluginRecord) -> list[McpDiscoveredTool]:
|
||||
manifest = record.plugin.manifest
|
||||
return self.mcp.start(
|
||||
manifest.plugin_id,
|
||||
manifest.backend,
|
||||
record.package_path,
|
||||
manifest.permissions,
|
||||
self._handle_mcp_unavailable,
|
||||
)
|
||||
|
||||
def _register_mcp_tool(
|
||||
self, record: _PluginRecord, discovered: McpDiscoveredTool
|
||||
) -> None:
|
||||
definition = discovered.definition
|
||||
arguments_model = _arguments_model_from_schema(
|
||||
definition.name, definition.parameters
|
||||
)
|
||||
plugin_id = record.plugin.manifest.plugin_id
|
||||
remote_name = discovered.remote_name
|
||||
|
||||
async def executor(
|
||||
arguments: BaseModel,
|
||||
context: ToolExecutionContext,
|
||||
) -> Any:
|
||||
return await self.mcp.call_tool(
|
||||
plugin_id,
|
||||
remote_name,
|
||||
# 省略的可选字段不能被补成 null;显式传入的 null 仍由
|
||||
# model_fields_set 保留并交给 MCP Server。
|
||||
arguments.model_dump(exclude_unset=True),
|
||||
request_id=context.tool_call_id or f"{context.run_id}:{definition.name}",
|
||||
)
|
||||
|
||||
self.registry.register(definition, arguments_model, executor)
|
||||
record.registered_tools.append(definition.name)
|
||||
record.mcp_remote_names[definition.name] = remote_name
|
||||
|
||||
def _handle_mcp_unavailable(self, plugin_id: str, message: str) -> None:
|
||||
with self._lock:
|
||||
record = self._records.get(plugin_id)
|
||||
if record is None:
|
||||
return
|
||||
for name in record.registered_tools:
|
||||
self.registry.unregister(name)
|
||||
record.registered_tools.clear()
|
||||
record.mcp_remote_names.clear()
|
||||
record.plugin.enabled = False
|
||||
record.plugin.status = PluginStatus.error
|
||||
record.plugin.error_message = message
|
||||
|
||||
def uninstall(self, plugin_id: str, dependent_skills: list[str] | None = None) -> None:
|
||||
with self._lock:
|
||||
self._uninstall(plugin_id, dependent_skills)
|
||||
|
||||
def _uninstall(
|
||||
self, plugin_id: str, dependent_skills: list[str] | None = None
|
||||
) -> None:
|
||||
record = self._record(plugin_id)
|
||||
if dependent_skills:
|
||||
raise ExtensionError(
|
||||
@@ -420,8 +587,13 @@ class PluginRuntime:
|
||||
status_code=409,
|
||||
details={"plugin_id": plugin_id, "skills": dependent_skills},
|
||||
)
|
||||
is_mcp = record.plugin.manifest.backend.type == "mcp"
|
||||
if record.plugin.enabled:
|
||||
self.disable(plugin_id)
|
||||
if is_mcp:
|
||||
# stop 只结束本次进程并保留状态供故障诊断;真正卸载时必须连同
|
||||
# 历史状态一起遗忘,避免同 ID 重装继承旧协商信息。
|
||||
self.mcp.remove(plugin_id)
|
||||
del self._records[plugin_id]
|
||||
|
||||
def _record(self, plugin_id: str) -> _PluginRecord:
|
||||
@@ -498,24 +670,18 @@ def _manifest_error(kind: str, exc: ValidationError) -> ExtensionError:
|
||||
|
||||
def _arguments_model(spec: DeclarativeToolSpec) -> type[BaseModel]:
|
||||
schema = spec.parameters or {"type": "object", "properties": {}}
|
||||
return _arguments_model_from_schema(spec.name, schema)
|
||||
|
||||
|
||||
def _arguments_model_from_schema(
|
||||
tool_name: str, schema: dict[str, Any]
|
||||
) -> type[BaseModel]:
|
||||
if schema.get("type", "object") != "object":
|
||||
raise ExtensionError("PLUGIN_TOOL_SCHEMA_INVALID", "Tool parameters must be an object schema.")
|
||||
properties = schema.get("properties", {})
|
||||
required = set(schema.get("required", []))
|
||||
fields: dict[str, tuple[Any, Any]] = {}
|
||||
types = {
|
||||
"string": str,
|
||||
"number": float,
|
||||
"integer": int,
|
||||
"boolean": bool,
|
||||
"array": list[Any],
|
||||
"object": dict[str, Any],
|
||||
}
|
||||
for name, field_schema in properties.items():
|
||||
annotation = types.get(field_schema.get("type"), Any)
|
||||
fields[name] = (annotation, ... if name in required else None)
|
||||
model_name = "PluginArgs_" + re.sub(r"\W+", "_", spec.name)
|
||||
return create_model(model_name, __config__=ConfigDict(extra="forbid"), **fields)
|
||||
model_name = "PluginArgs_" + re.sub(r"\W+", "_", tool_name)
|
||||
# 完整 JSON Schema 已在 ToolRegistry 中先行校验。参数载体不重复声明字段,
|
||||
# 从而完整保留 model_dump、连字符键、联合类型和动态属性等合法 JSON 键值。
|
||||
return create_model(model_name, __config__=ConfigDict(extra="allow"))
|
||||
|
||||
|
||||
def _validate_tool_schema(spec: DeclarativeToolSpec) -> None:
|
||||
@@ -536,3 +702,30 @@ def _validate_tool_schema(spec: DeclarativeToolSpec) -> None:
|
||||
"Tool parameters must be an object schema with object properties.",
|
||||
details={"tool": spec.name},
|
||||
)
|
||||
|
||||
|
||||
def _validate_backend(manifest: PluginManifest) -> None:
|
||||
backend = manifest.backend
|
||||
if backend.type == "mcp":
|
||||
if backend.transport != "stdio":
|
||||
raise ExtensionError(
|
||||
"MCP_CAPABILITY_UNSUPPORTED",
|
||||
"Phase C MCP Plugins must use stdio transport.",
|
||||
status_code=501,
|
||||
)
|
||||
if not backend.command or not backend.command.strip():
|
||||
raise ExtensionError(
|
||||
"EXTENSION_MANIFEST_INVALID",
|
||||
"MCP stdio backend requires a command.",
|
||||
)
|
||||
elif backend.command is not None or backend.args:
|
||||
raise ExtensionError(
|
||||
"EXTENSION_MANIFEST_INVALID",
|
||||
"Only MCP stdio backends may declare command or args.",
|
||||
)
|
||||
|
||||
|
||||
def _safe_extension_message(exc: Exception) -> str:
|
||||
if isinstance(exc, (ExtensionError, McpBridgeError)):
|
||||
return exc.message
|
||||
return f"Plugin Host operation failed: {type(exc).__name__}."
|
||||
|
||||
@@ -1,19 +1,31 @@
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from starlette.exceptions import HTTPException as StarletteHttpException
|
||||
|
||||
from app.config import get_settings
|
||||
from app.container import container
|
||||
from app.errors import ApiError, api_error_handler, http_error_handler, validation_error_handler
|
||||
from app.routes import router as api_router
|
||||
from app.schemas import HealthResponse, ServiceStatusResponse
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(_: FastAPI):
|
||||
yield
|
||||
# 第三方 MCP Server 必须跟随 AI Core 退出,不能遗留孤儿进程。
|
||||
container.plugins.shutdown()
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
title=settings.name,
|
||||
version=settings.version,
|
||||
description="AI 笔记软件的本地 AI Core 与 Agent Core 服务。",
|
||||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
app.add_middleware(
|
||||
|
||||
@@ -19,6 +19,7 @@ class EmbeddingProvider(Protocol):
|
||||
"""统一 Embedding 接口(与文档一致)。"""
|
||||
|
||||
model_id: str
|
||||
version: str
|
||||
dim: int
|
||||
|
||||
async def embed_documents(self, texts: list[str]) -> list[list[float]]: ...
|
||||
@@ -33,6 +34,7 @@ class HashEmbeddingProvider:
|
||||
"""
|
||||
|
||||
model_id = "hash-v1"
|
||||
version = "1"
|
||||
dim = EMBEDDING_DIM
|
||||
|
||||
async def embed_documents(self, texts: list[str]) -> list[list[float]]:
|
||||
|
||||
@@ -29,6 +29,8 @@ from app.textutils import make_snippet, match_query
|
||||
CANDIDATE_POOL = 50
|
||||
# 分页窗口上限:候选池至少覆盖 offset+limit,但设上限防止超大 offset 撑爆内存
|
||||
MAX_CANDIDATE_POOL = 200
|
||||
# FTS 全量取回上限:统一归一化 + 阈值过滤后再分页,保证阈值语义跨页一致
|
||||
FTS_FETCH_LIMIT = 5000
|
||||
# 带 metadata 过滤时放大召回倍数,缓解「先截断候选池再过滤」造成的漏召回
|
||||
OVERSCAN_FACTOR = 4
|
||||
|
||||
@@ -84,7 +86,7 @@ class RetrievalEngine:
|
||||
elif request.mode == SearchMode.vector:
|
||||
candidate_scores = vec_scores
|
||||
else: # hybrid:RRF 融合
|
||||
candidate_scores = rrf_fuse([fts_ranked, vec_ranked])
|
||||
candidate_scores = rrf_fuse([fts_ranked, vec_ranked], k=request.rrf_k)
|
||||
|
||||
if not candidate_scores:
|
||||
return self._empty(request)
|
||||
@@ -97,14 +99,23 @@ class RetrievalEngine:
|
||||
if not filtered:
|
||||
return self._empty(request)
|
||||
|
||||
# 4. 排序 / 精排
|
||||
# 4. 排序 / 精排:hybrid 先按融合分预排序,再对前 rerank_candidates 个候选做精排,
|
||||
# 剩余候选按融合分排在精排结果之后;rerank=False 时跳过精排直接按融合分排序。
|
||||
if request.mode == SearchMode.hybrid:
|
||||
pre_sorted = sorted(filtered, key=lambda h: -candidate_scores[h.block_id])
|
||||
if request.rerank:
|
||||
limit = request.rerank_candidates
|
||||
pool = pre_sorted if limit is None else pre_sorted[:limit]
|
||||
rest = [] if limit is None else pre_sorted[limit:]
|
||||
candidates = [
|
||||
RankedCandidate(block_id=h.block_id, score=candidate_scores[h.block_id], text=h.content)
|
||||
for h in filtered
|
||||
for h in pool
|
||||
]
|
||||
ranked = await self.reranker.rerank(request.query, candidates)
|
||||
ordered = [(c.block_id, c.score) for c in ranked]
|
||||
ordered += [(h.block_id, candidate_scores[h.block_id]) for h in rest]
|
||||
else:
|
||||
ordered = [(h.block_id, candidate_scores[h.block_id]) for h in pre_sorted]
|
||||
else:
|
||||
ordered = sorted(
|
||||
((h.block_id, candidate_scores[h.block_id]) for h in filtered),
|
||||
@@ -112,6 +123,8 @@ class RetrievalEngine:
|
||||
)
|
||||
|
||||
ordered = normalize_scores(ordered)
|
||||
# score_threshold:归一化后过滤低分结果(默认 0 不过滤)
|
||||
ordered = [(bid, score) for bid, score in ordered if score >= request.score_threshold]
|
||||
|
||||
# 5. 分页:total = 过滤后候选集大小。fts 已取全量(≤FTS_FETCH_LIMIT)故为真实命中数;
|
||||
# vector/hybrid 为 KNN 候选集,无全局 total。
|
||||
@@ -126,15 +139,18 @@ class RetrievalEngine:
|
||||
)
|
||||
|
||||
def _search_fts(self, request: SearchRequest) -> SearchResponse:
|
||||
"""FTS 专用路径:过滤、COUNT 与分页全部在 SQLite 中完成。"""
|
||||
"""FTS 专用路径:先取全量命中(≤FTS_FETCH_LIMIT),统一归一化 + 阈值过滤后再分页。
|
||||
|
||||
阈值过滤必须在计数与分页之前完成,否则 score_threshold 只作用于当前页,
|
||||
且返回的 total 与 items 数量不一致(如 items 为空但 total 非零)。"""
|
||||
match = match_query(request.query)
|
||||
if not match:
|
||||
return self._empty(request)
|
||||
|
||||
fts_hits, total = repository.fts_search_page(
|
||||
fts_hits, _ = repository.fts_search_page(
|
||||
match=match,
|
||||
limit=request.limit,
|
||||
offset=request.offset,
|
||||
limit=FTS_FETCH_LIMIT,
|
||||
offset=0,
|
||||
folders=request.folders,
|
||||
note_ids=request.note_ids,
|
||||
tags=request.tags,
|
||||
@@ -144,17 +160,18 @@ class RetrievalEngine:
|
||||
updated_to=request.updated_to,
|
||||
)
|
||||
if not fts_hits:
|
||||
return SearchResponse(
|
||||
query=request.query,
|
||||
mode=request.mode,
|
||||
page=PageMeta(total=total, limit=request.limit, offset=request.offset),
|
||||
)
|
||||
return self._empty(request)
|
||||
|
||||
hits = {h.block_id: h for h in repository.get_block_hits([hit.block_id for hit in fts_hits])}
|
||||
ordered = normalize_scores(
|
||||
[(hit.block_id, -hit.bm25) for hit in fts_hits if hit.block_id in hits]
|
||||
)
|
||||
items = [self._build_result(hits[block_id], request, score) for block_id, score in ordered]
|
||||
ordered = normalize_scores([(hit.block_id, -hit.bm25) for hit in fts_hits])
|
||||
ordered = [(bid, score) for bid, score in ordered if score >= request.score_threshold]
|
||||
total = len(ordered)
|
||||
page = ordered[request.offset : request.offset + request.limit]
|
||||
hits = {h.block_id: h for h in repository.get_block_hits([bid for bid, _ in page])}
|
||||
items = [
|
||||
self._build_result(hits[block_id], request, score)
|
||||
for block_id, score in page
|
||||
if block_id in hits
|
||||
]
|
||||
return SearchResponse(
|
||||
query=request.query,
|
||||
mode=request.mode,
|
||||
|
||||
@@ -24,6 +24,7 @@ class RerankerProvider(Protocol):
|
||||
"""统一 Reranker 接口:输入候选块,输出按相关性重排后的候选块。"""
|
||||
|
||||
model_id: str
|
||||
version: str
|
||||
|
||||
async def rerank(self, query: str, candidates: list[RankedCandidate]) -> list[RankedCandidate]: ...
|
||||
|
||||
@@ -32,6 +33,7 @@ class LexicalReranker:
|
||||
"""轻量精排:query 与块正文的词面重叠度,与归一化后的原始分数加权求和。"""
|
||||
|
||||
model_id = "lexical-v1"
|
||||
version = "1"
|
||||
|
||||
def __init__(self, lexical_weight: float = 0.5) -> None:
|
||||
self.lexical_weight = lexical_weight
|
||||
|
||||
@@ -35,6 +35,7 @@ class VectorStore(Protocol):
|
||||
async def upsert(self, records: list[VectorRecord]) -> None: ...
|
||||
async def delete(self, ids: list[str]) -> None: ...
|
||||
async def search(self, vector: list[float], *, top_k: int) -> list[VectorHit]: ...
|
||||
async def count(self) -> int: ...
|
||||
|
||||
|
||||
class SqliteVecStore:
|
||||
@@ -92,3 +93,10 @@ class SqliteVecStore:
|
||||
conn.execute("DELETE FROM vec_blocks")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
async def count(self) -> int:
|
||||
conn = connect()
|
||||
try:
|
||||
return conn.execute("SELECT COUNT(*) FROM vec_blocks").fetchone()[0]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
+210
-4
@@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
from collections.abc import AsyncIterator
|
||||
from datetime import datetime, timezone
|
||||
from uuid import uuid4
|
||||
@@ -11,6 +12,14 @@ from app.contracts import (
|
||||
AgentRunListResponse,
|
||||
AgentTraceResponse,
|
||||
ChatRequest,
|
||||
BenchmarkDatasetListResponse,
|
||||
BenchmarkEventType,
|
||||
BenchmarkKind,
|
||||
BenchmarkReport,
|
||||
BenchmarkRun,
|
||||
BenchmarkRunListResponse,
|
||||
BenchmarkStatus,
|
||||
RAGRunRequest,
|
||||
CredentialStatus,
|
||||
CredentialWriteRequest,
|
||||
ExtensionInstallRequest,
|
||||
@@ -32,6 +41,7 @@ from app.contracts import (
|
||||
PageMeta,
|
||||
PermissionDecisionRequest,
|
||||
Plugin,
|
||||
PluginHostStatus,
|
||||
PluginListResponse,
|
||||
PluginPermissionGrantRequest,
|
||||
ProviderConfig,
|
||||
@@ -59,6 +69,8 @@ from app.contracts import (
|
||||
WorkspaceSnapshot,
|
||||
)
|
||||
from app.agent import AgentCapacityError, AgentRunNotFoundError
|
||||
from app.benchmarks import datasets as benchmark_datasets
|
||||
from app.benchmarks import service as benchmark_service
|
||||
from app.container import container
|
||||
from app.errors import ApiError
|
||||
from app.extensions import ExtensionError
|
||||
@@ -130,6 +142,15 @@ def extension_call(operation):
|
||||
raise ApiError(exc.status_code, exc.code, exc.message, exc.details) from exc
|
||||
|
||||
|
||||
async def extension_call_async(operation):
|
||||
"""进程启动/关闭可能等待 stdio Host,移出 FastAPI 事件循环。"""
|
||||
|
||||
try:
|
||||
return await asyncio.to_thread(operation)
|
||||
except ExtensionError as exc:
|
||||
raise ApiError(exc.status_code, exc.code, exc.message, exc.details) from exc
|
||||
|
||||
|
||||
# Workspace (single configured Vault in Web development mode)
|
||||
@router.get("/workspace", response_model=WorkspaceInfo, tags=["Workspace"])
|
||||
async def get_workspace() -> WorkspaceInfo:
|
||||
@@ -483,7 +504,7 @@ async def install_plugin(request: ExtensionInstallRequest) -> Plugin:
|
||||
tags=["Plugins"],
|
||||
)
|
||||
async def enable_plugin(plugin_id: str) -> Plugin:
|
||||
return extension_call(lambda: container.plugins.enable(plugin_id))
|
||||
return await extension_call_async(lambda: container.plugins.enable(plugin_id))
|
||||
|
||||
|
||||
@router.post(
|
||||
@@ -492,7 +513,7 @@ async def enable_plugin(plugin_id: str) -> Plugin:
|
||||
tags=["Plugins"],
|
||||
)
|
||||
async def disable_plugin(plugin_id: str) -> Plugin:
|
||||
return extension_call(lambda: container.plugins.disable(plugin_id))
|
||||
return await extension_call_async(lambda: container.plugins.disable(plugin_id))
|
||||
|
||||
|
||||
@router.put(
|
||||
@@ -503,11 +524,37 @@ async def disable_plugin(plugin_id: str) -> Plugin:
|
||||
async def set_plugin_permissions(
|
||||
plugin_id: str, request: PluginPermissionGrantRequest
|
||||
) -> Plugin:
|
||||
return extension_call(
|
||||
return await extension_call_async(
|
||||
lambda: container.plugins.set_permissions(plugin_id, request.permissions)
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/plugins/{plugin_id}/host",
|
||||
response_model=PluginHostStatus,
|
||||
tags=["Plugins"],
|
||||
)
|
||||
async def get_plugin_host_status(plugin_id: str) -> PluginHostStatus:
|
||||
return extension_call(lambda: container.plugins.get_host_status(plugin_id))
|
||||
|
||||
|
||||
@router.post(
|
||||
"/plugins/{plugin_id}/host/restart",
|
||||
response_model=OperationResponse,
|
||||
status_code=202,
|
||||
tags=["Plugins"],
|
||||
)
|
||||
async def restart_plugin_host(plugin_id: str) -> OperationResponse:
|
||||
status = await extension_call_async(
|
||||
lambda: container.plugins.restart_host(plugin_id)
|
||||
)
|
||||
return OperationResponse(
|
||||
status="accepted",
|
||||
resource_id=plugin_id,
|
||||
message=f"Plugin Host status: {status.status.value}",
|
||||
)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/plugins/{plugin_id}",
|
||||
response_model=OperationResponse,
|
||||
@@ -516,7 +563,9 @@ async def set_plugin_permissions(
|
||||
async def uninstall_plugin(plugin_id: str) -> OperationResponse:
|
||||
plugin = extension_call(lambda: container.plugins.get(plugin_id))
|
||||
dependent_skills = container.skills.depending_on_tools(plugin.manifest.contributes.tools)
|
||||
extension_call(lambda: container.plugins.uninstall(plugin_id, dependent_skills))
|
||||
await extension_call_async(
|
||||
lambda: container.plugins.uninstall(plugin_id, dependent_skills)
|
||||
)
|
||||
return OperationResponse(status="completed", resource_id=plugin_id, message="uninstalled")
|
||||
|
||||
|
||||
@@ -795,3 +844,160 @@ async def get_index_job(job_id: str) -> IndexJob:
|
||||
if job is None:
|
||||
raise ApiError(404, "RESOURCE_NOT_FOUND", "index job not found", {"job_id": job_id})
|
||||
return job
|
||||
|
||||
|
||||
# Benchmark
|
||||
@router.get(
|
||||
"/benchmarks/datasets",
|
||||
response_model=BenchmarkDatasetListResponse,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def list_benchmark_datasets(
|
||||
kind: BenchmarkKind = Query(default=BenchmarkKind.rag),
|
||||
) -> BenchmarkDatasetListResponse:
|
||||
return BenchmarkDatasetListResponse(items=benchmark_datasets.list_datasets(kind))
|
||||
|
||||
|
||||
@router.post(
|
||||
"/benchmarks/rag/runs",
|
||||
response_model=BenchmarkRun,
|
||||
status_code=202,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def create_rag_benchmark(request: RAGRunRequest) -> BenchmarkRun:
|
||||
return await benchmark_service.create_rag_run(request)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/benchmarks/runs",
|
||||
response_model=BenchmarkRunListResponse,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def list_benchmark_runs(
|
||||
kind: BenchmarkKind | None = Query(default=None),
|
||||
status: BenchmarkStatus | None = Query(default=None),
|
||||
limit: int = Query(default=50, ge=1, le=100),
|
||||
offset: int = Query(default=0, ge=0),
|
||||
) -> BenchmarkRunListResponse:
|
||||
items, total = benchmark_service.list_runs(
|
||||
kind=kind, status=status, limit=limit, offset=offset
|
||||
)
|
||||
return BenchmarkRunListResponse(
|
||||
items=items, page=PageMeta(total=total, limit=limit, offset=offset)
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/benchmarks/runs/{run_id}",
|
||||
response_model=BenchmarkRun,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def get_benchmark_run(run_id: str) -> BenchmarkRun:
|
||||
run = benchmark_service.get_run(run_id)
|
||||
if run is None:
|
||||
raise ApiError(
|
||||
404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
|
||||
)
|
||||
return run
|
||||
|
||||
|
||||
@router.post(
|
||||
"/benchmarks/runs/{run_id}/cancel",
|
||||
response_model=OperationResponse,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def cancel_benchmark_run(run_id: str) -> OperationResponse:
|
||||
run = benchmark_service.cancel_run(run_id)
|
||||
if run is None:
|
||||
raise ApiError(
|
||||
404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
|
||||
)
|
||||
return OperationResponse(
|
||||
status="accepted",
|
||||
resource_id=run_id,
|
||||
message=f"Benchmark run status: {run.status.value}",
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/benchmarks/runs/{run_id}/events",
|
||||
response_class=StreamingResponse,
|
||||
responses={
|
||||
200: {
|
||||
"description": "BenchmarkEvent Server-Sent Events stream",
|
||||
"content": {"text/event-stream": {}},
|
||||
}
|
||||
},
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def benchmark_events(
|
||||
run_id: str,
|
||||
after_sequence: int = Query(default=-1, ge=-1),
|
||||
last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
|
||||
) -> StreamingResponse:
|
||||
if benchmark_service.get_run(run_id) is None:
|
||||
raise ApiError(
|
||||
404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
|
||||
)
|
||||
|
||||
# SSE 断线重连:Last-Event-ID 优先于 after_sequence,用于从上次收到的事件继续
|
||||
cursor = after_sequence
|
||||
if last_event_id is not None:
|
||||
try:
|
||||
cursor = int(last_event_id)
|
||||
except ValueError as exc:
|
||||
raise ApiError(
|
||||
400,
|
||||
"BENCHMARK_EVENT_CURSOR_INVALID",
|
||||
"Last-Event-ID must be an integer sequence.",
|
||||
{"last_event_id": last_event_id},
|
||||
) from exc
|
||||
if cursor < -1:
|
||||
raise ApiError(
|
||||
400,
|
||||
"BENCHMARK_EVENT_CURSOR_INVALID",
|
||||
"Last-Event-ID must be greater than or equal to -1.",
|
||||
)
|
||||
|
||||
async def stream() -> AsyncIterator[str]:
|
||||
# 先订阅(保证订阅之后产生的事件也能收到),再回放历史事件,最后实时输出新事件
|
||||
queue = benchmark_service.subscribe(run_id)
|
||||
last_sequence = cursor
|
||||
for event in benchmark_service.get_events(run_id):
|
||||
if event.sequence <= cursor:
|
||||
continue
|
||||
yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence)
|
||||
last_sequence = event.sequence
|
||||
if queue is None:
|
||||
return
|
||||
try:
|
||||
while True:
|
||||
event = await queue.get()
|
||||
if event.sequence <= last_sequence:
|
||||
continue
|
||||
yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence)
|
||||
last_sequence = event.sequence
|
||||
if event.event in (
|
||||
BenchmarkEventType.run_completed,
|
||||
BenchmarkEventType.run_failed,
|
||||
BenchmarkEventType.run_cancelled,
|
||||
):
|
||||
break
|
||||
finally:
|
||||
benchmark_service.unsubscribe(run_id, queue)
|
||||
|
||||
return StreamingResponse(stream(), media_type="text/event-stream")
|
||||
|
||||
|
||||
@router.get(
|
||||
"/benchmarks/runs/{run_id}/report",
|
||||
response_model=BenchmarkReport,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def get_benchmark_report(run_id: str) -> BenchmarkReport:
|
||||
report = benchmark_service.get_report(run_id)
|
||||
if report is None:
|
||||
raise ApiError(
|
||||
404, "BENCHMARK_RUN_NOT_FOUND", "benchmark report not found", {"run_id": run_id}
|
||||
)
|
||||
return report
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
{
|
||||
"dataset_id": "rag-core-v1",
|
||||
"kind": "rag",
|
||||
"version": "1.0.0",
|
||||
"description": "基础中文笔记检索集(对应 backend/data/vault 内置语料,重建索引后即可复现)",
|
||||
"cases": [
|
||||
{
|
||||
"case_id": "rag-vector-sim",
|
||||
"query": "向量数据库如何进行相似度检索",
|
||||
"expected_note_ids": ["note_c1454740a0e55ef5"],
|
||||
"expected_block_ids": ["blk_07c4c6bce0ec4d12", "blk_605fb3593809f224"],
|
||||
"citation_required": true,
|
||||
"tags": ["向量数据库", "检索"]
|
||||
},
|
||||
{
|
||||
"case_id": "rag-python-func",
|
||||
"query": "Python 如何定义函数",
|
||||
"expected_note_ids": ["note_424c3742c6f0e555"],
|
||||
"expected_block_ids": ["blk_45d48cae2fed40fe", "blk_0768d9c25c2ecf07"],
|
||||
"citation_required": true,
|
||||
"tags": ["python"]
|
||||
},
|
||||
{
|
||||
"case_id": "rag-citation",
|
||||
"query": "搜索结果如何定位到原文位置",
|
||||
"expected_note_ids": ["note_0c619caa30b1614c"],
|
||||
"expected_block_ids": ["blk_3f6fcead71c25fc6", "blk_9af7b12e9ce909fc"],
|
||||
"citation_required": true,
|
||||
"tags": ["RAG"]
|
||||
},
|
||||
{
|
||||
"case_id": "rag-hybrid",
|
||||
"query": "混合检索怎么融合全文和向量",
|
||||
"expected_note_ids": ["note_c1454740a0e55ef5"],
|
||||
"expected_block_ids": ["blk_82b45418dba9f720"],
|
||||
"citation_required": true,
|
||||
"tags": ["检索"]
|
||||
},
|
||||
{
|
||||
"case_id": "rag-tech-stack",
|
||||
"query": "这个项目用什么后端和检索技术",
|
||||
"expected_note_ids": ["note_3327e6cf18f3701f"],
|
||||
"expected_block_ids": ["blk_feb2a9c42e7d31ad"],
|
||||
"citation_required": false,
|
||||
"tags": ["项目"]
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
id: mcp-fixture
|
||||
name: MCP Fixture
|
||||
version: 1.0.0
|
||||
description: 阶段 C 离线联调 Fixture,覆盖 MCP Tool 生命周期与错误边界。
|
||||
permissions:
|
||||
- notes.read
|
||||
contributes:
|
||||
tools:
|
||||
- mcp-fixture.echo
|
||||
- mcp-fixture.fail
|
||||
- mcp-fixture.sleep
|
||||
- mcp-fixture.large
|
||||
- mcp-fixture.environment
|
||||
- mcp-fixture.exit
|
||||
backend:
|
||||
type: mcp
|
||||
transport: stdio
|
||||
command: python
|
||||
args: [server.py]
|
||||
startup_timeout_seconds: 5
|
||||
tool_timeout_seconds: 1
|
||||
@@ -0,0 +1,204 @@
|
||||
"""确定性的 MCP stdio 测试 Server;仅使用标准库,不依赖产品代码。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
WRITE_LOCK = threading.Lock()
|
||||
CANCELLED: dict[int, threading.Event] = {}
|
||||
MODE = sys.argv[1] if len(sys.argv) > 1 else "normal"
|
||||
|
||||
|
||||
def send(message: dict[str, Any]) -> None:
|
||||
with WRITE_LOCK:
|
||||
sys.stdout.write(json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def respond(request_id: int, result: dict[str, Any]) -> None:
|
||||
send({"jsonrpc": "2.0", "id": request_id, "result": result})
|
||||
|
||||
|
||||
def tool(name: str, description: str, properties: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
return {
|
||||
"name": name,
|
||||
"description": description,
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": properties or {},
|
||||
"required": list(properties or {}),
|
||||
"additionalProperties": False,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
TOOLS = {
|
||||
"echo": {
|
||||
**tool(
|
||||
"echo",
|
||||
"Return the provided text.",
|
||||
{
|
||||
"text": {"type": "string"},
|
||||
"suffix": {"type": ["string", "null"]},
|
||||
},
|
||||
),
|
||||
"_meta": {"notesagent/permission": "notes.read"},
|
||||
},
|
||||
"fail": tool("fail", "Return an MCP business error."),
|
||||
"sleep": tool("sleep", "Wait until completed or cancelled.", {"seconds": {"type": "number"}}),
|
||||
"large": tool("large", "Return a result larger than the host limit."),
|
||||
"environment": tool("environment", "Report whether host secrets leaked into the process."),
|
||||
"exit": tool("exit", "Terminate the fixture process."),
|
||||
}
|
||||
# suffix 是可选字段,用于验证 Host 不会把缺省值擅自补成 null。
|
||||
TOOLS["echo"]["inputSchema"]["required"] = ["text"]
|
||||
|
||||
|
||||
def call_tool(request_id: int, params: dict[str, Any]) -> None:
|
||||
name = params.get("name")
|
||||
arguments = params.get("arguments") or {}
|
||||
if name == "echo":
|
||||
text = str(arguments.get("text", ""))
|
||||
structured_content = {"echo": text}
|
||||
if "suffix" in arguments:
|
||||
structured_content["suffix"] = arguments["suffix"]
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": text}],
|
||||
"structuredContent": structured_content,
|
||||
"isError": False,
|
||||
},
|
||||
)
|
||||
return
|
||||
if name == "fail":
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": "fixture failure"}],
|
||||
"isError": True,
|
||||
},
|
||||
)
|
||||
return
|
||||
if name == "large":
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": "x" * 300_000}],
|
||||
"isError": False,
|
||||
},
|
||||
)
|
||||
return
|
||||
if name == "environment":
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": "environment checked"}],
|
||||
"structuredContent": {
|
||||
"has_openai_key": "OPENAI_API_KEY" in os.environ,
|
||||
"has_app_db_path": "APP_DB_PATH" in os.environ,
|
||||
},
|
||||
"isError": False,
|
||||
},
|
||||
)
|
||||
return
|
||||
if name == "exit":
|
||||
os._exit(17)
|
||||
if name == "sleep":
|
||||
cancelled = CANCELLED.setdefault(request_id, threading.Event())
|
||||
seconds = max(0.0, min(float(arguments.get("seconds", 0)), 30.0))
|
||||
if cancelled.wait(seconds):
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": "cancelled"}],
|
||||
"isError": True,
|
||||
},
|
||||
)
|
||||
else:
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"content": [{"type": "text", "text": "completed"}],
|
||||
"structuredContent": {"slept": seconds},
|
||||
"isError": False,
|
||||
},
|
||||
)
|
||||
CANCELLED.pop(request_id, None)
|
||||
return
|
||||
send(
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": request_id,
|
||||
"error": {"code": -32602, "message": f"Unknown tool: {name}"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
for line in sys.stdin:
|
||||
message = json.loads(line)
|
||||
method = message.get("method")
|
||||
request_id = message.get("id")
|
||||
params = message.get("params") or {}
|
||||
if method == "initialize" and isinstance(request_id, int):
|
||||
if MODE == "invalid-result":
|
||||
send({"jsonrpc": "2.0", "id": request_id, "result": None})
|
||||
continue
|
||||
if MODE == "oversized-stdout":
|
||||
# 不带换行,验证 Host 在读取完整内容前执行硬上限。
|
||||
sys.stdout.write("x" * (2 * 1024 * 1024 + 1))
|
||||
sys.stdout.flush()
|
||||
time.sleep(10)
|
||||
return
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"protocolVersion": params.get("protocolVersion"),
|
||||
"capabilities": (
|
||||
{} if MODE == "no-tools" else {"tools": {"listChanged": False}}
|
||||
),
|
||||
"serverInfo": {"name": "notesagent-mcp-fixture", "version": "1.0.0"},
|
||||
},
|
||||
)
|
||||
elif method == "tools/list" and isinstance(request_id, int):
|
||||
if MODE == "invalid-schema":
|
||||
respond(
|
||||
request_id,
|
||||
{
|
||||
"tools": [
|
||||
{
|
||||
"name": "broken",
|
||||
"description": "invalid schema",
|
||||
"inputSchema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
elif params.get("cursor") == "page-2":
|
||||
respond(
|
||||
request_id,
|
||||
{"tools": [TOOLS["large"], TOOLS["environment"], TOOLS["exit"]]},
|
||||
)
|
||||
else:
|
||||
respond(
|
||||
request_id,
|
||||
{"tools": [TOOLS["echo"], TOOLS["fail"], TOOLS["sleep"]], "nextCursor": "page-2"},
|
||||
)
|
||||
elif method == "tools/call" and isinstance(request_id, int):
|
||||
threading.Thread(target=call_tool, args=(request_id, params), daemon=True).start()
|
||||
elif method == "notifications/cancelled":
|
||||
cancelled_id = params.get("requestId")
|
||||
if isinstance(cancelled_id, int):
|
||||
CANCELLED.setdefault(cancelled_id, threading.Event()).set()
|
||||
elif method == "ping" and isinstance(request_id, int):
|
||||
respond(request_id, {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -93,6 +93,8 @@ def test_openapi_contains_documented_frontend_interfaces() -> None:
|
||||
"/api/skills",
|
||||
"/api/plugins",
|
||||
"/api/plugins/install",
|
||||
"/api/plugins/{plugin_id}/host",
|
||||
"/api/plugins/{plugin_id}/host/restart",
|
||||
"/api/plugins/{plugin_id}/enable",
|
||||
"/api/plugins/{plugin_id}/disable",
|
||||
"/api/providers/test",
|
||||
|
||||
@@ -0,0 +1,441 @@
|
||||
"""Benchmark 服务的单元与端到端测试。
|
||||
|
||||
沿用 conftest 的隔离机制:APP_DATA_DIR / DB / Vault 都指向临时目录,benchmark
|
||||
数据集也落在临时目录(settings.benchmark_datasets_path),不读写真实数据。
|
||||
|
||||
运行采用「创建即 queued + 后台 Task 执行」的异步模型,测试通过 _run 在同一事件循环内
|
||||
创建并等待后台任务结束,得到终态 BenchmarkRun 后再断言。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.benchmarks import datasets, metrics as m, service
|
||||
from app.config import get_settings
|
||||
from app.contracts import (
|
||||
BenchmarkKind,
|
||||
BenchmarkRun,
|
||||
BenchmarkStatus,
|
||||
RAGRunRequest,
|
||||
SearchMode,
|
||||
)
|
||||
from app.errors import ApiError
|
||||
|
||||
|
||||
def _write_dataset(dataset_id: str, cases: list[dict], *, kind: str = "rag") -> None:
|
||||
directory = get_settings().benchmark_datasets_path
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
payload = {
|
||||
"dataset_id": dataset_id,
|
||||
"kind": kind,
|
||||
"version": "1.0.0",
|
||||
"description": "test dataset",
|
||||
"cases": cases,
|
||||
}
|
||||
(directory / f"{dataset_id}.json").write_text(
|
||||
json.dumps(payload, ensure_ascii=False), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def _write_raw(dataset_id: str, raw: dict) -> None:
|
||||
directory = get_settings().benchmark_datasets_path
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
(directory / f"{dataset_id}.json").write_text(
|
||||
json.dumps(raw, ensure_ascii=False), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def _run(request: RAGRunRequest):
|
||||
"""创建运行并在同一事件循环内等待后台任务结束,返回终态 BenchmarkRun。"""
|
||||
from app.contracts import BenchmarkRun
|
||||
|
||||
async def _execute() -> BenchmarkRun:
|
||||
run = await service.create_rag_run(request)
|
||||
return await service.wait_for_run(run.run_id)
|
||||
|
||||
return asyncio.run(_execute())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 指标纯函数
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_hit_at_k_and_recall() -> None:
|
||||
retrieved = ["a", "b", "c"]
|
||||
expected = {"b", "z"}
|
||||
|
||||
assert m.hit_at_k(retrieved, expected, 1) is False
|
||||
assert m.hit_at_k(retrieved, expected, 2) is True
|
||||
assert m.recall_at_k(retrieved, expected, 5) == 0.5 # 只召回 b
|
||||
|
||||
|
||||
def test_recall_at_k_dedups_duplicate_notes() -> None:
|
||||
# 同一 Note 经多个 Block 重复出现,去重后 Recall 不应超过 1
|
||||
assert m.recall_at_k(["note-a", "note-a"], {"note-a"}, 2) == 1.0
|
||||
assert m.recall_at_k(["note-a", "note-a", "note-b"], {"note-a"}, 3) == 1.0
|
||||
|
||||
|
||||
def test_reciprocal_rank_and_citation_hit() -> None:
|
||||
assert m.reciprocal_rank(["x", "a", "b"], {"b"}) == 1 / 3
|
||||
assert m.reciprocal_rank(["x"], {"b"}) == 0.0
|
||||
assert m.citation_hit(["blk_1"], {"blk_1"}) is True
|
||||
assert m.citation_hit(["blk_2"], {"blk_1"}) is False
|
||||
assert m.citation_hit([], {"blk_1"}) is False
|
||||
|
||||
|
||||
def test_percentile() -> None:
|
||||
assert m.percentile([1.0, 2.0, 3.0, 4.0], 50.0) == 2.5
|
||||
assert m.percentile([], 50.0) == 0.0
|
||||
assert m.percentile([7.0], 95.0) == 7.0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Dataset 注册与校验
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_list_datasets_empty_by_default() -> None:
|
||||
assert datasets.list_datasets(BenchmarkKind.rag) == []
|
||||
|
||||
|
||||
def test_load_missing_dataset_raises() -> None:
|
||||
with pytest.raises(ApiError) as exc:
|
||||
datasets.load_dataset("does-not-exist", BenchmarkKind.rag)
|
||||
assert exc.value.status_code == 404
|
||||
assert exc.value.code == "BENCHMARK_DATASET_NOT_FOUND"
|
||||
|
||||
|
||||
def test_dataset_without_expected_ids_is_invalid() -> None:
|
||||
_write_dataset("bad-v1", [{"case_id": "x", "query": "q", "citation_required": False}])
|
||||
with pytest.raises(ApiError) as exc:
|
||||
datasets.load_dataset("bad-v1", BenchmarkKind.rag)
|
||||
assert exc.value.code == "BENCHMARK_DATASET_INVALID"
|
||||
|
||||
|
||||
def test_dataset_kind_mismatch_is_invalid() -> None:
|
||||
_write_dataset("agent-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}], kind="agent")
|
||||
with pytest.raises(ApiError) as exc:
|
||||
datasets.load_dataset("agent-v1", BenchmarkKind.rag)
|
||||
assert exc.value.code == "BENCHMARK_DATASET_INVALID"
|
||||
|
||||
|
||||
def test_citation_required_requires_expected_block_ids() -> None:
|
||||
# citation_required=true 却没有 expected_block_ids,无法计算 Citation Hit Rate,应拒绝
|
||||
_write_dataset(
|
||||
"cit-req-v1",
|
||||
[{"case_id": "x", "query": "q", "expected_note_ids": ["n"], "citation_required": True}],
|
||||
)
|
||||
with pytest.raises(ApiError) as exc:
|
||||
datasets.load_dataset("cit-req-v1", BenchmarkKind.rag)
|
||||
assert exc.value.code == "BENCHMARK_DATASET_INVALID"
|
||||
|
||||
|
||||
def test_list_datasets_skips_corrupted_structure() -> None:
|
||||
# 合法 JSON 但字段结构错误(cases: 42),列表接口应隔离该文件而非整体 500
|
||||
_write_raw("bad-structure", {"dataset_id": "bad-structure", "kind": "rag", "cases": 42})
|
||||
_write_dataset("good-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
|
||||
|
||||
infos = datasets.list_datasets(BenchmarkKind.rag)
|
||||
ids = {info.dataset_id for info in infos}
|
||||
assert "good-v1" in ids
|
||||
assert "bad-structure" not in ids
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 请求校验(空 / 重复 modes)
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_empty_modes_rejected() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
RAGRunRequest(dataset_id="x", modes=[])
|
||||
|
||||
|
||||
def test_duplicate_modes_rejected() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
RAGRunRequest(dataset_id="x", modes=[SearchMode.fts, SearchMode.fts])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# RAG Benchmark 端到端
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _single_note_case() -> tuple[str, str, dict]:
|
||||
from app.services import note_service
|
||||
|
||||
note = asyncio.run(
|
||||
note_service.create_note(
|
||||
title="向量库",
|
||||
markdown="向量数据库用于存储高维向量并支持近似最近邻检索。",
|
||||
folder="",
|
||||
tags=["向量"],
|
||||
)
|
||||
)
|
||||
case = {
|
||||
"case_id": "c1",
|
||||
"query": "向量数据库相似度检索",
|
||||
"expected_note_ids": [note.note_id],
|
||||
"expected_block_ids": [note.blocks[0].block_id],
|
||||
"citation_required": True,
|
||||
"tags": ["向量"],
|
||||
}
|
||||
return note.note_id, note.blocks[0].block_id, case
|
||||
|
||||
|
||||
def test_rag_benchmark_end_to_end() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("e2e-v1", [case])
|
||||
|
||||
run = _run(RAGRunRequest(dataset_id="e2e-v1", modes=[SearchMode.fts]))
|
||||
|
||||
assert run.status.value == "completed"
|
||||
assert run.dataset_hash.startswith("sha256:")
|
||||
assert run.metrics is not None
|
||||
|
||||
fts = run.metrics["fts"]
|
||||
assert fts["hit_at_1"] == 1.0
|
||||
assert fts["recall_at_k"] == 1.0
|
||||
assert fts["mrr"] == 1.0
|
||||
assert fts["citation_hit_rate"] == 1.0
|
||||
assert fts["p50_latency_ms"] >= 0.0
|
||||
assert fts["p95_latency_ms"] >= fts["p50_latency_ms"]
|
||||
|
||||
|
||||
def test_rag_benchmark_all_modes_produce_metrics() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("e2e-modes-v1", [case])
|
||||
|
||||
run = _run(RAGRunRequest(dataset_id="e2e-modes-v1"))
|
||||
assert run.status.value == "completed"
|
||||
|
||||
for mode in ("fts", "vector", "hybrid"):
|
||||
assert mode in run.metrics
|
||||
for key in ("hit_at_1", "hit_at_5", "recall_at_k", "mrr", "citation_hit_rate"):
|
||||
assert 0.0 <= run.metrics[mode][key] <= 1.0
|
||||
|
||||
|
||||
def test_config_snapshot_records_index_and_models() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("snapshot-v1", [case])
|
||||
|
||||
run = _run(RAGRunRequest(dataset_id="snapshot-v1", modes=[SearchMode.fts]))
|
||||
|
||||
snapshot = run.config_snapshot
|
||||
assert snapshot["index_meta"] is not None
|
||||
assert snapshot["embedding"]["version"]
|
||||
assert snapshot["embedding"]["dim"]
|
||||
assert snapshot["reranker"]["version"]
|
||||
assert snapshot["retrieval"]["rrf_k"] == 60
|
||||
|
||||
|
||||
def test_benchmark_report_and_events() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("report-v1", [case])
|
||||
|
||||
run = _run(RAGRunRequest(dataset_id="report-v1", modes=[SearchMode.fts]))
|
||||
report = service.get_report(run.run_id)
|
||||
events = service.get_events(run.run_id)
|
||||
|
||||
assert report is not None
|
||||
assert report.run_id == run.run_id
|
||||
assert len(report.cases) == 1
|
||||
assert report.cases[0].case_id == "c1"
|
||||
assert report.cases[0].hit_at_1 is True
|
||||
|
||||
assert events, "运行应产生事件"
|
||||
assert events[0].event.value == "RunStarted"
|
||||
assert events[-1].event.value == "RunCompleted"
|
||||
|
||||
|
||||
def test_cancel_completed_run_keeps_status() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("cancel-v1", [case])
|
||||
|
||||
run = _run(RAGRunRequest(dataset_id="cancel-v1", modes=[SearchMode.fts]))
|
||||
assert run.status.value == "completed"
|
||||
|
||||
cancelled = service.cancel_run(run.run_id)
|
||||
assert cancelled.status.value == "completed" # 已结束,不再变 cancelled
|
||||
|
||||
|
||||
def test_cancel_queued_run_marks_cancelled() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("cancel-queued-v1", [case])
|
||||
|
||||
async def _scenario():
|
||||
run = await service.create_rag_run(
|
||||
RAGRunRequest(dataset_id="cancel-queued-v1", modes=[SearchMode.fts])
|
||||
)
|
||||
service.cancel_run(run.run_id)
|
||||
return await service.wait_for_run(run.run_id)
|
||||
|
||||
run = asyncio.run(_scenario())
|
||||
assert run.status.value == "cancelled"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 指标聚合:Citation Hit Rate 只统计 citation_required 样本
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_citation_hit_rate_only_counts_citation_required() -> None:
|
||||
from app.benchmarks import rag as rag_module
|
||||
from app.contracts import RAGCaseResult
|
||||
|
||||
cases = [
|
||||
RAGCaseResult(
|
||||
case_id="a", mode=SearchMode.fts, repeat=0, latency_ms=1.0,
|
||||
citation_hit=True, citation_applicable=True,
|
||||
),
|
||||
RAGCaseResult(
|
||||
case_id="b", mode=SearchMode.fts, repeat=0, latency_ms=1.0,
|
||||
citation_hit=False, citation_applicable=False,
|
||||
),
|
||||
]
|
||||
metrics = rag_module._aggregate(cases, SearchMode.fts)
|
||||
# 只有 citation_applicable(citation_required=true)的样本计入分母
|
||||
assert metrics.citation_hit_rate == 1.0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 路由接入
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_benchmark_routes_wired() -> None:
|
||||
from app import routes
|
||||
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("route-v1", [case])
|
||||
|
||||
async def _scenario():
|
||||
listed = await routes.list_benchmark_datasets(BenchmarkKind.rag)
|
||||
assert any(item.dataset_id == "route-v1" for item in listed.items)
|
||||
|
||||
run = await routes.create_rag_benchmark(
|
||||
RAGRunRequest(dataset_id="route-v1", modes=[SearchMode.fts])
|
||||
)
|
||||
assert run.status.value == "queued"
|
||||
return await service.wait_for_run(run.run_id)
|
||||
|
||||
run = asyncio.run(_scenario())
|
||||
assert run.status.value == "completed"
|
||||
|
||||
got = asyncio.run(routes.get_benchmark_run(run.run_id))
|
||||
assert got.run_id == run.run_id
|
||||
|
||||
report = asyncio.run(routes.get_benchmark_report(run.run_id))
|
||||
assert report.cases[0].case_id == "c1"
|
||||
|
||||
|
||||
def test_benchmark_run_not_found_raises() -> None:
|
||||
from app import routes
|
||||
|
||||
with pytest.raises(ApiError) as exc:
|
||||
asyncio.run(routes.get_benchmark_run("benchmark_missing"))
|
||||
assert exc.value.code == "BENCHMARK_RUN_NOT_FOUND"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 审阅回归:索引兼容 / 容量 / 失败样本 / 取消事件 / 数据集隔离
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_create_rag_run_requires_built_index() -> None:
|
||||
# 空索引(无已索引 block)会让所有模式得到全 0 指标,应在创建时拒绝而非跑出误导结果
|
||||
_write_dataset("empty-index-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
|
||||
with pytest.raises(ApiError) as exc:
|
||||
asyncio.run(
|
||||
service.create_rag_run(
|
||||
RAGRunRequest(dataset_id="empty-index-v1", modes=[SearchMode.fts])
|
||||
)
|
||||
)
|
||||
assert exc.value.status_code == 409
|
||||
assert exc.value.code == "BENCHMARK_INDEX_INCOMPATIBLE"
|
||||
|
||||
|
||||
def test_capacity_exceeded_when_all_runs_active(monkeypatch) -> None:
|
||||
# 满容量且全为活动(非终态)run 时,无法淘汰,应拒绝创建而非删掉正在运行的 run
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("capacity-v1", [case])
|
||||
|
||||
monkeypatch.setattr(service, "MAX_RUNS", 1)
|
||||
fake_id = "benchmark_fake_active"
|
||||
service._runs[fake_id] = BenchmarkRun(
|
||||
run_id=fake_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
dataset_id="capacity-v1",
|
||||
dataset_hash="sha256:fake",
|
||||
status=BenchmarkStatus.queued,
|
||||
created_at=service._now(),
|
||||
)
|
||||
try:
|
||||
with pytest.raises(ApiError) as exc:
|
||||
asyncio.run(
|
||||
service.create_rag_run(
|
||||
RAGRunRequest(dataset_id="capacity-v1", modes=[SearchMode.fts])
|
||||
)
|
||||
)
|
||||
assert exc.value.status_code == 429
|
||||
assert exc.value.code == "BENCHMARK_CAPACITY_EXCEEDED"
|
||||
finally:
|
||||
service._runs.pop(fake_id, None)
|
||||
|
||||
|
||||
def test_failed_samples_counted_as_zero_in_aggregate() -> None:
|
||||
from app.benchmarks import rag as rag_module
|
||||
from app.contracts import RAGCaseResult
|
||||
|
||||
cases = [
|
||||
RAGCaseResult(
|
||||
case_id="ok", mode=SearchMode.fts, repeat=0, latency_ms=10.0,
|
||||
hit_at_1=True, recall=1.0, reciprocal_rank=1.0,
|
||||
citation_hit=True, citation_applicable=True,
|
||||
),
|
||||
RAGCaseResult(
|
||||
case_id="boom", mode=SearchMode.fts, repeat=0, latency_ms=0.0,
|
||||
error="RAG case evaluation failed.",
|
||||
error_code="BENCHMARK_CASE_EVALUATION_FAILED",
|
||||
),
|
||||
]
|
||||
metrics = rag_module._aggregate(cases, SearchMode.fts)
|
||||
|
||||
assert metrics.total_cases == 2
|
||||
assert metrics.successful_cases == 1
|
||||
assert metrics.failed_cases == 1
|
||||
assert metrics.failure_rate == 0.5
|
||||
# 失败样本按零分计入质量指标分母,汇总不虚高
|
||||
assert metrics.hit_at_1 == 0.5
|
||||
assert metrics.recall_at_k == 0.5
|
||||
# 延迟只统计成功样本
|
||||
assert metrics.p50_latency_ms == 10.0
|
||||
|
||||
|
||||
def test_cancel_emits_run_cancelled_event() -> None:
|
||||
_, _, case = _single_note_case()
|
||||
_write_dataset("cancel-event-v1", [case])
|
||||
|
||||
async def _scenario():
|
||||
run = await service.create_rag_run(
|
||||
RAGRunRequest(dataset_id="cancel-event-v1", modes=[SearchMode.fts])
|
||||
)
|
||||
service.cancel_run(run.run_id)
|
||||
return await service.wait_for_run(run.run_id)
|
||||
|
||||
run = asyncio.run(_scenario())
|
||||
assert run.status.value == "cancelled"
|
||||
events = service.get_events(run.run_id)
|
||||
assert events[-1].event.value == "RunCancelled"
|
||||
|
||||
|
||||
def test_load_dataset_ignores_corrupted_unrelated_files() -> None:
|
||||
# 无关文件损坏(非法 JSON / 顶层非对象)不应阻断目标数据集加载
|
||||
directory = get_settings().benchmark_datasets_path
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
(directory / "broken.json").write_text("{ not valid json", encoding="utf-8")
|
||||
(directory / "array.json").write_text('["a", "b"]', encoding="utf-8")
|
||||
_write_dataset("ok-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
|
||||
|
||||
dataset = datasets.load_dataset("ok-v1", BenchmarkKind.rag)
|
||||
assert dataset.dataset_id == "ok-v1"
|
||||
assert len(dataset.cases) == 1
|
||||
|
||||
|
||||
def test_load_dataset_top_level_must_be_object() -> None:
|
||||
_write_raw("array-top", ["a", "b"])
|
||||
with pytest.raises(ApiError) as exc:
|
||||
datasets.load_dataset("array-top", BenchmarkKind.rag)
|
||||
assert exc.value.code == "BENCHMARK_DATASET_INVALID"
|
||||
@@ -1,4 +1,7 @@
|
||||
import asyncio
|
||||
import shutil
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -12,14 +15,31 @@ from app.contracts import (
|
||||
ToolCall,
|
||||
)
|
||||
from app.extensions import ExtensionError
|
||||
from app.extensions.mcp import McpStdioClient
|
||||
from app.extensions.runtime import _arguments_model_from_schema
|
||||
from app.services import note_service
|
||||
from app.config import get_settings
|
||||
from app.config import BACKEND_DIR, get_settings
|
||||
|
||||
|
||||
MCP_FIXTURE = BACKEND_DIR / "extensions" / "fixtures" / "mcp-echo"
|
||||
|
||||
|
||||
def run(coroutine):
|
||||
return asyncio.run(coroutine)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mcp_container():
|
||||
container = build_container()
|
||||
installed = container.plugins.install(MCP_FIXTURE)
|
||||
assert installed.status == "permission_required"
|
||||
container.plugins.set_permissions("mcp-fixture", ["notes.read"])
|
||||
try:
|
||||
yield container
|
||||
finally:
|
||||
container.plugins.shutdown()
|
||||
|
||||
|
||||
def test_bundled_plugin_registers_tool_and_skill_is_ready() -> None:
|
||||
async def scenario() -> None:
|
||||
container = build_container()
|
||||
@@ -297,3 +317,300 @@ def test_attachment_and_transcription_tools_use_host_storage() -> None:
|
||||
assert transcription.output["text"] == "会议转写内容"
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
def test_mcp_stdio_host_discovers_namespaced_tools_and_maps_results(
|
||||
mcp_container, monkeypatch
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "must-not-enter-plugin-host")
|
||||
enabled = mcp_container.plugins.enable("mcp-fixture")
|
||||
status = mcp_container.plugins.get_host_status("mcp-fixture")
|
||||
definition = mcp_container.tools.get("mcp-fixture.echo").definition
|
||||
result = await mcp_container.tools.execute(
|
||||
ToolCall(
|
||||
tool_call_id="call_mcp_echo",
|
||||
name="mcp-fixture.echo",
|
||||
arguments={"text": "hello mcp"},
|
||||
),
|
||||
ToolExecutionContext(
|
||||
run_id="run_mcp_fixture", tool_call_id="call_mcp_echo"
|
||||
),
|
||||
)
|
||||
|
||||
assert enabled.status == "ready" and enabled.enabled is True
|
||||
assert status.status == "ready"
|
||||
environment = await mcp_container.tools.execute(
|
||||
ToolCall(
|
||||
tool_call_id="call_mcp_environment",
|
||||
name="mcp-fixture.environment",
|
||||
arguments={},
|
||||
),
|
||||
ToolExecutionContext(run_id="run_mcp_fixture"),
|
||||
)
|
||||
|
||||
assert status.tools_count == 6
|
||||
assert status.protocol_version == "2025-11-25"
|
||||
assert status.server_name == "notesagent-mcp-fixture"
|
||||
assert definition.permission == "notes.read"
|
||||
assert result.success is True
|
||||
assert result.output == {"echo": "hello mcp"}
|
||||
explicit_null = await mcp_container.tools.execute(
|
||||
ToolCall(
|
||||
tool_call_id="call_mcp_explicit_null",
|
||||
name="mcp-fixture.echo",
|
||||
arguments={"text": "null stays explicit", "suffix": None},
|
||||
),
|
||||
ToolExecutionContext(run_id="run_mcp_fixture"),
|
||||
)
|
||||
assert explicit_null.success is True
|
||||
assert explicit_null.output == {
|
||||
"echo": "null stays explicit",
|
||||
"suffix": None,
|
||||
}
|
||||
assert environment.success is True
|
||||
assert environment.output == {
|
||||
"has_openai_key": False,
|
||||
"has_app_db_path": False,
|
||||
}
|
||||
|
||||
disabled = mcp_container.plugins.disable("mcp-fixture")
|
||||
assert disabled.status == "disabled"
|
||||
assert mcp_container.plugins.get_host_status("mcp-fixture").status == "stopped"
|
||||
assert not mcp_container.tools.contains("mcp-fixture.echo")
|
||||
with pytest.raises(ExtensionError) as exc:
|
||||
mcp_container.plugins.restart_host("mcp-fixture")
|
||||
assert exc.value.code == "PLUGIN_HOST_UNAVAILABLE"
|
||||
assert mcp_container.plugins.get("mcp-fixture").status == "disabled"
|
||||
assert not mcp_container.tools.contains("mcp-fixture.echo")
|
||||
|
||||
mcp_container.plugins.uninstall("mcp-fixture")
|
||||
reinstalled = mcp_container.plugins.install(MCP_FIXTURE)
|
||||
fresh_status = mcp_container.plugins.get_host_status("mcp-fixture")
|
||||
assert reinstalled.status == "permission_required"
|
||||
assert fresh_status.status == "stopped"
|
||||
assert fresh_status.started_at is None
|
||||
assert fresh_status.protocol_version is None
|
||||
assert fresh_status.server_name is None
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
def test_agent_calls_mcp_tool_through_registry_and_writes_trace(mcp_container) -> None:
|
||||
async def scenario() -> None:
|
||||
mcp_container.plugins.enable("mcp-fixture")
|
||||
created = await mcp_container.agent.create_run(
|
||||
AgentRunCreateRequest(
|
||||
input='/tool mcp-fixture.echo {"text":"agent mcp"}',
|
||||
provider_id="mock",
|
||||
model="mock-1",
|
||||
allowed_tools=["mcp-fixture.echo"],
|
||||
)
|
||||
)
|
||||
completed = await mcp_container.agent.wait(created.run_id)
|
||||
trace = mcp_container.agent.get_trace(
|
||||
created.run_id, after_sequence=-1, limit=100
|
||||
)
|
||||
|
||||
assert completed.status == AgentRunStatus.completed
|
||||
assert completed.tool_results[0].success is True
|
||||
assert completed.tool_results[0].output == {"echo": "agent mcp"}
|
||||
assert any(
|
||||
item.event == "ToolCall" and item.data.get("name") == "mcp-fixture.echo"
|
||||
for item in trace.items
|
||||
)
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
def test_mcp_business_error_size_limit_and_timeout_are_structured(mcp_container) -> None:
|
||||
async def scenario() -> None:
|
||||
mcp_container.plugins.enable("mcp-fixture")
|
||||
context = ToolExecutionContext(run_id="run_mcp_errors")
|
||||
|
||||
failed = await mcp_container.tools.execute(
|
||||
ToolCall(tool_call_id="call_fail", name="mcp-fixture.fail", arguments={}),
|
||||
context,
|
||||
)
|
||||
oversized = await mcp_container.tools.execute(
|
||||
ToolCall(tool_call_id="call_large", name="mcp-fixture.large", arguments={}),
|
||||
context,
|
||||
)
|
||||
timed_out = await mcp_container.tools.execute(
|
||||
ToolCall(
|
||||
tool_call_id="call_sleep",
|
||||
name="mcp-fixture.sleep",
|
||||
arguments={"seconds": 5},
|
||||
),
|
||||
ToolExecutionContext(
|
||||
run_id="run_mcp_errors", tool_call_id="call_sleep"
|
||||
),
|
||||
)
|
||||
recovered = await mcp_container.tools.execute(
|
||||
ToolCall(
|
||||
tool_call_id="call_after_timeout",
|
||||
name="mcp-fixture.echo",
|
||||
arguments={"text": "still ready"},
|
||||
),
|
||||
context,
|
||||
)
|
||||
|
||||
assert failed.success is False
|
||||
assert failed.error_code == "MCP_TOOL_CALL_FAILED"
|
||||
assert failed.error_message == "fixture failure"
|
||||
assert oversized.success is False
|
||||
assert oversized.error_code == "MCP_TOOL_RESULT_TOO_LARGE"
|
||||
assert timed_out.success is False
|
||||
assert timed_out.error_code == "MCP_TOOL_CALL_FAILED"
|
||||
assert recovered.success is True
|
||||
assert mcp_container.plugins.get_host_status("mcp-fixture").status == "ready"
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
def test_mcp_cancel_releases_blocking_response_thread(
|
||||
mcp_container, monkeypatch
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
mcp_container.plugins.enable("mcp-fixture")
|
||||
released = threading.Event()
|
||||
original_wait = McpStdioClient.wait_response
|
||||
|
||||
def tracked_wait(self, *args, **kwargs):
|
||||
try:
|
||||
return original_wait(self, *args, **kwargs)
|
||||
finally:
|
||||
released.set()
|
||||
|
||||
monkeypatch.setattr(McpStdioClient, "wait_response", tracked_wait)
|
||||
task = asyncio.create_task(
|
||||
mcp_container.plugins.mcp.call_tool(
|
||||
"mcp-fixture",
|
||||
"sleep",
|
||||
{"seconds": 5},
|
||||
request_id="call_cancel_release",
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0.05)
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
deadline = time.monotonic() + 0.5
|
||||
while not released.is_set() and time.monotonic() < deadline:
|
||||
await asyncio.sleep(0.01)
|
||||
assert released.is_set(), "cancelled MCP wait must not occupy a worker until timeout"
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
def test_mcp_argument_model_preserves_json_schema_additional_properties() -> None:
|
||||
arguments_model = _arguments_model_from_schema(
|
||||
"mcp-fixture.dynamic",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"model_dump": {"type": "string"}},
|
||||
"required": ["model_dump"],
|
||||
"additionalProperties": {"type": "string"},
|
||||
},
|
||||
)
|
||||
|
||||
arguments = arguments_model.model_validate(
|
||||
{"model_dump": "method name remains data", "dynamic-key": "value"}
|
||||
)
|
||||
|
||||
assert arguments.model_dump() == {
|
||||
"model_dump": "method name remains data",
|
||||
"dynamic-key": "value",
|
||||
}
|
||||
|
||||
|
||||
def test_production_rejects_unsandboxed_mcp_host(monkeypatch) -> None:
|
||||
monkeypatch.setenv("APP_ENVIRONMENT", "production")
|
||||
get_settings.cache_clear()
|
||||
container = build_container()
|
||||
installed = container.plugins.install(MCP_FIXTURE)
|
||||
assert installed.status == "permission_required"
|
||||
container.plugins.set_permissions("mcp-fixture", ["notes.read"])
|
||||
try:
|
||||
with pytest.raises(ExtensionError) as exc:
|
||||
container.plugins.enable("mcp-fixture")
|
||||
assert exc.value.code == "MCP_TRUST_APPROVAL_REQUIRED"
|
||||
assert container.plugins.get_host_status("mcp-fixture").status == "stopped"
|
||||
assert not container.tools.contains("mcp-fixture.echo")
|
||||
finally:
|
||||
container.plugins.shutdown()
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_mcp_abnormal_exit_unregisters_tools_and_restart_recovers(mcp_container) -> None:
|
||||
async def scenario() -> None:
|
||||
mcp_container.plugins.enable("mcp-fixture")
|
||||
crashed = await mcp_container.tools.execute(
|
||||
ToolCall(tool_call_id="call_exit", name="mcp-fixture.exit", arguments={}),
|
||||
ToolExecutionContext(run_id="run_mcp_exit", tool_call_id="call_exit"),
|
||||
)
|
||||
|
||||
deadline = time.monotonic() + 2
|
||||
while mcp_container.tools.contains("mcp-fixture.echo") and time.monotonic() < deadline:
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
plugin = mcp_container.plugins.get("mcp-fixture")
|
||||
status = mcp_container.plugins.get_host_status("mcp-fixture")
|
||||
assert crashed.success is False
|
||||
assert crashed.error_code == "PLUGIN_HOST_UNAVAILABLE"
|
||||
assert plugin.status == "error" and plugin.enabled is False
|
||||
assert status.status == "unhealthy"
|
||||
assert not mcp_container.tools.contains("mcp-fixture.echo")
|
||||
|
||||
restarted = mcp_container.plugins.restart_host("mcp-fixture")
|
||||
assert restarted.status == "ready"
|
||||
assert restarted.tools_count == 6
|
||||
assert mcp_container.tools.contains("mcp-fixture.echo")
|
||||
|
||||
run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mode", "contributions", "expected_code"),
|
||||
[
|
||||
("no-tools", "[]", "MCP_CAPABILITY_UNSUPPORTED"),
|
||||
("invalid-schema", "[mcp-invalid.broken]", "MCP_TOOL_SCHEMA_INVALID"),
|
||||
("invalid-result", "[]", "MCP_INITIALIZE_FAILED"),
|
||||
("oversized-stdout", "[]", "PLUGIN_HOST_UNAVAILABLE"),
|
||||
],
|
||||
)
|
||||
def test_mcp_rejects_invalid_initialization_and_discovery(
|
||||
tmp_path, mode, contributions, expected_code
|
||||
) -> None:
|
||||
package = tmp_path / f"mcp-{mode}"
|
||||
package.mkdir()
|
||||
shutil.copyfile(MCP_FIXTURE / "server.py", package / "server.py")
|
||||
(package / "plugin.yaml").write_text(
|
||||
f"""
|
||||
id: mcp-invalid
|
||||
name: Invalid MCP Fixture
|
||||
version: 1.0.0
|
||||
contributes:
|
||||
tools: {contributions}
|
||||
backend:
|
||||
type: mcp
|
||||
transport: stdio
|
||||
command: python
|
||||
args: [server.py, {mode}]
|
||||
startup_timeout_seconds: 5
|
||||
tool_timeout_seconds: 1
|
||||
""".strip(),
|
||||
encoding="utf-8",
|
||||
)
|
||||
container = build_container()
|
||||
container.plugins.install(package)
|
||||
try:
|
||||
with pytest.raises(ExtensionError) as exc:
|
||||
container.plugins.enable("mcp-invalid")
|
||||
assert exc.value.code == expected_code
|
||||
assert container.plugins.get("mcp-invalid").status == "error"
|
||||
assert container.plugins.get_host_status("mcp-invalid").status == "error"
|
||||
assert not container.tools.contains("mcp-invalid.broken")
|
||||
finally:
|
||||
container.plugins.shutdown()
|
||||
|
||||
@@ -436,6 +436,44 @@ def test_fts_pagination_is_not_truncated_at_one_thousand(vault) -> None:
|
||||
assert len(response.items) == 10
|
||||
|
||||
|
||||
def test_fts_score_threshold_filters_before_total(vault) -> None:
|
||||
"""score_threshold 先于计数与分页生效:total 反映过滤后数量,与 items 一致。
|
||||
|
||||
高阈值过滤掉全部结果时 total==0 且 items 为空,杜绝「空页但 total>0」的
|
||||
不一致(审阅 P2-7)。
|
||||
"""
|
||||
from app.retrieval.engine import engine
|
||||
from app.services import note_service
|
||||
|
||||
# 10 个 block,含「目标」次数递增,bm25 分数各异,min-max 归一化后分数落在 [0,1]
|
||||
markdown = "\n\n".join(f"{'目标' * i} 分隔内容" for i in range(1, 11))
|
||||
asyncio.run(
|
||||
note_service.create_note(title="阈值过滤", markdown=markdown, folder="", tags=[])
|
||||
)
|
||||
|
||||
all_hits = asyncio.run(
|
||||
engine.search(
|
||||
SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.0)
|
||||
)
|
||||
)
|
||||
filtered = asyncio.run(
|
||||
engine.search(
|
||||
SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.5)
|
||||
)
|
||||
)
|
||||
none = asyncio.run(
|
||||
engine.search(
|
||||
SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=2.0)
|
||||
)
|
||||
)
|
||||
|
||||
assert all_hits.page.total >= 10
|
||||
assert 0 < filtered.page.total < all_hits.page.total # 阈值过滤掉部分而非全部
|
||||
assert filtered.page.total == len(filtered.items)
|
||||
assert none.page.total == 0
|
||||
assert none.items == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 审阅回归:PATCH tags 语义 / 向量-块一致性 / 过滤漏召回 / rebuild 语义与回滚
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -30,7 +30,9 @@
|
||||
|
||||
- [AI Core 与 Agent Core 开发说明](development/AI-Core与Agent-Core开发说明.md)
|
||||
- [Knowledge 与 Retrieval Core 开发说明](development/Knowledge与Retrieval-Core开发说明.md)
|
||||
- [Benchmark 开发说明](development/Benchmark开发说明.md)
|
||||
- [模型提供商与模型发现开发说明](development/模型提供商与模型发现开发说明.md)
|
||||
- [MCP Bridge 与 Plugin Host 开发说明](development/MCP-Bridge与Plugin-Host开发说明.md)
|
||||
- [前端壳子与接口层开发说明](development/前端壳子与接口层开发说明.md)
|
||||
- [前端写作体验优化开发说明](development/前端写作体验优化开发说明.md)
|
||||
- [前端视觉与轻量动效优化开发说明](development/前端视觉与轻量动效优化开发说明.md)
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
> 适用范围:桌面客户端、本地知识库、RAG、Agent、Skill、多模型接入、多模态处理与可选云同步
|
||||
> 目标读者:前端、Rust 桌面端、Python AI Core、算法、测试与后续接手项目的开发成员
|
||||
|
||||
> 实施状态更新:2026-09-01。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault,第二阶段 Agent Trace 持久化、分页快照和可恢复 SSE 已完成。后续继续接入真实音频处理、MCP、Plugin Command/Settings、Provider 协议增强、Benchmark、文档导出、主题包、Trace 可视化、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。
|
||||
> 实施状态更新:2026-09-03。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault;第二阶段 Agent Trace 持久化、分页快照、可恢复 SSE、stdio MCP Bridge 与隔离 Plugin Host 已完成;RAG Benchmark 检索评测(Dataset 加载、异步运行、SSE 进度、指标聚合与报告)已完成,Agent Benchmark 暂缓。后续继续接入真实音频处理、Plugin Command/Settings、Provider 协议增强、文档导出、主题包、Trace 可视化、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。
|
||||
|
||||
---
|
||||
|
||||
@@ -1040,10 +1040,18 @@ Plugin Host 负责:
|
||||
|
||||
内置 Plugin 可以使用相同的 Plugin Interface 注册能力,减少内置功能和社区扩展之间的接口差异。
|
||||
|
||||
Python 包形式的 MCP Server 推荐使用固定版本的 `uvx --isolated --from <package>==<version> <command>` 启动,以隔离依赖并避免污染 AI Core 环境;包内脚本和非 Python Server 仍可使用受控 `command + args`。`uvx` 的虚拟环境不是安全沙箱,不能限制文件、网络、子进程或系统调用。
|
||||
|
||||
面向社区或不可信 Plugin 开放前,Tauri/Rust Host 必须增加平台级沙箱、完整进程树回收、包来源/签名校验,并在首次安装或命令变化时向用户完整展示 executable 和参数、要求明确同意。当前 Python Host 的独立进程、环境裁剪和 Permission 只用于可信开发联调,不能替代这些生产安全门槛。
|
||||
|
||||
在该门槛完成前,后端仅允许 `APP_ENVIRONMENT=development` 启动未沙箱化 MCP Host;生产环境统一返回 `MCP_TRUST_APPROVAL_REQUIRED`。Python `uvx` Server 在开发模式首次运行可能联网解析依赖,生产版本必须在安装/更新阶段预取并验证固定版本,正常运行阶段只使用已经准备好的环境。
|
||||
|
||||
### 12.5 MCP Bridge
|
||||
|
||||
MCP Bridge 用于接入具有 MCP Server 接口的插件或外部工具服务。
|
||||
|
||||
当前已实现本地 stdio 首版:Plugin Runtime 在授权后的启用阶段启动独立 Server 进程,完成 `initialize`、capability negotiation、分页 `tools/list`、`tools/call`、取消、超时、异常退出和 Host Restart。实现接受 `2025-11-25`、`2025-06-18`、`2025-03-26` 与 `2024-11-05` 协议版本;Streamable HTTP、Resource、Prompt、Sampling 与操作系统级沙箱仍属于后续范围。
|
||||
|
||||
MCP Tool 进入系统后的调用路径为:
|
||||
|
||||
```text
|
||||
@@ -1062,7 +1070,7 @@ Tool Registry 仍使用项目自己的 `ToolDefinition` 和 `ToolResult`。MCP B
|
||||
|
||||
MCP 能力首先用于 Tool 和 Resource 类扩展。需要复杂 UI 的插件通过 Frontend Extension Slot 单独处理。
|
||||
|
||||
第二阶段 MCP Bridge 至少覆盖以下协议边界:
|
||||
当前 stdio MCP Bridge 已覆盖以下协议边界:
|
||||
|
||||
```text
|
||||
Server Process / Connection Lifecycle
|
||||
@@ -1074,7 +1082,7 @@ tools/call 与 ToolResult 映射
|
||||
健康检查与 Tool 注销
|
||||
```
|
||||
|
||||
首个宿主实现优先支持本地 `stdio` 传输;其他传输在兼容性测试后增加。外部 Server 的 Tool 名称进入项目注册表前添加 Plugin 命名空间,并校验 JSON Schema、权限和重复 ID。MCP 内容不得绕过项目自己的 Permission、超时、日志净化和结果大小限制。
|
||||
首个宿主实现支持本地 `stdio` 传输;其他传输在兼容性测试后增加。外部 Server 的 Tool 名称进入项目注册表前添加 Plugin 命名空间,并校验 JSON Schema、权限、重复 ID 及其与 Manifest Contribution 的一致性。MCP 调用复用项目自己的 Permission、超时、Agent Trace、日志净化和结果大小限制;子进程环境按白名单裁剪,不传入 Provider Key、Vault 或数据库路径。
|
||||
|
||||
### 12.6 Frontend Extension Slot
|
||||
|
||||
@@ -2092,9 +2100,13 @@ MRR
|
||||
Citation Hit Rate
|
||||
P50 Latency
|
||||
P95 Latency
|
||||
total_cases
|
||||
successful_cases
|
||||
failed_cases
|
||||
failure_rate
|
||||
```
|
||||
|
||||
Benchmark 参数、Embedding 模型、Reranker、数据集版本和运行环境需要一起记录,保证不同实验结果可以复现。
|
||||
失败样本按零分计入质量指标分母,报告同时输出样本构成字段标明实际分母。Benchmark 参数、Embedding 模型、Reranker、数据集版本和运行环境需要一起记录,保证不同实验结果可以复现。
|
||||
|
||||
### 20.3 Agent Benchmark
|
||||
|
||||
@@ -2321,7 +2333,7 @@ Markdown Workspace
|
||||
|
||||
第一阶段 Plugin Runtime 已完成安装、启用、停用、权限和声明式 Tool 注册,建立 Skill 调用 Plugin Tool 的基础链路。Command、Settings 和 MCP 执行不计入第一阶段完成项。
|
||||
|
||||
截至 2026-09-01,上述第一阶段后端链路和 Web 联调前端均已完成,第二阶段前置的 Workspace 去 Mock 联调及 Agent Trace 持久化/恢复接口也已完成。当前验证基线为后端 81 项测试、前端 27 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性,真实 Embedding 召回质量不属于该测试结论。
|
||||
截至 2026-09-01,上述第一阶段后端链路和 Web 联调前端均已完成;第二阶段前置的 Workspace 去 Mock 联调、Agent Trace 持久化/恢复接口以及 stdio MCP Bridge / Plugin Host 也已完成。当前验证基线为后端 92 项测试、前端 27 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性,真实 Embedding 召回质量不属于该测试结论。
|
||||
|
||||
第二阶段在既有 Contract 上接入:
|
||||
|
||||
@@ -2331,7 +2343,7 @@ Multimodal
|
||||
└── pyannote.audio
|
||||
|
||||
Extension / Model
|
||||
├── MCP Bridge
|
||||
├── MCP Bridge(stdio 首版已实现)
|
||||
├── Plugin Command Contribution
|
||||
├── Plugin Settings Contribution
|
||||
└── Provider Streaming / Tool Calling / Error Mapping 增强
|
||||
@@ -2353,7 +2365,7 @@ Frontend Extension
|
||||
└── Plugin Settings UI
|
||||
```
|
||||
|
||||
上述列表描述第二阶段技术范围,不表示能力已经实现。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界,不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
|
||||
上述列表描述第二阶段技术范围,其中 stdio MCP Bridge 已实现,其余能力以各自开发说明的状态为准。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界,不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
|
||||
|
||||
第三阶段处理:
|
||||
|
||||
@@ -2409,7 +2421,7 @@ Sync Server 按独立服务开发和部署,不进入桌面客户端核心启
|
||||
|
||||
Python AI Core 未来作为 Tauri Sidecar 运行,当前由开发命令独立启动,FastAPI 提供本地接口。Knowledge Core 管理笔记结构;Retrieval Core 当前通过 FTS5、`HashEmbeddingProvider`、sqlite-vec、RRF 和轻量 Reranker 跑通混合检索,真实 Embedding 与正式 Benchmark 在第二阶段接入;Agent Runtime 使用 Tool Registry 操作知识库和任务,并将扩展 Agent Trace Contract 供可视化和 Benchmark 共用;Skill Runtime 将提示词、工具、权限和检索参数组装为可复用 Agent 配置。
|
||||
|
||||
当前 Plugin Runtime 支持 Manifest、生命周期和声明式白名单 Tool Contribution;第二阶段通过 MCP Bridge 接入隔离 Tool,并增加 Command 与 Settings Contribution。Provider Adapter 当前实现 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama,第二阶段按统一行为测试完善 OpenAI Responses、Anthropic Messages 等协议。多模态目标方案使用 faster-whisper、pyannote.audio 和可选 emotion2vec;当前只读取 Host 预生成 transcript。
|
||||
当前 Plugin Runtime 支持 Manifest、生命周期和声明式白名单 Tool Contribution,并已通过 stdio MCP Bridge 接入独立进程 Tool、Host 状态与重启接口;Command 与 Settings Contribution 尚待后续阶段实现。Provider Adapter 当前实现 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama,第二阶段按统一行为测试完善 OpenAI Responses、Anthropic Messages 等协议。多模态目标方案使用 faster-whisper、pyannote.audio 和可选 emotion2vec;当前只读取 Host 预生成 transcript。
|
||||
|
||||
第二阶段内容输出以 Document AST、Exporter Adapter、Mermaid Renderer 和 Function Plot Renderer 为共同边界,支持 HTML、PDF、DOCX 与静态图导出。Theme Package 使用 Manifest、Design Token 和受限 CSS 实现本地导入;联网主题市场不属于本阶段核心依赖。API Key 在 Web 联调期由 Fernet 开发存储加密保存,桌面版迁移到 Tauri Stronghold。多设备同步的目标方案为独立、可自托管的 Sync Server,目前尚未实现;本地核心功能不依赖 Sync Server。
|
||||
|
||||
|
||||
@@ -746,8 +746,8 @@ Markdown
|
||||
- [ ] faster-whisper 能完成真实音频转写;
|
||||
- [ ] pyannote.audio 能生成说话人分段;
|
||||
- [ ] 两者能组合生成带时间戳和 Speaker 的 Transcript;
|
||||
- [ ] MCP Server 能通过 MCP Bridge 注册 Tool;
|
||||
- [ ] Agent 能调用 MCP Tool;
|
||||
- [x] MCP Server 能通过 MCP Bridge 注册 Tool;
|
||||
- [x] Agent 能调用 MCP Tool;
|
||||
- [ ] Plugin Command Contribution 后端可注册;
|
||||
- [ ] Plugin Settings Contribution 后端可解析;
|
||||
- [ ] Provider Adapter 的 Streaming / Tool Calling / Error Mapping 稳定;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# 后端接口契约(开发版)
|
||||
|
||||
> 更新日期:2026-08-31。本文档记录当前前后端联调使用的已实现接口;机器可读字段、校验规则和响应模型以 FastAPI 运行时生成的 OpenAPI 为准。第二阶段尚未实现的规划接口见 `第二阶段接口契约-开发版.md`,不要将规划路径视为当前服务能力。
|
||||
> 更新日期:2026-09-01。本文档记录当前前后端联调使用的已实现接口;机器可读字段、校验规则和响应模型以 FastAPI 运行时生成的 OpenAPI 为准。第二阶段尚未实现的规划接口见 `第二阶段接口契约-开发版.md`,不要将规划路径视为当前服务能力。
|
||||
|
||||
## 契约入口
|
||||
|
||||
@@ -76,6 +76,8 @@ Web 联调阶段只暴露后端通过 `APP_VAULT_PATH` 配置的单一 Vault,
|
||||
| POST | `/api/plugins/{plugin_id}/enable` | 启用 Plugin |
|
||||
| POST | `/api/plugins/{plugin_id}/disable` | 停用 Plugin |
|
||||
| PUT | `/api/plugins/{plugin_id}/permissions` | 设置 Plugin 已授权权限 |
|
||||
| GET | `/api/plugins/{plugin_id}/host` | 获取隔离 MCP Host 状态、工具数和协商信息 |
|
||||
| POST | `/api/plugins/{plugin_id}/host/restart` | 重启 MCP Host 并重新发现、校验和注册 Tool |
|
||||
| DELETE | `/api/plugins/{plugin_id}` | 卸载 Plugin |
|
||||
|
||||
### Provider
|
||||
@@ -174,7 +176,7 @@ RunCancelled
|
||||
|
||||
## 当前实现状态
|
||||
|
||||
更新至 2026-09-01:后端 81 项回归测试通过。
|
||||
更新至 2026-09-01:后端 92 项回归测试通过。
|
||||
|
||||
- Chat、Agent Run、Agent Events、Tool 列表、Provider 配置生命周期、模型列表和连接测试已经接入 AI Core。
|
||||
- Agent Run/Event 已持久化到 SQLite;SSE 帧携带 sequence `id`,断线后可以回放缺失事件。Trace API 与 Benchmark 共用同一事件事实,并在入库前执行 Secret 脱敏和结果限长。
|
||||
@@ -183,6 +185,7 @@ RunCancelled
|
||||
- Workspace 已接入后端配置的真实 Vault;文件树、笔记读写、文件/目录新建、重命名和删除不再使用前端 Mock Fallback。
|
||||
- Note Move 保留 `note_id`;Citation 的字符偏移统一使用 UTF-16 code unit,供浏览器编辑器直接定位。
|
||||
- Plugin 启用前必须通过权限接口记录授权,未知权限默认拒绝。
|
||||
- 本地 stdio MCP Server 已通过独立子进程接入 Plugin Runtime;Agent 只消费内部 Tool Contract。Host 支持 initialize、分页发现、调用、超时取消、状态查询、重启和异常退出后的 Tool 注销。
|
||||
- Attachment Tool 读取 Host 管理的 `attachments` 目录;音频接口读取 Host 生成的转写文本,真实本地语音模型在第二阶段接入。
|
||||
- 接入业务模块时保持当前路径和 Contract,不在 Router 中直接实现数据库、Provider 或 Agent 逻辑。
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
> 文档状态:接口冻结草案
|
||||
>
|
||||
> 更新日期:2026-08-31
|
||||
> 更新日期:2026-09-01
|
||||
>
|
||||
> 依据:`../architecture/第二阶段团队分工表.md`、`../architecture/AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`
|
||||
|
||||
@@ -45,8 +45,8 @@
|
||||
| Transcription | POST | `/api/media/transcriptions/{job_id}/notes` | 计划新增 | 将 Transcript 写入 Knowledge Core |
|
||||
| Agent Trace | GET | `/api/agent/runs/{run_id}/events` | 已实现 | 支持游标恢复并增加模型与权限事件 |
|
||||
| Agent Trace | GET | `/api/agent/runs/{run_id}/trace` | 已实现 | 分页读取可回放 Trace 快照 |
|
||||
| Plugin Host | GET | `/api/plugins/{plugin_id}/host` | 计划新增 | 获取 MCP Host 健康状态 |
|
||||
| Plugin Host | POST | `/api/plugins/{plugin_id}/host/restart` | 计划新增 | 重启异常 Host 并重新发现 Tool |
|
||||
| Plugin Host | GET | `/api/plugins/{plugin_id}/host` | 已实现 | 获取 MCP Host 健康状态 |
|
||||
| Plugin Host | POST | `/api/plugins/{plugin_id}/host/restart` | 已实现 | 重启异常 Host 并重新发现 Tool |
|
||||
| Plugin Command | GET | `/api/plugin-contributions/commands` | 计划新增 | 获取前端可展示的 Command |
|
||||
| Plugin Command | POST | `/api/plugin-contributions/commands/{command_id}/execute` | 计划新增 | 受控执行 Command |
|
||||
| Plugin Settings | GET | `/api/plugins/{plugin_id}/settings` | 计划新增 | 获取 Schema 与非敏感配置 |
|
||||
@@ -55,9 +55,9 @@
|
||||
| Provider | 现有路径 | `/api/providers/*`、`POST /api/chat` | 扩展 | 补齐协议能力和统一行为 |
|
||||
| Retrieval | GET/POST | `/api/index/status`、`/api/index/rebuild` | 扩展 | 暴露 Embedding 兼容状态并安全重建向量 |
|
||||
| Benchmark | GET | `/api/benchmarks/datasets` | 计划新增 | 枚举受控 Dataset |
|
||||
| Benchmark | POST | `/api/benchmarks/rag/runs` | 计划新增 | 创建 RAG Benchmark |
|
||||
| Benchmark | POST | `/api/benchmarks/agent/runs` | 计划新增 | 创建 Agent Benchmark |
|
||||
| Benchmark | GET | `/api/benchmarks/runs` | 计划新增 | 分页获取 Benchmark Run |
|
||||
| Benchmark | POST | `/api/benchmarks/rag/runs` | 已实现 | 创建 RAG Benchmark |
|
||||
| Benchmark | POST | `/api/benchmarks/agent/runs` | 暂缓 | 创建 Agent Benchmark(依赖 Agent Runtime 完成后交付) |
|
||||
| Benchmark | GET | `/api/benchmarks/runs` | 已实现 | 分页获取 Benchmark Run |
|
||||
| Benchmark | GET/POST | `/api/benchmarks/runs/{run_id}/*` | 计划新增 | 查询、订阅、取消和读取报告 |
|
||||
| Export | POST | `/api/exports` | 计划新增 | 创建 HTML/PDF/DOCX 导出任务 |
|
||||
| Export | GET | `/api/exports` | 计划新增 | 分页获取导出任务 |
|
||||
@@ -427,7 +427,31 @@ class McpBridge(Protocol):
|
||||
async def stop(self, plugin_id: str) -> None: ...
|
||||
```
|
||||
|
||||
首个实现支持本地 `stdio`。Host 负责 initialize、capability negotiation、进程生命周期、超时、取消、stderr 隔离和异常退出后的 Tool 注销。
|
||||
首个实现已支持本地 `stdio`,按 MCP `2025-11-25` 发起 initialize,并兼容 `2025-06-18`、`2025-03-26` 和 `2024-11-05` 协商结果。Host 负责 capability negotiation、分页 `tools/list`、进程生命周期、超时取消、stderr 隔离和异常退出后的 Tool 注销。stdio 消息使用 UTF-8 单行 JSON-RPC,并在完整行进入内存前执行有界读取;当前不实现 Streamable HTTP。
|
||||
|
||||
MCP Plugin 的 `backend` 增加:
|
||||
|
||||
```yaml
|
||||
backend:
|
||||
type: mcp
|
||||
transport: stdio
|
||||
command: uvx
|
||||
args: [--isolated, --from, example-mcp==1.2.3, example-mcp]
|
||||
startup_timeout_seconds: 60
|
||||
tool_timeout_seconds: 30
|
||||
```
|
||||
|
||||
命令通过参数数组直接启动,不经过 Shell。带路径的 executable 必须位于 Plugin 包内;PATH 中的命令可以按名称引用。Python 包形式的 MCP 推荐使用固定版本的 `uvx --isolated --from`,但 `uvx` 只隔离依赖而不是文件/网络/系统调用安全沙箱,非 Python Server 不强制使用。开发模式首次运行未缓存的 uvx 包可能联网解析,因此 startup 示例使用 60 秒;生产安装阶段必须预取并验证,运行阶段不得临时解析依赖。子进程只继承运行所需的系统环境变量,不继承 `OPENAI_API_KEY`、`APP_DB_PATH`、Vault 路径等宿主状态。Secret 注入留给阶段 D 的专用引用接口。
|
||||
|
||||
在平台沙箱和可信命令许可完成前,`APP_ENVIRONMENT != development` 时启用 MCP Plugin 必须返回 `403 MCP_TRUST_APPROVAL_REQUIRED`,不得启动进程或注册 Tool。该门禁由后端执行,不能只依赖前端提示或文档约定。
|
||||
|
||||
远端 Tool 的可选项目权限放在 MCP `_meta`:
|
||||
|
||||
```json
|
||||
{ "_meta": { "notesagent/permission": "notes.read" } }
|
||||
```
|
||||
|
||||
该权限必须属于项目已知权限并同时出现在 Plugin Manifest 中。发现结果必须与 `contributes.tools` 的命名空间 ID 完全一致;校验全部成功后才一次性发布到 Tool Registry。
|
||||
|
||||
当前 Web 开发接口 `POST /api/plugins/install` 使用 `package_path`。桌面 Host 接入后,`ExtensionInstallRequest` 增加 `package_id`,由文件选择器产生临时包句柄;`package_path` 只在明确的 development 环境保留并标记 deprecated,生产构建拒绝任意前端路径。
|
||||
|
||||
@@ -460,6 +484,8 @@ error
|
||||
|
||||
`POST /api/plugins/{plugin_id}/host/restart` 返回 `202 OperationResponse`。重启期间先注销旧 Tool,发现和校验全部成功后再一次性发布新 Tool 集合,避免半注册状态。
|
||||
|
||||
当前实现还返回协商后的 `protocol_version`、`server_name` 和 `server_version`。单条协议消息上限为 2 MiB,单次 Tool Result 上限为 256 KiB;超限分别按 Host/Result 错误处理。Server 异常退出或发送无效 stdout 消息时,Host 进入 `unhealthy`,Plugin 进入 `error`,相关 Tool 立即注销。取消会同时通知 Server、移除 pending request 并唤醒本地等待线程。Restart 不得把 `installed`、`disabled` 或 `permission_required` Plugin 隐式启用,这些状态必须走 Enable。
|
||||
|
||||
### 7.3 Command Contribution 列表
|
||||
|
||||
`GET /api/plugin-contributions/commands?location=command_palette`
|
||||
@@ -608,6 +634,7 @@ MCP_CAPABILITY_UNSUPPORTED
|
||||
MCP_TOOL_SCHEMA_INVALID
|
||||
MCP_TOOL_CALL_FAILED
|
||||
MCP_TOOL_RESULT_TOO_LARGE
|
||||
MCP_TRUST_APPROVAL_REQUIRED
|
||||
PLUGIN_COMMAND_NOT_FOUND
|
||||
PLUGIN_COMMAND_CONTEXT_INVALID
|
||||
PLUGIN_SETTINGS_SCHEMA_INVALID
|
||||
@@ -790,7 +817,7 @@ Dataset 从仓库或受控导入目录注册。API 不接受调用方提交任
|
||||
|
||||
配置快照必须记录 Embedding model ID/version/dimension、Reranker、索引版本、Dataset Hash 和运行环境。
|
||||
|
||||
### 9.5 创建 Agent Benchmark
|
||||
### 9.5 创建 Agent Benchmark(暂缓,未暴露接口)
|
||||
|
||||
`POST /api/benchmarks/agent/runs`
|
||||
|
||||
@@ -824,12 +851,16 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`:
|
||||
"metrics": null,
|
||||
"config_snapshot": {},
|
||||
"error": null,
|
||||
"error_code": null,
|
||||
"created_at": "2026-08-31T10:30:00Z",
|
||||
"started_at": null,
|
||||
"completed_at": null
|
||||
}
|
||||
```
|
||||
|
||||
`status` 取值:`queued` → `running` → `completed` | `failed` | `cancelled`。失败/取消时 `error` 与
|
||||
`error_code` 只返回项目错误码与安全消息,不暴露第三方堆栈。
|
||||
|
||||
公共接口:
|
||||
|
||||
| 方法 | 路径 | 用途 |
|
||||
@@ -840,6 +871,10 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`:
|
||||
| POST | `/api/benchmarks/runs/{run_id}/cancel` | 取消运行 |
|
||||
| GET | `/api/benchmarks/runs/{run_id}/report` | 获取结构化完整报告 |
|
||||
|
||||
SSE 事件流(`RunStarted` → `CaseCompleted`* → `RunCompleted` | `RunFailed` | `RunCancelled`):
|
||||
`GET /api/benchmarks/runs/{run_id}/events` 支持 `Last-Event-ID` 与 `?after_sequence=` 游标恢复,
|
||||
`RunCompleted` / `RunFailed` / `RunCancelled` 为终止事件,收到后即断流。
|
||||
|
||||
### 9.7 指标 Contract
|
||||
|
||||
RAG:
|
||||
@@ -852,10 +887,17 @@ RAG:
|
||||
"mrr": 0.81,
|
||||
"citation_hit_rate": 0.89,
|
||||
"p50_latency_ms": 24.5,
|
||||
"p95_latency_ms": 67.3
|
||||
"p95_latency_ms": 67.3,
|
||||
"total_cases": 50,
|
||||
"successful_cases": 48,
|
||||
"failed_cases": 2,
|
||||
"failure_rate": 0.04
|
||||
}
|
||||
```
|
||||
|
||||
失败样本按零分计入质量指标分母,`total_cases` / `successful_cases` / `failed_cases` /
|
||||
`failure_rate` 让报告明确实际分母;延迟仅统计成功样本。
|
||||
|
||||
Agent:
|
||||
|
||||
```json
|
||||
@@ -879,8 +921,10 @@ BENCHMARK_DATASET_NOT_FOUND
|
||||
BENCHMARK_DATASET_INVALID
|
||||
BENCHMARK_CONFIG_INVALID
|
||||
BENCHMARK_INDEX_INCOMPATIBLE
|
||||
BENCHMARK_CAPACITY_EXCEEDED
|
||||
BENCHMARK_RUN_NOT_FOUND
|
||||
BENCHMARK_RUN_FAILED
|
||||
BENCHMARK_CASE_EVALUATION_FAILED
|
||||
```
|
||||
|
||||
### 9.9 Retrieval Profile 与索引兼容
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
> 本文档用于团队开发和模块联调,记录当前已经落地的核心边界与使用方式。
|
||||
|
||||
> 更新日期:2026-09-01。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成;第二阶段 Agent Trace 持久化和可恢复 SSE 已落地,后端当前回归基线为 81 项测试通过。
|
||||
> 更新日期:2026-09-01。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成;第二阶段 Agent Trace 持久化、可恢复 SSE、stdio MCP Bridge 与隔离 Plugin Host 已落地,后端当前回归基线为 92 项测试通过。
|
||||
|
||||
## 当前实现
|
||||
|
||||
@@ -33,12 +33,14 @@ backend/app/
|
||||
│ ├── permissions.py 权限策略、确认请求和会话授权
|
||||
│ └── builtin_tools.py 无副作用的内置开发 Tool
|
||||
├── extensions/
|
||||
│ └── runtime.py Skill/Plugin Manifest、生命周期、依赖与 Tool Contribution
|
||||
│ ├── runtime.py Skill/Plugin Manifest、生命周期、依赖与 Tool Contribution
|
||||
│ └── mcp.py stdio JSON-RPC、MCP 生命周期、发现、调用与 Host 隔离
|
||||
└── container.py AI Core 依赖组装
|
||||
|
||||
backend/extensions/
|
||||
├── skills/knowledge-assistant/ 内置知识库 Skill
|
||||
└── plugins/text-tools/ 内置示例 Plugin
|
||||
├── plugins/text-tools/ 内置声明式 Plugin
|
||||
└── fixtures/mcp-echo/ 离线 MCP Server 联调 Fixture
|
||||
```
|
||||
|
||||
Router 只负责 HTTP/SSE 与错误转换,不实现 Agent、Tool 或 Provider 业务逻辑。
|
||||
@@ -57,6 +59,7 @@ Router 只负责 HTTP/SSE 与错误转换,不实现 Agent、Tool 或 Provider
|
||||
- SQLite Trace、分页快照与可恢复 SSE;
|
||||
- Skill Manifest、Prompt、Tool/Permission/模型能力解析;
|
||||
- Plugin Manifest、生命周期和 Tool Contribution;
|
||||
- stdio MCP Bridge、隔离进程生命周期、Tool 映射与 Host 健康状态;
|
||||
- Skill 调用内置 Tool 与 Plugin Tool;
|
||||
- 公共 Contract 和 API 接入。
|
||||
|
||||
@@ -65,7 +68,7 @@ Router 只负责 HTTP/SSE 与错误转换,不实现 Agent、Tool 或 Provider
|
||||
- Note、NoteBlock、Markdown Parser:由 Knowledge Core 提供;
|
||||
- FTS5、Vector、RRF、Reranker、Citation:由 Retrieval Core 提供;
|
||||
- 文件系统和 API Key 明文读取:由 Rust Host 提供;
|
||||
- MCP Plugin Host、Frontend Extension Slot:按技术基线放在第二阶段实现。
|
||||
- Frontend Extension Slot 与 Plugin Command/Settings:按第二阶段后续阶段实现。
|
||||
|
||||
## Provider
|
||||
|
||||
@@ -302,7 +305,7 @@ DELETE /api/skills/{skill_id}
|
||||
|
||||
### Plugin Runtime
|
||||
|
||||
第一阶段 Plugin Runtime 完成 Manifest 校验、安装、启用、停用、卸载和 Tool Contribution。第三方代码不会直接 import 到 AI Core;当前 Declarative Plugin Host 只执行宿主实现的白名单 handler,MCP Host 留到第二阶段。
|
||||
第一阶段 Plugin Runtime 完成 Manifest 校验、安装、启用、停用、卸载和声明式 Tool Contribution。阶段 C 增加 stdio MCP Bridge:第三方代码不会直接 import 到 AI Core,而由独立子进程运行,通过换行分隔 JSON-RPC 完成 initialize、Tool 发现和调用。
|
||||
|
||||
启用 Plugin 时将 Tool 注册到统一 Tool Registry,并标记 `source=plugin`;停用或异常时注销 Tool。启用中的 Skill 依赖某 Plugin Tool 时,Plugin 不能直接卸载。
|
||||
|
||||
@@ -315,11 +318,15 @@ GET /api/plugins/{plugin_id}
|
||||
POST /api/plugins/{plugin_id}/enable
|
||||
POST /api/plugins/{plugin_id}/disable
|
||||
PUT /api/plugins/{plugin_id}/permissions
|
||||
GET /api/plugins/{plugin_id}/host
|
||||
POST /api/plugins/{plugin_id}/host/restart
|
||||
DELETE /api/plugins/{plugin_id}
|
||||
```
|
||||
|
||||
Plugin Manifest 中的权限只是声明,不代表已经授权。带权限的 Plugin 安装后进入 `permission_required`,Host 必须通过权限接口记录用户授权,之后才能启用。JSON Schema 在安装阶段校验,Tool 调用时再次校验实际参数。
|
||||
|
||||
MCP Tool 进入 Registry 前统一增加 `<plugin_id>.<remote_name>` 命名空间。Server 声明的 `notesagent/permission` 必须属于已知权限并出现在 Plugin Manifest;发现集合还必须与 Manifest Contribution 完全一致。启用失败会回滚全部 Tool 并关闭子进程,异常退出会把 Plugin 标记为 `error` 并立即注销对应 Tool。详细实现和 Fixture 操作见 [MCP Bridge 与 Plugin Host 开发说明](MCP-Bridge与Plugin-Host开发说明.md)。
|
||||
|
||||
内置示例 `text-tools` 注册 `text.uppercase`。内置 `knowledge-assistant` Skill 同时声明 `notes.search` 和 `text.uppercase`,用于验证完整链路:
|
||||
|
||||
```text
|
||||
@@ -342,4 +349,4 @@ Skill Manifest
|
||||
- Task 已持久化到 SQLite;Attachment Tool 读取 Host 管理目录中的 UTF-8 文件。
|
||||
- `audio.transcribe` 当前消费 Host 预生成的 transcript;faster-whisper 与说话人分离仍按技术基线在第二阶段接入。
|
||||
- Extension 安装记录暂存内存;后续接入持久化 Registry 与版本升级流程。
|
||||
- 当前 Plugin Host 只支持内置声明式白名单 handler;MCP Bridge、独立进程健康检查与 UI Contribution 在第二阶段实现。
|
||||
- 当前 Plugin Host 支持内置声明式 handler 和本地 stdio MCP Server;Streamable HTTP、OS 级沙箱、Plugin Command/Settings 与 UI Contribution 留在后续阶段。
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
# Benchmark 开发说明
|
||||
|
||||
> 所属模块:Knowledge / Retrieval Core(后端,负责人 yxx)。RAG Benchmark 已交付;Agent Benchmark 暂缓,待 Agent Runtime 完成后在同一契约下补齐。
|
||||
|
||||
## 定位
|
||||
|
||||
Benchmark Service 用受控 Dataset 对检索引擎做可复现评测:创建即返回 queued、后台 asyncio.Task 执行、SSE 实时推送进度、结束后产出结构化报告。CLI、测试与前端报告页复用同一 Service,不各自实现指标。
|
||||
|
||||
## 接口
|
||||
|
||||
| 方法 | 路径 | 用途 |
|
||||
| --- | --- | --- |
|
||||
| GET | `/api/benchmarks/datasets?kind=rag` | 枚举受控目录下的 Dataset 元信息 |
|
||||
| POST | `/api/benchmarks/rag/runs` | 创建 RAG Benchmark(202) |
|
||||
| GET | `/api/benchmarks/runs?kind=&status=&limit=&offset=` | 分页获取运行记录 |
|
||||
| GET | `/api/benchmarks/runs/{run_id}` | 状态与指标摘要 |
|
||||
| GET | `/api/benchmarks/runs/{run_id}/events` | SSE 进度与 Case 结果 |
|
||||
| POST | `/api/benchmarks/runs/{run_id}/cancel` | 取消运行 |
|
||||
| GET | `/api/benchmarks/runs/{run_id}/report` | 结构化完整报告 |
|
||||
|
||||
Agent Benchmark 的 `/api/benchmarks/agent/runs` 未暴露(暂缓),不在 OpenAPI 注册占位接口。
|
||||
|
||||
## Dataset
|
||||
|
||||
Dataset 来自 `settings.benchmark_datasets_path`(默认 `backend/data/benchmarks`),API 不接受调用方提交任意路径。按文件名 stem 精确匹配 `{dataset_id}.json`,与请求无关文件的损坏(JSON 语法错误、UTF-8 解码错误、顶层非对象)不会阻断加载;只有目标文件本身损坏才返回 `BENCHMARK_DATASET_INVALID`。
|
||||
|
||||
RAG Case 结构:`case_id`、`query`、`expected_note_ids`、`expected_block_ids`、`citation_required`、`tags`。`citation_required=true` 时必须声明 `expected_block_ids`,否则无法计算 Citation Hit Rate。
|
||||
|
||||
## 运行生命周期
|
||||
|
||||
`queued → running → completed | failed | cancelled`。
|
||||
|
||||
- 创建时校验索引兼容性:索引非空、Embedding model/dim 与当前引擎一致、vector/hybrid 时向量索引非空;不满足返回 `BENCHMARK_INDEX_INCOMPATIBLE`(409),避免把环境/索引错误误判为检索质量差。
|
||||
- 内存注册表上限 `MAX_RUNS=100`,超限只淘汰终态 run;满容量且全为活动 run 时返回 `BENCHMARK_CAPACITY_EXCEEDED`(429)。
|
||||
- 失败/取消只向公开响应暴露项目错误码与安全消息,详细异常进入日志,不通过 HTTP/SSE 返回。
|
||||
|
||||
## 指标
|
||||
|
||||
RAG 按 (mode, case, repeat) 逐样本计算,再按 mode 聚合:
|
||||
|
||||
- 质量:`hit_at_1`、`hit_at_5`、`recall_at_k`、`mrr`、`citation_hit_rate`;
|
||||
- 延迟:`p50_latency_ms`、`p95_latency_ms`(仅统计成功样本);
|
||||
- 样本构成:`total_cases`、`successful_cases`、`failed_cases`、`failure_rate`。
|
||||
|
||||
失败样本按零分计入质量指标分母,报告据此可知实际分母,避免把执行失败误判为检索质量差。
|
||||
|
||||
## 事件与 SSE
|
||||
|
||||
事件流:`RunStarted → CaseCompleted* → RunCompleted | RunFailed | RunCancelled`。
|
||||
|
||||
`GET /api/benchmarks/runs/{run_id}/events` 支持 `Last-Event-ID` 与 `?after_sequence=` 游标恢复(复用 Agent SSE 的解析逻辑),`RunCompleted` / `RunFailed` / `RunCancelled` 为终止事件,收到后断流。
|
||||
|
||||
## 错误码
|
||||
|
||||
```text
|
||||
BENCHMARK_DATASET_NOT_FOUND
|
||||
BENCHMARK_DATASET_INVALID
|
||||
BENCHMARK_INDEX_INCOMPATIBLE
|
||||
BENCHMARK_CAPACITY_EXCEEDED
|
||||
BENCHMARK_RUN_NOT_FOUND
|
||||
BENCHMARK_RUN_FAILED
|
||||
BENCHMARK_CASE_EVALUATION_FAILED
|
||||
```
|
||||
|
||||
## 配置快照
|
||||
|
||||
报告与运行记录保存 `config_snapshot`:dataset hash/version、modes、retrieval 参数、Embedding model/version/dim、Reranker、索引元数据、App 版本与环境、Python 版本,保证不同实验结果可复现。
|
||||
|
||||
## 测试
|
||||
|
||||
```powershell
|
||||
cd backend
|
||||
uv run pytest -q
|
||||
```
|
||||
|
||||
`tests/test_benchmark.py` 覆盖数据集注册与校验、指标纯函数、端到端运行、取消、索引兼容、容量与失败样本聚合;`tests/test_retrieval.py` 覆盖 FTS 阈值与分页 total 一致性。
|
||||
@@ -3,7 +3,7 @@
|
||||
> 本文档用于团队开发和模块联调,记录 Knowledge Core / Retrieval Core 已经落地的
|
||||
> 模块边界、数据模型、接口与使用方式,对应分工表中的杨星萱。
|
||||
|
||||
> 更新日期:2026-09-01。第一阶段 Knowledge/Retrieval 主链路已经完成,并已接入 Agent Tool Registry;完整后端回归基线为 81 项测试通过。
|
||||
> 更新日期:2026-09-01。第一阶段 Knowledge/Retrieval 主链路已经完成,并已接入 Agent Tool Registry;完整后端回归基线为 92 项测试通过。
|
||||
|
||||
## 当前实现
|
||||
|
||||
@@ -198,7 +198,7 @@ cd backend
|
||||
uv run pytest -q
|
||||
```
|
||||
|
||||
当前后端完整测试共 71 个用例通过(单元 + 端到端)。测试通过 `tests/conftest.py` 的 autouse fixture 把
|
||||
当前后端完整测试共 120 个用例通过(单元 + 端到端)。测试通过 `tests/conftest.py` 的 autouse fixture 把
|
||||
数据目录/DB/Vault 重定向到临时目录,不读写真实 `backend/data`,任何本机状态下结果确定。
|
||||
|
||||
## 配置
|
||||
@@ -232,4 +232,6 @@ rag.search
|
||||
- Embedding / Reranker 为轻量实现,后续替换为真实模型(接口不变)。
|
||||
- 小语料下 hybrid 检索召回偏宽(向量 Top-K 覆盖全部 block),可加相关性阈值收紧。
|
||||
- 重建为同步 + 全量,后续接入增量索引与异步任务队列。
|
||||
- 检索 Benchmark 待建立。
|
||||
- RAG Benchmark 已建立:`POST /api/benchmarks/rag/runs` 创建即返回 queued、后台 Task 执行,
|
||||
通过 SSE 实时推送进度,报告含逐 Case 结果与 `total_cases` / `successful_cases` / `failed_cases` / `failure_rate`。
|
||||
- Agent Benchmark 暂缓,待 Agent Runtime 完成后交付。
|
||||
|
||||
@@ -0,0 +1,282 @@
|
||||
# MCP Bridge 与 Plugin Host 开发说明
|
||||
|
||||
> 更新日期:2026-09-01。本文记录第二阶段阶段 C 已实现的本地 stdio MCP Bridge、隔离 Plugin Host、Tool Contract 转换和离线测试方式。Plugin Command 与 Settings 属于阶段 D,不在本文实现范围内。
|
||||
|
||||
## 1. 目标与实现状态
|
||||
|
||||
阶段 C 的目标是让外部 MCP Server 进入既有 Plugin、Tool、Permission、Agent 和 Trace 链路,同时避免 Agent Runtime、前端或 Benchmark 直接依赖 MCP 原始消息。
|
||||
|
||||
当前链路:
|
||||
|
||||
```text
|
||||
Plugin Manifest
|
||||
→ Plugin Runtime
|
||||
→ 独立 stdio MCP Server 进程
|
||||
→ initialize / capability negotiation
|
||||
→ tools/list 分页发现与校验
|
||||
→ NotesAgent ToolDefinition
|
||||
→ Tool Registry / Permission Manager
|
||||
→ Agent Runtime / Agent Trace
|
||||
```
|
||||
|
||||
已经实现:
|
||||
|
||||
- 本地 stdio 子进程启动、关闭和异常退出检测;
|
||||
- UTF-8、换行分隔的 JSON-RPC 2.0 消息;
|
||||
- initialize、协议版本与 tools capability 协商;
|
||||
- `notifications/initialized`;
|
||||
- 分页 `tools/list`;
|
||||
- `tools/call`、业务错误与 JSON-RPC 错误转换;
|
||||
- 超时和 `notifications/cancelled`;
|
||||
- Tool 命名空间、JSON Schema、权限和 Manifest 集合校验;
|
||||
- Host 状态查询、重启和异常后的 Tool 自动注销;
|
||||
- stderr 隔离、环境变量裁剪、消息及结果大小限制;
|
||||
- 无网络、无密钥的确定性 MCP Fixture。
|
||||
|
||||
实现依据为 MCP 官方 [Lifecycle 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle)、[Transports 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports) 和 [Tools 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/server/tools)。
|
||||
|
||||
## 2. 代码位置
|
||||
|
||||
```text
|
||||
backend/app/extensions/mcp.py
|
||||
stdio 进程、JSON-RPC、MCP 生命周期、发现、调用和 Host 状态
|
||||
|
||||
backend/app/extensions/runtime.py
|
||||
Plugin Manifest、权限、MCP Tool 批量注册/回滚和生命周期集成
|
||||
|
||||
backend/app/agent/tools.py
|
||||
内部 Tool 参数校验、结构化执行错误和线程安全 Registry
|
||||
|
||||
backend/extensions/fixtures/mcp-echo/
|
||||
确定性 stdio MCP Server 与 Plugin Manifest
|
||||
```
|
||||
|
||||
## 3. Plugin Manifest
|
||||
|
||||
MCP Plugin 的后端配置示例:
|
||||
|
||||
```yaml
|
||||
id: example-mcp
|
||||
name: Example MCP
|
||||
version: 1.0.0
|
||||
permissions:
|
||||
- notes.read
|
||||
contributes:
|
||||
tools:
|
||||
- example-mcp.search
|
||||
backend:
|
||||
type: mcp
|
||||
transport: stdio
|
||||
command: uvx
|
||||
args: [--isolated, --from, example-mcp==1.2.3, example-mcp]
|
||||
startup_timeout_seconds: 60
|
||||
tool_timeout_seconds: 30
|
||||
```
|
||||
|
||||
约束:
|
||||
|
||||
- 阶段 C 只接受 `type: mcp` 与 `transport: stdio`;
|
||||
- 命令和参数通过数组直接传给 `subprocess.Popen`,不经过 Shell;
|
||||
- Python 包形式的 MCP Server 推荐使用 `uvx --isolated --from <package>==<version> <command>`,固定版本并与 NotesAgent 项目环境隔离;
|
||||
- Plugin 包内自带且不需要第三方依赖的 Python 脚本可以使用 `python server.py`;Node、Rust 等 Server 继续使用各自受控启动器,因此 Host 不强制所有 MCP 都经过 `uvx`;
|
||||
- PATH 中的 executable 使用名称,例如 `uvx`、`python`、`node`;
|
||||
- manifest 中带目录的 executable 必须解析到 Plugin 包内部;
|
||||
- `contributes.tools` 使用 `<plugin_id>.<remote_name>`;
|
||||
- 安装阶段只读 Manifest,不启动第三方进程;
|
||||
- 完成用户授权后,`enable` 才启动 Host。
|
||||
|
||||
## 4. 生命周期
|
||||
|
||||
### 4.1 启动
|
||||
|
||||
启用 MCP Plugin 时依次执行:
|
||||
|
||||
1. 检查 Plugin 声明权限是否全部获得授权;
|
||||
2. 检查 Manifest 声明的 Tool ID 是否与现有 Registry 冲突;
|
||||
3. 启动独立 stdio Server;
|
||||
4. 发送 `initialize`;
|
||||
5. 校验协商版本和 `tools` capability;
|
||||
6. 发送 `notifications/initialized`;
|
||||
7. 分页读取 `tools/list`;
|
||||
8. 校验全部 Tool;
|
||||
9. 确认发现集合与 Manifest 完全一致;
|
||||
10. 将完整集合注册到 Tool Registry;
|
||||
11. Plugin 和 Host 进入 `ready`。
|
||||
|
||||
任何步骤失败都会注销本轮已注册 Tool、关闭子进程并把 Plugin 标记为 `error`,不会留下半启用状态。
|
||||
|
||||
### 4.2 停止与异常退出
|
||||
|
||||
停用、卸载或应用关闭时,先注销 Tool,再关闭 stdin,等待 Server 正常退出。超时后依次 terminate 和 kill。
|
||||
|
||||
Server 异常退出、stdout 出现非 JSON-RPC 内容或发送超大协议消息时:
|
||||
|
||||
- 未完成请求返回 `PLUGIN_HOST_UNAVAILABLE`;
|
||||
- Host 进入 `unhealthy`;
|
||||
- Plugin 进入 `error`;
|
||||
- 对应 Tool 从 Registry 中立即注销;
|
||||
- 用户可以调用 Host Restart 接口重新协商和发现。
|
||||
|
||||
Server 发送 `notifications/tools/list_changed` 时不会直接信任新集合。当前实现先把 Host 标记为不健康并注销旧 Tool,要求通过 Restart 重新执行完整发现与校验。
|
||||
|
||||
## 5. Tool Contract 转换
|
||||
|
||||
MCP Tool:
|
||||
|
||||
```json
|
||||
{
|
||||
"name": "search",
|
||||
"description": "Search notes",
|
||||
"inputSchema": { "type": "object", "properties": {} },
|
||||
"_meta": { "notesagent/permission": "notes.read" }
|
||||
}
|
||||
```
|
||||
|
||||
进入系统后转换为:
|
||||
|
||||
```json
|
||||
{
|
||||
"name": "example-mcp.search",
|
||||
"description": "Search notes",
|
||||
"parameters": { "type": "object", "properties": {} },
|
||||
"permission": "notes.read",
|
||||
"source": "plugin"
|
||||
}
|
||||
```
|
||||
|
||||
转换规则:
|
||||
|
||||
- 远端名称必须能转换为合法且稳定的项目 Tool ID;
|
||||
- `inputSchema` 必须是有效的 object JSON Schema;
|
||||
- `additionalProperties`、`patternProperties` 等动态字段先由完整 JSON Schema 校验,Pydantic 参数载体不会再次误拒绝合法字段;
|
||||
- `_meta.notesagent/permission` 必须属于项目已知权限;
|
||||
- Tool 权限必须同时出现在 Plugin Manifest 中;
|
||||
- Agent 仍通过 Tool Registry 执行参数校验、Permission、超时和 Trace;
|
||||
- MCP `structuredContent` 存在时映射为内部 output;否则保留为受控 `content` 数组;
|
||||
- MCP `isError: true` 映射为 `MCP_TOOL_CALL_FAILED`;
|
||||
- 结果超过 256 KiB 映射为 `MCP_TOOL_RESULT_TOO_LARGE`。
|
||||
|
||||
## 6. 隔离与安全边界
|
||||
|
||||
当前隔离是“独立进程 + 协议边界”,不是完整的操作系统沙箱。
|
||||
|
||||
`uvx` 解决的是 Python 工具依赖隔离:它等价于 `uv tool run`,在 uv 缓存中使用可丢弃的独立虚拟环境。它不会限制 Server 读取用户文件、访问网络、创建子进程或调用系统 API,因此不能代替安全沙箱。当前开发模式下,首次 `enable` 尚未缓存的包可能访问包索引,因此示例使用 60 秒启动上限;生产实现不得依赖该行为,必须在用户确认后的安装/更新阶段预取和验证固定版本,运行阶段只启动已准备好的环境。
|
||||
|
||||
已经执行的保护:
|
||||
|
||||
- 第三方模块不 import 到 AI Core;
|
||||
- 子进程 `cwd` 固定为 Plugin 包目录;
|
||||
- 不使用 Shell 拼接命令;
|
||||
- 不把 Provider API Key、`APP_DB_PATH`、Vault 路径和其他宿主环境变量传入子进程;
|
||||
- stderr 与 JSON-RPC stdout 分离,stderr 不进入 API 和 Agent Trace;
|
||||
- stdout 只能发送合法 MCP JSON-RPC;
|
||||
- stdout 在读取完整行前即应用有界读取,单条协议消息上限 2 MiB;stderr 也按固定大小分块读取;
|
||||
- 单次 Tool Result 上限 256 KiB;
|
||||
- MCP Tool 不绕过 Permission Manager 和 Agent Tool Timeout。
|
||||
- 调用被 Agent 取消时,同时通知 Server 并唤醒本地 pending Queue,阻塞线程不会继续占用线程池直至远端超时。
|
||||
|
||||
当前尚未提供容器、受限系统账户、seccomp、Windows AppContainer 或 macOS Sandbox,因此 Plugin 进程仍具有当前操作系统用户授予的一般文件访问能力。正式社区插件分发或“一键安装”前必须完成以下安全门槛:
|
||||
|
||||
- 由 Tauri/Rust Host 统一启动进程并提供平台级文件、网络、子进程和资源配额限制;
|
||||
- 安装/更新时完整展示 executable 与全部参数,明确警告并要求用户主动确认;
|
||||
- 固定包来源和版本,增加包哈希/签名与可信发布者校验;
|
||||
- 默认禁止访问 Vault、凭据和宿主环境,只通过声明 Permission 与受控 Host API 授权;
|
||||
- 关闭 Host 时终止完整进程树,不只结束直接子进程。
|
||||
|
||||
在这些门槛完成前,当前 MCP Host 只适用于内置 Fixture、团队可信插件和开发联调;不得把它描述为可以安全执行任意社区代码。上述安装确认要求遵循 MCP [SEP-1024](https://modelcontextprotocol.io/seps/1024-mcp-client-security-requirements-for-local-server-);`uvx` 行为依据 uv 官方 [Using tools](https://docs.astral.sh/uv/guides/tools/) 文档。
|
||||
|
||||
后端通过 `APP_ENVIRONMENT` 强制该边界:只有 `development` 可以启动当前未沙箱化的 MCP Host;其他环境返回 `403 MCP_TRUST_APPROVAL_REQUIRED`,且不会创建进程或注册 Tool。后续 Tauri/Rust Host 提供沙箱与绑定完整命令摘要的可信许可后,再替换此临时门禁。
|
||||
|
||||
## 7. Host API
|
||||
|
||||
```http
|
||||
GET /api/plugins/{plugin_id}/host
|
||||
POST /api/plugins/{plugin_id}/host/restart
|
||||
```
|
||||
|
||||
状态响应包含:
|
||||
|
||||
```text
|
||||
plugin_id
|
||||
backend_type / transport
|
||||
status
|
||||
tools_count
|
||||
started_at / last_seen_at
|
||||
protocol_version
|
||||
server_name / server_version
|
||||
error
|
||||
```
|
||||
|
||||
状态值:
|
||||
|
||||
```text
|
||||
stopped
|
||||
starting
|
||||
ready
|
||||
unhealthy
|
||||
error
|
||||
```
|
||||
|
||||
Restart 返回 `202 OperationResponse`。接口返回前已完成本地 Host 重启和 Tool 重新发现;`message` 中给出最终 Host 状态。Restart 只用于运行中或异常 Host;用户主动停用、尚未启用或等待授权的 Plugin 返回 `409 PLUGIN_HOST_UNAVAILABLE`,必须通过 Enable 明确启动。
|
||||
|
||||
## 8. 离线 Fixture
|
||||
|
||||
Fixture 位于:
|
||||
|
||||
```text
|
||||
backend/extensions/fixtures/mcp-echo
|
||||
```
|
||||
|
||||
它提供:
|
||||
|
||||
- `mcp-fixture.echo`:返回 structuredContent;
|
||||
- `mcp-fixture.fail`:返回 `isError: true`;
|
||||
- `mcp-fixture.sleep`:验证超时和取消;
|
||||
- `mcp-fixture.large`:验证结果大小上限;
|
||||
- `mcp-fixture.environment`:验证宿主 Secret/路径没有进入子进程;
|
||||
- `mcp-fixture.exit`:验证异常退出、Tool 注销和 Restart。
|
||||
|
||||
Fixture 的 `tools/list` 使用两页响应,用于覆盖分页发现。测试还会启动缺少 tools capability、返回无效 Schema/initialize result,以及输出超长无换行 stdout 的变体。
|
||||
|
||||
## 9. 验证
|
||||
|
||||
```powershell
|
||||
cd backend
|
||||
uv run python -m compileall -q app
|
||||
uv run pytest
|
||||
|
||||
cd ../frontend
|
||||
pnpm test
|
||||
pnpm type-check
|
||||
pnpm build
|
||||
```
|
||||
|
||||
阶段 C 新增测试覆盖:
|
||||
|
||||
- initialize、版本和 capability negotiation;
|
||||
- 分页 `tools/list` 与命名空间映射;
|
||||
- Permission、JSON Schema 与 Contribution 集合;
|
||||
- Tool 成功、业务错误、结果过大和超时;
|
||||
- Agent 取消后 pending 等待线程及时释放;
|
||||
- `additionalProperties` 动态参数保持 JSON Schema 语义;
|
||||
- Agent Runtime 调用 MCP Tool 并写入正式 Trace;
|
||||
- Secret/Vault 环境隔离;
|
||||
- Server 异常退出、Tool 注销和 Host Restart;
|
||||
- 缺少 capability、无效 initialize result、无效 MCP Schema 和超长无换行 stdout;
|
||||
- disabled Plugin 不会被 Host Restart 隐式重新启用;
|
||||
- OpenAPI 发布 Host 状态和重启路径。
|
||||
|
||||
## 10. 当前边界与后续阶段
|
||||
|
||||
阶段 C 不包含:
|
||||
|
||||
- Streamable HTTP MCP transport;
|
||||
- Resources、Prompts、Sampling、Elicitation 和 MCP Tasks;
|
||||
- Plugin Command 与 Settings Contribution;
|
||||
- Secret Reference 注入;
|
||||
- Plugin Registry 持久化、签名与社区来源校验;
|
||||
- 操作系统级沙箱;
|
||||
- 一键安装前的完整命令展示与确认 UI;
|
||||
- Tool 列表热更新的无中断替换。
|
||||
|
||||
阶段 D 将在当前 Plugin Runtime 上继续增加 Command、Settings、Secret Contract 和命名空间 Storage,不修改 Agent 使用内部 Tool Contract 的原则。
|
||||
@@ -187,12 +187,12 @@ pnpm build
|
||||
```text
|
||||
pnpm build passed
|
||||
pnpm test 27 passed
|
||||
uv run pytest 81 passed
|
||||
uv run pytest 92 passed
|
||||
preview smoke HTTP 200
|
||||
git diff --check passed
|
||||
```
|
||||
|
||||
当前前端使用 Vitest 执行 Store、Workspace API Adapter、SSE 恢复游标、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试;`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告,不影响 81 项测试结果,也不涉及产品代码。
|
||||
当前前端使用 Vitest 执行 Store、Workspace API Adapter、SSE 恢复游标、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试;`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告,不影响 92 项测试结果,也不涉及产品代码。
|
||||
|
||||
Vite 当前会提示 Chat 与 Workspace 的部分异步 Chunk 超过 500 kB,这是 Milkdown、CodeMirror、KaTeX 和 Shiki 等编辑/渲染依赖带来的性能优化项,不影响构建成功或功能正确性;进入桌面打包前应通过手动分包或更细粒度动态加载继续优化。
|
||||
|
||||
|
||||
@@ -104,4 +104,4 @@ pnpm build
|
||||
|
||||
自动化验证覆盖 Provider 预设、OpenAI-Compatible `/models` 请求与鉴权头、模型映射、前端自动刷新、排序去重及按 Provider 隔离错误。生产构建同时执行 Vue 和 TypeScript 类型检查。
|
||||
|
||||
当前完整回归基线:后端 81 项测试、前端 27 项测试通过,前端类型检查和生产构建通过。Provider 配置目前仍保存在内存 Registry,AI Core 重启后需要重新创建;凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现,设置页正式预设不会使用这两种协议。
|
||||
当前完整回归基线:后端 92 项测试、前端 27 项测试通过,前端类型检查和生产构建通过。Provider 配置目前仍保存在内存 Registry,AI Core 重启后需要重新创建;凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现,设置页正式预设不会使用这两种协议。
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
> 审阅范围:FastAPI、Knowledge / Retrieval Core、Agent Core、Extension Core、Provider Adapter、公共接口和后端开发文档。
|
||||
> 文档用途:记录问题形成原因、实际影响、修复判断和落地方案,供后续开发文档、比赛材料与技术博客使用。
|
||||
|
||||
> 2026-09-01 状态补充:本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析、Fernet 加密存储和 Agent Trace 持久化,当前完整后端回归基线为 81 项测试通过。
|
||||
> 2026-09-01 状态补充:本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析、Fernet 加密存储、Agent Trace 持久化和 stdio MCP Plugin Host,当前完整后端回归基线为 92 项测试通过。
|
||||
|
||||
## 1. 审阅结论
|
||||
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
allowBuilds:
|
||||
esbuild: true
|
||||
@@ -255,6 +255,22 @@ export type PluginStatus =
|
||||
| 'dependency_missing'
|
||||
| 'permission_required'
|
||||
|
||||
export type PluginHostState = 'stopped' | 'starting' | 'ready' | 'unhealthy' | 'error'
|
||||
|
||||
export interface PluginHostStatus {
|
||||
plugin_id: string
|
||||
backend_type: 'mcp' | 'internal_rpc' | 'none'
|
||||
transport: 'stdio' | 'http' | 'none'
|
||||
status: PluginHostState
|
||||
tools_count: number
|
||||
started_at?: string | null
|
||||
last_seen_at?: string | null
|
||||
protocol_version?: string | null
|
||||
server_name?: string | null
|
||||
server_version?: string | null
|
||||
error?: string | null
|
||||
}
|
||||
|
||||
export interface PluginContribution {
|
||||
type: 'tool' | 'command' | 'importer' | 'exporter' | 'sidebar_panel' | 'settings_section'
|
||||
id: string
|
||||
@@ -539,7 +555,14 @@ export interface ApiPlugin {
|
||||
panels: string[]
|
||||
settings_sections: string[]
|
||||
}
|
||||
backend: { type: 'mcp' | 'internal_rpc' | 'none'; transport: 'stdio' | 'http' | 'none' }
|
||||
backend: {
|
||||
type: 'mcp' | 'internal_rpc' | 'none'
|
||||
transport: 'stdio' | 'http' | 'none'
|
||||
command?: string | null
|
||||
args?: string[]
|
||||
startup_timeout_seconds?: number
|
||||
tool_timeout_seconds?: number
|
||||
}
|
||||
}
|
||||
status: PluginStatus
|
||||
enabled: boolean
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import apiClient from './apiClient'
|
||||
import type { ApiPlugin, OperationResponse, Plugin, PluginContribution } from '@/contracts'
|
||||
import type { ApiPlugin, OperationResponse, Plugin, PluginContribution, PluginHostStatus } from '@/contracts'
|
||||
|
||||
function toPlugin(plugin: ApiPlugin): Plugin {
|
||||
const { manifest } = plugin
|
||||
@@ -54,6 +54,14 @@ export async function grantPluginPermissions(pluginId: string, permissions: stri
|
||||
return toPlugin(await apiClient.put<ApiPlugin>(`/api/plugins/${pluginId}/permissions`, { permissions }))
|
||||
}
|
||||
|
||||
export async function getPluginHostStatus(pluginId: string): Promise<PluginHostStatus> {
|
||||
return apiClient.get(`/api/plugins/${pluginId}/host`)
|
||||
}
|
||||
|
||||
export async function restartPluginHost(pluginId: string): Promise<OperationResponse> {
|
||||
return apiClient.post(`/api/plugins/${pluginId}/host/restart`)
|
||||
}
|
||||
|
||||
export async function uninstallPlugin(pluginId: string): Promise<OperationResponse> {
|
||||
return apiClient.delete(`/api/plugins/${pluginId}`)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user