From bf1ec16ef6247567485b9c18cc3417015565ecb8 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 24 Aug 2026 23:18:11 +0800 Subject: [PATCH] =?UTF-8?q?perf(eval):=20case/rule/campaign=20=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E6=89=A7=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - EvalEngine case 循环改 asyncio.gather + semaphore（默认 3 并发） - 规则评估并行（默认 5 并发），LLM 评分耗时从串行求和降为最慢一条 - _resolve_model 改 async + 双重检查锁，保护共享模型缓存 - _case_errors / case_outcomes 并发写入加状态锁 - Campaign occurrence 派生并行（默认 2 并发），claim 拒绝时提前收敛 - 新增配置：max_concurrent_cases=3 / max_concurrent_rules=5 / max_concurrent_runs=2 SQLite StaticPool 单连接下 DB 写入仍天然串行，并行收益集中在 channel I/O 与 LLM 调用的等待重叠。 --- backend/agenteval/config/settings.py | 12 ++ .../agenteval/evaluation/campaign_runner.py | 23 +++- backend/agenteval/evaluation/engine.py | 107 ++++++++++++------ backend/agenteval/services/runs.py | 2 + 4 files changed, 105 insertions(+), 39 deletions(-) diff --git a/backend/agenteval/config/settings.py b/backend/agenteval/config/settings.py index f1481df..3eaa47c 100644 --- a/backend/agenteval/config/settings.py +++ b/backend/agenteval/config/settings.py @@ -81,6 +81,18 @@ class Settings(BaseSettings): default=30.0, description="Seconds to wait for the target's reply per turn before recording no-reply.", ) + max_concurrent_cases: int = Field( + default=3, + description="Max cases running concurrently within a single eval run.", + ) + max_concurrent_rules: int = Field( + default=5, + description="Max rules evaluated concurrently within a single case.", + ) + max_concurrent_runs: int = Field( + default=2, + description="Max child Runs spawned concurrently per campaign plan entry.", + ) # ── Webhook ──────────────────────────────────────────────────── webhook_url: Optional[str] = Field( diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 79d672b..8627c52 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -21,6 +21,7 @@ from typing import Awaitable, Callable, Optional from sqlmodel import Session +from agenteval.config import get_settings from agenteval.evaluation.campaign_lifecycle import complete_campaign from agenteval.evaluation.campaign_lifecycle import start_campaign as start_campaign_lifecycle from agenteval.evaluation.campaign_scheduler import ( @@ -121,7 +122,12 @@ async def _execute_child_run( target = TargetRepository(session).get(campaign.target_id) if not target: raise ValueError(f"campaign target missing: {campaign.target_id}") - engine = EvalEngine(target=target, scenario=scenario, triggered_by=RunTrigger.CAMPAIGN) + engine = EvalEngine( + target=target, + scenario=scenario, + triggered_by=RunTrigger.CAMPAIGN, + max_concurrent_cases=get_settings().max_concurrent_cases, + ) await engine.run(existing_run=run) @@ -194,14 +200,18 @@ async def _advance_campaign( result = AdvanceResult(finished=decision.finished) errors: list[dict] = list(campaign.summary.scheduler.errors) if campaign.summary else [] + run_semaphore = asyncio.Semaphore(max(1, get_settings().max_concurrent_runs)) for due in decision.due: # Cancellation stops further spawning; runs already in flight finish. if cancel_event is not None and cancel_event.is_set(): break claim_rejected = False - try: - for occurrence_index in range(due.entry.count): - run_id = await _spawn_child_run( + + async def _spawn_one(occurrence_index: int) -> Optional[str]: + async with run_semaphore: + if claim_rejected: + return None + return await _spawn_child_run( campaign, due.entry.scenario_id, plan_index=due.index, @@ -209,6 +219,11 @@ async def _advance_campaign( session=session, execute_child_run=execute_child_run, ) + + try: + tasks = [_spawn_one(i) for i in range(due.entry.count)] + run_ids = await asyncio.gather(*tasks) + for run_id in run_ids: if run_id is None: claim_rejected = True break diff --git a/backend/agenteval/evaluation/engine.py b/backend/agenteval/evaluation/engine.py index d161bcc..94d6059 100644 --- a/backend/agenteval/evaluation/engine.py +++ b/backend/agenteval/evaluation/engine.py @@ -114,6 +114,10 @@ class EvalEngine: ) self.triggered_by = triggered_by self._case_semaphore = asyncio.Semaphore(max(1, max_concurrent_cases)) + self._rule_semaphore = asyncio.Semaphore( + max(1, get_settings().max_concurrent_rules), + ) + self._state_lock = asyncio.Lock() # Collects fatal case-level errors (e.g. dynamic message generation # failures) so their cause is persisted into run.summary — not just # emitted transiently over WebSocket. @@ -155,7 +159,7 @@ class EvalEngine: total_cases = len(self.scenario.cases) case_outcomes: dict[str, CaseOutcome] = {} - for idx, case in enumerate(self.scenario.cases, start=1): + async def _case_worker(idx: int, case: Case) -> None: self._check_cancel() await self._emit( progress_callback, @@ -172,7 +176,8 @@ class EvalEngine: case, progress_callback, ) - case_outcomes[case.id] = outcome + async with self._state_lock: + case_outcomes[case.id] = outcome await self._emit( progress_callback, "case_end", @@ -186,6 +191,21 @@ class EvalEngine: }, ) + workers = [ + _case_worker(idx, case) + for idx, case in enumerate(self.scenario.cases, start=1) + ] + try: + await asyncio.gather(*workers) + except Exception: + raise + + async with self._state_lock: + resolved_snapshot = { + purpose.value: config.snapshot() + for purpose, config in self._resolved_models.items() + } + results = self.run_repo.get_results(run.id) turns = self.run_repo.get_turns(run.id) summary = build_run_summary( @@ -193,11 +213,7 @@ class EvalEngine: latencies=[t.latency_ms for t in turns if t.latency_ms is not None], rule_passes=[r.passed for r in results], case_errors=self._case_errors or None, - model_configs=( - {purpose.value: config.snapshot() for purpose, config in self._resolved_models.items()} - if self._resolved_models - else None - ), + model_configs=resolved_snapshot or None, ) run.status = RunStatus.COMPLETED run.completed_at = utc_now() @@ -445,19 +461,36 @@ class EvalEngine: implicit_outcomes: list[RuleOutcome] = [] all_rules = [(cfg, False) for cfg in rules_config] + [(cfg, True) for cfg in implicit_config] - for rule_config, is_implicit in all_rules: - purpose = RULE_PURPOSE.get(rule_config.type) - try: - model_config = self._resolve_model(purpose) if purpose else None - rule = get_rule( - rule_config.type, - rule_config.params, - model_config=model_config, - gateway=self.model_gateway if model_config else None, - ) - result = await rule.evaluate(case, dialog) - except Exception as exc: - result = RuleResult(passed=False, reason=f"模型配置解析失败: {exc}") + + async def _eval_one( + rule_config: EvalRuleConfig, + is_implicit: bool, + ) -> tuple[EvalRuleConfig, bool, RuleResult]: + async with self._rule_semaphore: + purpose = RULE_PURPOSE.get(rule_config.type) + try: + model_config = await self._resolve_model(purpose) if purpose else None + rule = get_rule( + rule_config.type, + rule_config.params, + model_config=model_config, + gateway=self.model_gateway if model_config else None, + ) + result = await rule.evaluate(case, dialog) + except Exception as exc: + result = RuleResult(passed=False, reason=f"模型配置解析失败: {exc}") + return rule_config, is_implicit, result + + rule_results = await asyncio.gather( + *[_eval_one(cfg, is_impl) for cfg, is_impl in all_rules] + ) + + passed_count = 0 + total_count = 0 + explicit_outcomes: list[RuleOutcome] = [] + implicit_outcomes: list[RuleOutcome] = [] + + for rule_config, is_implicit, result in rule_results: reason = f"[期望] {result.reason}" if is_implicit else result.reason eval_result = EvalResult( id=str(uuid.uuid4()), @@ -513,7 +546,8 @@ class EvalEngine: async def _fail(msg: str) -> list[str]: # Persist the reason into run-level case_errors (surfaced in summary), # not just a transient WebSocket emit that's lost after the run. - self._case_errors.append({"case_id": case.id, "stage": "generate_messages", "error": msg}) + async with self._state_lock: + self._case_errors.append({"case_id": case.id, "stage": "generate_messages", "error": msg}) await self._emit(progress_callback, "error", {"error": msg, "case_id": case.id}) return [] @@ -533,7 +567,7 @@ class EvalEngine: ] try: - model_config = self._resolve_model(ModelPurpose.GENERATOR) + model_config = await self._resolve_model(ModelPurpose.GENERATOR) if model_config: content = await self.model_gateway.chat(model_config, messages_payload, temperature=0.7) else: @@ -589,23 +623,26 @@ class EvalEngine: raise ValueError("LLM 返回内容为空或无法解析") return content - def _resolve_model(self, purpose: ModelPurpose | None) -> ModelRuntimeConfig | None: + async def _resolve_model(self, purpose: ModelPurpose | None) -> ModelRuntimeConfig | None: if purpose is None: return None if purpose in self._resolved_models: return self._resolved_models[purpose] - config_id = self.scenario.model_bindings.get(purpose) - if not config_id: - return None - expected = { - ModelPurpose.GENERATOR: ModelCapability.CHAT, - ModelPurpose.JUDGE: ModelCapability.CHAT, - ModelPurpose.EMBEDDING: ModelCapability.EMBEDDING, - ModelPurpose.MODERATION: ModelCapability.MODERATION, - }[purpose] - runtime = self.model_service.resolve(config_id, expected) - self._resolved_models[purpose] = runtime - return runtime + async with self._state_lock: + if purpose in self._resolved_models: + return self._resolved_models[purpose] + config_id = self.scenario.model_bindings.get(purpose) + if not config_id: + return None + expected = { + ModelPurpose.GENERATOR: ModelCapability.CHAT, + ModelPurpose.JUDGE: ModelCapability.CHAT, + ModelPurpose.EMBEDDING: ModelCapability.EMBEDDING, + ModelPurpose.MODERATION: ModelCapability.MODERATION, + }[purpose] + runtime = self.model_service.resolve(config_id, expected) + self._resolved_models[purpose] = runtime + return runtime # ── helpers ──────────────────────────────────────────────────────── diff --git a/backend/agenteval/services/runs.py b/backend/agenteval/services/runs.py index cb980d5..de02087 100644 --- a/backend/agenteval/services/runs.py +++ b/backend/agenteval/services/runs.py @@ -10,6 +10,7 @@ import asyncio from sqlmodel import Session +from agenteval.config import get_settings from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts from agenteval.evaluation.engine import EvalEngine from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger @@ -61,6 +62,7 @@ async def execute_run( scenario=scenario, session=session, cancel_token=cancel_token, + max_concurrent_cases=get_settings().max_concurrent_cases, ) await engine.run(progress_callback=on_progress, existing_run=existing_run) # Fire webhook after run completes (non-blocking, best-effort)