HaiLin/ai-ticket-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add intelligent support workflow

Browse Source
This commit is contained in:
hailin
2026-07-27 19:00:35 +08:00
parent 93065c0a00
commit 1a49ce4eae
38 changed files with 5644 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
backend/apps/tickets/tests/test_stage_3_1.py Normal file
View File

@@ -0,0 +1,362 @@
import json
from types import SimpleNamespace
import pytest
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.core.management import call_command
from apps.agents.models import AgentDefinition, AgentRun
from apps.agents.tests.helpers import create_deployment
from apps.audit.models import OperationAuditLog
from apps.knowledge.models import KnowledgeBase, KnowledgeItem
from apps.knowledge.services import index_resolved_ticket, rebuild_item_chunks
from apps.modelhub.services import CompletionResult
from apps.solutions.models import Solution
from apps.tickets.models import (
Ticket,
TicketCategory,
TicketSupportAnalysis,
)
from apps.tickets.support_analysis import (
SupportAnalysisError,
validate_support_analysis_output,
)
def publish_support_agent():
deployment = create_deployment()
call_command("publish_support_analysis", deployment_id=deployment.id)
@pytest.mark.django_db
def test_support_analysis_uses_only_published_scoped_references_and_is_audited(
admin_client,
django_capture_on_commit_callbacks,
monkeypatch,
):
publish_support_agent()
monkeypatch.setattr(
"apps.agents.tasks.execute_agent_run.delay",
lambda run_id: SimpleNamespace(id=f"task-{run_id}"),
)
category, _ = TicketCategory.objects.get_or_create(
code="technical_issue",
defaults={"name": "技术问题"},
)
base = KnowledgeBase.objects.create(name="正式知识", enabled=True)
published_item = KnowledgeItem.objects.create(
knowledge_base=base,
title="订单接口超时排查",
problem="订单接口返回 TIMEOUT-504",
applicable_scope="生产环境订单系统",
root_cause="下游连接池耗尽",
troubleshooting_steps="检查连接池和慢请求",
solution="恢复连接池并限制慢请求",
status="published",
)
rebuild_item_chunks(published_item)
draft_item = KnowledgeItem.objects.create(
knowledge_base=base,
title="未审核订单结论",
problem="TIMEOUT-504",
solution="未经审核的操作",
status="draft",
)
published_solution = Solution.objects.create(
title="订单连接池恢复方案",
problem_signature="订单接口 TIMEOUT-504",
related_system="订单系统",
environment_scope="生产环境",
category=category,
root_cause="连接池耗尽",
solution="检查并恢复连接池",
status="published",
allow_auto_resolve=True,
)
draft_solution = Solution.objects.create(
title="未发布方案",
problem_signature="TIMEOUT-504",
category=category,
root_cause="未知",
solution="未经审核",
status="draft",
)
historical = Ticket.objects.create(
ticket_no="TK202607270201",
title="订单接口历史超时",
description="订单系统曾返回 TIMEOUT-504",
category=category,
status="resolved",
environment="生产环境",
related_system="订单系统",
)
index_resolved_ticket(historical)
ticket = Ticket.objects.create(
ticket_no="TK202607270202",
title="订单接口持续超时",
description="生产环境订单接口返回 TIMEOUT-504",
category=category,
status="in_progress",
environment="生产环境",
related_system="订单系统",
)
with django_capture_on_commit_callbacks(execute=True):
response = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "queued"
assert payload["run_public_id"]
references = payload["references"]
assert {item["source_type"] for item in references} == {
"knowledge",
"solution",
"historical_ticket",
}
assert published_item.id in {
item["source_id"] for item in references if item["source_type"] == "knowledge"
}
assert draft_item.id not in {
item["source_id"] for item in references if item["source_type"] == "knowledge"
}
knowledge_reference = next(
item for item in references if item["source_type"] == "knowledge"
)
assert knowledge_reference["retrieval"]["channels"] == [
"portable_text",
"portable_vector",
]
history_reference = next(
item for item in references if item["source_type"] == "historical_ticket"
)
assert history_reference["retrieval"]["channels"] == [
"portable_text",
"portable_vector",
]
assert TicketSupportAnalysis.objects.get(id=payload["id"]).retrieval_policy[
"policy_version"
] == (
"published_scope_hybrid_rerank_v2"
)
assert published_solution.id in {
item["source_id"] for item in references if item["source_type"] == "solution"
}
assert draft_solution.id not in {
item["source_id"] for item in references if item["source_type"] == "solution"
}
assert historical.id in {
item["source_id"] for item in references if item["source_type"] == "historical_ticket"
}
analysis = TicketSupportAnalysis.objects.select_related("run").get()
assert analysis.run.task_id == f"task-{analysis.run_id}"
assert analysis.retrieval_policy["allowed_statuses"]["knowledge"] == ["published"]
assert ticket.events.filter(
event_type="support_analysis_started",
agent_run_id=analysis.run_id,
).exists()
assert OperationAuditLog.objects.filter(
module="tickets",
action="support_analysis_started",
target_id=str(ticket.id),
).exists()
@pytest.mark.django_db
def test_support_analysis_reuses_active_run_and_rejects_closed_ticket(
admin_client,
django_capture_on_commit_callbacks,
monkeypatch,
):
publish_support_agent()
monkeypatch.setattr(
"apps.agents.tasks.execute_agent_run.delay",
lambda run_id: SimpleNamespace(id=f"task-{run_id}"),
)
ticket = Ticket.objects.create(
ticket_no="TK202607270203",
title="同步任务失败",
description="同步任务返回 SYNC-500",
status="pending",
)
with django_capture_on_commit_callbacks(execute=True):
first = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
second = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert first.status_code == 200
assert second.status_code == 200
assert first.json()["id"] == second.json()["id"]
assert TicketSupportAnalysis.objects.filter(ticket=ticket).count() == 1
assert AgentRun.objects.filter(definition__code="support-analysis").count() == 1
ticket.status = "closed"
ticket.save(update_fields=["status", "updated_at"])
TicketSupportAnalysis.objects.filter(ticket=ticket).delete()
response = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert response.status_code == 400
assert "已结束工单" in response.json()["detail"]
@pytest.mark.django_db
def test_support_analysis_output_cannot_escape_frozen_references(
admin_client,
django_capture_on_commit_callbacks,
monkeypatch,
):
publish_support_agent()
monkeypatch.setattr(
"apps.agents.tasks.execute_agent_run.delay",
lambda run_id: SimpleNamespace(id=f"task-{run_id}"),
)
solution = Solution.objects.create(
title="人工确认方案",
problem_signature="登录失败 AUTH-401",
root_cause="凭据失效",
solution="重新授权",
status="published",
allow_auto_resolve=False,
)
ticket = Ticket.objects.create(
ticket_no="TK202607270204",
title="登录失败",
description="登录返回 AUTH-401",
)
with django_capture_on_commit_callbacks(execute=True):
response = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert response.status_code == 200
analysis = TicketSupportAnalysis.objects.select_related("run__definition").get()
solution_reference = next(
item for item in analysis.references if item["source_type"] == "solution"
)
valid_output = {
"analysis_summary": "凭据可能已失效。",
"troubleshooting_steps": ["检查授权状态"],
"candidate_solution_ids": [solution.id],
"customer_reply_draft": "我们正在核对授权状态。",
"citations": [solution_reference["rank"]],
"auto_resolve_recommended": False,
}
validate_support_analysis_output(analysis.run, valid_output)
invalid_citation = {**valid_output, "citations": [999]}
with pytest.raises(SupportAnalysisError, match="无效引用"):
validate_support_analysis_output(analysis.run, invalid_citation)
invalid_solution = {**valid_output, "candidate_solution_ids": [solution.id + 999]}
with pytest.raises(SupportAnalysisError, match="检索范围外"):
validate_support_analysis_output(analysis.run, invalid_solution)
unsafe_auto_resolve = {**valid_output, "auto_resolve_recommended": True}
with pytest.raises(SupportAnalysisError, match="风险策略"):
validate_support_analysis_output(analysis.run, unsafe_auto_resolve)
@pytest.mark.django_db
def test_support_analysis_requires_permissions_and_published_agent(client):
call_command("bootstrap_governance")
user = get_user_model().objects.create_user(username="stage-31-no-permission")
user.user_permissions.add(
Permission.objects.get(
content_type__app_label="system",
codename="ticket.view",
)
)
ticket = Ticket.objects.create(
ticket_no="TK202607270205",
title="无法登录",
description="登录失败",
)
client.force_login(user)
list_response = client.get(f"/api/tickets/{ticket.id}/support-analyses")
response = client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert list_response.status_code == 403
assert response.status_code == 403
assert not TicketSupportAnalysis.objects.exists()
@pytest.mark.django_db
def test_support_analysis_reports_missing_published_agent(admin_client):
ticket = Ticket.objects.create(
ticket_no="TK202607270206",
title="无法登录",
description="登录失败",
)
response = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert response.status_code == 400
assert "尚未发布支持分析 Agent" in response.json()["detail"]
@pytest.mark.django_db
@pytest.mark.parametrize(
("citation", "expected_status", "expected_error"),
[
(1, "succeeded", ""),
(999, "failed", "untrusted_support_output"),
],
)
def test_support_analysis_runtime_enforces_frozen_citations(
admin_client,
django_capture_on_commit_callbacks,
monkeypatch,
citation,
expected_status,
expected_error,
):
publish_support_agent()
solution = Solution.objects.create(
title="登录授权恢复方案",
problem_signature="登录失败 AUTH-401",
root_cause="授权凭据失效",
solution="重新授权",
status="published",
)
ticket = Ticket.objects.create(
ticket_no=f"TK2026072703{citation}",
title="登录失败",
description="登录返回 AUTH-401",
)
model_output = {
"analysis_summary": "授权凭据可能失效。",
"troubleshooting_steps": ["核对授权状态"],
"candidate_solution_ids": [solution.id],
"customer_reply_draft": "我们正在核对授权状态。",
"citations": [citation],
"auto_resolve_recommended": False,
}
monkeypatch.setattr(
"apps.modelhub.services._request_completion",
lambda *args, **kwargs: CompletionResult(
json.dumps(model_output, ensure_ascii=False),
10,
20,
),
)
with django_capture_on_commit_callbacks(execute=True):
response = admin_client.post(f"/api/tickets/{ticket.id}/support-analyses")
assert response.status_code == 200
run = AgentRun.objects.get(public_id=response.json()["run_public_id"])
assert run.status == expected_status
assert run.error_code == expected_error
assert run.events.filter(
event_type="completed" if expected_status == "succeeded" else "failed"
).exists()
@pytest.mark.django_db
def test_publish_support_analysis_command_is_idempotent():
deployment = create_deployment()
call_command("publish_support_analysis", deployment_id=deployment.id)
call_command("publish_support_analysis", deployment_id=deployment.id)
definition = AgentDefinition.objects.get(code="support-analysis")
assert definition.active_version is not None
assert definition.versions.count() == 1
assert definition.active_version.steps.count() == 1

262
backend/apps/tickets/tests/test_stage_3_3.py Normal file
View File

@@ -0,0 +1,262 @@
import pytest
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.core.management import call_command
from django.test import override_settings
from apps.agents.models import AgentDefinition, AgentRun
from apps.agents.services import build_config_snapshot
from apps.agents.tests.helpers import create_deployment
from apps.solutions.models import Solution, TicketResolution, TicketSolutionLink
from apps.tickets.models import (
CustomerReplyRevision,
CustomerReplyTask,
Ticket,
TicketCategory,
TicketSupportAnalysis,
)
from apps.tickets.support_analysis import process_support_analysis_result
from apps.workitems.models import WorkItem
def build_completed_support_analysis(
actor,
*,
score=0.96,
auto_resolve_recommended=True,
allow_auto_resolve=True,
requires_human_action=False,
allowed_actions=None,
):
deployment = create_deployment()
call_command("publish_support_analysis", deployment_id=deployment.id)
definition = AgentDefinition.objects.select_related("active_version").get(
code="support-analysis"
)
version = definition.active_version
if allowed_actions is not None:
version.allowed_actions = allowed_actions
version.save(update_fields=["allowed_actions", "updated_at"])
category, _ = TicketCategory.objects.get_or_create(
code="business_issue",
defaults={"name": "业务问题"},
)
solution = Solution.objects.create(
title="订单状态恢复方案",
problem_signature="订单状态未更新",
related_system="订单系统",
environment_scope="生产环境",
category=category,
root_cause="状态回写延迟",
solution="重新同步订单状态并核对结果",
status="published",
allow_auto_resolve=allow_auto_resolve,
requires_human_action=requires_human_action,
)
ticket = Ticket.objects.create(
ticket_no=f"TK-STAGE33-{Solution.objects.count():04d}",
title="订单状态未更新",
description="生产环境订单状态长时间未更新",
category=category,
status="pending",
related_system="订单系统",
environment="生产环境",
)
reference = {
"rank": 1,
"source_type": "solution",
"source_id": solution.id,
"source_version": solution.updated_at.isoformat(),
"title": solution.title,
"excerpt": solution.solution,
"score": score,
"applicable_scope": {
"category_code": category.code,
"related_system": solution.related_system,
"environment": solution.environment_scope,
},
"auto_resolve_policy": {
"allow_auto_resolve": allow_auto_resolve,
"requires_human_action": requires_human_action,
"requires_external_confirmation": False,
"requires_production_change": False,
},
}
output = {
"analysis_summary": "订单状态回写延迟。",
"troubleshooting_steps": ["核对订单状态"],
"candidate_solution_ids": [solution.id],
"customer_reply_draft": "您好，订单状态已完成核对，请刷新后查看。",
"citations": [1],
"auto_resolve_recommended": auto_resolve_recommended,
}
run = AgentRun.objects.create(
definition=definition,
agent_version=version,
status="succeeded",
input_payload={},
final_output=output,
config_snapshot=build_config_snapshot(version),
created_by=actor,
)
analysis = TicketSupportAnalysis.objects.create(
ticket=ticket,
run=run,
query=ticket.description,
references=[reference],
retrieval_policy={"policy_version": "test"},
created_by=actor,
)
return analysis, solution, ticket
@pytest.mark.django_db
@override_settings(TICKET_AUTO_RESOLVE_ENABLED=True)
def test_support_analysis_auto_resolves_through_domain_service_and_is_idempotent(
admin_client,
):
actor = get_user_model().objects.get(is_superuser=True)
analysis, solution, ticket = build_completed_support_analysis(actor)
process_support_analysis_result(analysis.run)
process_support_analysis_result(analysis.run)
analysis.refresh_from_db()
ticket.refresh_from_db()
solution.refresh_from_db()
resolution = TicketResolution.objects.get(ticket=ticket)
assert ticket.status == "resolved"
assert resolution.source == "agent"
assert resolution.linked_solution == solution
assert resolution.agent_run_id == analysis.run_id
assert analysis.applied_solution == solution
assert analysis.automation_decision["status"] == "auto_resolved"
assert analysis.automation_decision["applied"] is True
assert solution.usage_count == 1
assert TicketSolutionLink.objects.filter(
ticket=ticket,
solution=solution,
link_type="adopted",
agent_run_id=analysis.run_id,
).count() == 1
assert CustomerReplyTask.objects.filter(ticket=ticket, status="draft").count() == 1
assert WorkItem.objects.filter(
ticket=ticket,
task_type="ticket_customer_reply",
status="active",
).count() == 1
assert ticket.events.filter(event_type="support_automation_evaluated").count() == 1
@pytest.mark.django_db
@pytest.mark.parametrize(
("settings_enabled", "score", "requires_human_action", "allowed_actions", "reason"),
[
(False, 0.96, False, ["ticket.resolve"], "global_policy_disabled"),
(True, 0.70, False, ["ticket.resolve"], "score_below_threshold"),
(True, 0.96, True, ["ticket.resolve"], "risk_policy_blocked"),
(True, 0.96, False, [], "agent_action_not_allowed"),
],
)
def test_support_analysis_policy_blocks_auto_resolution_but_keeps_reply_task(
admin_client,
settings,
settings_enabled,
score,
requires_human_action,
allowed_actions,
reason,
):
settings.TICKET_AUTO_RESOLVE_ENABLED = settings_enabled
actor = get_user_model().objects.get(is_superuser=True)
analysis, _solution, ticket = build_completed_support_analysis(
actor,
score=score,
requires_human_action=requires_human_action,
allowed_actions=allowed_actions,
)
process_support_analysis_result(analysis.run)
analysis.refresh_from_db()
ticket.refresh_from_db()
assert ticket.status == "pending"
assert analysis.applied_solution_id is None
assert analysis.automation_decision["status"] == "manual_recommendation"
assert reason in analysis.automation_decision["reasons"]
assert CustomerReplyTask.objects.filter(ticket=ticket, status="draft").exists()
@pytest.mark.django_db
def test_auto_resolution_requires_initiator_permission(settings):
settings.TICKET_AUTO_RESOLVE_ENABLED = True
call_command("bootstrap_governance")
actor = get_user_model().objects.create_user(username="stage-33-limited")
actor.user_permissions.add(
Permission.objects.get(
content_type__app_label="system",
codename="agent.use",
)
)
analysis, _solution, ticket = build_completed_support_analysis(actor)
process_support_analysis_result(analysis.run)
analysis.refresh_from_db()
ticket.refresh_from_db()
assert ticket.status == "pending"
assert "initiator_permission_denied" in analysis.automation_decision["reasons"]
@pytest.mark.django_db
def test_customer_reply_edit_history_stale_write_and_sent_confirmation(
admin_client,
):
actor = get_user_model().objects.get(is_superuser=True)
analysis, _solution, ticket = build_completed_support_analysis(
actor,
auto_resolve_recommended=False,
)
process_support_analysis_result(analysis.run)
task = CustomerReplyTask.objects.get(ticket=ticket)
response = admin_client.patch(
f"/api/tickets/{ticket.id}/reply-tasks/{task.id}",
data={
"final_text": "您好，我们已完成订单状态核对，请刷新页面确认。",
"expected_revision": 1,
},
content_type="application/json",
)
assert response.status_code == 200
assert response.json()["revision"] == 2
assert len(response.json()["revisions"]) == 2
assert CustomerReplyRevision.objects.get(task=task, version=2).source == "human"
stale = admin_client.patch(
f"/api/tickets/{ticket.id}/reply-tasks/{task.id}",
data={"final_text": "过期覆盖", "expected_revision": 1},
content_type="application/json",
)
assert stale.status_code == 409
sent = admin_client.post(
f"/api/tickets/{ticket.id}/reply-tasks/{task.id}/confirm-sent",
data={"expected_revision": 2},
content_type="application/json",
)
assert sent.status_code == 200
assert sent.json()["status"] == "sent"
task.refresh_from_db()
ticket.refresh_from_db()
assert task.sent_by == actor
assert task.sent_at is not None
assert ticket.replied_to_customer is True
assert ticket.support_last_reply_at is not None
assert WorkItem.objects.get(
ticket=ticket,
task_type="ticket_customer_reply",
).status == "completed"
assert ticket.events.filter(event_type="customer_reply_sent").exists()

112
backend/apps/tickets/tests/test_stage_3_4.py Normal file
View File

@@ -0,0 +1,112 @@
from datetime import timedelta
import pytest
from django.contrib.auth import get_user_model
from django.utils import timezone
from apps.tickets.services import transition_ticket
from apps.tickets.tests.test_stage_3_3 import build_completed_support_analysis
from apps.workitems.models import WorkItem
from apps.workitems.services import sync_support_analysis_work_item
@pytest.mark.django_db
def test_resolved_ticket_creates_close_task_and_closing_completes_it(admin_client):
actor = get_user_model().objects.get(is_superuser=True)
_, _, ticket = build_completed_support_analysis(actor)
transition_ticket(
ticket,
"resolved",
operator=actor,
resolution="人工确认解决方案",
)
close_task = WorkItem.objects.get(
ticket=ticket,
task_type="ticket_close",
)
assert close_task.status == "active"
assert close_task.due_at is not None
assert close_task.due_at > timezone.now()
transition_ticket(ticket, "closed", operator=actor)
close_task.refresh_from_db()
assert close_task.status == "completed"
assert close_task.completed_by == actor
assert close_task.completed_at is not None
@pytest.mark.django_db
def test_support_analysis_failure_task_is_deduplicated_and_success_completes_it(
admin_client,
):
actor = get_user_model().objects.get(is_superuser=True)
analysis, _, _ = build_completed_support_analysis(actor)
run = analysis.run
run.status = "failed"
run.error_code = "step_failed"
run.error_message = "模型调用失败"
run.save(update_fields=["status", "error_code", "error_message"])
sync_support_analysis_work_item(run)
sync_support_analysis_work_item(run)
task = WorkItem.objects.get(
ticket=analysis.ticket,
task_type="support_analysis_failed",
)
assert task.status == "active"
assert task.payload["agent_run_public_id"] == str(run.public_id)
assert task.payload["error_code"] == "step_failed"
run.status = "succeeded"
run.save(update_fields=["status"])
sync_support_analysis_work_item(run)
task.refresh_from_db()
assert task.status == "completed"
@pytest.mark.django_db
def test_work_item_metrics_report_active_overdue_due_soon_and_completed(admin_client):
actor = get_user_model().objects.get(is_superuser=True)
_, _, ticket = build_completed_support_analysis(actor)
now = timezone.now()
WorkItem.objects.create(
task_type="ticket_close",
object_type="ticket",
object_id=str(ticket.id),
ticket=ticket,
title="已超时",
due_at=now - timedelta(minutes=1),
)
WorkItem.objects.create(
task_type="support_analysis_failed",
object_type="ticket",
object_id=str(ticket.id),
ticket=ticket,
title="即将超时",
due_at=now + timedelta(hours=1),
)
WorkItem.objects.create(
task_type="ticket_customer_reply",
object_type="customer_reply",
object_id="completed",
ticket=ticket,
title="按时完成",
status="completed",
due_at=now,
completed_at=now - timedelta(minutes=1),
)
response = admin_client.get("/api/tasks/metrics")
assert response.status_code == 200
payload = response.json()
assert payload["active_count"] == 2
assert payload["overdue_count"] == 1
assert payload["due_soon_count"] == 1
assert payload["completed_within_sla_count"] == 1
assert payload["by_task_type"]["ticket_close"] == 1

277
backend/apps/tickets/tests/test_stage_3_acceptance.py Normal file
View File

@@ -0,0 +1,277 @@
import json
import pytest
from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.test import override_settings
from apps.agents.models import AgentDefinition, AgentRun
from apps.agents.services import build_config_snapshot
from apps.agents.tests.helpers import create_deployment
from apps.audit.models import OperationAuditLog
from apps.solutions.models import Solution
from apps.tickets.models import (
SupportEvaluationCase,
Ticket,
TicketCategory,
TicketSupportAnalysis,
)
from apps.tickets.support_analysis import process_support_analysis_result
from apps.tickets.support_evaluation import (
SupportEvaluationError,
build_stage_3_acceptance_report,
build_support_quality_metrics,
import_support_evaluation_cases,
record_support_evaluation_result,
)
def build_support_analysis(
actor,
*,
suffix: str,
allow_auto_resolve: bool,
candidate_present: bool,
):
deployment = create_deployment(suffix=suffix)
call_command("publish_support_analysis", deployment_id=deployment.id)
definition = AgentDefinition.objects.select_related("active_version").get(
code="support-analysis"
)
version = definition.active_version
category, _ = TicketCategory.objects.get_or_create(
code="business_issue",
defaults={"name": "业务问题"},
)
solution = Solution.objects.create(
title=f"订单状态恢复方案{suffix}",
problem_signature="订单状态未更新",
related_system="订单系统",
environment_scope="生产环境",
category=category,
root_cause="状态回写延迟",
solution="重新同步订单状态并核对结果",
status="published",
allow_auto_resolve=allow_auto_resolve,
)
ticket = Ticket.objects.create(
ticket_no=f"TK-ACCEPT-{suffix}",
title="订单状态未更新",
description="生产环境订单状态长时间未更新",
category=category,
status="pending",
related_system="订单系统",
environment="生产环境",
)
reference = {
"rank": 1,
"source_type": "solution",
"source_id": solution.id,
"source_version": solution.updated_at.isoformat(),
"title": solution.title,
"excerpt": solution.solution,
"score": 0.96,
"applicable_scope": {
"category_code": category.code,
"related_system": solution.related_system,
"environment": solution.environment_scope,
},
}
output = {
"analysis_summary": "订单状态回写延迟。",
"troubleshooting_steps": ["核对订单状态"],
"candidate_solution_ids": [solution.id] if candidate_present else [],
"customer_reply_draft": "您好，订单状态已完成核对。",
"citations": [1],
"auto_resolve_recommended": candidate_present,
}
run = AgentRun.objects.create(
definition=definition,
agent_version=version,
status="succeeded",
input_payload={},
final_output=output,
config_snapshot=build_config_snapshot(version),
created_by=actor,
)
analysis = TicketSupportAnalysis.objects.create(
ticket=ticket,
run=run,
query=ticket.description,
references=[reference],
retrieval_policy={"policy_version": "test"},
created_by=actor,
)
process_support_analysis_result(run)
analysis.refresh_from_db()
return analysis, solution
@pytest.mark.django_db
def test_import_support_evaluation_cases_validates_real_case_contract(tmp_path):
path = tmp_path / "support-cases.json"
path.write_text(
json.dumps(
{
"schema_version": 1,
"cases": [
{
"key": "real-auto",
"name": "真实自动解决案例",
"source_kind": "anonymized_real",
"input_manifest": {"ticket_summary": "订单状态未更新"},
"expected_solution_ids": [12],
"expected_auto_resolve": True,
"expected_manual_fallback": False,
}
],
},
ensure_ascii=False,
),
encoding="utf-8",
)
assert import_support_evaluation_cases(path) == (1, 0)
case = SupportEvaluationCase.objects.get(key="real-auto")
assert case.expected_solution_ids == [12]
assert import_support_evaluation_cases(path) == (0, 1)
invalid = tmp_path / "invalid.json"
invalid.write_text(
json.dumps(
{
"schema_version": 1,
"cases": [
{
"key": "invalid",
"source_kind": "anonymized_real",
"input_manifest": {"ticket_summary": "问题"},
"expected_solution_ids": [],
"expected_auto_resolve": True,
"expected_manual_fallback": True,
}
],
}
),
encoding="utf-8",
)
with pytest.raises(SupportEvaluationError):
import_support_evaluation_cases(invalid)
@pytest.mark.django_db
@override_settings(TICKET_AUTO_RESOLVE_ENABLED=True)
def test_stage_3_real_case_metrics_cover_match_auto_reuse_and_fallback():
actor = get_user_model().objects.create_superuser(
username="stage3-reviewer",
password="test-pass",
)
auto_analysis, auto_solution = build_support_analysis(
actor,
suffix="auto",
allow_auto_resolve=True,
candidate_present=True,
)
manual_analysis, manual_solution = build_support_analysis(
actor,
suffix="manual",
allow_auto_resolve=False,
candidate_present=True,
)
no_match_analysis, _ = build_support_analysis(
actor,
suffix="no-match",
allow_auto_resolve=True,
candidate_present=False,
)
cases = [
SupportEvaluationCase.objects.create(
key="real-auto",
name="允许自动解决",
source_kind="anonymized_real",
input_manifest={"ticket_summary": "自动解决"},
expected_solution_ids=[auto_solution.id],
expected_auto_resolve=True,
expected_manual_fallback=False,
),
SupportEvaluationCase.objects.create(
key="real-manual",
name="风险策略人工降级",
source_kind="anonymized_real",
input_manifest={"ticket_summary": "人工降级"},
expected_solution_ids=[manual_solution.id],
expected_auto_resolve=False,
expected_manual_fallback=True,
),
SupportEvaluationCase.objects.create(
key="real-no-match",
name="无正确方案",
source_kind="anonymized_real",
input_manifest={"ticket_summary": "无匹配"},
expected_solution_ids=[],
expected_auto_resolve=False,
expected_manual_fallback=True,
),
]
results = [
record_support_evaluation_result(case, analysis, reviewer=actor)
for case, analysis in zip(
cases,
[auto_analysis, manual_analysis, no_match_analysis],
strict=True,
)
]
assert all(result.passed for result in results)
metrics = build_support_quality_metrics()
assert metrics["evaluation"]["solution_match_rate"] == 1.0
assert metrics["evaluation"]["eligible_auto_resolve_rate"] == 1.0
assert metrics["evaluation"]["incorrect_auto_reuse_rate"] == 0.0
assert metrics["evaluation"]["manual_fallback_accuracy"] == 1.0
assert metrics["evaluation"]["citation_validity_rate"] == 1.0
assert all(
value is True
for key, value in metrics["targets"].items()
if key.endswith("_met")
)
report = build_stage_3_acceptance_report()
assert report["exit_ready"] is True
assert report["status"] == "passed"
assert report["reasons"] == []
@pytest.mark.django_db
@override_settings(TICKET_AUTO_RESOLVE_ENABLED=True)
def test_support_evaluation_api_records_audited_result(admin_client):
actor = get_user_model().objects.get(is_superuser=True)
analysis, solution = build_support_analysis(
actor,
suffix="api",
allow_auto_resolve=True,
candidate_present=True,
)
case = SupportEvaluationCase.objects.create(
key="real-api",
name="API 验收",
source_kind="anonymized_real",
input_manifest={"ticket_summary": "API"},
expected_solution_ids=[solution.id],
expected_auto_resolve=True,
expected_manual_fallback=False,
)
response = admin_client.post(
"/api/tickets/support-evaluations",
data={"case_key": case.key, "analysis_id": analysis.id},
content_type="application/json",
)
assert response.status_code == 200
assert response.json()["passed"] is True
assert OperationAuditLog.objects.filter(
action="support_quality_evaluated",
target_id=str(analysis.id),
).exists()
metrics = admin_client.get("/api/tickets/support-quality-metrics")
assert metrics.status_code == 200
assert metrics.json()["evaluation"]["reviewed_case_count"] == 1