solahqb/AgentEvalTool
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

perf(eval): case/rule/campaign 并发执行

Browse Source 
- 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 调用的等待重叠。
This commit is contained in:
sinohqb 2026-08-24 23:18:11 +08:00
parent ca208232c7
commit bf1ec16ef6
4 changed files with 105 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
backend/agenteval/config/settings.py
View File

@ -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(

23
backend/agenteval/evaluation/campaign_runner.py
View File

@ -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

107
backend/agenteval/evaluation/engine.py
View File

@ -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 ────────────────────────────────────────────────────────

2
backend/agenteval/services/runs.py
View File

@ -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)