505 lines
17 KiB
Python
505 lines
17 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from celery.exceptions import SoftTimeLimitExceeded
|
|
from django.utils import timezone
|
|
|
|
from apps.agents.models import AgentRun
|
|
from apps.agents.services import build_config_snapshot, execute_agent_run
|
|
from apps.agents.structured_metrics import build_structured_output_metrics
|
|
from apps.agents.tasks import cleanup_expired_agent_runs
|
|
from apps.agents.tasks import execute_agent_run as execute_agent_run_task
|
|
from apps.audit.models import OperationAuditLog
|
|
from apps.modelhub.services import CompletionResult
|
|
|
|
from .helpers import create_agent, create_deployment
|
|
|
|
|
|
def create_run(version):
|
|
return AgentRun.objects.create(
|
|
definition=version.definition,
|
|
agent_version=version,
|
|
sandbox=version.status == "draft",
|
|
input_payload={"problem": "无法登录"},
|
|
config_snapshot=build_config_snapshot(version),
|
|
total_steps=version.steps.count(),
|
|
)
|
|
|
|
|
|
def test_ordered_runtime_records_steps_usage_and_final_output(db, monkeypatch):
|
|
_, version = create_agent()
|
|
outputs = iter(
|
|
[
|
|
CompletionResult('{"summary":"账号异常"}', 10, 5),
|
|
CompletionResult('{"reply":"请重置密码"}', 12, 6),
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: next(outputs),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "succeeded"
|
|
assert run.final_output == {"reply": "请重置密码"}
|
|
assert run.input_tokens == 22
|
|
assert run.output_tokens == 11
|
|
assert list(run.steps.values_list("code", flat=True)) == ["structure", "reply"]
|
|
assert run.steps.count() == 2
|
|
assert sum(step.attempts.count() for step in run.steps.all()) == 2
|
|
first_invocation = run.steps.first().attempts.get().model_invocation
|
|
assert first_invocation.model_snapshot["structured_output"]["mode"] == "native_json_schema"
|
|
assert first_invocation.model_snapshot["structured_output"]["schema_hash"]
|
|
assert first_invocation.input_snapshot["output_schema"]
|
|
metrics = build_structured_output_metrics(
|
|
deployment_id=version.model_deployment_id,
|
|
)
|
|
assert metrics["initial_count"] == 2
|
|
assert metrics["first_pass_rate"] == 1
|
|
assert metrics["repair_count"] == 0
|
|
assert metrics["invocation_count"] == 2
|
|
assert run.progress_stage == "completed"
|
|
assert run.progress_message == "草稿生成完成"
|
|
|
|
|
|
def test_invalid_structure_is_repaired_once(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
outputs = iter(
|
|
[
|
|
CompletionResult("not-json", 4, 2),
|
|
CompletionResult('{"summary":"已修复"}', 5, 3),
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: next(outputs),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
step = run.steps.get()
|
|
assert run.status == "succeeded"
|
|
assert step.attempts.count() == 2
|
|
assert list(step.attempts.values_list("attempt_type", flat=True)) == [
|
|
"initial",
|
|
"structure_repair",
|
|
]
|
|
metrics = build_structured_output_metrics(skill_version_id=step.skill_version_id)
|
|
assert metrics["initial_count"] == 1
|
|
assert metrics["first_pass_rate"] == 0
|
|
assert metrics["repair_count"] == 1
|
|
assert metrics["repair_rate"] == 1
|
|
|
|
|
|
def test_flattened_fact_output_never_passes_as_final_result(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
skill_version = version.steps.get().skill_version
|
|
fact_object = {
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {"type": "string"},
|
|
"description": {"type": "string"},
|
|
},
|
|
"required": ["title", "description"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.output_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"facts": fact_object,
|
|
"evidence": fact_object,
|
|
"missing_fields": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
"required": ["facts", "evidence", "missing_fields"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.save(update_fields=["output_schema"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: CompletionResult(
|
|
'{"title":"登录失败","evidence":{"title":"无法登录"},"missing_fields":[]}',
|
|
10,
|
|
5,
|
|
),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
step = run.steps.get()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_invalid"
|
|
assert step.attempts.count() == 2
|
|
assert run.final_output is None
|
|
|
|
|
|
def test_missing_nested_fact_field_never_passes_as_final_result(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
skill_version = version.steps.get().skill_version
|
|
fact_object = {
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {"type": "string"},
|
|
"problem_performance": {"type": "string"},
|
|
},
|
|
"required": ["title", "problem_performance"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.output_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"facts": fact_object,
|
|
"evidence": fact_object,
|
|
"missing_fields": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
"required": ["facts", "evidence", "missing_fields"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.save(update_fields=["output_schema"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: CompletionResult(
|
|
'{"facts":{"title":"登录失败"},'
|
|
'"evidence":{"title":"无法登录"},"missing_fields":[]}',
|
|
10,
|
|
5,
|
|
),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_invalid"
|
|
assert run.steps.get().attempts.count() == 2
|
|
assert run.final_output is None
|
|
|
|
|
|
def test_old_flattened_draft_fields_never_pass_as_final_result(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
skill_version = version.steps.get().skill_version
|
|
skill_version.output_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"fields": {
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {"type": "string"},
|
|
"description": {"type": "string"},
|
|
},
|
|
"required": ["title", "description"],
|
|
"additionalProperties": False,
|
|
},
|
|
"missing_fields": {"type": "array", "items": {"type": "string"}},
|
|
"questions": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
"required": ["fields", "missing_fields", "questions"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.save(update_fields=["output_schema"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: CompletionResult(
|
|
'{"title":"登录失败","description":"",'
|
|
'"missing_fields":["description"],"follow_up_questions":["请补充现象"]}',
|
|
10,
|
|
5,
|
|
),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_invalid"
|
|
assert run.steps.get().attempts.count() == 2
|
|
assert run.final_output is None
|
|
|
|
|
|
def test_wrong_evidence_properties_never_pass_as_final_result(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
skill_version = version.steps.get().skill_version
|
|
evidence_schema = {
|
|
"type": "object",
|
|
"properties": {"title": {"type": "string"}},
|
|
"required": ["title"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.output_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"facts": evidence_schema,
|
|
"evidence": evidence_schema,
|
|
"missing_fields": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
"required": ["facts", "evidence", "missing_fields"],
|
|
"additionalProperties": False,
|
|
}
|
|
skill_version.save(update_fields=["output_schema"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: CompletionResult(
|
|
'{"facts":{"title":"登录失败"},'
|
|
'"evidence":{"source":"旧证据字段"},"missing_fields":[]}',
|
|
10,
|
|
5,
|
|
),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_invalid"
|
|
assert run.steps.get().attempts.count() == 2
|
|
assert run.final_output is None
|
|
|
|
|
|
def test_unverified_model_fails_before_any_model_call(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
version.model_deployment.structured_output_test_status = "untested"
|
|
version.model_deployment.save(update_fields=["structured_output_test_status"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: pytest.fail("未验证模型不应被调用"),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
step = run.steps.get()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_not_verified"
|
|
assert step.error_code == "structured_output_not_verified"
|
|
assert step.attempts.count() == 0
|
|
|
|
|
|
def test_second_invalid_structure_fails_and_stops(db, monkeypatch):
|
|
_, version = create_agent()
|
|
outputs = iter(
|
|
[
|
|
CompletionResult("not-json", 4, 2),
|
|
CompletionResult("still-not-json", 5, 3),
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: next(outputs),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "failed"
|
|
assert run.error_code == "structured_output_invalid"
|
|
assert run.final_output is None
|
|
assert run.steps.count() == 1
|
|
step = run.steps.get()
|
|
assert step.status == "failed"
|
|
assert step.attempts.count() == 2
|
|
assert step.attempts.get(attempt_type="structure_repair").validation_error
|
|
|
|
|
|
def test_cancellation_before_first_step_stops_run(db):
|
|
_, version = create_agent()
|
|
run = create_run(version)
|
|
run.cancellation_requested_at = timezone.now()
|
|
run.save(update_fields=["cancellation_requested_at"])
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "cancelled"
|
|
assert run.progress_stage == "cancelled"
|
|
assert run.steps.count() == 0
|
|
|
|
|
|
def test_cancellation_after_model_response_stops_following_steps(db, monkeypatch):
|
|
_, version = create_agent()
|
|
run = create_run(version)
|
|
|
|
def request(*args, **kwargs):
|
|
AgentRun.objects.filter(id=run.id).update(cancellation_requested_at=timezone.now())
|
|
return CompletionResult('{"summary":"已完成当前请求"}', 3, 2)
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
|
|
execute_agent_run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "cancelled"
|
|
assert list(run.steps.values_list("code", flat=True)) == ["structure"]
|
|
assert run.steps.get().status == "succeeded"
|
|
assert run.final_output is None
|
|
|
|
|
|
def test_multimodal_request_keeps_base64_out_of_database_snapshots(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
version.model_deployment.capability = "vision_chat"
|
|
version.model_deployment.provider.image_egress_approved = True
|
|
version.model_deployment.provider.save(update_fields=["image_egress_approved"])
|
|
version.model_deployment.save(update_fields=["capability"])
|
|
captured = {}
|
|
|
|
def request(deployment, messages, **kwargs):
|
|
captured["messages"] = messages
|
|
return CompletionResult('{"summary":"识别到截图"}', 8, 3)
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
run = create_run(version)
|
|
run.input_payload["image_refs"] = [
|
|
{"object_path": "ticket-drafts/proof.png", "mime_type": "image/png"}
|
|
]
|
|
run.save(update_fields=["input_payload"])
|
|
|
|
execute_agent_run(
|
|
run.id,
|
|
image_parts=[
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": "data:image/png;base64,c2Vuc2l0aXZl"},
|
|
}
|
|
],
|
|
)
|
|
|
|
invocation = run.steps.get().attempts.get(attempt_type="initial").model_invocation
|
|
assert "base64" in str(captured["messages"])
|
|
assert "base64" not in str(invocation.input_snapshot)
|
|
assert invocation.input_snapshot["image_refs"][0]["object_path"].endswith("proof.png")
|
|
|
|
|
|
def test_step_model_override_controls_deployment_and_image_egress(db, monkeypatch):
|
|
_, version = create_agent()
|
|
vision = create_deployment(
|
|
capability="vision_chat",
|
|
provider=version.model_deployment.provider,
|
|
suffix="-vision",
|
|
)
|
|
vision.provider.image_egress_approved = True
|
|
vision.provider.save(update_fields=["image_egress_approved"])
|
|
first_binding = version.steps.get(code="structure")
|
|
first_binding.model_deployment = vision
|
|
first_binding.save(update_fields=["model_deployment"])
|
|
captured = []
|
|
outputs = iter(
|
|
[
|
|
CompletionResult('{"summary":"识别到截图"}', 8, 3),
|
|
CompletionResult('{"reply":"请重置密码"}', 7, 2),
|
|
]
|
|
)
|
|
|
|
def request(deployment, messages, **kwargs):
|
|
captured.append((deployment.id, messages))
|
|
return next(outputs)
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
run = create_run(version)
|
|
run.input_payload["image_refs"] = [
|
|
{"object_path": "ticket-drafts/proof.png", "mime_type": "image/png"}
|
|
]
|
|
run.save(update_fields=["input_payload"])
|
|
|
|
execute_agent_run(
|
|
run.id,
|
|
image_parts=[
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": "data:image/png;base64,c2Vuc2l0aXZl"},
|
|
}
|
|
],
|
|
)
|
|
|
|
run.refresh_from_db()
|
|
steps = list(run.steps.select_related("model_deployment"))
|
|
assert run.status == "succeeded"
|
|
assert [item[0] for item in captured] == [vision.id, version.model_deployment_id]
|
|
assert isinstance(captured[0][1][-1]["content"], list)
|
|
assert isinstance(captured[1][1][-1]["content"], str)
|
|
assert steps[0].model_deployment_id == vision.id
|
|
assert steps[1].model_deployment_id == version.model_deployment_id
|
|
assert run.config_snapshot["steps"][0]["model"] == {
|
|
"deployment_id": vision.id,
|
|
"inherited": False,
|
|
"provider_name": vision.provider.name,
|
|
"model_name": vision.model_name,
|
|
"capability": "vision_chat",
|
|
"structured_output_mode": "native_json_schema",
|
|
"structured_output_test_status": "passed",
|
|
}
|
|
assert run.config_snapshot["steps"][1]["model"]["inherited"] is True
|
|
assert run.config_snapshot["steps"][1]["model"]["capability"] == "text_chat"
|
|
|
|
|
|
def test_unapproved_vision_step_rejects_image_egress(db, monkeypatch):
|
|
_, version = create_agent(two_steps=False)
|
|
version.model_deployment.capability = "vision_chat"
|
|
version.model_deployment.save(update_fields=["capability"])
|
|
monkeypatch.setattr(
|
|
"apps.modelhub.services._request_completion",
|
|
lambda *args, **kwargs: pytest.fail("未授权图片不应进入模型调用"),
|
|
)
|
|
run = create_run(version)
|
|
|
|
execute_agent_run(
|
|
run.id,
|
|
image_parts=[
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": "data:image/png;base64,c2Vuc2l0aXZl"},
|
|
}
|
|
],
|
|
)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "failed"
|
|
assert "未授权图片出境" in run.error_message
|
|
assert run.steps.get().attempts.count() == 0
|
|
|
|
|
|
def test_soft_time_limit_marks_run_timed_out_and_audits(db, monkeypatch):
|
|
_, version = create_agent()
|
|
run = create_run(version)
|
|
monkeypatch.setattr(
|
|
"apps.agents.tasks.execute_agent_run_service",
|
|
lambda *_: (_ for _ in ()).throw(SoftTimeLimitExceeded()),
|
|
)
|
|
|
|
execute_agent_run_task.run(run.id)
|
|
|
|
run.refresh_from_db()
|
|
assert run.status == "timed_out"
|
|
assert run.error_code == "run_timeout"
|
|
assert run.progress_stage == "timed_out"
|
|
assert OperationAuditLog.objects.filter(
|
|
target_id=str(run.public_id),
|
|
action="run_finished",
|
|
result="timed_out",
|
|
).exists()
|
|
|
|
|
|
def test_cleanup_removes_expired_runs_but_keeps_audit(db, settings):
|
|
_, version = create_agent()
|
|
run = create_run(version)
|
|
AgentRun.objects.filter(id=run.id).update(created_at=timezone.now() - timedelta(days=91))
|
|
|
|
deleted = cleanup_expired_agent_runs()
|
|
|
|
assert deleted == 1
|
|
assert not AgentRun.objects.filter(id=run.id).exists()
|
|
assert OperationAuditLog.objects.filter(
|
|
module="agents",
|
|
action="cleanup_runs",
|
|
request_snapshot__deleted_count=1,
|
|
).exists()
|