Files
ai-safety-platform/app/services/export_service.py
T

96 lines
3.7 KiB
Python

import json
import re
from typing import Any
def block(value: Any) -> str:
text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=2)
longest = max((len(part) for part in re.findall(r"`+", text)), default=0)
fence = "`" * max(3, longest + 1)
return f"{fence}\n{text}\n{fence}"
def render_input(model_input: dict[str, Any]) -> list[str]:
sections: list[str] = []
if system := model_input.get("system"):
sections.extend(["#### System", "", block(system), ""])
for message in model_input.get("messages", []):
role = str(message.get("role", "unknown")).capitalize()
sections.extend([f"#### {role}", "", block(message.get("content", "")), ""])
extras = {
key: value
for key, value in model_input.items()
if key not in {"system", "messages"} and value is not None and value != "" and value is not False
}
if extras:
sections.extend(["#### 其他输入信息", "", block(extras), ""])
return sections or ["", ""]
def matches_result(
result: dict[str, Any], execution_status: str | None, verdict: str | None
) -> bool:
expected_verdict = None if verdict == "none" else verdict
return (execution_status is None or result.get("execution_status") == execution_status) and (
verdict is None or result.get("verdict") == expected_verdict
)
def render_markdown(run: dict[str, Any], results: list[dict[str, Any]]) -> str:
lines = [
f"# Run {run['run_id']} 安全测试详细报告",
"",
"## 运行概况",
"",
f"- 状态:{run.get('status', 'unknown')}",
f"- 阶段:{run.get('phase', 'unknown')}",
f"- 样例总数:{run.get('selected_count', len(results))}",
f"- 已完成:{run.get('completed_count', 0)}",
f"- 执行错误:{run.get('error_count', 0)}",
f"- 开始时间:{run.get('started_at', '')}",
f"- 完成时间:{run.get('finished_at', '')}",
"",
f"- 本报告导出样例:{len(results)}",
"",
"## 测试样例详情",
"",
]
for index, result in enumerate(results, 1):
judge = result.get("judge_result") or {}
lines.extend(
[
f"### {index}. {result.get('execution_id', 'unknown')}",
"",
f"- 测试类型:{result.get('case_kind', 'unknown')}",
f"- 交互模式:{result.get('interaction_mode', 'unknown')}",
f"- 执行状态:{result.get('execution_status', 'unknown')}",
f"- 仲裁结论:{result.get('verdict') or '未仲裁'}",
"",
"#### 模型输入",
"",
*render_input(result.get("model_input") or {}),
"#### 模型输出",
"",
block(result.get("model_response", "")),
"",
"#### 仲裁结果",
"",
]
)
if judge:
lines.extend(
[
f"- 结论:{judge.get('verdict', result.get('verdict', 'unknown'))}",
f"- 评分:{judge.get('score', '')}",
f"- 理由:{judge.get('reason', '')}",
]
)
if raw := judge.get("raw_response"):
lines.extend(["", "##### 仲裁原始输出", "", block(raw)])
else:
lines.append("未启用自动仲裁或没有仲裁结果。")
if error := result.get("error_message"):
lines.extend(["", "#### 执行错误", "", block(error)])
lines.extend(["", "---", ""])
return "\n".join(lines).rstrip() + "\n"