fix(backend): 落实 PR #11 第二轮评审意见
- Benchmark 容量淘汰只删终态 run，满容量且全活动时返回 BENCHMARK_CAPACITY_EXCEEDED - 创建 run 前校验索引兼容性（BENCHMARK_INDEX_INCOMPATIBLE） - 取消 run 补发 RunCancelled 终止事件；失败分支脱敏（BENCHMARK_RUN_FAILED） - 失败样本计入汇总分母，报告输出 total/successful/failed/failure_rate - load_dataset 按文件名隔离无关损坏文件，顶层非对象拒绝 - FTS score_threshold 先于计数/分页，total 与 items 一致 - Benchmark SSE 支持 Last-Event-ID 游标 - 移除 Agent Benchmark 501 占位接口 - 同步第二阶段接口契约与开发说明文档 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -172,11 +172,23 @@ def list_datasets(kind: BenchmarkKind) -> list[BenchmarkDatasetInfo]:
|
||||
|
||||
|
||||
def load_dataset(dataset_id: str, kind: BenchmarkKind) -> RAGDataset:
|
||||
"""按 id 加载并校验数据集；找不到抛 BENCHMARK_DATASET_NOT_FOUND。"""
|
||||
"""按文件名加载并校验数据集；找不到抛 BENCHMARK_DATASET_NOT_FOUND。
|
||||
|
||||
只读取与请求 dataset_id 同名的文件（{dataset_id}.json），无关文件的损坏（JSON 语法
|
||||
错误、UTF-8 解码错误、顶层非对象）不会阻断目标数据集加载；只有目标文件本身损坏
|
||||
才抛 BENCHMARK_DATASET_INVALID。按现有文件 stem 精确匹配，不拼接调用方传入的路径。
|
||||
"""
|
||||
for path in _dataset_files():
|
||||
raw, raw_bytes = _read_json(path)
|
||||
if raw.get("dataset_id") != dataset_id:
|
||||
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,
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
"""RAG Benchmark Runner：调用检索引擎对数据集逐 Case 求值并聚合指标。
|
||||
|
||||
只读操作，直接复用 app.retrieval.engine 的 search()，不旁路检索链路。指标按
|
||||
(mode, case, repeat) 逐样本计算，再按 mode 聚合；失败样本保留在报告中但不计入汇总，
|
||||
避免异常样本污染指标。
|
||||
(mode, case, repeat) 逐样本计算，再按 mode 聚合；失败样本按零分计入质量指标分母，
|
||||
避免把执行失败误判为检索质量（同时保留 total/successful/failed/failure_rate）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
|
||||
@@ -22,6 +23,8 @@ from app.contracts import (
|
||||
)
|
||||
from app.retrieval.engine import engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BenchmarkCancelled(Exception):
|
||||
"""运行在 Case 之间被取消时抛出，用于中断后台执行并标记 cancelled。"""
|
||||
@@ -75,12 +78,19 @@ async def _evaluate_one(
|
||||
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,
|
||||
error=str(exc),
|
||||
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]
|
||||
@@ -107,19 +117,27 @@ async def _evaluate_one(
|
||||
|
||||
def _aggregate(cases: list[RAGCaseResult], mode: SearchMode) -> RAGMetrics:
|
||||
samples = [c for c in cases if c.mode == mode]
|
||||
ok = [c for c in samples if c.error is None]
|
||||
if not ok:
|
||||
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 ok]
|
||||
# citation_hit_rate 只统计声明了 expected_block_ids 的样本
|
||||
citation_samples = [c for c in ok if c.citation_applicable]
|
||||
# 延迟只统计成功样本；失败样本按零分计入质量指标分母，避免汇总虚高
|
||||
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.hit_at_1 else 0.0 for c in ok]),
|
||||
hit_at_5=m.mean([1.0 if c.hit_at_5 else 0.0 for c in ok]),
|
||||
recall_at_k=m.mean([c.recall for c in ok]),
|
||||
mrr=m.mean([c.reciprocal_rank for c in ok]),
|
||||
citation_hit_rate=m.mean([1.0 if c.citation_hit else 0.0 for c in citation_samples]),
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -9,6 +9,7 @@ RAG Benchmark 采用「创建即返回 queued、后台 Task 异步执行」的
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from uuid import uuid4
|
||||
@@ -28,10 +29,13 @@ from app.contracts import (
|
||||
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] = {}
|
||||
@@ -45,16 +49,31 @@ def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _remember(run: BenchmarkRun) -> None:
|
||||
_runs[run.run_id] = run
|
||||
while len(_runs) > MAX_RUNS:
|
||||
oldest = next(iter(_runs))
|
||||
_runs.pop(oldest, None)
|
||||
_events.pop(oldest, None)
|
||||
_reports.pop(oldest, None)
|
||||
_tasks.pop(oldest, None)
|
||||
_subscribers.pop(oldest, None)
|
||||
_cancel_flags.pop(oldest, None)
|
||||
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:
|
||||
@@ -83,12 +102,57 @@ def _config_snapshot(request: RAGRunRequest, dataset: RAGDataset) -> dict:
|
||||
}
|
||||
|
||||
|
||||
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,
|
||||
@@ -99,7 +163,7 @@ async def create_rag_run(request: RAGRunRequest) -> BenchmarkRun:
|
||||
config_snapshot=snapshot,
|
||||
created_at=_now(),
|
||||
)
|
||||
_remember(run)
|
||||
_runs[run_id] = run
|
||||
_events[run_id] = []
|
||||
_subscribers[run_id] = []
|
||||
_cancel_flags[run_id] = asyncio.Event()
|
||||
@@ -155,6 +219,7 @@ async def _execute_rag(
|
||||
"completed_at": _now(),
|
||||
}
|
||||
)
|
||||
emit(BenchmarkEventType.run_cancelled, {"status": BenchmarkStatus.cancelled.value})
|
||||
_reports[run_id] = BenchmarkReport(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
@@ -166,15 +231,21 @@ async def _execute_rag(
|
||||
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": str(exc),
|
||||
"error": "Benchmark run failed.",
|
||||
"error_code": "BENCHMARK_RUN_FAILED",
|
||||
"completed_at": _now(),
|
||||
}
|
||||
)
|
||||
emit(BenchmarkEventType.run_failed, {"error": str(exc)})
|
||||
emit(
|
||||
BenchmarkEventType.run_failed,
|
||||
{"error": "Benchmark run failed.", "error_code": "BENCHMARK_RUN_FAILED"},
|
||||
)
|
||||
_reports[run_id] = BenchmarkReport(
|
||||
run_id=run_id,
|
||||
kind=BenchmarkKind.rag,
|
||||
@@ -182,7 +253,8 @@ async def _execute_rag(
|
||||
dataset_hash=dataset.content_hash,
|
||||
status=BenchmarkStatus.failed,
|
||||
config_snapshot=snapshot,
|
||||
error=str(exc),
|
||||
error="Benchmark run failed.",
|
||||
error_code="BENCHMARK_RUN_FAILED",
|
||||
)
|
||||
finish()
|
||||
return
|
||||
|
||||
@@ -720,6 +720,11 @@ class RAGMetrics(Contract):
|
||||
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):
|
||||
@@ -745,6 +750,7 @@ class BenchmarkRun(Contract):
|
||||
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
|
||||
@@ -760,6 +766,7 @@ class BenchmarkEventType(str, Enum):
|
||||
case_completed = "CaseCompleted"
|
||||
run_completed = "RunCompleted"
|
||||
run_failed = "RunFailed"
|
||||
run_cancelled = "RunCancelled"
|
||||
|
||||
|
||||
class BenchmarkEvent(Contract):
|
||||
@@ -785,6 +792,7 @@ class RAGCaseResult(Contract):
|
||||
# 该 Case 是否声明了 expected_block_ids（决定是否计入 citation_hit_rate 分母）
|
||||
citation_applicable: bool = False
|
||||
error: str | None = None
|
||||
error_code: str | None = None
|
||||
|
||||
|
||||
class BenchmarkReport(Contract):
|
||||
@@ -797,3 +805,4 @@ class BenchmarkReport(Contract):
|
||||
metrics: dict[str, Any] = Field(default_factory=dict)
|
||||
cases: list[RAGCaseResult] = Field(default_factory=list)
|
||||
error: str | None = None
|
||||
error_code: str | None = None
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -137,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,
|
||||
@@ -155,18 +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]
|
||||
)
|
||||
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]
|
||||
items = [self._build_result(hits[block_id], request, score) for block_id, score in ordered]
|
||||
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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
+28
-16
@@ -72,7 +72,7 @@ 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, not_implemented
|
||||
from app.errors import ApiError
|
||||
from app.extensions import ExtensionError
|
||||
from app.providers.registry import ProviderNotFoundError
|
||||
from app.providers.factory import UnsupportedProviderError
|
||||
@@ -868,18 +868,6 @@ async def create_rag_benchmark(request: RAGRunRequest) -> BenchmarkRun:
|
||||
return await benchmark_service.create_rag_run(request)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/benchmarks/agent/runs",
|
||||
response_model=BenchmarkRun,
|
||||
status_code=202,
|
||||
tags=["Benchmark"],
|
||||
)
|
||||
async def create_agent_benchmark() -> BenchmarkRun:
|
||||
# Agent Benchmark 基础设施在 RAG Benchmark 之后单独交付，先占位契约
|
||||
not_implemented("Agent Benchmark")
|
||||
raise AssertionError("unreachable")
|
||||
|
||||
|
||||
@router.get(
|
||||
"/benchmarks/runs",
|
||||
response_model=BenchmarkRunListResponse,
|
||||
@@ -945,18 +933,38 @@ async def cancel_benchmark_run(run_id: str) -> OperationResponse:
|
||||
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 = after_sequence
|
||||
last_sequence = cursor
|
||||
for event in benchmark_service.get_events(run_id):
|
||||
if event.sequence <= after_sequence:
|
||||
if event.sequence <= cursor:
|
||||
continue
|
||||
yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence)
|
||||
last_sequence = event.sequence
|
||||
@@ -969,7 +977,11 @@ async def benchmark_events(
|
||||
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):
|
||||
if event.event in (
|
||||
BenchmarkEventType.run_completed,
|
||||
BenchmarkEventType.run_failed,
|
||||
BenchmarkEventType.run_cancelled,
|
||||
):
|
||||
break
|
||||
finally:
|
||||
benchmark_service.unsubscribe(run_id, queue)
|
||||
|
||||
Reference in New Issue
Block a user