GuangYu/QuestionProject
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持 Agent 并发批测与洞察导出

Browse Source
This commit is contained in:
Nelson
2026-08-13 17:25:22 +08:00
parent 3faecab6ac
commit f952d6dc58
22 changed files with 666 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
ai_knowledge_base_v2/apps/backend/app/services/agent_batch_test_service.py
View File

@@ -19,6 +19,7 @@ from app.models.knowledge import Knowledge
MAX_BATCH_QUESTIONS = 200
MAX_QUESTION_LENGTH = 2000
MAX_JOB_CONCURRENCY = 10
TEMPLATE_HEADERS = ["序号", "问题"]
@@ -105,10 +106,25 @@ class AgentBatchTestService:
"reasoningVisible": 0,
"responseDepth": int(config.get("responseDepth", 35)),
}
execution_mode = str(config.get("executionMode") or "sequential")
if execution_mode not in {"sequential", "concurrent"}:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="批量测试执行模式不正确")
try:
requested_concurrency = int(config.get("concurrencyLimit") or 5)
except (TypeError, ValueError) as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="并发数格式不正确") from exc
concurrency_limit = 1 if execution_mode == "sequential" else requested_concurrency
if execution_mode == "concurrent" and not 2 <= concurrency_limit <= MAX_JOB_CONCURRENCY:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"单个任务并发数必须在 2 到 {MAX_JOB_CONCURRENCY} 之间",
)
job = AgentBatchTest(
name=str(config.get("name") or filename.rsplit(".", 1)[0] or "Agent 批量测试")[:160],
original_filename=filename[:255],
total_count=len(rows),
execution_mode=execution_mode,
concurrency_limit=concurrency_limit,
prompt_content=prompt_content,
model_id=model.id,
model_name=model.display_name or model.model_name,
@@ -183,13 +199,14 @@ class AgentBatchTestService:
summary.append(["任务名称", job.name])
summary.append(["任务状态", job.status])
summary.append(["问题总数", job.total_count])
summary.append(["执行方式", "逐条串行" if job.execution_mode == "sequential" else f"并发测试（{job.concurrency_limit} 路）"])
summary.append(["成功", job.success_count])
summary.append(["失败", job.failed_count])
summary.append(["测试模型", job.model_name])
summary.append(["知识库", "、".join(json.loads(job.knowledge_names or "[]"))])
summary.append(["创建时间", job.created_at])
summary.append(["完成时间", job.finished_at])
_style_sheet(summary, widths=(20, 86), table_ref="A1:B10", table_name="AgentBatchSummary")
_style_sheet(summary, widths=(20, 86), table_ref="A1:B11", table_name="AgentBatchSummary")
summary.column_dimensions["B"].width = 86
stream = BytesIO()
workbook.save(stream)
@@ -205,6 +222,8 @@ def job_dict(job: AgentBatchTest) -> dict:
"status": job.status,
"totalCount": job.total_count,
"processedCount": job.processed_count,
"executionMode": job.execution_mode,
"concurrencyLimit": job.concurrency_limit,
"successCount": job.success_count,
"failedCount": job.failed_count,
"progress": round(job.processed_count * 100 / job.total_count) if job.total_count else 0,

117
ai_knowledge_base_v2/apps/backend/app/services/agent_batch_test_worker.py
View File

@@ -33,13 +33,43 @@ class AgentBatchTestWorker:
logger.info("agent batch test worker disabled")
return
worker_id = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
active: set[asyncio.Task] = set()
worker_concurrency = max(1, min(settings.agent_batch_worker_concurrency, 50))
next_recovery_at = 0.0
while True:
try:
processed = await cls.run_once(worker_id)
if time.monotonic() >= next_recovery_at:
with SessionLocal() as db:
recovered_batch_ids = cls.recover_stale_items(db)
db.commit()
for batch_id in recovered_batch_ids:
cls.refresh_job(db, batch_id)
next_recovery_at = time.monotonic() + 60
active = {task for task in active if not task.done()}
claimed = False
while len(active) < worker_concurrency:
with SessionLocal() as db:
item_id = cls.claim_next(db, worker_id=worker_id)
if item_id is None:
break
claimed = True
task = asyncio.create_task(cls.execute_claimed(item_id=item_id, worker_id=worker_id))
task.add_done_callback(cls._log_task_failure)
active.add(task)
if active:
await asyncio.wait(
active,
timeout=max(1, settings.agent_batch_poll_seconds),
return_when=asyncio.FIRST_COMPLETED,
)
processed = True
else:
processed = claimed
except Exception:
processed = False
logger.exception("agent batch test worker iteration failed")
await asyncio.sleep(0 if processed else max(1, settings.agent_batch_poll_seconds))
if not processed:
await asyncio.sleep(max(1, settings.agent_batch_poll_seconds))
@classmethod
async def run_once(cls, worker_id: str) -> bool:
@@ -59,35 +89,64 @@ class AgentBatchTestWorker:
@staticmethod
def claim_next(db: Session, *, worker_id: str, now: datetime | None = None) -> int | None:
current = now or _now()
item = db.scalar(
select(AgentBatchTestItem)
.join(AgentBatchTest, AgentBatchTest.id == AgentBatchTestItem.batch_id)
.where(
AgentBatchTest.status.in_(("pending", "running")),
AgentBatchTestItem.status == "pending",
AgentBatchTestItem.attempt_count < AgentBatchTestItem.max_attempts,
or_(AgentBatchTestItem.next_run_at.is_(None), AgentBatchTestItem.next_run_at <= current),
jobs = list(
db.scalars(
select(AgentBatchTest)
.where(AgentBatchTest.status.in_(("pending", "running")))
.order_by(AgentBatchTest.created_at.asc(), AgentBatchTest.id.asc())
.limit(50)
.with_for_update(skip_locked=True)
)
.order_by(AgentBatchTestItem.batch_id.asc(), AgentBatchTestItem.row_number.asc())
.with_for_update(skip_locked=True)
.limit(1)
)
if item is None:
db.rollback()
return None
job = db.get(AgentBatchTest, item.batch_id)
item.status = "running"
item.attempt_count += 1
item.locked_at = current
item.locked_by = worker_id
item.started_at = current
if job is not None and job.status == "pending":
job.status = "running"
job.started_at = current
db.add(job)
db.add(item)
db.commit()
return item.id
for job in jobs:
running_count = db.scalar(
select(func.count(AgentBatchTestItem.id)).where(
AgentBatchTestItem.batch_id == job.id,
AgentBatchTestItem.status == "running",
)
) or 0
job_limit = 1 if job.execution_mode == "sequential" else max(1, job.concurrency_limit)
if running_count >= job_limit:
continue
item = db.scalar(
select(AgentBatchTestItem)
.where(
AgentBatchTestItem.batch_id == job.id,
AgentBatchTestItem.status == "pending",
AgentBatchTestItem.attempt_count < AgentBatchTestItem.max_attempts,
or_(AgentBatchTestItem.next_run_at.is_(None), AgentBatchTestItem.next_run_at <= current),
)
.order_by(AgentBatchTestItem.row_number.asc(), AgentBatchTestItem.id.asc())
.with_for_update(skip_locked=True)
.limit(1)
)
if item is None:
continue
item.status = "running"
item.attempt_count += 1
item.locked_at = current
item.locked_by = worker_id
item.started_at = current
if job.status == "pending":
job.status = "running"
job.started_at = current
db.add(job)
db.add(item)
db.commit()
return item.id
db.rollback()
return None
@staticmethod
def _log_task_failure(task: asyncio.Task) -> None:
if task.cancelled():
return
error = task.exception()
if error is not None:
logger.error(
"agent batch test execution task failed",
exc_info=(type(error), error, error.__traceback__),
)
@classmethod
async def execute_claimed(cls, *, item_id: int, worker_id: str) -> None:

167
ai_knowledge_base_v2/apps/backend/app/services/question_insight_export_service.py Normal file
View File

@@ -0,0 +1,167 @@
from __future__ import annotations
from datetime import datetime
from io import BytesIO
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.worksheet.table import Table, TableStyleInfo
class QuestionInsightExportService:
"""Render a complete question-insight snapshot as an operator-friendly workbook."""
@staticmethod
def build(result: dict) -> BytesIO:
workbook = Workbook()
insight_sheet = workbook.active
insight_sheet.title = "洞察结果"
_append_insights(insight_sheet, result.get("items") or [])
sample_sheet = workbook.create_sheet("相似问法与样例")
_append_samples(sample_sheet, result.get("items") or [])
summary_sheet = workbook.create_sheet("统计说明")
_append_summary(summary_sheet, result)
stream = BytesIO()
workbook.save(stream)
stream.seek(0)
return stream
def _append_insights(sheet, items: list[dict]) -> None:
headers = [
"排名", "问题组", "分类", "出现次数", "用户数", "会话数", "关联AI请求",
"无知识命中", "失败请求", "需补知识/查召回", "建议动作", "高频词",
"首次出现", "最后出现",
]
sheet.append(headers)
for item in items:
sheet.append([
item.get("rank"),
_excel_safe(item.get("title")),
_excel_safe(item.get("categoryLabel")),
item.get("count", 0),
item.get("userCount", 0),
item.get("sessionCount", 0),
item.get("aiRequestCount", 0),
item.get("noHitCount", 0),
item.get("failedCount", 0),
"是" if item.get("needsKnowledgeFollowUp") else "否",
_excel_safe(item.get("suggestedAction")),
_excel_safe("、".join(item.get("topTerms") or [])),
_excel_datetime(item.get("firstSeenAt")),
_excel_datetime(item.get("lastSeenAt")),
])
_style_table(
sheet,
widths=(8, 44, 14, 12, 10, 10, 14, 12, 12, 18, 54, 36, 20, 20),
table_name="QuestionInsightResults",
)
for row in sheet.iter_rows(min_row=2):
for index in (1, 10, 11):
row[index].alignment = Alignment(vertical="top", wrap_text=True)
for index in (12, 13):
row[index].number_format = "yyyy-mm-dd hh:mm:ss"
def _append_samples(sheet, items: list[dict]) -> None:
headers = [
"问题组排名", "问题组", "记录类型", "相似问法/清洗后问题", "出现次数",
"用户", "手机号", "用户ID", "会话ID", "消息ID", "原始问题", "提问时间",
]
sheet.append(headers)
for item in items:
for variant in item.get("variants") or []:
sheet.append([
item.get("rank"), _excel_safe(item.get("title")), "相似问法",
_excel_safe(variant.get("text")), variant.get("count", 0),
"", "", "", "", "", "", "",
])
for sample in item.get("samples") or []:
sheet.append([
item.get("rank"), _excel_safe(item.get("title")), "原始样例",
_excel_safe(sample.get("cleaned")), "",
_excel_safe(sample.get("userName")), _excel_safe(sample.get("userPhone")),
sample.get("userId"), sample.get("sessionId"), sample.get("messageId"),
_excel_safe(sample.get("raw")), _excel_datetime(sample.get("createdAt")),
])
_style_table(
sheet,
widths=(12, 38, 12, 46, 12, 16, 18, 12, 12, 12, 72, 20),
table_name="QuestionInsightSamples",
)
for row in sheet.iter_rows(min_row=2):
for index in (1, 3, 10):
row[index].alignment = Alignment(vertical="top", wrap_text=True)
row[11].number_format = "yyyy-mm-dd hh:mm:ss"
def _append_summary(sheet, result: dict) -> None:
range_info = result.get("range") or {}
summary = result.get("summary") or {}
sheet.append(["项目", "内容"])
rows = [
("开始时间", range_info.get("dateFrom") or "未限制"),
("结束时间", range_info.get("dateTo") or "未限制"),
("单次清洗上限", range_info.get("maxMessages", 0)),
("最低频次", summary.get("minCount", 0)),
("纳入清洗消息", summary.get("scannedMessages", 0)),
("有效问题", summary.get("cleanedQuestions", 0)),
("过滤低价值", summary.get("filteredMessages", 0)),
("全部问题组", summary.get("clusterCount", 0)),
("导出问题组", summary.get("visibleClusterCount", 0)),
("清洗规则版本", summary.get("cleanerVersion", "")),
("导出时间", datetime.now()),
("说明", "导出结果按所选日期范围和最低频次生成，包含全部符合条件的问题组，不受页面分页影响。"),
]
for label, value in rows:
sheet.append([label, _excel_safe(value) if isinstance(value, str) else _excel_datetime(value)])
_style_table(sheet, widths=(22, 88), table_name="QuestionInsightSummary")
for row in sheet.iter_rows(min_row=2):
row[1].alignment = Alignment(vertical="top", wrap_text=True)
if isinstance(row[1].value, datetime):
row[1].number_format = "yyyy-mm-dd hh:mm:ss"
def _style_table(sheet, *, widths: tuple[int, ...], table_name: str) -> None:
sheet.freeze_panes = "A2"
sheet.sheet_view.showGridLines = False
header_fill = PatternFill("solid", fgColor="176B57")
for cell in sheet[1]:
cell.fill = header_fill
cell.font = Font(color="FFFFFF", bold=True)
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
sheet.row_dimensions[1].height = 30
for index, width in enumerate(widths, start=1):
sheet.column_dimensions[_column_letter(index)].width = width
if sheet.max_row > 1:
ref = f"A1:{_column_letter(len(widths))}{sheet.max_row}"
table = Table(displayName=table_name, ref=ref)
table.tableStyleInfo = TableStyleInfo(
name="TableStyleMedium4",
showRowStripes=True,
showFirstColumn=False,
showLastColumn=False,
)
sheet.add_table(table)
def _column_letter(index: int) -> str:
result = ""
while index:
index, remainder = divmod(index - 1, 26)
result = chr(65 + remainder) + result
return result
def _excel_safe(value: object | None) -> str:
text = str(value or "")
return f"'{text}" if text.lstrip().startswith(("=", "+", "-", "@")) else text
def _excel_datetime(value: object) -> object:
if isinstance(value, datetime) and value.tzinfo is not None:
return value.replace(tzinfo=None)
return value