diff --git a/backend/app.py b/backend/app.py index c0dc2ac..3d795b5 100644 --- a/backend/app.py +++ b/backend/app.py @@ -27,6 +27,10 @@ from db.session import sync_default_tools, sync_interface_definitions from services.knowledge import recover_interrupted_documents from services.post_call.worker import recover_interrupted_analyses, run_analysis_worker from services.webhooks.worker import recover_interrupted_webhooks, run_webhook_worker +from services.test_runs.orchestrator import ( + recover_interrupted_test_runs, + test_run_orchestrator, +) from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -41,6 +45,8 @@ from routes import ( model_registry, node_types, site_settings, + test_cases, + test_runs, tools, voice_webrtc, voice_ws, @@ -54,6 +60,7 @@ async def lifespan(_app: FastAPI): await recover_interrupted_documents() await recover_interrupted_analyses() await recover_interrupted_webhooks() + await recover_interrupted_test_runs() analysis_worker = asyncio.create_task( run_analysis_worker(), name="post-call-analysis-worker" ) @@ -68,6 +75,7 @@ async def lifespan(_app: FastAPI): await asyncio.gather( analysis_worker, webhook_worker, return_exceptions=True ) + await test_run_orchestrator.shutdown() await voice_webrtc.shutdown_active_sessions() @@ -91,6 +99,8 @@ app.include_router(mcp_servers.router) app.include_router(model_registry.router) app.include_router(node_types.router) app.include_router(site_settings.router) +app.include_router(test_cases.router) +app.include_router(test_runs.router) app.include_router(tools.router) app.include_router(voice_webrtc.router) app.include_router(voice_ws.router) diff --git a/backend/db/models.py b/backend/db/models.py index 0361785..4230dc0 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -422,3 +422,115 @@ class WebhookDelivery(Base): updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now() ) + + +class TestSuite(Base): + """A persisted group of text-based assistant test cases.""" + + __tablename__ = "test_suites" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + name: Mapped[str] = mapped_column(String(128)) + description: Mapped[str] = mapped_column(String(2048), default="") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class TestCase(Base): + """One editable test definition; nested turns remain one JSON document.""" + + __tablename__ = "test_cases" + __table_args__ = ( + UniqueConstraint("suite_id", "sort_order", name="uq_test_case_suite_order"), + ) + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + suite_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("test_suites.id", ondelete="CASCADE"), + index=True, + ) + name: Mapped[str] = mapped_column(String(128)) + description: Mapped[str] = mapped_column(String(2048), default="") + input_mode: Mapped[str] = mapped_column( + String(40), default="fixed_script_text", index=True + ) + definition: Mapped[dict] = mapped_column(JSONB, default=dict) + sort_order: Mapped[int] = mapped_column(Integer, default=0) + last_result: Mapped[str] = mapped_column(String(16), default="not_run") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class TestRun(Base): + """One immutable batch-run request plus its aggregate lifecycle state.""" + + __tablename__ = "test_runs" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + assistant_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("assistants.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + assistant_name: Mapped[str] = mapped_column(String(128), default="") + title: Mapped[str] = mapped_column(String(256), default="批量测试") + status: Mapped[str] = mapped_column(String(16), index=True, default="queued") + config: Mapped[dict] = mapped_column(JSONB, default=dict) + stop_reason: Mapped[str | None] = mapped_column(String(32), nullable=True) + cancel_requested: Mapped[bool] = mapped_column(Boolean, default=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + +class TestRunCase(Base): + """Run-local test case snapshot and detailed execution result.""" + + __tablename__ = "test_run_cases" + __table_args__ = ( + UniqueConstraint("run_id", "test_case_id", name="uq_test_run_case"), + ) + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + run_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("test_runs.id", ondelete="CASCADE"), + index=True, + ) + test_case_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("test_cases.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + test_case_name: Mapped[str] = mapped_column(String(128)) + suite_id: Mapped[str | None] = mapped_column(String(40), nullable=True) + position: Mapped[int] = mapped_column(Integer, default=0) + case_snapshot: Mapped[dict] = mapped_column(JSONB, default=dict) + status: Mapped[str] = mapped_column(String(16), index=True, default="waiting") + attempt_count: Mapped[int] = mapped_column(Integer, default=0) + max_attempts: Mapped[int] = mapped_column(Integer, default=1) + result: Mapped[dict] = mapped_column(JSONB, default=dict) + execution_error: Mapped[dict | None] = mapped_column(JSONB, nullable=True) + started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/backend/migrations/versions/20260810_0016_add_test_runs.py b/backend/migrations/versions/20260810_0016_add_test_runs.py new file mode 100644 index 0000000..fd6cede --- /dev/null +++ b/backend/migrations/versions/20260810_0016_add_test_runs.py @@ -0,0 +1,112 @@ +"""add persisted test suites, cases, and batch runs + +Revision ID: 20260810_0016 +Revises: 20260807_0015 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "20260810_0016" +down_revision: str | Sequence[str] | None = "20260807_0015" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "test_suites", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("name", sa.String(length=128), nullable=False), + sa.Column("description", sa.String(length=2048), server_default="", nullable=False), + sa.Column("assistant_id", sa.String(length=40), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["assistant_id"], ["assistants.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_test_suites_assistant_id", "test_suites", ["assistant_id"]) + + op.create_table( + "test_cases", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("suite_id", sa.String(length=40), nullable=False), + sa.Column("name", sa.String(length=128), nullable=False), + sa.Column("description", sa.String(length=2048), server_default="", nullable=False), + sa.Column("input_mode", sa.String(length=40), server_default="fixed_script_text", nullable=False), + sa.Column("definition", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.Column("sort_order", sa.Integer(), server_default="0", nullable=False), + sa.Column("last_result", sa.String(length=16), server_default="not_run", nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["suite_id"], ["test_suites.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("suite_id", "sort_order", name="uq_test_case_suite_order"), + ) + op.create_index("ix_test_cases_suite_id", "test_cases", ["suite_id"]) + op.create_index("ix_test_cases_input_mode", "test_cases", ["input_mode"]) + + op.create_table( + "test_runs", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("assistant_id", sa.String(length=40), nullable=True), + sa.Column("assistant_name", sa.String(length=128), server_default="", nullable=False), + sa.Column("title", sa.String(length=256), server_default="批量测试", nullable=False), + sa.Column("status", sa.String(length=16), server_default="queued", nullable=False), + sa.Column("config", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.Column("stop_reason", sa.String(length=32), nullable=True), + sa.Column("cancel_requested", sa.Boolean(), server_default=sa.text("false"), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["assistant_id"], ["assistants.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_test_runs_assistant_id", "test_runs", ["assistant_id"]) + op.create_index("ix_test_runs_status", "test_runs", ["status"]) + + op.create_table( + "test_run_cases", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("run_id", sa.String(length=40), nullable=False), + sa.Column("test_case_id", sa.String(length=40), nullable=True), + sa.Column("test_case_name", sa.String(length=128), nullable=False), + sa.Column("suite_id", sa.String(length=40), nullable=True), + sa.Column("position", sa.Integer(), server_default="0", nullable=False), + sa.Column("case_snapshot", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.Column("status", sa.String(length=16), server_default="waiting", nullable=False), + sa.Column("attempt_count", sa.Integer(), server_default="0", nullable=False), + sa.Column("max_attempts", sa.Integer(), server_default="1", nullable=False), + sa.Column("result", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.Column("execution_error", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["run_id"], ["test_runs.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["test_case_id"], ["test_cases.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("run_id", "test_case_id", name="uq_test_run_case"), + ) + op.create_index("ix_test_run_cases_run_id", "test_run_cases", ["run_id"]) + op.create_index("ix_test_run_cases_test_case_id", "test_run_cases", ["test_case_id"]) + op.create_index("ix_test_run_cases_status", "test_run_cases", ["status"]) + + +def downgrade() -> None: + op.drop_index("ix_test_run_cases_status", table_name="test_run_cases") + op.drop_index("ix_test_run_cases_test_case_id", table_name="test_run_cases") + op.drop_index("ix_test_run_cases_run_id", table_name="test_run_cases") + op.drop_table("test_run_cases") + op.drop_index("ix_test_runs_status", table_name="test_runs") + op.drop_index("ix_test_runs_assistant_id", table_name="test_runs") + op.drop_table("test_runs") + op.drop_index("ix_test_cases_input_mode", table_name="test_cases") + op.drop_index("ix_test_cases_suite_id", table_name="test_cases") + op.drop_table("test_cases") + op.drop_index("ix_test_suites_assistant_id", table_name="test_suites") + op.drop_table("test_suites") diff --git a/backend/migrations/versions/20260810_0017_remove_test_suite_assistant.py b/backend/migrations/versions/20260810_0017_remove_test_suite_assistant.py new file mode 100644 index 0000000..126415b --- /dev/null +++ b/backend/migrations/versions/20260810_0017_remove_test_suite_assistant.py @@ -0,0 +1,43 @@ +"""remove obsolete assistant binding from test suites + +Revision ID: 20260810_0017 +Revises: 20260810_0016 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260810_0017" +down_revision: str | Sequence[str] | None = "20260810_0016" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.drop_index("ix_test_suites_assistant_id", table_name="test_suites") + op.drop_column("test_suites", "assistant_id") + + +def downgrade() -> None: + op.add_column( + "test_suites", + sa.Column("assistant_id", sa.String(length=40), nullable=True), + ) + op.create_foreign_key( + "fk_test_suites_assistant_id_assistants", + "test_suites", + "assistants", + ["assistant_id"], + ["id"], + ondelete="SET NULL", + ) + op.create_index( + "ix_test_suites_assistant_id", + "test_suites", + ["assistant_id"], + ) diff --git a/backend/routes/test_cases.py b/backend/routes/test_cases.py new file mode 100644 index 0000000..1a48b1c --- /dev/null +++ b/backend/routes/test_cases.py @@ -0,0 +1,369 @@ +"""Persistent test suite and fixed-text test case CRUD.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from uuid import uuid4 + +from db.models import TestCase, TestRunCase, TestSuite +from db.session import get_session +from fastapi import APIRouter, Depends, HTTPException, Query +from services.auth import require_admin +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession +from test_schemas import ( + TestCaseBulkDeleteIn, + TestCaseDefinition, + TestCaseOrderIn, + TestCaseOut, + TestCaseWrite, + TestSuiteCreate, + TestSuiteOut, + TestSuiteUpdate, +) + + +router = APIRouter( + tags=["tests"], + dependencies=[Depends(require_admin)], +) + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +async def _touch_suite(session: AsyncSession, suite_id: str) -> None: + suite = await session.get(TestSuite, suite_id) + if suite is not None: + suite.updated_at = datetime.now(UTC) + + +async def _suite_out(session: AsyncSession, row: TestSuite) -> TestSuiteOut: + total = await session.scalar( + select(func.count()).select_from(TestCase).where(TestCase.suite_id == row.id) + ) + run = await session.scalar( + select(func.count(func.distinct(TestRunCase.run_id))).where( + TestRunCase.suite_id == row.id + ) + ) + passed = await session.scalar( + select(func.count()) + .select_from(TestCase) + .where(TestCase.suite_id == row.id, TestCase.last_result == "pass") + ) + return TestSuiteOut( + id=row.id, + name=row.name, + description=row.description, + case_count=int(total or 0), + passed_count=int(passed or 0), + run_count=int(run or 0), + updated_at=row.updated_at, + ) + + +def _case_out(row: TestCase) -> TestCaseOut: + definition = TestCaseDefinition.model_validate(row.definition or {}) + return TestCaseOut( + id=row.id, + suite_id=row.suite_id, + name=row.name, + description=row.description, + input_mode=row.input_mode, + last_result=row.last_result, + sort_order=row.sort_order, + updated_at=row.updated_at, + **definition.model_dump(), + ) + + +def _apply_case_write(row: TestCase, body: TestCaseWrite) -> None: + row.name = body.name.strip() + row.description = body.description.strip() + row.input_mode = body.input_mode + row.definition = body.definition().model_dump(mode="json", by_alias=True) + + +async def _renumber_cases(session: AsyncSession, suite_id: str) -> None: + rows = ( + await session.execute( + select(TestCase) + .where(TestCase.suite_id == suite_id) + .order_by(TestCase.sort_order, TestCase.created_at, TestCase.id) + ) + ).scalars().all() + offset = len(rows) + 1 + for index, row in enumerate(rows): + row.sort_order = offset + index + await session.flush() + for index, row in enumerate(rows): + row.sort_order = index + + +@router.get("/api/test-suites", response_model=list[TestSuiteOut]) +async def list_test_suites(session: AsyncSession = Depends(get_session)): + rows = ( + await session.execute(select(TestSuite).order_by(TestSuite.updated_at.desc())) + ).scalars().all() + return [await _suite_out(session, row) for row in rows] + + +@router.post("/api/test-suites", response_model=TestSuiteOut) +async def create_test_suite( + body: TestSuiteCreate, + session: AsyncSession = Depends(get_session), +): + row = TestSuite( + id=_new_id("suite"), + name=body.name.strip(), + description=body.description.strip(), + ) + session.add(row) + await session.commit() + await session.refresh(row) + return await _suite_out(session, row) + + +@router.get("/api/test-suites/{suite_id}", response_model=TestSuiteOut) +async def get_test_suite( + suite_id: str, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestSuite, suite_id) + if row is None: + raise HTTPException(404, "测试集不存在") + return await _suite_out(session, row) + + +@router.put("/api/test-suites/{suite_id}", response_model=TestSuiteOut) +async def update_test_suite( + suite_id: str, + body: TestSuiteUpdate, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestSuite, suite_id) + if row is None: + raise HTTPException(404, "测试集不存在") + row.name = body.name.strip() + row.description = body.description.strip() + await session.commit() + await session.refresh(row) + return await _suite_out(session, row) + + +@router.delete("/api/test-suites/{suite_id}") +async def delete_test_suite( + suite_id: str, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestSuite, suite_id) + if row is None: + raise HTTPException(404, "测试集不存在") + await session.delete(row) + await session.commit() + return {"ok": True} + + +@router.post("/api/test-suites/{suite_id}/duplicate", response_model=TestSuiteOut) +async def duplicate_test_suite( + suite_id: str, + session: AsyncSession = Depends(get_session), +): + source = await session.get(TestSuite, suite_id) + if source is None: + raise HTTPException(404, "测试集不存在") + copied = TestSuite( + id=_new_id("suite"), + name=f"{source.name}（副本）", + description=source.description, + ) + session.add(copied) + source_cases = ( + await session.execute( + select(TestCase) + .where(TestCase.suite_id == suite_id) + .order_by(TestCase.sort_order) + ) + ).scalars().all() + for index, source_case in enumerate(source_cases): + session.add( + TestCase( + id=_new_id("tc"), + suite_id=copied.id, + name=source_case.name, + description=source_case.description, + input_mode=source_case.input_mode, + definition=dict(source_case.definition or {}), + sort_order=index, + last_result="not_run", + ) + ) + await session.commit() + await session.refresh(copied) + return await _suite_out(session, copied) + + +@router.get("/api/test-cases", response_model=list[TestCaseOut]) +async def list_test_cases( + suite_id: str | None = Query(default=None, alias="suiteId"), + session: AsyncSession = Depends(get_session), +): + statement = select(TestCase) + if suite_id: + statement = statement.where(TestCase.suite_id == suite_id) + rows = ( + await session.execute( + statement.order_by(TestCase.suite_id, TestCase.sort_order, TestCase.name) + ) + ).scalars().all() + return [_case_out(row) for row in rows] + + +@router.post("/api/test-suites/{suite_id}/cases", response_model=TestCaseOut) +async def create_test_case( + suite_id: str, + body: TestCaseWrite, + session: AsyncSession = Depends(get_session), +): + if await session.get(TestSuite, suite_id) is None: + raise HTTPException(404, "测试集不存在") + max_order = await session.scalar( + select(func.max(TestCase.sort_order)).where(TestCase.suite_id == suite_id) + ) + row = TestCase( + id=_new_id("tc"), + suite_id=suite_id, + name=body.name.strip(), + description=body.description.strip(), + input_mode=body.input_mode, + definition=body.definition().model_dump(mode="json", by_alias=True), + sort_order=int(max_order if max_order is not None else -1) + 1, + last_result="not_run", + ) + session.add(row) + await _touch_suite(session, suite_id) + await session.commit() + await session.refresh(row) + return _case_out(row) + + +@router.post("/api/test-cases/bulk-delete") +async def bulk_delete_test_cases( + body: TestCaseBulkDeleteIn, + session: AsyncSession = Depends(get_session), +): + rows = ( + await session.execute(select(TestCase).where(TestCase.id.in_(body.case_ids))) + ).scalars().all() + suite_ids = {row.suite_id for row in rows} + for row in rows: + await session.delete(row) + await session.flush() + for suite_id in suite_ids: + await _renumber_cases(session, suite_id) + await _touch_suite(session, suite_id) + await session.commit() + return {"ok": True, "deleted": len(rows)} + + +@router.put("/api/test-suites/{suite_id}/case-order") +async def reorder_test_cases( + suite_id: str, + body: TestCaseOrderIn, + session: AsyncSession = Depends(get_session), +): + rows = ( + await session.execute( + select(TestCase) + .where(TestCase.suite_id == suite_id) + .order_by(TestCase.sort_order) + ) + ).scalars().all() + if {row.id for row in rows} != set(body.case_ids) or len(rows) != len(body.case_ids): + raise HTTPException(422, "排序列表必须包含测试集内全部用例且不能重复") + by_id = {row.id: row for row in rows} + offset = len(rows) + 1 + for index, row in enumerate(rows): + row.sort_order = offset + index + await session.flush() + for index, case_id in enumerate(body.case_ids): + by_id[case_id].sort_order = index + await _touch_suite(session, suite_id) + await session.commit() + return {"ok": True} + + +@router.get("/api/test-cases/{case_id}", response_model=TestCaseOut) +async def get_test_case( + case_id: str, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestCase, case_id) + if row is None: + raise HTTPException(404, "测试用例不存在") + return _case_out(row) + + +@router.put("/api/test-cases/{case_id}", response_model=TestCaseOut) +async def update_test_case( + case_id: str, + body: TestCaseWrite, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestCase, case_id) + if row is None: + raise HTTPException(404, "测试用例不存在") + _apply_case_write(row, body) + await _touch_suite(session, row.suite_id) + await session.commit() + await session.refresh(row) + return _case_out(row) + + +@router.delete("/api/test-cases/{case_id}") +async def delete_test_case( + case_id: str, + session: AsyncSession = Depends(get_session), +): + row = await session.get(TestCase, case_id) + if row is None: + raise HTTPException(404, "测试用例不存在") + suite_id = row.suite_id + await session.delete(row) + await session.flush() + await _renumber_cases(session, suite_id) + await _touch_suite(session, suite_id) + await session.commit() + return {"ok": True} + + +@router.post("/api/test-cases/{case_id}/duplicate", response_model=TestCaseOut) +async def duplicate_test_case( + case_id: str, + session: AsyncSession = Depends(get_session), +): + source = await session.get(TestCase, case_id) + if source is None: + raise HTTPException(404, "测试用例不存在") + max_order = await session.scalar( + select(func.max(TestCase.sort_order)).where( + TestCase.suite_id == source.suite_id + ) + ) + copied = TestCase( + id=_new_id("tc"), + suite_id=source.suite_id, + name=f"{source.name}（副本）", + description=source.description, + input_mode=source.input_mode, + definition=dict(source.definition or {}), + sort_order=int(max_order if max_order is not None else -1) + 1, + last_result="not_run", + ) + session.add(copied) + await _touch_suite(session, source.suite_id) + await session.commit() + await session.refresh(copied) + return _case_out(copied) diff --git a/backend/routes/test_runs.py b/backend/routes/test_runs.py new file mode 100644 index 0000000..fec5c7b --- /dev/null +++ b/backend/routes/test_runs.py @@ -0,0 +1,240 @@ +"""Start, inspect, and cancel persisted batch text test runs.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from uuid import uuid4 + +from db.models import ( + Assistant, + ModelResource, + TestCase, + TestRun, + TestRunCase, + TestSuite, +) +from db.session import get_session +from fastapi import APIRouter, Depends, HTTPException +from services.auth import require_admin +from services.config_resolver import resolve_runtime_config +from services.test_runs.orchestrator import test_run_orchestrator +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from test_schemas import ( + BatchExecutionError, + BatchRunCaseOut, + BatchRunCreate, + BatchRunSnapshotOut, + BatchTurnResult, + BatchEvaluationResult, + TestCaseDefinition, +) + + +router = APIRouter( + prefix="/api/test-runs", + tags=["tests"], + dependencies=[Depends(require_admin)], +) + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +async def _run_out( + session: AsyncSession, + run: TestRun, +) -> BatchRunSnapshotOut: + rows = ( + await session.execute( + select(TestRunCase) + .where(TestRunCase.run_id == run.id) + .order_by(TestRunCase.position) + ) + ).scalars().all() + cases: list[BatchRunCaseOut] = [] + for row in rows: + result = dict(row.result or {}) + cases.append( + BatchRunCaseOut( + id=str(row.test_case_id or row.case_snapshot.get("id") or row.id), + name=row.test_case_name, + status=row.status, + turns=[ + BatchTurnResult.model_validate(item) + for item in result.get("turns") or [] + ], + overall_criteria=[ + BatchEvaluationResult.model_validate(item) + for item in result.get("overallCriteria") + or result.get("overall_criteria") + or [] + ], + attempt_count=row.attempt_count, + max_attempts=row.max_attempts, + execution_error=( + BatchExecutionError.model_validate(row.execution_error) + if row.execution_error + else None + ), + ) + ) + return BatchRunSnapshotOut( + id=run.id, + status=run.status, + title=run.title, + assistant_name=run.assistant_name, + config=dict(run.config or {}), + cases=cases, + started_at=run.started_at or run.created_at, + finished_at=run.finished_at, + stop_reason=run.stop_reason, + ) + + +@router.post("", response_model=BatchRunSnapshotOut) +async def create_test_run( + body: BatchRunCreate, + session: AsyncSession = Depends(get_session), +): + assistant = await session.get(Assistant, body.assistant_id) + if assistant is None: + raise HTTPException(404, "被测助手不存在") + if assistant.runtime_mode != "pipeline": + raise HTTPException(422, "第一版批量测试只支持 Pipeline 运行模式") + if assistant.type not in {"prompt", "workflow"}: + raise HTTPException(422, f"第一版批量测试暂不支持 {assistant.type} 类型助手") + + evaluator_resource = await session.get( + ModelResource, + body.evaluator_model_resource_id, + ) + if ( + evaluator_resource is None + or evaluator_resource.capability != "LLM" + or not evaluator_resource.enabled + ): + raise HTTPException(422, "评估模型不存在、已停用或不是 LLM 资源") + try: + runtime_config = await resolve_runtime_config(session, assistant.id) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + + rows = ( + await session.execute(select(TestCase).where(TestCase.id.in_(body.case_ids))) + ).scalars().all() + by_id = {row.id: row for row in rows} + missing = [case_id for case_id in body.case_ids if case_id not in by_id] + if missing: + raise HTTPException(422, "测试用例不存在: " + "、".join(missing)) + ordered_cases = [by_id[case_id] for case_id in body.case_ids] + tools_by_id = {tool.id: tool for tool in runtime_config.tools} + for row in ordered_cases: + if row.input_mode != "fixed_script_text": + raise HTTPException(422, f"用例“{row.name}”不是可运行的固定文字模式") + try: + definition = TestCaseDefinition.model_validate(row.definition or {}) + except ValueError as exc: + raise HTTPException(422, f"用例“{row.name}”校验失败: {exc}") from exc + for turn in definition.turns: + for behavior in turn.behaviors: + if behavior.type != "tool_call": + continue + tool = tools_by_id.get(behavior.tool_id) + if tool is None or tool.function_name != behavior.function_name: + raise HTTPException( + 422, + f"用例“{row.name}”引用的工具 {behavior.function_name} " + "不属于当前被测助手", + ) + + suite_ids = list(dict.fromkeys(row.suite_id for row in ordered_cases)) + suites = ( + await session.execute(select(TestSuite).where(TestSuite.id.in_(suite_ids))) + ).scalars().all() + suite_by_id = {row.id: row for row in suites} + if body.title: + title = body.title.strip() + elif len(suite_ids) == 1 and suite_ids[0] in suite_by_id: + title = f"{suite_by_id[suite_ids[0]].name} · {len(ordered_cases)} 个用例" + else: + title = f"批量测试 · {len(ordered_cases)} 个用例" + + now = datetime.now(UTC) + run = TestRun( + id=_new_id("run"), + assistant_id=assistant.id, + assistant_name=assistant.name, + title=title, + status="queued", + config={ + "suiteCount": len(suite_ids), + "evaluatorModelResourceId": evaluator_resource.id, + "evaluatorModelResourceName": evaluator_resource.name, + "evaluatorModel": str( + (evaluator_resource.values or {}).get("modelId") or "" + ), + **body.config.model_dump(mode="json", by_alias=True), + }, + started_at=now, + ) + session.add(run) + for position, row in enumerate(ordered_cases): + definition = TestCaseDefinition.model_validate(row.definition or {}) + snapshot = { + "id": row.id, + "suiteId": row.suite_id, + "name": row.name, + "description": row.description, + "inputMode": row.input_mode, + **definition.model_dump(mode="json", by_alias=True), + } + session.add( + TestRunCase( + id=_new_id("rc"), + run_id=run.id, + test_case_id=row.id, + test_case_name=row.name, + suite_id=row.suite_id, + position=position, + case_snapshot=snapshot, + status="waiting", + attempt_count=0, + max_attempts=body.config.error_retry_count + 1, + result={"turns": [], "overallCriteria": []}, + ) + ) + await session.commit() + await session.refresh(run) + test_run_orchestrator.start(run.id) + return await _run_out(session, run) + + +@router.get("/{run_id}", response_model=BatchRunSnapshotOut) +async def get_test_run( + run_id: str, + session: AsyncSession = Depends(get_session), +): + run = await session.get(TestRun, run_id) + if run is None: + raise HTTPException(404, "批量测试运行不存在") + return await _run_out(session, run) + + +@router.post("/{run_id}/cancel", response_model=BatchRunSnapshotOut) +async def cancel_test_run( + run_id: str, + session: AsyncSession = Depends(get_session), +): + run = await session.get(TestRun, run_id) + if run is None: + raise HTTPException(404, "批量测试运行不存在") + if run.status in {"completed", "cancelled"}: + return await _run_out(session, run) + run.cancel_requested = True + run.stop_reason = "manual" + await session.commit() + await test_run_orchestrator.cancel(run_id) + await session.refresh(run) + return await _run_out(session, run) diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 1af2e10..55b4bbe 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -99,6 +99,9 @@ class BrainRuntime: Callable[[bool, dict[str, Any]], Awaitable[None]] | None ) = None flow_global_functions: list[Any] = field(default_factory=list) + # Tests replace external side effects without changing Pipecat or stored + # tool resources. Production leaves this unset and uses ToolExecutor. + tool_executor_factory: Callable[[Any], Any] | None = None @dataclass(frozen=True) diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index f33eb9b..01724ca 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -103,6 +103,8 @@ class PromptBrain(BaseBrain): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: self._runtime = runtime + if runtime.tool_executor_factory is not None: + self._tools = runtime.tool_executor_factory(self._store) self._tools.set_client_tools(runtime.client_tools) self._actions = ActionRunner( self._tools, @@ -651,14 +653,27 @@ class PromptBrain(BaseBrain): if not kind: raise ValueError(f"系统工具 {tool.id} 缺少有效 kind") if kind == "end_conversation": - return self._make_end_call_tool(tool, runtime) - if kind == "update_state": - return self._make_update_state_tool(tool, runtime) - if kind == "skip_turn": - return self._make_skip_turn_tool(tool) - if kind == "request_human_handoff": - return self._make_handoff_tool(tool, runtime) - raise ValueError(f"未知系统工具: {kind}") + schema, handler = self._make_end_call_tool(tool, runtime) + elif kind == "update_state": + schema, handler = self._make_update_state_tool(tool, runtime) + elif kind == "skip_turn": + schema, handler = self._make_skip_turn_tool(tool) + elif kind == "request_human_handoff": + schema, handler = self._make_handoff_tool(tool, runtime) + else: + raise ValueError(f"未知系统工具: {kind}") + + if runtime.tool_executor_factory is None: + return schema, handler + + async def mock_system_tool(params: FunctionCallParams) -> None: + result = await self._tools.execute( + tool, + dict(params.arguments or {}), + ) + await params.result_callback(result) + + return schema, mock_system_tool def _make_update_state_tool(self, tool, runtime: BrainRuntime): """更新已声明的动态变量(会话状态),并让模型继续当前回答。""" diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index a450eeb..d4c7ee7 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -214,7 +214,12 @@ class WorkflowBrain(BaseBrain): self._cfg = cfg self._runtime = runtime self._store = DynamicVariableStore.from_config(cfg) - self._tools = ToolExecutor(self._store, client_tools=runtime.client_tools) + self._tools = ( + runtime.tool_executor_factory(self._store) + if runtime.tool_executor_factory is not None + else ToolExecutor(self._store, client_tools=runtime.client_tools) + ) + self._tools.set_client_tools(runtime.client_tools) self._actions = ActionRunner( self._tools, is_session_ending=lambda: runtime.call_end.ending, @@ -934,18 +939,28 @@ class WorkflowBrain(BaseBrain): """Build one platform-owned tool scoped to the active Agent node.""" kind = system_tool_kind(tool.definition or {}) if kind == "update_state": - return self._workflow_update_state_tool( + schema = self._workflow_update_state_tool( tool, node_id, state_variable_names=state_variable_names, ) - if kind == "skip_turn": - return self._workflow_skip_turn_tool(tool) - if kind == "request_human_handoff": - return self._workflow_handoff_tool(tool, node_id) - if kind == "end_conversation": - return self._workflow_end_conversation_tool(tool, node_id) - raise ValueError(f"系统工具 {tool.id} 缺少有效 kind") + elif kind == "skip_turn": + schema = self._workflow_skip_turn_tool(tool) + elif kind == "request_human_handoff": + schema = self._workflow_handoff_tool(tool, node_id) + elif kind == "end_conversation": + schema = self._workflow_end_conversation_tool(tool, node_id) + else: + raise ValueError(f"系统工具 {tool.id} 缺少有效 kind") + + runtime = self._require_runtime() + if runtime.tool_executor_factory is None: + return schema + + async def mock_system_tool(args, _flow_manager): + return await self._tools.execute(tool, dict(args or {})) + + return replace(schema, handler=mock_system_tool) def _workflow_update_state_tool( self, diff --git a/backend/services/test_runs/__init__.py b/backend/services/test_runs/__init__.py new file mode 100644 index 0000000..28fb868 --- /dev/null +++ b/backend/services/test_runs/__init__.py @@ -0,0 +1 @@ +"""Fixed-text test execution domain.""" diff --git a/backend/services/test_runs/errors.py b/backend/services/test_runs/errors.py new file mode 100644 index 0000000..74f31f6 --- /dev/null +++ b/backend/services/test_runs/errors.py @@ -0,0 +1,24 @@ +"""Structured failures shared by the text runner and batch orchestrator.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class TestExecutionError(RuntimeError): + code: str + message: str + stage: str + retryable: bool = False + + def __post_init__(self) -> None: + RuntimeError.__init__(self, self.message) + + def as_dict(self) -> dict[str, object]: + return { + "code": self.code, + "message": self.message, + "stage": self.stage, + "retryable": self.retryable, + } diff --git a/backend/services/test_runs/evaluator.py b/backend/services/test_runs/evaluator.py new file mode 100644 index 0000000..08498c1 --- /dev/null +++ b/backend/services/test_runs/evaluator.py @@ -0,0 +1,374 @@ +"""Deterministic and LLM-backed assertions for text test results.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from typing import Any + +import httpx +from models import RuntimeModelResource + +from services.test_runs.errors import TestExecutionError +from services.test_runs.text_runner import RawToolCall, RawTurnResult, TextCaseResult +from test_schemas import ( + BatchEvaluationResult, + BatchTurnResult, + FixedInputTurn, + OverallCriterion, + ReplyExpectedBehavior, + TestCaseDefinition, + ToolCallExpectedBehavior, + ToolParamAssertion, +) + + +@dataclass(frozen=True) +class EvaluatedCase: + turns: list[BatchTurnResult] + overall_criteria: list[BatchEvaluationResult] + + @property + def passed(self) -> bool: + evaluations = [ + item + for turn in self.turns + for item in turn.evaluations + ] + self.overall_criteria + return all(item.status == "pass" for item in evaluations) + + +class LLMJudge: + """Small OpenAI-compatible JSON judge using a dedicated model resource.""" + + def __init__( + self, + resource: RuntimeModelResource, + *, + timeout_seconds: float = 60, + ): + self._resource = resource + self._timeout_seconds = timeout_seconds + + async def evaluate( + self, + *, + criteria: str, + subject: str, + context: str, + ) -> tuple[bool, str, str]: + values = self._resource.values or {} + secrets = self._resource.secrets or {} + base_url = str(values.get("apiUrl") or "").rstrip("/") + api_key = str(secrets.get("apiKey") or "") + model = str(values.get("modelId") or "") + if not base_url or not api_key or not model: + raise TestExecutionError( + code="EVALUATOR_NOT_CONFIGURED", + message="LLM 评估所需的模型资源未完整配置", + stage="evaluation", + ) + prompt = ( + "你是严格的自动化测试评估器。只根据提供的内容判断标准是否满足。\n" + "必须只输出一个 JSON 对象，不要输出 Markdown：\n" + '{"passed":true或false,"reason":"简短中文理由","actual":"实际表现摘要"}\n\n' + f"评估对象：{subject}\n" + f"评估标准：{criteria}\n\n" + f"待评估内容：\n{context}" + ) + payload: dict[str, Any] = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "temperature": 0, + } + extra_body = values.get("extraBody") + if isinstance(extra_body, dict): + payload.update( + { + key: value + for key, value in extra_body.items() + if key not in {"model", "messages"} + } + ) + try: + async with httpx.AsyncClient(timeout=self._timeout_seconds) as client: + response = await client.post( + f"{base_url}/chat/completions", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json=payload, + ) + response.raise_for_status() + body = response.json() + content = body["choices"][0]["message"]["content"] + if isinstance(content, list): + content = "".join( + str(part.get("text") or "") + for part in content + if isinstance(part, dict) + ) + parsed = self._parse_json(str(content or "")) + return ( + bool(parsed.get("passed")), + str(parsed.get("reason") or "LLM 未提供判断理由"), + str(parsed.get("actual") or ""), + ) + except TestExecutionError: + raise + except (httpx.HTTPError, KeyError, IndexError, TypeError, ValueError) as exc: + raise TestExecutionError( + code="EVALUATOR_REQUEST_FAILED", + message=f"LLM 评估失败: {exc}", + stage="evaluation", + retryable=True, + ) from exc + + @staticmethod + def _parse_json(value: str) -> dict[str, Any]: + text = value.strip() + if text.startswith("```"): + text = re.sub(r"^```(?:json)?\s*", "", text) + text = re.sub(r"\s*```$", "", text) + start = text.find("{") + end = text.rfind("}") + if start < 0 or end < start: + raise ValueError("评估模型没有返回 JSON") + parsed = json.loads(text[start : end + 1]) + if ( + not isinstance(parsed, dict) + or not isinstance(parsed.get("passed"), bool) + ): + raise ValueError("评估模型返回格式不正确") + return parsed + + +class EvaluationEngine: + def __init__(self, judge_resource: RuntimeModelResource): + self._judge = LLMJudge(judge_resource) + + async def evaluate( + self, + definition: TestCaseDefinition, + execution: TextCaseResult, + ) -> EvaluatedCase: + raw_by_id = {turn.id: turn for turn in execution.turns} + turns: list[BatchTurnResult] = [] + for index, expected_turn in enumerate(definition.turns): + raw = raw_by_id.get(expected_turn.id) or RawTurnResult( + id=expected_turn.id, + index=index, + user_input=expected_turn.user_input, + assistant_reply="", + tool_calls=[], + ) + evaluations = [ + await self._evaluate_behavior(behavior, raw) + for behavior in expected_turn.behaviors + ] + turns.append( + BatchTurnResult( + id=raw.id, + index=index, + user_input=raw.user_input, + assistant_reply=raw.assistant_reply, + tool_calls=[item.as_result() for item in raw.tool_calls], + evaluations=evaluations, + ) + ) + + transcript = self._format_transcript(execution, definition) + overall = [ + await self._evaluate_overall(criterion, transcript) + for criterion in definition.overall_criteria + ] + return EvaluatedCase(turns=turns, overall_criteria=overall) + + async def _evaluate_behavior( + self, + behavior: ReplyExpectedBehavior | ToolCallExpectedBehavior, + turn: RawTurnResult, + ) -> BatchEvaluationResult: + if behavior.type == "reply": + return await self._evaluate_reply(behavior, turn.assistant_reply) + return await self._evaluate_tool(behavior, turn.tool_calls) + + async def _evaluate_reply( + self, + behavior: ReplyExpectedBehavior, + reply: str, + ) -> BatchEvaluationResult: + if behavior.assertion_type == "llm": + passed, reason, actual = await self._judge.evaluate( + criteria=behavior.llm_criteria, + subject="当前轮助手回复", + context=reply or "（助手没有生成文字回复）", + ) + return BatchEvaluationResult( + id=behavior.id, + label="回复 · LLM 判断", + kind="reply", + status="pass" if passed else "fail", + expected=behavior.llm_criteria, + actual=actual or reply, + reason="" if passed else reason, + ) + + normalized_reply = reply.casefold() + matches = [keyword.casefold() in normalized_reply for keyword in behavior.keywords] + matched = all(matches) if behavior.keyword_match_mode == "all" else any(matches) + passed = not matched if behavior.negate_keywords else matched + relation = ( + "不应包含" + if behavior.negate_keywords + else "应包含全部" + if behavior.keyword_match_mode == "all" + else "应至少包含其一" + ) + return BatchEvaluationResult( + id=behavior.id, + label="回复 · 关键词", + kind="reply", + status="pass" if passed else "fail", + expected=f"{relation}：{'、'.join(behavior.keywords)}", + actual=reply or "（无文字回复）", + reason="" if passed else "实际回复未满足关键词规则。", + ) + + async def _evaluate_tool( + self, + behavior: ToolCallExpectedBehavior, + calls: list[RawToolCall], + ) -> BatchEvaluationResult: + matching_calls = [ + call for call in calls if call.function_name == behavior.function_name + ] + count = len(matching_calls) + if behavior.expectation == "not_called": + passed = count == 0 + return BatchEvaluationResult( + id=behavior.id, + label=f"工具 · {behavior.function_name}", + kind="tool_call", + status="pass" if passed else "fail", + expected=f"不应调用 {behavior.function_name}", + actual=f"实际调用 {count} 次 {behavior.function_name}", + reason="" if passed else "检测到本轮不应发生的工具调用。", + ) + + count_passed = count >= behavior.min_calls and ( + behavior.max_calls is None or count <= behavior.max_calls + ) + params_passed = True + parameter_reason = "" + if count_passed and behavior.param_assertions: + params_passed = False + reasons: list[str] = [] + for call in matching_calls: + call_passed, call_reason = await self._call_matches_parameters( + call, behavior.param_assertions + ) + if call_passed: + params_passed = True + break + reasons.append(call_reason) + parameter_reason = "；".join(item for item in reasons if item) + + passed = count_passed and params_passed + count_text = ( + f"{behavior.min_calls} 次以上" + if behavior.max_calls is None + else f"{behavior.min_calls} 次" + if behavior.min_calls == behavior.max_calls + else f"{behavior.min_calls}–{behavior.max_calls} 次" + ) + reason = "" + if not count_passed: + reason = "工具调用次数不符合要求。" + elif not params_passed: + reason = parameter_reason or "没有一次工具调用满足全部参数断言。" + return BatchEvaluationResult( + id=behavior.id, + label=f"工具 · {behavior.function_name}", + kind="tool_call", + status="pass" if passed else "fail", + expected=f"应调用 {behavior.function_name} {count_text}", + actual=f"实际调用 {count} 次 {behavior.function_name}", + reason=reason, + ) + + async def _call_matches_parameters( + self, + call: RawToolCall, + assertions: list[ToolParamAssertion], + ) -> tuple[bool, str]: + arguments = call.arguments if isinstance(call.arguments, dict) else {} + for assertion in assertions: + if assertion.name not in arguments: + return False, f"缺少参数 {assertion.name}" + actual = arguments[assertion.name] + if assertion.match_mode == "exact": + try: + expected: Any = json.loads(assertion.value) + except json.JSONDecodeError: + expected = assertion.value + if actual != expected and str(actual) != assertion.value: + return False, f"参数 {assertion.name} 与期望值不一致" + elif assertion.match_mode == "regex": + actual_text = ( + actual + if isinstance(actual, str) + else json.dumps(actual, ensure_ascii=False, default=str) + ) + if re.search(assertion.value, actual_text) is None: + return False, f"参数 {assertion.name} 未匹配正则表达式" + else: + passed, reason, _actual = await self._judge.evaluate( + criteria=assertion.value, + subject=f"工具参数 {assertion.name}", + context=json.dumps(actual, ensure_ascii=False, default=str), + ) + if not passed: + return False, reason + return True, "" + + async def _evaluate_overall( + self, + criterion: OverallCriterion, + transcript: str, + ) -> BatchEvaluationResult: + passed, reason, actual = await self._judge.evaluate( + criteria=criterion.criteria, + subject="完整测试对话", + context=transcript, + ) + return BatchEvaluationResult( + id=criterion.id, + label=criterion.name, + kind="overall", + status="pass" if passed else "fail", + expected=criterion.criteria, + actual=actual or "已评估完整对话", + reason="" if passed else reason, + ) + + @staticmethod + def _format_transcript( + execution: TextCaseResult, + definition: TestCaseDefinition, + ) -> str: + if execution.transcript: + labels = {"user": "User", "assistant": "Agent"} + return "\n".join( + f"{labels.get(item['role'], item['role'])}: {item['content']}" + for item in execution.transcript + ) + raw_by_id = {turn.id: turn for turn in execution.turns} + lines: list[str] = [] + for turn in definition.turns: + raw = raw_by_id.get(turn.id) + lines.append(f"User: {turn.user_input}") + lines.append(f"Agent: {raw.assistant_reply if raw else ''}") + return "\n".join(lines) diff --git a/backend/services/test_runs/mock_tools.py b/backend/services/test_runs/mock_tools.py new file mode 100644 index 0000000..559f630 --- /dev/null +++ b/backend/services/test_runs/mock_tools.py @@ -0,0 +1,85 @@ +"""Test-only tool executor that never reaches external side effects.""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any + +from models import RuntimeTool +from services.runtime_variables import DynamicVariableStore +from services.tool_executor import ToolExecutionError, ToolExecutor +from test_schemas import ToolCallExpectedBehavior + + +class MockToolExecutor(ToolExecutor): + """Resolve the active turn's configured mocks and reject everything else.""" + + def __init__(self, store: DynamicVariableStore): + super().__init__(store) + self._mocks_by_tool_id: dict[str, ToolCallExpectedBehavior] = {} + self._mocks_by_function: dict[str, ToolCallExpectedBehavior] = {} + self.unmocked_calls: list[str] = [] + + def set_turn_behaviors( + self, + behaviors: list[ToolCallExpectedBehavior], + ) -> None: + self._mocks_by_tool_id = {item.tool_id: item for item in behaviors} + self._mocks_by_function = {item.function_name: item for item in behaviors} + self.unmocked_calls = [] + + async def execute( + self, + tool: RuntimeTool, + arguments: dict[str, Any] | None = None, + *, + result_assignments: dict[str, str] | None = None, + ) -> dict[str, Any]: + self.register_secrets(tool) + behavior = self._mocks_by_tool_id.get(tool.id) or self._mocks_by_function.get( + tool.function_name + ) + if behavior is None: + self.unmocked_calls.append(tool.function_name) + raise ToolExecutionError( + f"UNMOCKED_TOOL_CALL: 工具 {tool.function_name} 未配置 Mock 返回值" + ) + + delay_seconds = behavior.mock_response.delay_ms / 1000 + if delay_seconds: + await asyncio.sleep(delay_seconds) + + payload = json.loads(behavior.mock_response.body) + status = "ok" if behavior.mock_response.outcome == "success" else "error" + if isinstance(payload, dict): + result: dict[str, Any] = dict(payload) + result["status"] = status + else: + result = {"status": status, "data": payload} + if result.get("status") != "ok": + return {**result, "updated_variables": []} + return self._apply_result_assignments( + tool, + result, + result_assignments=result_assignments, + ) + + +class TextClientToolPort: + """Automatically acknowledge built-in message stages in text tests.""" + + async def call( + self, + function_name: str, + arguments: dict[str, Any], + **_kwargs: Any, + ) -> dict[str, Any]: + if function_name == "show_message": + return { + "status": "ok", + "data": {"action": "confirmed", "arguments": arguments}, + } + raise ToolExecutionError( + f"UNMOCKED_CLIENT_TOOL: 客户端工具 {function_name} 不能脱离 Mock 执行" + ) diff --git a/backend/services/test_runs/orchestrator.py b/backend/services/test_runs/orchestrator.py new file mode 100644 index 0000000..7da4a0f --- /dev/null +++ b/backend/services/test_runs/orchestrator.py @@ -0,0 +1,410 @@ +"""In-process MVP scheduler for persisted batch test runs.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from datetime import UTC, datetime + +from loguru import logger +from sqlalchemy import select + +from db.models import ModelResource, TestCase, TestRun, TestRunCase +from db.session import SessionLocal +from models import RuntimeModelResource +from services.config_resolver import resolve_runtime_config +from services.test_runs.errors import TestExecutionError +from services.test_runs.evaluator import EvaluationEngine +from services.test_runs.text_runner import TextPipelineRunner +from test_schemas import TestCaseDefinition + + +@dataclass +class _RunControl: + stop_event: asyncio.Event = field(default_factory=asyncio.Event) + reason: str | None = None + case_tasks: set[asyncio.Task[None]] = field(default_factory=set) + + +class TestRunOrchestrator: + """Run cases concurrently while keeping PostgreSQL as the source of truth.""" + + def __init__(self) -> None: + self._run_tasks: dict[str, asyncio.Task[None]] = {} + self._controls: dict[str, _RunControl] = {} + + def start(self, run_id: str) -> None: + current = self._run_tasks.get(run_id) + if current is not None and not current.done(): + return + control = _RunControl() + self._controls[run_id] = control + task = asyncio.create_task(self._execute_run(run_id, control), name=f"test-run-{run_id}") + self._run_tasks[run_id] = task + task.add_done_callback(lambda _task: self._forget(run_id)) + + async def cancel(self, run_id: str) -> None: + control = self._controls.get(run_id) + if control is None: + return + control.reason = "manual" + control.stop_event.set() + for task in tuple(control.case_tasks): + task.cancel() + + async def shutdown(self) -> None: + tasks = [task for task in self._run_tasks.values() if not task.done()] + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + self._run_tasks.clear() + self._controls.clear() + + def _forget(self, run_id: str) -> None: + self._run_tasks.pop(run_id, None) + self._controls.pop(run_id, None) + + async def _execute_run(self, run_id: str, control: _RunControl) -> None: + try: + async with SessionLocal() as session: + run = await session.get(TestRun, run_id) + if run is None: + return + run.status = "running" + run.started_at = run.started_at or datetime.now(UTC) + await session.commit() + try: + cfg = await resolve_runtime_config(session, str(run.assistant_id or "")) + except Exception as exc: + await self._fail_run_setup(run_id, str(exc)) + return + config = dict(run.config or {}) + evaluator_resource_id = str( + config.get("evaluatorModelResourceId") or "" + ) + evaluator_row = await session.get( + ModelResource, + evaluator_resource_id, + ) + if ( + evaluator_row is None + or evaluator_row.capability != "LLM" + or not evaluator_row.enabled + ): + await self._fail_run_setup( + run_id, + "评估模型不存在、已停用或不是 LLM 资源", + ) + return + evaluator_resource = RuntimeModelResource( + id=evaluator_row.id, + name=evaluator_row.name, + capability=evaluator_row.capability, + interface_type=evaluator_row.interface_type, + values=evaluator_row.values or {}, + secrets=evaluator_row.secrets or {}, + support_image_input=bool(evaluator_row.support_image_input), + ) + rows = ( + await session.execute( + select(TestRunCase) + .where(TestRunCase.run_id == run_id) + .order_by(TestRunCase.position) + ) + ).scalars().all() + + semaphore = asyncio.Semaphore(int(config.get("concurrency") or 3)) + tasks = { + asyncio.create_task( + self._execute_case( + run_id, + row.id, + cfg, + evaluator_resource, + config, + semaphore, + control, + ), + name=f"test-run-case-{row.id}", + ) + for row in rows + } + control.case_tasks = tasks + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + await self._finish_run(run_id, control) + except asyncio.CancelledError: + await self._interrupt_run(run_id) + raise + except Exception as exc: # noqa: BLE001 - background task must persist failure + logger.exception(f"批量测试运行失败: run_id={run_id}: {exc}") + await self._fail_run_setup(run_id, str(exc)) + + async def _execute_case( + self, + run_id: str, + run_case_id: str, + cfg, + evaluator_resource: RuntimeModelResource, + config: dict, + semaphore: asyncio.Semaphore, + control: _RunControl, + ) -> None: + async with semaphore: + if control.stop_event.is_set(): + await self._mark_skipped(run_case_id) + return + + async with SessionLocal() as session: + row = await session.get(TestRunCase, run_case_id) + if row is None: + return + snapshot = dict(row.case_snapshot or {}) + definition = TestCaseDefinition.model_validate(snapshot) + max_attempts = row.max_attempts + + for attempt in range(1, max_attempts + 1): + if control.stop_event.is_set(): + await self._mark_skipped(run_case_id) + return + await self._mark_running(run_case_id, attempt) + try: + timeout_seconds = int( + config.get("timeoutSecs") + or config.get("timeout_secs") + or 60 + ) + async with asyncio.timeout(timeout_seconds): + execution = await TextPipelineRunner().run( + cfg.model_copy(deep=True), + definition, + timeout_seconds=timeout_seconds, + ) + evaluated = await EvaluationEngine(evaluator_resource).evaluate( + definition, + execution, + ) + except asyncio.CancelledError: + await self._mark_skipped(run_case_id) + raise + except TimeoutError: + error = TestExecutionError( + code="CASE_TIMEOUT", + message=f"测试用例执行与评估超过 {timeout_seconds} 秒", + stage="pipeline", + retryable=True, + ) + if attempt < max_attempts and not control.stop_event.is_set(): + continue + await self._mark_error(run_case_id, error) + if str( + config.get("errorStrategy") + or config.get("error_strategy") + ) == "stop_on_error": + await self._request_stop(run_id, control, "execution_error") + return + except TestExecutionError as exc: + if exc.retryable and attempt < max_attempts and not control.stop_event.is_set(): + continue + await self._mark_error(run_case_id, exc) + if str(config.get("errorStrategy") or config.get("error_strategy")) == "stop_on_error": + await self._request_stop(run_id, control, "execution_error") + return + except Exception as exc: # noqa: BLE001 - normalize unknown runtime errors + error = TestExecutionError( + code="UNEXPECTED_EXECUTION_ERROR", + message=str(exc) or type(exc).__name__, + stage="pipeline", + retryable=False, + ) + await self._mark_error(run_case_id, error) + if str(config.get("errorStrategy") or config.get("error_strategy")) == "stop_on_error": + await self._request_stop(run_id, control, "execution_error") + return + + status = "pass" if evaluated.passed else "fail" + await self._mark_evaluated(run_case_id, status, evaluated) + if status == "fail" and str( + config.get("failureStrategy") or config.get("failure_strategy") + ) == "stop_on_fail": + await self._request_stop(run_id, control, "assertion_failure") + return + + async def _request_stop( + self, + run_id: str, + control: _RunControl, + reason: str, + ) -> None: + if control.reason is None: + control.reason = reason + control.stop_event.set() + async with SessionLocal() as session: + run = await session.get(TestRun, run_id) + if run is not None and run.stop_reason is None: + run.stop_reason = reason + await session.commit() + + async def _mark_running(self, run_case_id: str, attempt: int) -> None: + async with SessionLocal() as session: + row = await session.get(TestRunCase, run_case_id) + if row is None: + return + row.status = "running" + row.attempt_count = attempt + row.started_at = row.started_at or datetime.now(UTC) + row.finished_at = None + row.execution_error = None + await session.commit() + + async def _mark_evaluated(self, run_case_id: str, status: str, evaluated) -> None: + async with SessionLocal() as session: + row = await session.get(TestRunCase, run_case_id) + if row is None: + return + row.status = status + row.result = { + "turns": [ + item.model_dump(mode="json", by_alias=True) + for item in evaluated.turns + ], + "overallCriteria": [ + item.model_dump(mode="json", by_alias=True) + for item in evaluated.overall_criteria + ], + } + row.execution_error = None + row.finished_at = datetime.now(UTC) + if row.test_case_id: + test_case = await session.get(TestCase, row.test_case_id) + if test_case is not None: + test_case.last_result = status + await session.commit() + + async def _mark_error( + self, + run_case_id: str, + error: TestExecutionError, + ) -> None: + async with SessionLocal() as session: + row = await session.get(TestRunCase, run_case_id) + if row is None: + return + row.status = "error" + row.execution_error = error.as_dict() + row.finished_at = datetime.now(UTC) + if row.test_case_id: + test_case = await session.get(TestCase, row.test_case_id) + if test_case is not None: + test_case.last_result = "fail" + await session.commit() + + async def _mark_skipped(self, run_case_id: str) -> None: + async with SessionLocal() as session: + row = await session.get(TestRunCase, run_case_id) + if row is None or row.status in {"pass", "fail", "error", "skipped"}: + return + row.status = "skipped" + row.finished_at = datetime.now(UTC) + await session.commit() + + async def _finish_run(self, run_id: str, control: _RunControl) -> None: + async with SessionLocal() as session: + run = await session.get(TestRun, run_id) + if run is None: + return + waiting = ( + await session.execute( + select(TestRunCase).where( + TestRunCase.run_id == run_id, + TestRunCase.status.in_(["waiting", "running"]), + ) + ) + ).scalars().all() + for row in waiting: + row.status = "skipped" + row.finished_at = datetime.now(UTC) + manual = bool(run.cancel_requested or control.reason == "manual") + run.status = "cancelled" if manual else "completed" + run.stop_reason = "manual" if manual else control.reason or run.stop_reason + run.finished_at = datetime.now(UTC) + await session.commit() + + async def _fail_run_setup(self, run_id: str, message: str) -> None: + error = TestExecutionError( + code="RUN_SETUP_FAILED", + message=message or "批量测试初始化失败", + stage="pipeline", + retryable=False, + ) + async with SessionLocal() as session: + run = await session.get(TestRun, run_id) + if run is None: + return + rows = ( + await session.execute( + select(TestRunCase).where( + TestRunCase.run_id == run_id, + TestRunCase.status.in_(["waiting", "running"]), + ) + ) + ).scalars().all() + now = datetime.now(UTC) + for row in rows: + row.status = "error" + row.execution_error = error.as_dict() + row.finished_at = now + run.status = "completed" + run.stop_reason = "execution_error" + run.finished_at = now + await session.commit() + + async def _interrupt_run(self, run_id: str) -> None: + await _mark_run_interrupted(run_id, "后端服务停止，运行已中断") + + +async def _mark_run_interrupted(run_id: str, message: str) -> None: + async with SessionLocal() as session: + run = await session.get(TestRun, run_id) + if run is None or run.status not in {"queued", "running"}: + return + rows = ( + await session.execute( + select(TestRunCase).where( + TestRunCase.run_id == run_id, + TestRunCase.status.in_(["waiting", "running"]), + ) + ) + ).scalars().all() + now = datetime.now(UTC) + for row in rows: + if row.status == "running": + row.status = "error" + row.execution_error = { + "code": "SERVER_RESTARTED", + "message": message, + "stage": "pipeline", + "retryable": True, + } + else: + row.status = "skipped" + row.finished_at = now + run.status = "completed" + run.stop_reason = "execution_error" + run.finished_at = now + await session.commit() + + +async def recover_interrupted_test_runs() -> None: + async with SessionLocal() as session: + run_ids = ( + await session.execute( + select(TestRun.id).where(TestRun.status.in_(["queued", "running"])) + ) + ).scalars().all() + for run_id in run_ids: + await _mark_run_interrupted(str(run_id), "后端服务重启，运行已中断") + + +test_run_orchestrator = TestRunOrchestrator() diff --git a/backend/services/test_runs/text_runner.py b/backend/services/test_runs/text_runner.py new file mode 100644 index 0000000..b70b320 --- /dev/null +++ b/backend/services/test_runs/text_runner.py @@ -0,0 +1,687 @@ +"""Transportless Pipecat runner for one persisted fixed-text test case.""" + +from __future__ import annotations + +import asyncio +import json +import time +from dataclasses import dataclass, field +from typing import Any, Awaitable, Callable +from uuid import uuid4 + +from loguru import logger +from models import AssistantConfig +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, + CancelFrame, + EndFrame, + ErrorFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + ManuallySwitchServiceFrame, + TTSSpeakFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineParams, PipelineWorker +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.llm_service import FunctionCallParams +from pipecat.workers.runner import WorkerRunner + +from db.session import SessionLocal +from services.brains import BrainRuntime, build_brain +from services.knowledge import search as search_knowledge +from services.pipecat.call_lifecycle import ( + CallEndCoordinator, + FixedSpeechPlaybackMarkerFrame, +) +from services.pipecat.processors import ( + KnowledgeRetrievalProcessor, + UserTurnRoutingProcessor, +) +from services.pipecat.service_factory import config_with_resource +from services.pipecat.workflow_services import build_workflow_llm_switcher +from services.test_runs.errors import TestExecutionError +from services.test_runs.mock_tools import MockToolExecutor, TextClientToolPort +from test_schemas import ContextTurn, FixedInputTurn, TestCaseDefinition + + +IDLE_SETTLE_SECONDS = 0.2 +WORKER_START_TIMEOUT_SECONDS = 5.0 +INTERNAL_CANCEL_REASON = "text_test_case_complete" +AUTOMATIC_KNOWLEDGE_HINT = ( + "你已连接内部知识库。系统会在每轮用户问题前自动提供相关资料；" + "回答资料事实时只根据检索内容，资料不足要明确说明。" +) +ON_DEMAND_KNOWLEDGE_HINT = ( + "你已连接内部知识库。当用户问题涉及可能存在于业务知识库中的事实时，" + "先调用 search_knowledge_base 检索；回答资料事实时只根据检索内容，" + "资料不足要明确说明。" +) + + +@dataclass +class RawToolCall: + id: str + function_name: str + arguments: Any + result: Any = None + outcome: str = "error" + duration_ms: int = 0 + _started_at: float = field(default_factory=time.monotonic, repr=False) + + def as_result(self) -> dict[str, Any]: + return { + "id": self.id, + "functionName": self.function_name, + "argumentsJson": json.dumps( + self.arguments, ensure_ascii=False, indent=2, default=str + ), + "resultJson": json.dumps( + self.result, ensure_ascii=False, indent=2, default=str + ), + "outcome": self.outcome, + "durationMs": self.duration_ms, + } + + +@dataclass +class RawTurnResult: + id: str + index: int + user_input: str + assistant_reply: str + tool_calls: list[RawToolCall] + + +@dataclass +class TextCaseResult: + turns: list[RawTurnResult] + transcript: list[dict[str, str]] + + +@dataclass +class _ResponseWindow: + active_llm: int = 0 + active_assistant: int = 0 + pending_tools: set[str] = field(default_factory=set) + activity_count: int = 0 + last_activity_at: float = field(default_factory=time.monotonic) + + def touch(self) -> None: + self.activity_count += 1 + self.last_activity_at = time.monotonic() + + @property + def idle(self) -> bool: + return self.active_llm == 0 and self.active_assistant == 0 and not self.pending_tools + + +class _TextCaptureProcessor(FrameProcessor): + """Capture assistant/tool output and expose a reliable response boundary.""" + + def __init__(self, context: LLMContext): + super().__init__() + self.context = context + self.window = _ResponseWindow() + self.outputs: list[str] = [] + self.tool_calls: list[RawToolCall] = [] + self._tool_by_id: dict[str, RawToolCall] = {} + self.error: TestExecutionError | None = None + self.on_fixed_speech_complete: Callable[[], Awaitable[None]] | None = None + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TTSSpeakFrame): + text = frame.text.strip() + if text: + self.outputs.append(text) + if frame.append_to_context: + self.context.add_message({"role": "assistant", "content": text}) + self.window.touch() + if self.on_fixed_speech_complete is not None: + await self.on_fixed_speech_complete() + return + + if isinstance(frame, FixedSpeechPlaybackMarkerFrame): + await frame.completion.mark_played() + self.window.touch() + return + + if isinstance(frame, LLMFullResponseStartFrame): + self.window.active_llm += 1 + self.window.touch() + elif isinstance(frame, LLMFullResponseEndFrame): + self.window.active_llm = max(0, self.window.active_llm - 1) + self.window.touch() + await self.push_frame(frame, direction) + # There is no TTS/output transport. This releases provider/tool + # follow-up logic that normally waits for a speaking boundary. + await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + return + elif isinstance(frame, FunctionCallInProgressFrame): + call = RawToolCall( + id=frame.tool_call_id, + function_name=frame.function_name, + arguments=frame.arguments or {}, + ) + self.tool_calls.append(call) + self._tool_by_id[call.id] = call + self.window.pending_tools.add(call.id) + self.window.touch() + elif isinstance(frame, FunctionCallResultFrame): + call = self._tool_by_id.get(frame.tool_call_id) + if call is None: + call = RawToolCall( + id=frame.tool_call_id, + function_name=frame.function_name, + arguments=frame.arguments or {}, + ) + self.tool_calls.append(call) + self._tool_by_id[call.id] = call + call.result = frame.result + call.duration_ms = max(0, round((time.monotonic() - call._started_at) * 1000)) + status = frame.result.get("status") if isinstance(frame.result, dict) else None + call.outcome = "error" if status in {"error", "failed", "timeout"} else "success" + self.window.pending_tools.discard(call.id) + self.window.touch() + elif isinstance(frame, ErrorFrame): + self.error = TestExecutionError( + code="PIPELINE_ERROR", + message=frame.error or "Pipeline 执行失败", + stage="model" if frame.processor else "pipeline", + retryable=True, + ) + self.window.touch() + elif isinstance(frame, (EndFrame, CancelFrame)): + self.window.touch() + + await self.push_frame(frame, direction) + + +def _context_messages(context_turns: list[ContextTurn]) -> list[dict[str, Any]]: + messages: list[dict[str, Any]] = [] + latest_call_by_name: dict[str, str] = {} + for index, turn in enumerate(context_turns): + if turn.role in {"agent", "user"}: + messages.append( + { + "role": "assistant" if turn.role == "agent" else "user", + "content": turn.content, + } + ) + continue + + call_id = turn.tool_call_id or latest_call_by_name.get(turn.tool_name or "") + call_id = call_id or f"context_tool_{index}_{uuid4().hex[:8]}" + if turn.role == "tool_call": + latest_call_by_name[turn.tool_name or ""] = call_id + messages.append( + { + "role": "assistant", + "tool_calls": [ + { + "id": call_id, + "type": "function", + "function": { + "name": turn.tool_name, + "arguments": json.dumps( + json.loads(turn.content), ensure_ascii=False + ), + }, + } + ], + } + ) + else: + result_payload = json.loads(turn.content) + if turn.is_error: + result_payload = ( + {**result_payload, "status": "error"} + if isinstance(result_payload, dict) + else {"status": "error", "data": result_payload} + ) + messages.append( + { + "role": "tool", + "tool_call_id": call_id, + "content": json.dumps(result_payload, ensure_ascii=False), + } + ) + return messages + + +async def _wait_for_quiescence( + capture: _TextCaptureProcessor, + runner_task: asyncio.Task[None], + activity_marker: int, + *, + require_activity: bool = True, +) -> None: + while True: + if capture.error is not None: + raise capture.error + if runner_task.done(): + await runner_task + return + has_activity = capture.window.activity_count > activity_marker + settled = time.monotonic() - capture.window.last_activity_at >= IDLE_SETTLE_SECONDS + if capture.window.idle and settled and (has_activity or not require_activity): + return + await asyncio.sleep(0.05) + + +class TextPipelineRunner: + """Execute all fixed input turns in one isolated, transportless pipeline.""" + + async def run( + self, + cfg: AssistantConfig, + definition: TestCaseDefinition, + *, + timeout_seconds: int, + ) -> TextCaseResult: + try: + async with asyncio.timeout(timeout_seconds): + return await self._run(cfg, definition) + except TimeoutError as exc: + raise TestExecutionError( + code="PIPELINE_TIMEOUT", + message=f"测试用例执行超过 {timeout_seconds} 秒", + stage="pipeline", + retryable=True, + ) from exc + + async def _run( + self, + cfg: AssistantConfig, + definition: TestCaseDefinition, + ) -> TextCaseResult: + if cfg.runtimeMode != "pipeline": + raise TestExecutionError( + code="UNSUPPORTED_RUNTIME_MODE", + message="第一版文本测试只支持 Pipeline 运行模式", + stage="pipeline", + ) + if cfg.type not in {"prompt", "workflow"}: + raise TestExecutionError( + code="UNSUPPORTED_ASSISTANT_TYPE", + message=f"第一版文本测试暂不支持 {cfg.type} 类型助手", + stage="pipeline", + ) + + brain = build_brain(cfg) + seed_messages = _context_messages(definition.context_turns) + knowledge_config = cfg.knowledge_retrieval_config or {} + knowledge_mode = str(knowledge_config.get("mode") or "automatic") + + def with_knowledge_hint(text: str) -> str: + if cfg.type != "prompt" or not cfg.knowledge_base_id: + return text + hint = ( + AUTOMATIC_KNOWLEDGE_HINT + if knowledge_mode == "automatic" + else ON_DEMAND_KNOWLEDGE_HINT + ) + return "\n\n".join(part for part in (text, hint) if part) + + system_prompt = with_knowledge_hint(brain.system_prompt(cfg)) + context = LLMContext( + messages=( + seed_messages + if cfg.type == "workflow" + else [{"role": "system", "content": system_prompt}, *seed_messages] + ) + ) + + graph_settings = cfg.graph.get("settings") or {} + default_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultLlmResourceId") or "") + ) + llm_cfg = ( + config_with_resource(cfg, default_resource) + if cfg.type == "workflow" and default_resource is not None + else cfg + ) + llm = brain.build_llm(llm_cfg, context) + llm_services: dict[str, FrameProcessor] = {} + current_llm_service: FrameProcessor = llm + if cfg.type == "workflow": + llm, llm_services, current_llm_service = build_workflow_llm_switcher(cfg, llm) + + aggregators = LLMContextAggregatorPair(context) + user_aggregator = aggregators.user() + assistant_aggregator = aggregators.assistant() + capture = _TextCaptureProcessor(context) + knowledge = KnowledgeRetrievalProcessor( + cfg.knowledge_base_id if knowledge_mode == "automatic" else None, + top_n=int(knowledge_config.get("top_n", knowledge_config.get("topN", 5))), + score_threshold=float( + knowledge_config.get( + "score_threshold", knowledge_config.get("scoreThreshold", 0.0) + ) + ), + ) + pipeline = Pipeline( + [ + user_aggregator, + UserTurnRoutingProcessor(brain), + knowledge, + llm, + capture, + assistant_aggregator, + ] + ) + worker = PipelineWorker( + pipeline, + params=PipelineParams(enable_metrics=False), + enable_rtvi=False, + enable_turn_tracking=False, + idle_timeout_secs=None, + ) + runner = WorkerRunner(handle_sigint=False, check_dangling_tasks=False) + worker_started = asyncio.Event() + + @worker.event_handler("on_pipeline_started") + async def on_pipeline_started(_worker, _frame): + worker_started.set() + + executor_holder: dict[str, MockToolExecutor] = {} + + def create_test_executor(store) -> MockToolExecutor: + executor = MockToolExecutor(store) + executor_holder["executor"] = executor + return executor + + async def queue_call_end(reason: str) -> None: + logger.debug(f"文本测试收到会话结束请求: {reason}") + await worker.queue_frame(EndFrame(reason=reason)) + + call_end = CallEndCoordinator(queue_call_end) + + async def finish_fixed_speech_if_ending() -> None: + if call_end.ending: + await call_end.finish() + + capture.on_fixed_speech_complete = finish_fixed_speech_if_ending + current_service = current_llm_service + + async def switch_services( + llm_resource_id: str | None, + _asr_resource_id: str | None, + _tts_resource_id: str | None, + ) -> None: + nonlocal current_service + target = ( + llm_services.get(llm_resource_id) + if llm_resource_id + else current_llm_service + ) + if target is None: + raise ValueError(f"Workflow LLM 资源未加载:{llm_resource_id}") + if target is current_service: + return + await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) + current_service = target + + def set_system_prompt(text: str) -> None: + messages = context.get_messages() + text = with_knowledge_hint(text) + if messages and messages[0].get("role") == "system": + messages[0] = {"role": "system", "content": text} + else: + messages.insert(0, {"role": "system", "content": text}) + + knowledge_schema: FunctionSchema | None = None + if ( + cfg.type == "prompt" + and cfg.knowledge_base_id + and knowledge_mode == "on_demand" + ): + knowledge_schema = FunctionSchema( + name="search_knowledge_base", + description=( + "在当前助手绑定的知识库中检索资料。" + f"知识库：{cfg.knowledge_base_name}。" + f"{cfg.knowledge_base_description}" + ), + properties={ + "query": { + "type": "string", + "description": "完整问题或检索关键词", + } + }, + required=["query"], + ) + + async def search_bound_knowledge(params: FunctionCallParams) -> None: + query = str(params.arguments.get("query") or "").strip() + if not query: + await params.result_callback( + {"status": "error", "message": "检索问题为空"} + ) + return + try: + async with SessionLocal() as session: + results = await search_knowledge( + session, + cfg.knowledge_base_id or "", + query, + top_k=int( + knowledge_config.get( + "top_n", + knowledge_config.get("topN", 5), + ) + ), + score_threshold=float( + knowledge_config.get( + "score_threshold", + knowledge_config.get("scoreThreshold", 0.0), + ) + ), + ) + await params.result_callback( + {"status": "ok", "results": results} + ) + except Exception as exc: # noqa: BLE001 - return tool errors to LLM + logger.warning(f"文本测试知识库检索失败: {exc}") + await params.result_callback( + {"status": "error", "message": "知识库检索暂时不可用"} + ) + + llm.register_function("search_knowledge_base", search_bound_knowledge) + + def set_tools(schemas=None) -> None: + visible = list(schemas or []) + if knowledge_schema is not None: + visible.append(knowledge_schema) + if visible: + context.set_tools(ToolsSchema(standard_tools=visible)) + else: + context.set_tools() + + await brain.setup( + cfg, + BrainRuntime( + context=context, + llm=llm, + queue_frame=worker.queue_frame, + set_system_prompt=set_system_prompt, + set_tools=set_tools, + call_end=call_end, + session_id=f"test_{uuid4().hex}", + client_tools=TextClientToolPort(), + worker=worker, + context_aggregator=aggregators, + transport=None, + switch_services=switch_services, + set_knowledge_scope=knowledge.set_scope, + set_vision_scope=lambda _scope: None, + vision_function=None, + set_input_enabled=lambda _enabled: None, + apply_turn_config=lambda _enabled, _config: asyncio.sleep(0), + flow_global_functions=[], + tool_executor_factory=create_test_executor, + ), + ) + executor = executor_holder["executor"] + + @assistant_aggregator.event_handler("on_assistant_turn_started") + async def on_assistant_turn_started(_aggregator): + capture.window.active_assistant += 1 + capture.window.touch() + await brain.on_assistant_text_start(uuid4().hex) + + @assistant_aggregator.event_handler("on_assistant_turn_stopped") + async def on_assistant_turn_stopped(_aggregator, message): + content = str(message.content or "").strip() + capture.window.active_assistant = max( + 0, capture.window.active_assistant - 1 + ) + if content: + capture.outputs.append(content) + capture.window.touch() + await brain.on_assistant_text_end( + uuid4().hex, + content, + bool(getattr(message, "interrupted", False)), + ) + if call_end.ending: + # Text tests have no audio transport. A fully aggregated + # response is therefore the equivalent of playback ending. + await call_end.finish() + + all_tool_behaviors = [ + behavior + for turn in definition.turns + for behavior in turn.behaviors + if behavior.type == "tool_call" + ] + executor.set_turn_behaviors(all_tool_behaviors) + + runner_task: asyncio.Task[None] | None = None + try: + await brain.run_preflight() + if executor.unmocked_calls: + raise TestExecutionError( + code="UNMOCKED_TOOL_CALL", + message="启动阶段调用了未配置 Mock 的工具: " + + "、".join(executor.unmocked_calls), + stage="tool", + ) + + await runner.add_workers(worker) + runner_task = asyncio.create_task( + runner.run(), name=f"text-test-{uuid4().hex[:12]}" + ) + await asyncio.wait_for( + worker_started.wait(), timeout=WORKER_START_TIMEOUT_SECONDS + ) + + marker = capture.window.activity_count + await brain.on_connected(greeting_pending=False) + await brain.on_client_ready() + await _wait_for_quiescence( + capture, + runner_task, + marker, + require_activity=False, + ) + # Startup messages initialize the session but are not a scripted + # user turn result in the fixed-input editor contract. + capture.outputs.clear() + capture.tool_calls.clear() + + results: list[RawTurnResult] = [] + for index, turn in enumerate(definition.turns): + if worker.has_finished(): + raise TestExecutionError( + code="PIPELINE_ENDED_EARLY", + message=f"Pipeline 在第 {index + 1} 轮输入前已经结束", + stage="pipeline", + retryable=False, + ) + result = await self._run_turn( + worker, + runner_task, + capture, + executor, + turn, + index, + ) + results.append(result) + + transcript = [ + { + "role": str(message.get("role") or ""), + "content": str(message.get("content") or ""), + } + for message in context.get_messages() + if message.get("role") in {"user", "assistant"} + and str(message.get("content") or "").strip() + ] + return TextCaseResult(turns=results, transcript=transcript) + except TestExecutionError: + raise + except asyncio.CancelledError: + raise + except Exception as exc: + raise TestExecutionError( + code="PIPELINE_EXECUTION_FAILED", + message=str(exc) or type(exc).__name__, + stage="model", + retryable=True, + ) from exc + finally: + if not worker.has_finished(): + await worker.cancel(reason=INTERNAL_CANCEL_REASON) + if runner_task is not None: + await asyncio.gather(runner_task, return_exceptions=True) + + async def _run_turn( + self, + worker: PipelineWorker, + runner_task: asyncio.Task[None], + capture: _TextCaptureProcessor, + executor: MockToolExecutor, + turn: FixedInputTurn, + index: int, + ) -> RawTurnResult: + tool_behaviors = [ + behavior for behavior in turn.behaviors if behavior.type == "tool_call" + ] + executor.set_turn_behaviors(tool_behaviors) + activity_marker = capture.window.activity_count + output_marker = len(capture.outputs) + tool_marker = len(capture.tool_calls) + await worker.queue_frame( + LLMMessagesAppendFrame( + messages=[{"role": "user", "content": turn.user_input}], + run_llm=True, + ) + ) + await _wait_for_quiescence(capture, runner_task, activity_marker) + if executor.unmocked_calls: + raise TestExecutionError( + code="UNMOCKED_TOOL_CALL", + message="本轮调用了未配置 Mock 的工具: " + + "、".join(executor.unmocked_calls), + stage="tool", + ) + return RawTurnResult( + id=turn.id, + index=index, + user_input=turn.user_input, + assistant_reply="\n\n".join(capture.outputs[output_marker:]).strip(), + tool_calls=list(capture.tool_calls[tool_marker:]), + ) diff --git a/backend/test_schemas.py b/backend/test_schemas.py new file mode 100644 index 0000000..297836e --- /dev/null +++ b/backend/test_schemas.py @@ -0,0 +1,386 @@ +"""Contracts for persisted text test cases and batch execution results.""" + +from __future__ import annotations + +import json +import re +from datetime import datetime +from typing import Annotated, Any, Literal, Union + +from pydantic import Field, field_validator, model_validator + +from schemas import CamelModel + + +TestCaseInputMode = Literal[ + "fixed_script_text", + "fixed_script_turn_voice", + "fixed_script_continuous_voice", + "user_sim_text", + "user_sim_voice", +] +ContextRole = Literal["agent", "user", "tool_call", "tool_result"] +EvaluationKind = Literal["reply", "tool_call", "overall"] +EvaluationStatus = Literal["pass", "fail"] +CaseStatus = Literal["waiting", "running", "pass", "fail", "error", "skipped"] +RunStatus = Literal["queued", "running", "completed", "cancelled"] +ErrorStage = Literal["pipeline", "model", "tool", "evaluation"] +StopReason = Literal["manual", "assertion_failure", "execution_error"] + + +class ContextTurn(CamelModel): + role: ContextRole + content: str = Field(max_length=20_000) + tool_name: str | None = Field(default=None, max_length=128) + tool_call_id: str | None = Field(default=None, max_length=128) + is_error: bool | None = None + + @model_validator(mode="after") + def validate_content(self): + self.content = self.content.strip() + self.tool_call_id = (self.tool_call_id or "").strip() or None + if not self.content: + raise ValueError("上下文内容不能为空") + if self.role in {"tool_call", "tool_result"}: + self.tool_name = (self.tool_name or "").strip() + if not self.tool_name: + raise ValueError("工具上下文必须填写工具名称") + try: + payload = json.loads(self.content) + except json.JSONDecodeError as exc: + raise ValueError("工具上下文必须是有效 JSON") from exc + if self.role == "tool_call" and not isinstance(payload, dict): + raise ValueError("Tool Call 参数必须是 JSON 对象") + return self + + +class ToolParamAssertion(CamelModel): + name: str = Field(min_length=1, max_length=128) + match_mode: Literal["exact", "regex", "llm"] + value: str = Field(min_length=1, max_length=4_000) + + @field_validator("name", "value") + @classmethod + def strip_required_text(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("参数名称和值不能为空") + return value + + @model_validator(mode="after") + def validate_regex(self): + if self.match_mode == "regex": + try: + re.compile(self.value) + except re.error as exc: + raise ValueError(f"正则表达式无效: {exc}") from exc + return self + + +class ToolMockResponse(CamelModel): + outcome: Literal["success", "error"] = "success" + body: str = Field(default='{"status":"ok"}', max_length=100_000) + delay_ms: int = Field(default=0, ge=0, le=60_000) + + @model_validator(mode="after") + def validate_json_body(self): + if not self.body.strip(): + raise ValueError("Mock 工具返回值不能为空") + try: + json.loads(self.body) + except json.JSONDecodeError as exc: + raise ValueError("Mock 工具返回值必须是有效 JSON") from exc + return self + + +class ReplyExpectedBehavior(CamelModel): + id: str = Field(min_length=1, max_length=128) + type: Literal["reply"] = "reply" + assertion_type: Literal["keyword", "llm"] + keywords: list[str] = Field(default_factory=list, max_length=50) + keyword_match_mode: Literal["any", "all"] = "any" + negate_keywords: bool = False + llm_criteria: str = Field(default="", max_length=4_000) + + @model_validator(mode="after") + def validate_assertion(self): + self.keywords = [item.strip() for item in self.keywords if item.strip()] + self.llm_criteria = self.llm_criteria.strip() + if self.assertion_type == "keyword" and not self.keywords: + raise ValueError("关键词断言至少需要一个关键词") + if self.assertion_type == "llm" and not self.llm_criteria: + raise ValueError("LLM 判断要求不能为空") + return self + + +class ToolCallExpectedBehavior(CamelModel): + id: str = Field(min_length=1, max_length=128) + type: Literal["tool_call"] = "tool_call" + tool_id: str = Field(min_length=1, max_length=40) + function_name: str = Field(min_length=1, max_length=128) + expectation: Literal["called", "not_called"] + min_calls: int = Field(default=1, ge=0, le=100) + max_calls: int | None = Field(default=None, ge=0, le=100) + param_assertions: list[ToolParamAssertion] = Field(default_factory=list, max_length=50) + mock_response: ToolMockResponse = Field(default_factory=ToolMockResponse) + + @model_validator(mode="after") + def validate_call_expectation(self): + self.tool_id = self.tool_id.strip() + self.function_name = self.function_name.strip() + if not self.tool_id or not self.function_name: + raise ValueError("必须选择有效的工具") + if self.expectation == "called" and self.min_calls < 1: + raise ValueError("应调用工具时最少调用次数必须大于 0") + if self.max_calls is not None and self.max_calls < self.min_calls: + raise ValueError("最多调用次数不能小于最少调用次数") + return self + + +ExpectedBehavior = Annotated[ + Union[ReplyExpectedBehavior, ToolCallExpectedBehavior], + Field(discriminator="type"), +] + + +class FixedInputTurn(CamelModel): + id: str = Field(min_length=1, max_length=128) + user_input: str = Field(min_length=1, max_length=20_000) + behaviors: list[ExpectedBehavior] = Field(default_factory=list, max_length=100) + + @field_validator("user_input") + @classmethod + def strip_user_input(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("用户输入不能为空") + return value + + +class OverallCriterion(CamelModel): + id: str = Field(min_length=1, max_length=128) + type: Literal["llm"] = "llm" + name: str = Field(min_length=1, max_length=80) + criteria: str = Field(min_length=1, max_length=2_000) + + @field_validator("name", "criteria") + @classmethod + def strip_criterion(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("整体评估标准不能为空") + return value + + +class TestCaseDefinition(CamelModel): + context_turns: list[ContextTurn] = Field(default_factory=list, max_length=100) + turns: list[FixedInputTurn] = Field(min_length=1, max_length=100) + overall_criteria: list[OverallCriterion] = Field(default_factory=list, max_length=50) + + @model_validator(mode="after") + def validate_has_expectation(self): + if not any(turn.behaviors for turn in self.turns) and not self.overall_criteria: + raise ValueError("至少配置一项有效的预期行为或整体评估标准") + return self + + @model_validator(mode="after") + def validate_tool_context_pairs(self): + pending: list[tuple[str | None, str]] = [] + seen_ids: set[str] = set() + for index, turn in enumerate(self.context_turns): + if turn.role == "tool_call": + call_id = (turn.tool_call_id or "").strip() or None + if call_id and call_id in seen_ids: + raise ValueError(f"上下文第 {index + 1} 条 Tool Call ID 重复") + if call_id: + seen_ids.add(call_id) + pending.append((call_id, turn.tool_name or "")) + continue + if turn.role != "tool_result": + continue + + result_id = (turn.tool_call_id or "").strip() or None + matched_index = -1 + for pending_index in range(len(pending) - 1, -1, -1): + call_id, tool_name = pending[pending_index] + if result_id: + matches = call_id == result_id + else: + matches = tool_name == turn.tool_name + if matches: + matched_index = pending_index + break + if matched_index < 0: + raise ValueError( + f"上下文第 {index + 1} 条 Tool Result 没有匹配的 Tool Call" + ) + _call_id, tool_name = pending.pop(matched_index) + if tool_name != turn.tool_name: + raise ValueError( + f"上下文第 {index + 1} 条 Tool Result 的工具名称不匹配" + ) + + if pending: + names = "、".join(tool_name for _call_id, tool_name in pending) + raise ValueError(f"上下文 Tool Call 缺少对应的 Tool Result：{names}") + return self + + +class TestSuiteCreate(CamelModel): + name: str = Field(min_length=1, max_length=128) + description: str = Field(default="", max_length=2_048) + + @field_validator("name") + @classmethod + def strip_name(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("测试集名称不能为空") + return value + + +class TestSuiteUpdate(TestSuiteCreate): + pass + + +class TestSuiteOut(CamelModel): + id: str + name: str + description: str + case_count: int = 0 + passed_count: int = 0 + run_count: int = 0 + updated_at: datetime + + +class TestCaseWrite(CamelModel): + name: str = Field(min_length=1, max_length=128) + description: str = Field(default="", max_length=2_048) + input_mode: TestCaseInputMode = "fixed_script_text" + context_turns: list[ContextTurn] = Field(default_factory=list, max_length=100) + turns: list[FixedInputTurn] = Field(min_length=1, max_length=100) + overall_criteria: list[OverallCriterion] = Field(default_factory=list, max_length=50) + + @field_validator("name") + @classmethod + def strip_name(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("测试用例名称不能为空") + return value + + @model_validator(mode="after") + def validate_mvp_definition(self): + if self.input_mode != "fixed_script_text": + raise ValueError("第一版只支持固定脚本 · 文字") + TestCaseDefinition( + context_turns=self.context_turns, + turns=self.turns, + overall_criteria=self.overall_criteria, + ) + return self + + def definition(self) -> TestCaseDefinition: + return TestCaseDefinition( + context_turns=self.context_turns, + turns=self.turns, + overall_criteria=self.overall_criteria, + ) + + +class TestCaseOut(TestCaseWrite): + id: str + suite_id: str + last_result: Literal["pass", "fail", "not_run"] + sort_order: int + updated_at: datetime + + +class TestCaseOrderIn(CamelModel): + case_ids: list[str] = Field(min_length=1, max_length=1_000) + + +class TestCaseBulkDeleteIn(CamelModel): + case_ids: list[str] = Field(min_length=1, max_length=1_000) + + +class BatchRunConfig(CamelModel): + concurrency: int = Field(default=3, ge=1, le=20) + timeout_secs: int = Field(default=60, ge=1, le=600) + failure_strategy: Literal["continue", "stop_on_fail"] = "continue" + error_retry_count: int = Field(default=0, ge=0, le=3) + error_strategy: Literal["continue", "stop_on_error"] = "continue" + + +class BatchRunCreate(CamelModel): + assistant_id: str = Field(min_length=1, max_length=40) + evaluator_model_resource_id: str = Field(min_length=1, max_length=40) + case_ids: list[str] = Field(min_length=1, max_length=1_000) + config: BatchRunConfig = Field(default_factory=BatchRunConfig) + title: str | None = Field(default=None, max_length=256) + + @field_validator("case_ids") + @classmethod + def unique_case_ids(cls, values: list[str]) -> list[str]: + normalized = list(dict.fromkeys(value.strip() for value in values if value.strip())) + if not normalized: + raise ValueError("至少选择一个测试用例") + return normalized + + +class BatchExecutionError(CamelModel): + code: str + message: str + stage: ErrorStage + retryable: bool + + +class BatchEvaluationResult(CamelModel): + id: str + label: str + kind: EvaluationKind + status: EvaluationStatus + expected: str + actual: str + reason: str = "" + + +class BatchToolCallRecord(CamelModel): + id: str + function_name: str + arguments_json: str + result_json: str + outcome: Literal["success", "error"] + duration_ms: int = 0 + + +class BatchTurnResult(CamelModel): + id: str + index: int + user_input: str + assistant_reply: str + tool_calls: list[BatchToolCallRecord] = Field(default_factory=list) + evaluations: list[BatchEvaluationResult] = Field(default_factory=list) + + +class BatchRunCaseOut(CamelModel): + id: str + name: str + status: CaseStatus + turns: list[BatchTurnResult] = Field(default_factory=list) + overall_criteria: list[BatchEvaluationResult] = Field(default_factory=list) + attempt_count: int + max_attempts: int + execution_error: BatchExecutionError | None = None + + +class BatchRunSnapshotOut(CamelModel): + id: str + status: RunStatus + title: str + assistant_name: str + config: dict[str, Any] + cases: list[BatchRunCaseOut] + started_at: datetime + finished_at: datetime | None + stop_reason: StopReason | None diff --git a/frontend/src/components/assistant-editor/debug-auto-test.tsx b/frontend/src/components/assistant-editor/debug-auto-test.tsx index b730b4a..c39cc56 100644 --- a/frontend/src/components/assistant-editor/debug-auto-test.tsx +++ b/frontend/src/components/assistant-editor/debug-auto-test.tsx @@ -66,15 +66,10 @@ export type DebugTestCase = { failAtTurn?: number; }; -/** 测试集：Debug Drawer 只按「当前助手关联」浏览，不做管理 */ +/** 测试集：Debug Drawer 只浏览和选择，不做管理。 */ export type DebugTestSuite = { id: string; name: string; - /** - * 关联助手 id。MVP 用 `"*"` 表示「当前正在调试的助手都可见」； - * 接上真实 API 后改成具体 assistantId 列表。 - */ - assistantIds: string[]; caseIds: string[]; }; @@ -295,15 +290,11 @@ export const MOCK_TEST_CASES: DebugTestCase[] = [ ), ]; -/** - * 当前助手关联的测试集（MVP mock）。 - * `"*"` = 任意正在调试的助手都可见，避免把「系统全部测试集」甩进 Drawer。 - */ +/** 测试集（MVP mock）；测试集本身不绑定助手。 */ export const MOCK_TEST_SUITES: DebugTestSuite[] = [ { id: "suite-accident-basic", name: "事故快处基础流程", - assistantIds: ["*"], caseIds: [ "tc-dual-car", "tc-no-injury", @@ -314,7 +305,6 @@ export const MOCK_TEST_SUITES: DebugTestSuite[] = [ { id: "suite-clarify", name: "异常输入与澄清", - assistantIds: ["*"], caseIds: [ "tc-hello", "tc-noise", @@ -326,13 +316,11 @@ export const MOCK_TEST_SUITES: DebugTestSuite[] = [ { id: "suite-tools", name: "工具与系统联动", - assistantIds: ["*"], caseIds: ["tc-injury-transfer", "tc-transfer-tool", "tc-end-call-tool"], }, { id: "suite-realtime-voice", name: "实时语音交互", - assistantIds: ["*"], caseIds: ["tc-hello", "tc-noise", "tc-barge-in", "tc-long-silence"], }, ]; @@ -347,15 +335,6 @@ function findSuite(id: string | null): DebugTestSuite | null { return MOCK_TEST_SUITES.find((item) => item.id === id) ?? null; } -/** 只返回当前助手关联的测试集 */ -function suitesForAssistant(assistantId: string | null): DebugTestSuite[] { - return MOCK_TEST_SUITES.filter((suite) => { - if (suite.assistantIds.includes("*")) return true; - if (!assistantId) return false; - return suite.assistantIds.includes(assistantId); - }); -} - function casesInSuites(suites: DebugTestSuite[]): DebugTestCase[] { const seen = new Set(); const cases: DebugTestCase[] = []; @@ -657,23 +636,18 @@ function TestCasePickerPopover({ onOpenChange, selectedId, onSelect, - assistantId, trigger, }: { open: boolean; onOpenChange: (open: boolean) => void; selectedId: string | null; onSelect: (id: string) => void; - assistantId: string | null; trigger: React.ReactNode; }) { const [query, setQuery] = useState(""); const [activeSuiteId, setActiveSuiteId] = useState(null); - const suites = useMemo( - () => suitesForAssistant(assistantId), - [assistantId], - ); + const suites = MOCK_TEST_SUITES; const scopedCases = useMemo(() => casesInSuites(suites), [suites]); const activeSuite = findSuite(activeSuiteId); @@ -810,7 +784,7 @@ function TestCasePickerPopover({ {suites.length === 0 ? (
- 当前助手还没有关联测试集 + 暂无测试集
) : ( suites.map((suite) => ( @@ -851,7 +825,6 @@ function TestCasePickerPopover({ export function AutoTestRunBar({ state, - assistantId, onSelectCase, onClearCase, onStart, @@ -859,7 +832,6 @@ export function AutoTestRunBar({ onRerun, }: { state: AutoTestRunState; - assistantId: string | null; onSelectCase: (id: string) => void; onClearCase: () => void; onStart: () => void; @@ -877,7 +849,6 @@ export function AutoTestRunBar({ onOpenChange={setPickerOpen} selectedId={state.selectedId} onSelect={onSelectCase} - assistantId={assistantId} trigger={ } > + {runError && ( +

+ {runError} +

+ )} ); @@ -517,7 +556,7 @@ export function BatchTestPage() { } > + {runError && ( +

+ {runError} +

+ )} @@ -578,8 +630,12 @@ export function BatchTestPage() { aria-describedby="batch-run-readiness" onClick={handleStart} > - - 开始批量测试 + {starting ? ( + + ) : ( + + )} + {starting ? "正在启动…" : "开始批量测试"} } @@ -593,6 +649,15 @@ export function BatchTestPage() { } /> + {runError && ( +

+ {runError} +

+ )} +
{ @@ -603,36 +668,77 @@ export function BatchTestPage() { } title="运行目标" - description="选择本次批量测试要对齐的助手配置" + description="分别选择生成回复的助手和判断结果的 LLM" > - +
+ + + +
@@ -647,12 +753,26 @@ export function BatchTestPage() { title="选择测试范围" description="搜索测试集或用例，按需组合本次运行范围" > - + {loadingScope ? ( +
+ + 正在加载测试范围… +
+ ) : scopeError ? ( +

+ {scopeError} +

+ ) : ( + + )} diff --git a/frontend/src/components/pages/TestCasesPage.tsx b/frontend/src/components/pages/TestCasesPage.tsx index a3344bb..d2579e2 100644 --- a/frontend/src/components/pages/TestCasesPage.tsx +++ b/frontend/src/components/pages/TestCasesPage.tsx @@ -48,41 +48,24 @@ import { import { Input } from "@/components/ui/input"; import { ListToolbar } from "@/components/ui/list-toolbar"; import { SearchInput } from "@/components/ui/search-input"; -import { - Select, - SelectContent, - SelectItem, - SelectTrigger, - SelectValue, -} from "@/components/ui/select"; import { cloneOverallCriteria, cloneTurns, createEmptyFixedInputTurn, - createTestCase, - createTestSuite, DEFAULT_INPUT_MODE, - duplicateTestCase, - duplicateTestSuite, getTestCaseValidationMessage, - getTestSuite, - listTestCases, - listTestSuites, normalizeOverallCriteria, - removeTestCase, - removeTestCases, - removeTestSuite, - reorderTestCases, - suiteCaseStats, TEST_CASE_INPUT_MODE_LABEL, TEST_CASE_INPUT_MODE_SHORT_LABEL, - updateTestCase, - updateTestSuite, type TestCase, type TestCaseInputMode, type TestSuite, } from "@/data/test-suites"; -import { assistantsApi, type Assistant } from "@/lib/api"; +import { + testCasesApi, + testSuitesApi, + type TestCaseWrite, +} from "@/lib/api"; // 路由驱动: // /test/cases → list @@ -109,16 +92,33 @@ export function TestCasesPage(props: TestCasesPageProps) { function SuiteListView() { const router = useRouter(); - const [suites, setSuites] = useState(() => listTestSuites()); + const [suites, setSuites] = useState([]); + const [loading, setLoading] = useState(true); + const [loadError, setLoadError] = useState(""); const [search, setSearch] = useState(""); const [currentPage, setCurrentPage] = useState(1); const [deletingId, setDeletingId] = useState(null); + async function reloadSuites() { + try { + setLoadError(""); + setSuites(await testSuitesApi.list()); + } catch (error) { + setLoadError(error instanceof Error ? error.message : "加载测试集失败"); + } finally { + setLoading(false); + } + } + + useEffect(() => { + void reloadSuites(); + }, []); + const filtered = useMemo(() => { const keyword = search.trim().toLowerCase(); return suites.filter((suite) => { if (!keyword) return true; - return [suite.name, suite.assistantName, suite.id] + return [suite.name, suite.description, suite.id] .join(" ") .toLowerCase() .includes(keyword); @@ -136,13 +136,16 @@ function SuiteListView() { router.push(`/test/cases/${suite.id}`); } - function duplicateSuite(suite: TestSuite) { - const copied = duplicateTestSuite(suite.id); - if (!copied) return; - setSuites(listTestSuites()); + async function duplicateSuite(suite: TestSuite) { + try { + await testSuitesApi.duplicate(suite.id); + await reloadSuites(); + } catch (error) { + setLoadError(error instanceof Error ? error.message : "复制测试集失败"); + } } - function removeSuite(suite: TestSuite) { + async function removeSuite(suite: TestSuite) { if ( !window.confirm( `确定删除测试集“${suite.name}”及其全部测试用例吗？`, @@ -151,9 +154,14 @@ function SuiteListView() { return; } setDeletingId(suite.id); - removeTestSuite(suite.id); - setSuites(listTestSuites()); - setDeletingId(null); + try { + await testSuitesApi.remove(suite.id); + await reloadSuites(); + } catch (error) { + setLoadError(error instanceof Error ? error.message : "删除测试集失败"); + } finally { + setDeletingId(null); + } } return ( @@ -186,14 +194,26 @@ function SuiteListView() { } /> + {loadError && ( +

+ {loadError} +

+ )} + rows={paginated} rowKey={(suite) => suite.id} onRowClick={openSuite} empty={{ - title: suites.length === 0 ? "暂无测试集" : "未找到匹配的测试集", + title: loading + ? "正在加载测试集…" + : suites.length === 0 + ? "暂无测试集" + : "未找到匹配的测试集", description: - suites.length === 0 + loading + ? "请稍候。" + : suites.length === 0 ? "点击右上角「新建测试集」开始。" : "请调整关键词后再试。", }} @@ -222,19 +242,12 @@ function SuiteListView() { ), }, - { - key: "assistant", - header: "关联助手", - width: "md:w-[160px]", - cellClassName: "text-muted-foreground", - cell: (suite) => suite.assistantName || "—", - }, { key: "caseCount", header: "用例数", width: "md:w-[96px]", cellClassName: "tabular-nums text-muted-foreground", - cell: (suite) => suiteCaseStats(suite.id).total, + cell: (suite) => suite.caseCount ?? 0, }, { key: "actions", @@ -313,42 +326,31 @@ function SuiteListView() { function SuiteCreateView() { const router = useRouter(); - const [assistants, setAssistants] = useState([]); - const [loadingAssistants, setLoadingAssistants] = useState(true); const [name, setName] = useState(""); - const [assistantName, setAssistantName] = useState(""); const [creating, setCreating] = useState(false); + const [createError, setCreateError] = useState(""); - useEffect(() => { - void (async () => { - try { - const list = await assistantsApi.list(); - setAssistants(list); - if (list[0]) setAssistantName(list[0].name); - } catch { - setAssistantName("视频快处助手"); - } finally { - setLoadingAssistants(false); - } - })(); - }, []); - - function confirmCreate() { + async function confirmCreate() { if (!name.trim() || creating) return; setCreating(true); - const saved = createTestSuite({ - name, - description: "", - assistantName, - }); - router.push(`/test/cases/${saved.id}`); + setCreateError(""); + try { + const saved = await testSuitesApi.create({ + name: name.trim(), + description: "", + }); + router.push(`/test/cases/${saved.id}`); + } catch (error) { + setCreateError(error instanceof Error ? error.message : "创建测试集失败"); + setCreating(false); + } } return ( - -
-
- 关联助手 -
- {loadingAssistants ? ( -
- - 正在加载助手… -
- ) : assistants.length > 0 ? ( - - ) : ( - setAssistantName(event.target.value)} - placeholder="助手名称" - className="border-hairline-strong bg-background text-foreground placeholder:text-muted-soft" - /> - )} -
- + {createError && ( +

+ {createError} +

+ )}
diff --git a/frontend/src/data/batch-run.ts b/frontend/src/data/batch-run.ts index 4852497..9cc1af5 100644 --- a/frontend/src/data/batch-run.ts +++ b/frontend/src/data/batch-run.ts @@ -1,14 +1,4 @@ -/** - * 批量测试运行 — 前端 mock。 - * 真实执行引擎接入前，用本地状态模拟进度与结果。 - */ - -import type { - ExpectedBehavior, - ReplyExpectedBehavior, - TestCase, - ToolCallExpectedBehavior, -} from "@/data/test-suites"; +/** 批量测试运行的前后端共享结果契约与纯展示辅助函数。 */ export type BatchCaseStatus = | "waiting" @@ -34,24 +24,13 @@ export const BATCH_ERROR_STRATEGY_LABEL: Record = { stop_on_error: "标记错误并停止", }; -type BatchExecutionError = { +export type BatchExecutionError = { code: string; message: string; stage: "pipeline" | "model" | "tool" | "evaluation"; retryable: boolean; }; -export type BatchRunCase = { - id: string; - name: string; - status: BatchCaseStatus; - turns: BatchTurnResult[]; - overallCriteria: BatchEvaluationResult[]; - attemptCount: number; - maxAttempts: number; - executionError: BatchExecutionError | null; -}; - export type BatchEvaluationResult = { id: string; label: string; @@ -80,13 +59,30 @@ export type BatchTurnResult = { evaluations: BatchEvaluationResult[]; }; +export type BatchRunCase = { + id: string; + name: string; + status: BatchCaseStatus; + turns: BatchTurnResult[]; + overallCriteria: BatchEvaluationResult[]; + attemptCount: number; + maxAttempts: number; + executionError: BatchExecutionError | null; +}; + export type BatchRunPhase = "config" | "running" | "completed"; +export type BatchRunStatus = "queued" | "running" | "completed" | "cancelled"; export type BatchRunSnapshot = { + id: string; + status: BatchRunStatus; title: string; assistantName: string; config: { suiteCount: number; + evaluatorModelResourceId: string; + evaluatorModelResourceName: string; + evaluatorModel: string; concurrency: number; timeoutSecs: number; failureStrategy: BatchFailureStrategy; @@ -99,246 +95,6 @@ export type BatchRunSnapshot = { stopReason: "manual" | "assertion_failure" | "execution_error" | null; }; -export type BatchMockExecutionPlan = Record; - -/** 预定部分用例失败，方便演示失败展开态 */ -function shouldFail(index: number, item: TestCase): boolean { - if (item.lastResult === "fail") return true; - return index % 4 === 2; -} - -/** 稳定产生少量执行错误，便于检查重试和继续/停止流程。 */ -function shouldError(index: number): boolean { - return index % 5 === 3; -} - -function replyExpectation(behavior: ReplyExpectedBehavior): string { - if (behavior.assertionType === "llm") return behavior.llmCriteria; - const relation = behavior.negateKeywords - ? "不应包含" - : behavior.keywordMatchMode === "all" - ? "应包含全部" - : "应至少包含其一"; - return `${relation}：${behavior.keywords.join("、")}`; -} - -function toolExpectation(behavior: ToolCallExpectedBehavior): string { - if (behavior.expectation === "not_called") { - return `不应调用 ${behavior.functionName}`; - } - const count = - behavior.maxCalls === null - ? `${behavior.minCalls} 次以上` - : behavior.minCalls === behavior.maxCalls - ? `${behavior.minCalls} 次` - : `${behavior.minCalls}–${behavior.maxCalls} 次`; - return `应调用 ${behavior.functionName} ${count}`; -} - -function mockAssistantReply(behaviors: ExpectedBehavior[]): string { - const positiveKeyword = behaviors.find( - (behavior): behavior is ReplyExpectedBehavior => - behavior.type === "reply" && - behavior.assertionType === "keyword" && - !behavior.negateKeywords && - behavior.keywords.length > 0, - ); - if (positiveKeyword) { - return `好的，我已了解。请继续说明${positiveKeyword.keywords.slice(0, 2).join("和")}。`; - } - return "好的，我已记录当前信息，我们继续处理。"; -} - -function mockToolArguments(behavior: ToolCallExpectedBehavior): string { - const entries = behavior.paramAssertions.map((item) => [ - item.name, - item.matchMode === "exact" ? item.value : `mock_${item.name}`, - ]); - return JSON.stringify(Object.fromEntries(entries), null, 2); -} - -function createBehaviorEvaluation( - behavior: ExpectedBehavior, - fail: boolean, -): { - evaluation: BatchEvaluationResult; - toolCalls: BatchToolCallRecord[]; -} { - if (behavior.type === "reply") { - const reason = fail - ? behavior.assertionType === "keyword" - ? "实际回复未满足关键词规则。" - : "LLM 判断认为回复未达到该项语义要求。" - : ""; - return { - evaluation: { - id: behavior.id, - label: - behavior.assertionType === "keyword" - ? "回复 · 关键词" - : "回复 · LLM 判断", - kind: "reply", - status: fail ? "fail" : "pass", - expected: replyExpectation(behavior), - actual: fail - ? "已为您转接人工处理。" - : "实际回复满足本条内容要求。", - reason, - }, - toolCalls: [], - }; - } - - const shouldRecordCall = - behavior.expectation === "called" ? !fail : fail; - const toolCalls: BatchToolCallRecord[] = shouldRecordCall - ? [ - { - id: `call_${behavior.id}`, - functionName: behavior.functionName, - argumentsJson: mockToolArguments(behavior), - resultJson: behavior.mockResponse.body, - outcome: behavior.mockResponse.outcome, - durationMs: behavior.mockResponse.delayMs + 84, - }, - ] - : []; - const actual = shouldRecordCall - ? `实际调用 1 次 ${behavior.functionName}` - : `实际调用 0 次 ${behavior.functionName}`; - const reason = fail - ? behavior.expectation === "not_called" - ? "检测到本轮不应发生的工具调用。" - : "未检测到满足次数要求的工具调用。" - : ""; - return { - evaluation: { - id: behavior.id, - label: `工具 · ${behavior.functionName}`, - kind: "tool_call", - status: fail ? "fail" : "pass", - expected: toolExpectation(behavior), - actual, - reason, - }, - toolCalls, - }; -} - -function createCaseResult( - item: TestCase, - caseIndex: number, - maxAttempts: number, -): { result: BatchRunCase; plannedError: BatchExecutionError | null } { - const plannedError = shouldError(caseIndex); - const plannedFailure = !plannedError && shouldFail(caseIndex, item); - let failureAssigned = false; - const turns = item.turns.map((turn, turnIndex) => { - const toolCalls: BatchToolCallRecord[] = []; - const evaluations = turn.behaviors.map((behavior) => { - const fail = plannedFailure && !failureAssigned; - if (fail) failureAssigned = true; - const result = createBehaviorEvaluation(behavior, fail); - toolCalls.push(...result.toolCalls); - return result.evaluation; - }); - return { - id: turn.id, - index: turnIndex, - userInput: turn.userInput, - assistantReply: mockAssistantReply(turn.behaviors), - toolCalls, - evaluations, - }; - }); - const overallCriteria = item.overallCriteria.map((criterion) => { - const fail = plannedFailure && !failureAssigned; - if (fail) failureAssigned = true; - return { - id: criterion.id, - label: criterion.name, - kind: "overall" as const, - status: fail ? ("fail" as const) : ("pass" as const), - expected: criterion.criteria, - actual: fail - ? "整段对话未完整达到该业务目标。" - : "整段对话达到该业务目标。", - reason: fail ? "LLM 判断认为整段对话未满足该项标准。" : "", - }; - }); - return { - result: { - id: item.id, - name: item.name, - status: "waiting", - turns, - overallCriteria, - attemptCount: 0, - maxAttempts, - executionError: null, - }, - plannedError: plannedError - ? { - code: "PIPELINE_TIMEOUT", - message: "等待 pipeline 完成回复时超过单用例超时时间。", - stage: "pipeline", - retryable: true, - } - : null, - }; -} - -export function createMockBatchRun(input: { - title: string; - assistantName: string; - cases: TestCase[]; - concurrency: number; - timeoutSecs: number; - failureStrategy: BatchFailureStrategy; - errorRetryCount: number; - errorStrategy: BatchErrorStrategy; -}): { - snapshot: BatchRunSnapshot; - executionPlan: BatchMockExecutionPlan; -} { - const preparedCases = input.cases.map((item, index) => - createCaseResult(item, index, input.errorRetryCount + 1), - ); - const executionPlan = Object.fromEntries( - preparedCases.flatMap(({ result, plannedError }) => - plannedError ? [[result.id, plannedError]] : [], - ), - ); - - return { - snapshot: { - title: input.title, - assistantName: input.assistantName, - config: { - suiteCount: new Set(input.cases.map((item) => item.suiteId)).size, - concurrency: input.concurrency, - timeoutSecs: input.timeoutSecs, - failureStrategy: input.failureStrategy, - errorRetryCount: input.errorRetryCount, - errorStrategy: input.errorStrategy, - }, - startedAt: new Date().toISOString(), - finishedAt: null, - stopReason: null, - cases: preparedCases.map(({ result }) => result), - }, - executionPlan, - }; -} - -export function hasEvaluationFailure(item: BatchRunCase): boolean { - const evaluations = [ - ...item.turns.flatMap((turn) => turn.evaluations), - ...item.overallCriteria, - ]; - return evaluations.some((evaluation) => evaluation.status === "fail"); -} - export function firstEvaluationFailureReason(item: BatchRunCase): string { const evaluations = [ ...item.turns.flatMap((turn) => turn.evaluations), @@ -358,15 +114,13 @@ export function countByStatus(cases: BatchRunCase[]) { waiting: 0, skipped: 0, }; - for (const item of cases) { - counts[item.status] += 1; - } + for (const item of cases) counts[item.status] += 1; return counts; } export function formatRunTime(iso: string): string { const date = new Date(iso); - const pad = (n: number) => String(n).padStart(2, "0"); + const pad = (value: number) => String(value).padStart(2, "0"); return `${date.getFullYear()}-${pad(date.getMonth() + 1)}-${pad(date.getDate())} ${pad(date.getHours())}:${pad(date.getMinutes())}:${pad(date.getSeconds())}`; } diff --git a/frontend/src/data/test-suites.ts b/frontend/src/data/test-suites.ts index e375d8a..ccd0889 100644 --- a/frontend/src/data/test-suites.ts +++ b/frontend/src/data/test-suites.ts @@ -1,5 +1,5 @@ /** - * 测试集 / 测试用例 — 管理页用的本地 mock。 + * 测试集 / 测试用例的前后端共享契约与编辑器辅助函数。 * 两层：Test Suite → Test Case。 * MVP 仅运行固定文字脚本；其它输入模式保留为产品路线提示。 */ @@ -12,7 +12,7 @@ export type TestCaseInputMode = | "user_sim_text" | "user_sim_voice"; -type TestCaseResult = "pass" | "fail" | "not_run"; +export type TestCaseResult = "pass" | "fail" | "not_run"; export type AssertionType = "keyword" | "llm"; @@ -105,8 +105,9 @@ export type TestSuite = { id: string; name: string; description: string; - /** 关联助手展示名；MVP 直接存文案，接 API 后可改成 assistantId */ - assistantName: string; + caseCount: number; + passedCount: number; + runCount: number; updatedAt: string; }; @@ -160,12 +161,60 @@ export function getTestCaseValidationMessage( } if (context.role === "tool_call" || context.role === "tool_result") { try { - JSON.parse(context.content); + const payload = JSON.parse(context.content); + if ( + context.role === "tool_call" && + (payload === null || Array.isArray(payload) || typeof payload !== "object") + ) { + return `上下文第 ${index + 1} 条 Tool Call 参数必须是 JSON 对象`; + } } catch { return `上下文第 ${index + 1} 条工具数据必须是有效 JSON`; } } } + const pendingToolCalls: Array<{ id: string | null; toolName: string }> = []; + const seenToolCallIds = new Set(); + for (let index = 0; index < item.contextTurns.length; index += 1) { + const context = item.contextTurns[index]; + if (context.role === "tool_call") { + const id = context.toolCallId?.trim() || null; + if (id && seenToolCallIds.has(id)) { + return `上下文第 ${index + 1} 条 Tool Call ID 重复`; + } + if (id) seenToolCallIds.add(id); + pendingToolCalls.push({ id, toolName: context.toolName?.trim() ?? "" }); + continue; + } + if (context.role !== "tool_result") continue; + const resultId = context.toolCallId?.trim() || null; + let matchedIndex = -1; + for ( + let pendingIndex = pendingToolCalls.length - 1; + pendingIndex >= 0; + pendingIndex -= 1 + ) { + const pending = pendingToolCalls[pendingIndex]; + if ( + resultId ? pending.id === resultId : pending.toolName === context.toolName?.trim() + ) { + matchedIndex = pendingIndex; + break; + } + } + if (matchedIndex < 0) { + return `上下文第 ${index + 1} 条 Tool Result 没有匹配的 Tool Call`; + } + const [matched] = pendingToolCalls.splice(matchedIndex, 1); + if (matched.toolName !== context.toolName?.trim()) { + return `上下文第 ${index + 1} 条 Tool Result 的工具名称不匹配`; + } + } + if (pendingToolCalls.length > 0) { + return `上下文 Tool Call 缺少对应的 Tool Result：${pendingToolCalls + .map((item) => item.toolName) + .join("、")}`; + } const emptyTurnIndex = item.turns.findIndex( (turn) => !turn.userInput.trim(), ); @@ -311,26 +360,19 @@ export function getExpectedBehaviorValidationMessage( return null; } -let turnSeq = 1; -let behaviorSeq = 1; -let criterionSeq = 1; - -function nextTurnId() { - const id = `turn_${String(turnSeq).padStart(4, "0")}`; - turnSeq += 1; - return id; +function newEditorId(prefix: string) { + const randomPart = globalThis.crypto?.randomUUID + ? globalThis.crypto.randomUUID().replaceAll("-", "") + : `${Date.now().toString(36)}${Math.random().toString(36).slice(2)}`; + return `${prefix}_${randomPart}`; } function nextBehaviorId() { - const id = `beh_${String(behaviorSeq).padStart(4, "0")}`; - behaviorSeq += 1; - return id; + return newEditorId("beh"); } function nextCriterionId() { - const id = `criterion_${String(criterionSeq).padStart(4, "0")}`; - criterionSeq += 1; - return id; + return newEditorId("criterion"); } export function createOverallCriterion( @@ -365,7 +407,7 @@ export function createEmptyFixedInputTurn( userInput = "", ): FixedInputTurn { return { - id: nextTurnId(), + id: newEditorId("turn"), userInput, behaviors: [], }; @@ -435,10 +477,6 @@ function cloneBehavior(behavior: ExpectedBehavior): ExpectedBehavior { }; } -function cloneBehaviorWithNewId(behavior: ExpectedBehavior): ExpectedBehavior { - return { ...cloneBehavior(behavior), id: nextBehaviorId() }; -} - export function cloneTurns(turns: FixedInputTurn[]): FixedInputTurn[] { return turns.map((turn) => ({ id: turn.id, @@ -447,480 +485,6 @@ export function cloneTurns(turns: FixedInputTurn[]): FixedInputTurn[] { })); } -/** 新建用例时复制轮次并分配新 id */ -function cloneTurnsWithNewIds(turns: FixedInputTurn[]): FixedInputTurn[] { - return turns.map((turn) => ({ - id: nextTurnId(), - userInput: turn.userInput, - behaviors: turn.behaviors.map(cloneBehaviorWithNewId), - })); -} - -const INITIAL_SUITES: TestSuite[] = [ - { - id: "suite_001", - name: "事故基础流程", - description: "核心业务流程和正常事故处理", - assistantName: "视频快处助手", - updatedAt: "2026-08-05T10:20:00+08:00", - }, - { - id: "suite_002", - name: "异常输入与澄清", - description: "模糊表达、误打断、主动唤醒等", - assistantName: "视频快处助手", - updatedAt: "2026-08-04T16:40:00+08:00", - }, - { - id: "suite_003", - name: "实时语音交互", - description: "打断、延迟、VAD、语音链路等", - assistantName: "视频快处助手", - updatedAt: "2026-08-03T09:15:00+08:00", - }, -]; - -type RawCaseSeed = Omit & { - inputMode?: TestCaseInputMode; - overallCriteria?: OverallCriterion[]; -}; - -function seedReplyTurn( - seedId: string, - userInput: string, - expectation: Omit, -): FixedInputTurn { - return { - id: `turn_seed_${seedId}`, - userInput, - behaviors: [ - { - id: `beh_seed_${seedId}`, - type: "reply", - ...expectation, - }, - ], - }; -} - -const RAW_CASE_SEEDS: RawCaseSeed[] = [ - { - id: "tc_001", - suiteId: "suite_001", - name: "正常双车事故开场", - description: "开场后用户补充事故经过", - lastResult: "pass", - contextTurns: [], - turns: [ - seedReplyTurn("001", "我这里刚刚撞了一下。", { - assertionType: "keyword", - keywords: ["经过", "描述", "受伤"], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: "", - }), - ], - updatedAt: "2026-08-05T10:18:00+08:00", - }, - { - id: "tc_002", - suiteId: "suite_001", - name: "有人伤转人工", - description: "用户提到人伤时应引导转人工", - lastResult: "pass", - contextTurns: [], - turns: [ - seedReplyTurn("002", "有人受伤了，流血不止。", { - assertionType: "llm", - keywords: [], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: - "Agent 应识别人员受伤风险，明确告知将转接人工，并避免继续推进普通快处流程。", - }), - ], - overallCriteria: [ - { - id: "criterion_seed_001", - type: "llm", - name: "人伤流程正确", - criteria: - "整段对话应正确识别人员受伤场景，及时并准确转接人工处理。", - }, - { - id: "criterion_seed_002", - type: "llm", - name: "不推进普通快处", - criteria: - "确认存在人员受伤后，不应继续引导用户进入普通事故快处流程。", - }, - ], - updatedAt: "2026-08-05T10:12:00+08:00", - }, - { - id: "tc_101", - suiteId: "suite_002", - name: "用户说“喂”", - description: "验证主动唤醒回复且业务状态不推进", - lastResult: "pass", - contextTurns: [], - turns: [ - seedReplyTurn("101", "喂", { - assertionType: "keyword", - keywords: ["我在", "请说", "继续"], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: "", - }), - ], - updatedAt: "2026-08-04T16:35:00+08:00", - }, - { - id: "tc_102", - suiteId: "suite_002", - name: "用户只说“嗯”", - description: "短促确认不应误推进流程", - lastResult: "fail", - contextTurns: [], - turns: [ - seedReplyTurn("102", "嗯", { - assertionType: "llm", - keywords: [], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: - "Agent 应当回应用户的主动唤醒，同时继续引导用户描述事故情况，不应该无响应。", - }), - ], - updatedAt: "2026-08-04T16:20:00+08:00", - }, - { - id: "tc_103", - suiteId: "suite_002", - name: "模糊事故描述", - description: "地点含糊时应主动澄清", - lastResult: "not_run", - contextTurns: [], - turns: [ - seedReplyTurn("103", "就在那边……撞了一下。", { - assertionType: "keyword", - keywords: ["哪里", "路口", "路名", "再说"], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: "", - }), - ], - updatedAt: "2026-08-04T15:30:00+08:00", - }, - { - id: "tc_201", - suiteId: "suite_003", - name: "用户打断播报", - description: "播报中打断后正确切换聆听并承接", - lastResult: "pass", - contextTurns: [], - turns: [ - seedReplyTurn("201", "等一下，对方走了。", { - assertionType: "llm", - keywords: [], - keywordMatchMode: "any", - negateKeywords: false, - llmCriteria: - "Agent 应立即停止原播报思路，确认已听到用户新信息，并围绕「对方离开」继续询问。", - }), - ], - updatedAt: "2026-08-03T09:10:00+08:00", - }, -]; - -/** 按套件出现顺序写入 sortOrder，并补齐可选字段。 */ -const INITIAL_CASES: TestCase[] = (() => { - const counters = new Map(); - return RAW_CASE_SEEDS.map((item) => { - const order = counters.get(item.suiteId) ?? 0; - counters.set(item.suiteId, order + 1); - return { - ...item, - inputMode: item.inputMode ?? DEFAULT_INPUT_MODE, - turns: cloneTurns(item.turns), - overallCriteria: cloneOverallCriteria(item.overallCriteria), - sortOrder: order, - }; - }); -})(); - -/** 会话内可变的 mock 仓库（刷新页面会重置） */ -let suites = [...INITIAL_SUITES]; -let cases = structuredClone(INITIAL_CASES); -let suiteSeq = 4; -let caseSeq = 300; - -function nowIso() { - return new Date().toISOString(); -} - -function nextSuiteId() { - const id = `suite_${String(suiteSeq).padStart(3, "0")}`; - suiteSeq += 1; - return id; -} - -function nextCaseId() { - const id = `tc_${String(caseSeq).padStart(3, "0")}`; - caseSeq += 1; - return id; -} - -export function listTestSuites(): TestSuite[] { - return [...suites].sort( - (a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime(), - ); -} - -export function getTestSuite(id: string): TestSuite | null { - return suites.find((item) => item.id === id) ?? null; -} - -function getTestCase(id: string): TestCase | null { - return cases.find((item) => item.id === id) ?? null; -} - -export function listTestCases(suiteId: string): TestCase[] { - return cases - .filter((item) => item.suiteId === suiteId) - .sort( - (a, b) => - a.sortOrder - b.sortOrder || a.name.localeCompare(b.name, "zh-CN"), - ); -} - -export function suiteCaseStats(suiteId: string): { - total: number; - passed: number; - run: number; -} { - const items = cases.filter((item) => item.suiteId === suiteId); - const run = items.filter((item) => item.lastResult !== "not_run"); - const passed = run.filter((item) => item.lastResult === "pass"); - return { total: items.length, passed: passed.length, run: run.length }; -} - -export function createTestSuite(input: { - name: string; - description: string; - assistantName: string; -}): TestSuite { - const suite: TestSuite = { - id: nextSuiteId(), - name: input.name.trim(), - description: input.description.trim(), - assistantName: input.assistantName.trim() || "未关联助手", - updatedAt: nowIso(), - }; - suites = [suite, ...suites]; - return suite; -} - -export function updateTestSuite( - id: string, - patch: Partial>, -): TestSuite | null { - const index = suites.findIndex((item) => item.id === id); - if (index < 0) return null; - const next = { - ...suites[index], - ...patch, - updatedAt: nowIso(), - }; - suites = [...suites.slice(0, index), next, ...suites.slice(index + 1)]; - return next; -} - -export function removeTestSuite(id: string): boolean { - const before = suites.length; - suites = suites.filter((item) => item.id !== id); - cases = cases.filter((item) => item.suiteId !== id); - return suites.length < before; -} - -/** 复制测试集及其全部用例；名称加「（副本）」 */ -export function duplicateTestSuite(id: string): TestSuite | null { - const source = getTestSuite(id); - if (!source) return null; - - const copied = createTestSuite({ - name: `${source.name}（副本）`, - description: source.description, - assistantName: source.assistantName, - }); - - const sourceCases = listTestCases(id); - const clonedCases: TestCase[] = sourceCases.map((item, index) => { - const turns = cloneTurnsWithNewIds(item.turns); - return { - id: nextCaseId(), - suiteId: copied.id, - name: item.name, - description: item.description, - inputMode: item.inputMode, - lastResult: "not_run" as const, - contextTurns: item.contextTurns.map((turn) => ({ ...turn })), - turns, - overallCriteria: cloneOverallCriteria(item.overallCriteria), - sortOrder: index, - updatedAt: nowIso(), - }; - }); - cases = [...cases, ...clonedCases]; - return copied; -} - -export function createTestCase(input: { - suiteId: string; - name?: string; - description?: string; -}): TestCase | null { - if (!getTestSuite(input.suiteId)) return null; - const maxOrder = cases - .filter((item) => item.suiteId === input.suiteId) - .reduce((max, item) => Math.max(max, item.sortOrder), -1); - const turns = [createEmptyFixedInputTurn()]; - const item: TestCase = { - id: nextCaseId(), - suiteId: input.suiteId, - name: (input.name ?? "未命名用例").trim() || "未命名用例", - description: (input.description ?? "").trim(), - inputMode: DEFAULT_INPUT_MODE, - lastResult: "not_run", - contextTurns: [], - turns, - overallCriteria: [], - sortOrder: maxOrder + 1, - updatedAt: nowIso(), - }; - cases = [...cases, item]; - updateTestSuite(input.suiteId, {}); - return item; -} - -/** 复制单条用例；名称加「（副本）」，排在同套件末尾 */ -export function duplicateTestCase(id: string): TestCase | null { - const source = getTestCase(id); - if (!source || !getTestSuite(source.suiteId)) return null; - - const maxOrder = cases - .filter((item) => item.suiteId === source.suiteId) - .reduce((max, item) => Math.max(max, item.sortOrder), -1); - - const turns = cloneTurnsWithNewIds(source.turns); - const copied: TestCase = { - id: nextCaseId(), - suiteId: source.suiteId, - name: `${source.name}（副本）`, - description: source.description, - inputMode: source.inputMode, - lastResult: "not_run", - contextTurns: source.contextTurns.map((turn) => ({ ...turn })), - turns, - overallCriteria: cloneOverallCriteria(source.overallCriteria), - sortOrder: maxOrder + 1, - updatedAt: nowIso(), - }; - cases = [...cases, copied]; - updateTestSuite(source.suiteId, {}); - return copied; -} - -type TestCasePatch = Partial< - Pick< - TestCase, - | "name" - | "description" - | "inputMode" - | "contextTurns" - | "turns" - | "overallCriteria" - | "lastResult" - > ->; - -export function updateTestCase( - id: string, - patch: TestCasePatch, -): TestCase | null { - const index = cases.findIndex((item) => item.id === id); - if (index < 0) return null; - const current = cases[index]; - const nextTurns = - patch.turns !== undefined - ? cloneTurns( - patch.turns.length > 0 ? patch.turns : [createEmptyFixedInputTurn()], - ) - : current.turns; - const nextOverallCriteria = - patch.overallCriteria !== undefined - ? cloneOverallCriteria(patch.overallCriteria) - : cloneOverallCriteria(current.overallCriteria); - const next: TestCase = { - ...current, - ...patch, - turns: nextTurns, - overallCriteria: nextOverallCriteria, - updatedAt: nowIso(), - }; - cases = [...cases.slice(0, index), next, ...cases.slice(index + 1)]; - updateTestSuite(next.suiteId, {}); - return next; -} - -export function removeTestCase(id: string): boolean { - const existing = cases.find((item) => item.id === id); - if (!existing) return false; - cases = cases.filter((item) => item.id !== id); - renumberSortOrder(existing.suiteId); - updateTestSuite(existing.suiteId, {}); - return true; -} - -/** 批量删除；返回受影响的 suiteId（若有） */ -export function removeTestCases(ids: string[]): string | null { - const idSet = new Set(ids); - const affected = cases.find((item) => idSet.has(item.id)); - if (!affected) return null; - const suiteId = affected.suiteId; - cases = cases.filter((item) => !idSet.has(item.id)); - renumberSortOrder(suiteId); - updateTestSuite(suiteId, {}); - return suiteId; -} - -/** - * 按给定 id 顺序写回 sortOrder（应包含该套件全部用例 id）。 - * 用于拖拽结束后持久化顺序。 - */ -export function reorderTestCases(suiteId: string, orderedIds: string[]): void { - const orderMap = new Map(orderedIds.map((id, index) => [id, index])); - cases = cases.map((item) => { - if (item.suiteId !== suiteId) return item; - const nextOrder = orderMap.get(item.id); - if (nextOrder === undefined) return item; - return { ...item, sortOrder: nextOrder }; - }); - updateTestSuite(suiteId, {}); -} - -function renumberSortOrder(suiteId: string) { - const ordered = cases - .filter((item) => item.suiteId === suiteId) - .sort((a, b) => a.sortOrder - b.sortOrder); - const orderMap = new Map(ordered.map((item, index) => [item.id, index])); - cases = cases.map((item) => { - if (item.suiteId !== suiteId) return item; - const nextOrder = orderMap.get(item.id); - return nextOrder === undefined ? item : { ...item, sortOrder: nextOrder }; - }); -} - export function formatUpdatedAt(value?: string | null) { if (!value) return "—"; const date = new Date(value); diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 2eee52c..e74b98b 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -6,6 +6,15 @@ */ import { loginPathWithReturnTo } from "@/lib/auth-redirect"; +import type { BatchRunSnapshot } from "@/data/batch-run"; +import type { + ContextTurn, + FixedInputTurn, + OverallCriterion, + TestCase, + TestCaseInputMode, + TestSuite, +} from "@/data/test-suites"; const LOOPBACK_HOSTS = new Set(["localhost", "127.0.0.1", "[::1]"]); @@ -769,3 +778,94 @@ export const webrtcApi = { iceServers: () => request<{ iceServers: IceServerConfig[] }>("/api/webrtc/ice-servers"), }; + +// ---------- 测试用例与批量运行 ---------- +export type TestSuiteWrite = { + name: string; + description: string; +}; + +export type TestCaseWrite = { + name: string; + description: string; + inputMode: TestCaseInputMode; + contextTurns: ContextTurn[]; + turns: FixedInputTurn[]; + overallCriteria: OverallCriterion[]; +}; + +export const testSuitesApi = { + list: () => request("/api/test-suites"), + get: (id: string) => request(`/api/test-suites/${id}`), + create: (body: TestSuiteWrite) => + request("/api/test-suites", { + method: "POST", + body: JSON.stringify(body), + }), + update: (id: string, body: TestSuiteWrite) => + request(`/api/test-suites/${id}`, { + method: "PUT", + body: JSON.stringify(body), + }), + duplicate: (id: string) => + request(`/api/test-suites/${id}/duplicate`, { method: "POST" }), + remove: (id: string) => + request<{ ok: boolean }>(`/api/test-suites/${id}`, { method: "DELETE" }), +}; + +export const testCasesApi = { + list: (suiteId?: string) => + request( + `/api/test-cases${suiteId ? `?suiteId=${encodeURIComponent(suiteId)}` : ""}`, + ), + get: (id: string) => request(`/api/test-cases/${id}`), + create: (suiteId: string, body: TestCaseWrite) => + request(`/api/test-suites/${suiteId}/cases`, { + method: "POST", + body: JSON.stringify(body), + }), + update: (id: string, body: TestCaseWrite) => + request(`/api/test-cases/${id}`, { + method: "PUT", + body: JSON.stringify(body), + }), + duplicate: (id: string) => + request(`/api/test-cases/${id}/duplicate`, { method: "POST" }), + remove: (id: string) => + request<{ ok: boolean }>(`/api/test-cases/${id}`, { method: "DELETE" }), + bulkRemove: (caseIds: string[]) => + request<{ ok: boolean; deleted: number }>("/api/test-cases/bulk-delete", { + method: "POST", + body: JSON.stringify({ caseIds }), + }), + reorder: (suiteId: string, caseIds: string[]) => + request<{ ok: boolean }>(`/api/test-suites/${suiteId}/case-order`, { + method: "PUT", + body: JSON.stringify({ caseIds }), + }), +}; + +export const batchRunsApi = { + create: (body: { + assistantId: string; + evaluatorModelResourceId: string; + caseIds: string[]; + title?: string; + config: { + concurrency: number; + timeoutSecs: number; + failureStrategy: "continue" | "stop_on_fail"; + errorRetryCount: number; + errorStrategy: "continue" | "stop_on_error"; + }; + }) => + request("/api/test-runs", { + method: "POST", + body: JSON.stringify(body), + }), + get: (id: string) => request(`/api/test-runs/${id}`), + cancel: (id: string) => + request(`/api/test-runs/${id}/cancel`, { + method: "POST", + }), +};