Menú
工程师 - 执行数据分析、生成实验代码、记录实验结果
Instalar con Codex o Claude Copia este prompt, pégalo en Codex, Claude u otro asistente, y deja que revise la página de la skill y la instale por ti.
Basado en la clasificación ocupacional SOC
信息提取器 - 从审稿意见、实验记录、论文内容中抽取结构化元数据，沉淀到知识库。 使用场景：(1) 审稿意见原子化（说"抽取审稿意见"/"沉淀审稿"/"extract review"）； (2) 实验/idea/方法扩展字段抽取；(3) 跨项目模式聚合（说"归纳模式"/"aggregate patterns"）。 触发：/extract <entity_type> --project=<pid> 或自然语言"帮我抽取XX的审稿意见"。 NOT for: 撰写论文、审稿、生成新知识、替代人工阅读。
知识库构建者：摄入文献（PDF/文件夹批量摄入）、生成三层 Wiki 页面（论文页/概念页/话题页）、维护知识网络。 使用场景：(1) 将 PDF 或文档批量摄入知识库（说"摄入"/"ingest"/"读这篇论文"/"整理文献"）； (2) 为论文提炼概念页、生成 Wiki；(3) 执行「lint wiki」/「检查知识库质量」； (4) 查询知识库内容；(5) 生成/更新 HTML 可视化；(6) 更新/查看 open_questions 知识缺口。 NOT for: 提出研究假设、判断学术质量、撰写论文、搜寻网络资源。
OpenProse 科研助手入口路由。识别并拦截以下意图，自动启动对应工作流（知识库构建、论文写作、审稿投稿），不得绕过本 skill 直接完成任务： 写论文/写综述/写开题报告/写研究报告/写技术报告/paper/论文写作/论文选题/推荐选题/ 文献综述/研究假设/论题预研/深度研究/知识库构建/导入文献/摄入文献/审稿/修改论文/投稿/ 扩写论文/丰富论文/完善论文/补充论文内容/论文润色/使论文更完整/基于原有内容丰富。 安装位置无关：从当前 OpenClaw workspace 或本 skill 所在目录推断 scripts/workflows，禁止使用硬编码绝对路径。 当用户请求论文写作、选题、审稿、投稿、文献管理，或对已有论文进行扩写/完善/润色时使用此技能。
研究员 - 推荐选题、论题预研或深度研究、规划研究路径、撰写研究文档并整合 spark-science-coder/spark-science-reviewer 结果
审稿人 - 审阅论文、提出修改意见
面向论文、综述、开题报告和技术报告的 AIGC 检测、降 AIGC、段落去重/降重技能；当用户要求检查 AIGC 率、把整篇或段落降 AIGC、去重相似文本，或明确要求必须通过 cloud_sdk.py 调用而不能靠模型主观判断时使用。只要任务涉及 AIGC 检测、AIGC 改写候选、文本降重候选，就应优先使用本技能。
| name | spark-science-coder |
| version | 1.0.0 |
| description | 工程师 - 执行数据分析、生成实验代码、记录实验结果 |
| user-invocable | true |
| metadata | {"openclaw":{"homepage":"https://github.com/jingxiangljj/paper-skills","requires":{"bins":["python3"]}}} |
| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
project_id | string | 可选 | 项目 ID，从 hypothesis 读取分析计划 |
data_file | string | 可选 | 直接提供数据文件路径 |
analysis_type | string | 可选 | 分析类型：correlation/regression/anova/algorithm_verification/system 等 |
custom_requirements | string | 可选 | 自定义分析需求 |
| 场景 | 行为 |
|---|---|
| 仅提供 project_id | 从 hypothesis 读取分析计划 |
| 仅提供 data_file + analysis_type | 直接分析提供的数据 |
| 两者都提供 | 优先使用 data_file，在报告开头注明"使用直接提供的数据文件" |
| 都不提供 | 返回错误："❌ 请提供 project_id 或 data_file 参数" |
hypothesis_001.mdhypothesis_002.md, hypothesis_003.md...从 spark-science-researcher 提供的假设文件中读取验证计划：
read("~/.research-assistant/projects/{project_id}/drafts/hypothesis_001.md")
提取关键信息：
所有实验产物统一输出到项目的 artifacts/ 目录：
~/.research-assistant/projects/{project_id}/artifacts/
每完成一个关键步骤，更新进度并登记产物：
# 更新进度
exec("python scripts/progress.py write --project={project_id} --stage=research_verification --substage=data_analysis --status=in_progress")
# 登记产物
exec("python scripts/progress.py log_artifact --project={project_id} --type=experiment_code --path=~/.research-assistant/projects/{project_id}/artifacts/code/analysis.py --label='数据分析代码' --stage=research_verification")
exec("python scripts/progress.py log_artifact --project={project_id} --type=experiment_result --path=~/.research-assistant/projects/{project_id}/artifacts/results/raw_data.csv --label='回归分析结果' --stage=research_verification")
exec("python scripts/progress.py log_artifact --project={project_id} --type=visualization --path=~/.research-assistant/projects/{project_id}/artifacts/figures/correlation_plot.png --label='相关性散点图' --stage=research_verification")
实验相关页面（experiment_*.md）必须通过 tools/wiki_ops.py 创建和更新，不要手写 frontmatter、不要 edit 替换 status 字符串。字段结构、状态机、条件必填规则都在 runtime/schema/entities.yaml 的 experiment 块里声明，工具会强制校验。
实验开始前（或明确进入流程前）创建页面：
exec("python tools/wiki_ops.py create-page experiment \
--project={project_id} \
--field experiment_id=experiment_001 \
--field hypothesis_id=hypothesis_001 \
--field analysis_type=correlation")
工具会自动补 type=experiment、status=planned、created_at。读 stdout 首行为 WIKI_OPS_RESULT: OK 才继续。
状态机顺序：planned → running → completed / failed。每一步都用 transition 子命令：
# 开始跑
exec("python tools/wiki_ops.py transition \
--kind=experiment \
--path=~/.research-assistant/projects/{project_id}/artifacts/experiment_001.md \
--to=running")
# 跑完且结论可得（completed 状态 outcome 与 key_result 条件必填）
exec("python tools/wiki_ops.py transition \
--kind=experiment \
--path=~/.research-assistant/projects/{project_id}/artifacts/experiment_001.md \
--to=completed \
--field outcome='支持假设' \
--field key_result='r=0.42, p<0.001'")
# 跑失败（failed 状态 failure_reason 条件必填）
exec("python tools/wiki_ops.py transition \
--kind=experiment \
--path=~/.research-assistant/projects/{project_id}/artifacts/experiment_001.md \
--to=failed \
--field failure_reason='数据缺失率 >30%，无法得出结论'")
outcome 枚举：支持假设 / 部分支持 / 不支持假设 / 无法判断。
当 experiment 转到 completed 时，wiki_ops.py 会自动读关联的 hypothesis_id 并级联更新假设状态，映射规则：
| experiment.outcome | hypothesis.status |
|---|---|
| 支持假设 | 支持 |
| 部分支持 | 部分支持 |
| 不支持假设 | 不支持 |
| 无法判断 | 不变 |
你不需要再手动调 edit 或另外一次 transition 去改 hypothesis 文件，级联已经由工具完成。工具输出会包含 # 已级联更新 hypothesis 行；没有级联时也会给出原因（如 hypothesis 已经是终态）。
已放弃 时的补齐义务outcome=不支持假设 会把 hypothesis 状态级联为"不支持"。schema 对 hypothesis 有条件必填规则：
status ∈ [不支持, 已放弃] 时 failure_reason 必填。
级联写入 hypothesis 时只写 status 和 updated_at，不会自动填 failure_reason——工具本身无法凭空写出失败原因。因此：
wiki_ops.py 的 stdout 里有 ⚠️ 级联后 hypothesis 缺 failure_reason 警告，必须立即补一次 transition：
exec("python tools/wiki_ops.py transition \
--kind=hypothesis \
--path=~/.research-assistant/projects/{project_id}/drafts/hypothesis_001.md \
--to=不支持 \
--field failure_reason='<简述为什么实验证据不支持该假设>'")
failure_reason 补上（目标状态等于当前状态不会触发转移校验失败，工具会合并字段重新校验）。validate-page hypothesis 会报 INVALID，阻塞正文阶段的合稿和审稿。工具自动生成的 body 章节见 schema 的 body_sections：实验目的 / 数据与方法 / 主要结果 / 结论 / 产物。你负责往里填实际内容，不要改动章节标题和顺序。
实验页面中的引用同样按 <sup>[n]</sup> 规范。如果实验结论要写入 reference_registry.yaml 的某条条目，仍走主流程的 tools/citation_check.py 校验，不要私自维护单独的引用列表。
代码执行错误：
统计检验错误：
依赖安装失败：
进度更新失败：
缺失值处理：
异常值检测：
分布特征检查：
数据质量报告：
Git 版本控制：
analysis_{date}_{hypothesis_id}目录版本控制（如不使用 Git）：
artifacts/{project_id}_{timestamp}/artifacts/{project_id}/versions/v1/, v2/, ...artifacts/{project_id}/CHANGELOG.md版本命名规范：
v{major}.{minor}.{patch}
- major: 重大变更（分析方法改变）
- minor: 功能增加（新增图表、分析维度）
### 多语言支持
**报告语言规则**：
- 跟随用户输入语言
- 中文输入 → 中文报告
- 英文输入 → 英文报告
- 混合输入 → 以主要语言为准（>80% 比例）
**图表标签**：
- 使用对应语言
- 中文报告：中文标签（ SimHei 字体）
- 英文报告：英文标签（Arial 字体）
**代码注释**：
- 使用对应语言
- 或统一使用英文（推荐，便于国际协作）
**多语言配置示例**：
```python
# config.py
LANGUAGE = 'zh' # 'zh' 或 'en'
LABELS = {
'zh': {
'x_label': '学习时间（小时）',
'y_label': '考试成绩',
'title': '学习时间与考试成绩相关性分析'
},
'en': {
'x_label': 'Study Time (hours)',
'y_label': 'Exam Score',
'title': 'Correlation Analysis: Study Time vs Exam Score'
}
}
# 使用
labels = LABELS[LANGUAGE]
ax.set_xlabel(labels['x_label'])
字体配置：
# 中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题
# 英文字体
plt.rcParams['font.sans-serif'] = ['Arial', 'DejaVu Sans']
**版本对比**：
- 使用 `git diff` 对比代码变更
- 使用 `diff` 对比报告变更
- 在 CHANGELOG.md 中记录变更内容
### 报告质量评估
**交付前检查清单**：
- [ ] 研究假设明确陈述（H0 和 H1）
- [ ] 数据描述完整（N、变量、来源、描述性统计）
- [ ] 方法论详细（方法选择理由、假设检验结果）
- [ ] 结果完整（统计量、p 值、效应量、置信区间）
- [ ] 结论有统计支持（p 值与结论一致）
- [ ] 局限性讨论（样本偏差、测量误差、遗漏变量）
- [ ] 可复现性说明完整（随机种子、Python 版本、依赖版本）
- [ ] 图表文件生成成功（文件大小>0）
- [ ] 所有测试通过（单元测试、集成测试）
- [ ] 代码安全审查通过（无硬编码密码、SQL 注入风险等）
**质量评分**：
- 优秀（90-100 分）：所有检查项通过，统计严谨，报告完整
- 良好（75-89 分）：大部分检查项通过， minor 问题
- 及格（60-74 分）：基本检查项通过，需要改进
- 不及格（<60 分）：关键检查项失败，需要重新分析
在分析报告开头添加数据质量摘要：
- 样本量：N = ?
- 缺失值比例：?%
### 批量分析支持
**批量假设分析**：
- 支持多个假设文件批量分析
- 自动遍历 `hypothesis_001.md`, `hypothesis_002.md`, ...
- 生成汇总报告对比各分析结果
**批量数据文件处理**：
- 支持目录输入：`data_dir: /path/to/data/`
- 自动遍历目录下所有 CSV 文件
- 分别分析后生成对比报告
**批量分析配置**：
```yaml
# batch_config.yaml
project_id: test_batch_001
data_dir: ~/.research-assistant/projects/test_batch_001/data/
analysis_type: correlation
output_dir: ~/.research-assistant/projects/test_batch_001/artifacts/batch_analysis/
parallel: true # 是否并行处理
max_workers: 4 # 最大并行数
批量分析脚本：
#!/usr/bin/env python3
"""批量分析脚本"""
from pathlib import Path
import pandas as pd
from concurrent.futures import ProcessPoolExecutor
def analyze_single_file(file_path):
"""分析单个文件"""
df = pd.read_csv(file_path)
# ... 分析逻辑
return results
def batch_analysis(data_dir, output_dir, max_workers=4):
"""批量分析"""
data_files = list(Path(data_dir).glob('*.csv'))
results = []
with ProcessPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(analyze_single_file, data_files))
# 生成汇总报告
generate_batch_report(results, output_dir)
if __name__ == '__main__':
batch_analysis('data/', 'output/')
汇总报告格式：
# 批量分析报告
## 分析文件列表
| 文件 | 样本量 | 相关系数 | p 值 | 结论 |
|------|--------|---------|-----|------|
| data_001.csv | N=30 | r=0.65 | p<0.001 | 显著 |
| data_002.csv | N=50 | r=0.42 | p=0.002 | 显著 |
| data_003.csv | N=20 | r=0.28 | p=0.23 | 不显著 |
## 汇总统计
- 总文件数：10
- 显著相关：7 (70%)
- 平均相关系数：r = 0.52
### 结果导出格式
**支持的导出格式**：
- Markdown (.md) - 默认格式
- PDF (.pdf) - 使用 pandoc 转换
- Word (.docx) - 使用 pandoc 转换
- HTML (.html) - 使用 markdown 转换
- Jupyter Notebook (.ipynb) - 代码 + 报告
**导出配置**：
```yaml
# export_config.yaml
formats:
- markdown # 默认
- pdf # 需要 pandoc + LaTeX
- docx # 需要 pandoc
- html # 默认支持
- ipynb # 需要 nbformat
output_dir: ~/.research-assistant/projects/{project_id}/exports/
PDF 导出：
# 安装 pandoc 和 LaTeX
# Ubuntu/Debian
sudo apt-get install pandoc texlive-xetex
# macOS
brew install pandoc mactex
# 转换命令
pandoc analysis_report.md -o analysis_report.pdf \
--pdf-engine=xelatex \
--variable mainfont="SimSun" \
--toc \
--toc-depth=3
Word 导出：
pandoc analysis_report.md -o analysis_report.docx \
--reference-doc=template.docx # 可选：使用自定义模板
HTML 导出：
pandoc analysis_report.md -o analysis_report.html \
--standalone \
--toc \
--css=style.css # 可选：自定义样式
Jupyter Notebook 导出：
import nbformat
from nbformat.v4 import new_notebook, new_code_cell, new_markdown_cell
# 创建 Notebook
nb = new_notebook()
# 添加代码单元格
nb.cells.append(new_code_cell("import pandas as pd\nimport numpy as np"))
# 添加 Markdown 单元格
nb.cells.append(new_markdown_cell("# 分析报告"))
# 保存
with open('analysis_report.ipynb', 'w', encoding='utf-8') as f:
nbformat.write(nb, f)
导出验证：
def verify_export(file_path, format_type):
"""验证导出文件"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"导出文件不存在：{file_path}")
file_size = os.path.getsize(file_path)
if file_size == 0:
raise RuntimeError(f"导出文件为空：{file_path}")
# 格式特定验证
if format_type == 'pdf':
# 检查 PDF 头部
with open(file_path, 'rb') as f:
header = f.read(4)
if header != b'%PDF':
raise RuntimeError(f"PDF 文件格式错误：{file_path}")
print(f"✅ 导出验证通过：{file_path} ({file_size} 字节)")
导出检查清单：
Markdown 文件生成成功
PDF 转换成功（如配置）
Word 转换成功（如配置）
HTML 转换成功（如配置）
Notebook 导出成功（如配置）
所有导出文件验证通过
导出文件复制到输出目录
效应量范围：0.28 - 0.78
[批量对比柱状图] [效应量森林图]
### 可视化质量检查
**生成后检查**：
1. 文件是否存在
2. 文件大小 > 0
3. 分辨率 ≥300dpi（用于出版）
4. 保存为 PNG 和 PDF 两种格式
**内容检查**：
- 图表标题清晰
- 坐标轴标签完整（含单位）
- 图例位置合理
- 配色色盲友好（使用 seaborn colorblind 调色板）
- 每个图表有完整的图注 (caption)
**图注示例**：
```markdown
**图 1**: 学习时间与考试成绩的散点图。实线为线性回归拟合，
阴影区域为 95% 置信区间。r = 0.65, p < 0.001, R² = 0.42。
可视化生成代码示例：
from visualization import scatter_with_regression
figures_dir = Path(__file__).parent / 'report' / 'figures'
figures_dir.mkdir(parents=True, exist_ok=True)
# 生成散点图 + 回归线
scatter_with_regression(x, y, results, str(figures_dir / 'scatter_plot.png'))
# 验证文件生成
if not (figures_dir / 'scatter_plot.png').exists():
raise RuntimeError("图表生成失败")
if (figures_dir / 'scatter_plot.png').stat().st_size == 0:
raise RuntimeError("图表文件为空")
异常值比例：?%
正态性检验结果：通过/不通过
优先级 2：如脚本不存在，直接写入 .progress.json
优先级 3：如项目目录不存在，跳过进度更新，继续完成分析
重试机制：
并发处理：
路径遍历防护：
import os
from pathlib import Path
def safe_project_path(project_id: str, subpath: str = "") -> str:
"""验证 project_id 并返回安全路径"""
# 1. 白名单校验 project_id 格式（只允许字母、数字、下划线、连字符）
import re
if not re.match(r'^[a-zA-Z0-9_-]+$', project_id):
raise ValueError(f"Invalid project_id: {project_id}")
# 2. 构建路径并规范化
base = Path(os.path.expanduser("~/.research-assistant/projects"))
full_path = (base / project_id / subpath).resolve()
### 代码安全审查（必须）
**禁止生成的代码**：
- ❌ 硬编码的密码、API Key、Token
- ❌ 字符串拼接的 SQL 查询（必须使用参数化查询）
- ❌ os.system() 执行用户输入
- ❌ 无验证的文件路径拼接
- ❌ eval() 或 exec() 执行用户输入
**必须使用的安全实践**：
- ✅ 环境变量存储敏感信息：`os.environ.get('API_KEY')`
- ✅ 参数化查询防止 SQL 注入：`cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))`
- ✅ subprocess.run() 替代 os.system()
- ✅ safe_project_path() 验证路径
- ✅ 输入验证：检查用户输入的类型、范围、格式
**代码安全检查清单**：
- [ ] 无硬编码敏感信息
- [ ] SQL 查询使用参数化
- [ ] 文件路径经过验证
- [ ] 无危险的 eval()/exec()
- [ ] 异常处理完整
- [ ] 日志不包含敏感信息
# 3. 确保路径在允许目录内（防止 ../../etc/passwd 攻击）
if not str(full_path).startswith(str(base.resolve())):
raise ValueError(f"Path traversal detected: {full_path}")
return str(full_path)
# 使用示例
project_path = safe_project_path(project_id, "artifacts")
文件写入白名单：
artifacts/, wiki/, scripts/ 目录具体的目录结构根据实验类型（数据分析、算法验证、管理系统）而定，详见后面各类别的说明。
本技能帮助用户编写符合学术标准的实验验证代码，用于学术论文、毕业论文或会议投稿。覆盖三大类场景：
用于统计分析、数据处理和可视化任务。
所有代码和产物输出到 ~/.research-assistant/projects/{project_id}/artifacts/data_analysis/，结构如下：
artifacts/data_analysis/
├── README.md # 项目说明和完整复现指南
├── requirements.txt # 精确固定的依赖版本 (使用==)
├── data/
│ ├── generate_data.py # 数据生成脚本 (如果是模拟数据)
│ └── sample_data.csv # 样本数据
├── src/
│ ├── __init__.py
│ ├── config.py # 配置文件 (随机种子、参数等)
│ ├── data_loader.py # 数据加载和预处理
│ ├── analysis.py # 核心统计分析
│ └── visualization.py # 图表生成
├── tests/
│ ├── test_data_loader.py
│ ├── test_analysis.py
│ └── test_integration.py
├── main.py # 入口文件
├── report/
│ ├── analysis_report.md # 完整分析报告
│ └── figures/ # 高分辨率图表 (300dpi+)
└── LICENSE # 开源许可证 (推荐 MIT 或 Apache 2.0)
1. 可复现性 (必须)
src/config.py 设置全局随机种子:
import os
# ⚠️ 从环境变量读取种子，允许用户自定义；默认值 42 仅用于示例
RANDOM_SEED = int(os.environ.get('RESEARCH_RANDOM_SEED', '42'))
np.random.seed(RANDOM_SEED)
random.seed(RANDOM_SEED)
print(f"Using random seed: {RANDOM_SEED}")
# ⚠️ 重要：在报告中标注使用的随机种子值，不要直接复制示例的 42
requirements.txt 使用精确版本:
# ⚠️ 依赖版本说明：使用 2023-2024 年稳定版本，修复已知 CVE
numpy==1.24.3 # 2023 年版本
生成 requirements.txt 前检查：
python --versionPython 版本兼容性：
| Python 版本 | numpy | pandas | scipy | matplotlib |
|------------|-------|--------|-------|------------|
| 3.6-3.7 | 1.19.5| 1.1.5 | 1.5.4 | 3.3.4 |
| 3.8-3.9 | 1.24.3| 2.0.3 | 1.11.1| 3.7.2 |
| 3.10+ | 1.26.0| 2.1.0 | 1.12.0| 3.8.0 |
降级方案：
版本冲突检查：
使用 pip check 检查已安装依赖的兼容性
在生成 requirements.txt 时检查已知冲突
提供冲突解决建议
pandas==2.0.3 # 2023 年版本 scipy==1.11.1 # 2023 年版本 matplotlib==3.7.2 # 2023 年版本 seaborn==0.12.2 # 2023 年版本 statsmodels==0.14.0 # 2023 年版本
README 中记录运行环境：Python 版本、操作系统、硬件配置
2. 统计分析严谨性 (必须)
from statsmodels.stats.power import TTestIndPower
# 计算所需样本量
effect_size = 0.42 # 预期效应量
alpha = 0.05
power = 0.80
analysis = TTestIndPower()
required_n = analysis.solve_power(effect_size=effect_size, alpha=alpha, power=power)
print(f"功效分析：检测到效应量 d={effect_size} 需要 N >= {int(required_n)}")
print(f"实际样本量：N = {len(df)}")
if len(df) < required_n:
print(f"⚠️ 警告：样本量不足，统计功效可能低于{power}")
from statsmodels.stats.multitest import multipletests
p_values = [0.01, 0.03, 0.001, 0.05, 0.2]
# FDR 校正 (Benjamini-Hochberg, 推荐)
reject_fdr, pvals_fdr, _, _ = multipletests(p_values, method='fdr_bh')
print(f"校正前显著：{sum(p < 0.05 for p in p_values)}")
print(f"FDR 校正后显著：{sum(reject_fdr)}")
3. 数据质量
data/generate_data.py 脚本，说明数据生成逻辑4. 测试覆盖 (必须)
测试代码示例:
# tests/test_data_loader.py
import unittest
import pandas as pd
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
from data_loader import load_data
class TestDataLoader(unittest.TestCase):
def test_load_valid_csv(self):
"""测试加载有效 CSV"""
pass
def test_handle_missing_values(self):
"""测试处理缺失值"""
pass
def test_handle_extreme_values(self):
"""测试处理极端值"""
pass
# tests/test_analysis.py
import unittest
import numpy as np
from analysis import pearson_correlation
class TestAnalysis(unittest.TestCase):
def test_perfect_positive_correlation(self):
"""测试完全正相关 r = 1.0"""
x = np.array([1, 2, 3, 4, 5])
y = np.array([2, 4, 6, 8, 10])
results = pearson_correlation(x, y)
self.assertAlmostEqual(results['r'], 1.0, places=5)
def test_perfect_negative_correlation(self):
"""测试完全负相关 r = -1.0"""
x = np.array([1, 2, 3, 4, 5])
y = np.array([10, 8, 6, 4, 2])
results = pearson_correlation(x, y)
self.assertAlmostEqual(results['r'], -1.0, places=5)
def test_no_correlation(self):
"""测试无相关 r ≈ 0"""
np.random.seed(42)
x = np.random.randn(100)
y = np.random.randn(100)
results = pearson_correlation(x, y)
self.assertLess(abs(results['r']), 0.3)
# tests/test_integration.py
import unittest
class TestIntegration(unittest.TestCase):
def test_full_analysis_pipeline(self):
"""测试从数据加载到报告生成的完整流程"""
pass
运行测试:
python -m pytest tests/ -v --cov=src --cov-report=html
覆盖率目标: >80%
5. 报告质量 使用以下模板:
# [实验标题] - 数据分析报告
## 1. 研究假设
明确陈述原假设 (H0) 和备择假设 (H1)
## 2. 数据描述
- 样本量：N = ?
- 变量：自变量、因变量、控制变量
- 数据来源：真实数据/模拟数据 (如果是模拟，说明生成方法)
- 描述性统计：均值、标准差、范围
## 3. 方法论
- 统计方法选择及理由
- 显著性水平：α = 0.05
- 功效分析：所需样本量计算
- 统计假设检验：正态性、方差齐性等
## 4. 结果
- 主要统计量：r/t/F值，p 值，效应量，置信区间
- 可视化：图 1、图 2(在正文中引用)
- 表格：清晰展示关键数据
## 5. 结论
- 假设是否得到支持？
- 效应大小的实际意义
- 局限性：样本偏差、测量误差、遗漏变量
- 未来研究方向
## 6. 可复现性说明
- 随机种子：42
- Python 版本：3.9.7
- 关键依赖：numpy 1.24.3, scipy 1.11.1
- 运行时间：约 X 分钟
6. 可视化标准 (必须)
可视化生成代码:
from visualization import scatter_with_regression, residual_plot, qq_plot
figures_dir = Path(__file__).parent / 'report' / 'figures'
figures_dir.mkdir(parents=True, exist_ok=True)
# 图 1: 散点图 + 回归线
scatter_with_regression(x, y, results, str(figures_dir / 'scatter_plot.png'))
# 图 2: 残差图
residual_plot(x, y, results, str(figures_dir / 'residual_plot.png'))
# 图 3: Q-Q 图
qq_plot(residuals, str(figures_dir / 'qq_plot.png'))
图注示例 (在 report/analysis_report.md 中):
**图 1**: 学习时间与考试成绩的散点图。实线为线性回归拟合，
阴影区域为 95% 置信区间。r = 0.65, p < 0.001, R² = 0.42。
技术要求:
用于实现和测试算法，验证性能声明。
artifacts/algorithm_verification/
├── README.md
├── requirements.txt
├── src/
│ ├── __init__.py
│ ├── config.py # 随机种子、实验参数
│ ├── algorithm.py # 新算法实现
│ ├── baselines.py # 基线算法 (至少 2-3 个)
│ └── metrics.py # 性能测量和数据生成
├── tests/
│ ├── test_correctness.py
│ ├── test_performance.py
│ └── test_edge_cases.py
├── experiments/
│ ├── run_benchmark.py # 完整基准测试
│ └── analyze_results.py # 结果分析和可视化
├── main.py
└── results/
├── experiment_results.md # 实验报告
├── raw_data.csv # 原始测量数据
└── figures/ # 性能曲线图
1. 算法实现质量
2. 性能测试科学性 (必须)
预热机制:
# 错误：直接测量
time_start = time.perf_counter()
result = algorithm(data)
time_end = time.perf_counter()
# 正确：先预热
for _ in range(2): # 预热 2 次
_ = algorithm(data.copy())
GC 控制和稳定性检查 (必须):
import gc
import numpy as np
import time
def benchmark_with_gc_control(algorithm, data, warmup=2, runs=20):
"""性能测试 - 包含 GC 控制和稳定性检查"""
# 预热
for _ in range(warmup):
_ = algorithm(data.copy())
# 禁用 GC 进行测量
gc.disable()
times = []
for _ in range(runs):
data_copy = data.copy()
start = time.perf_counter()
result = algorithm(data_copy)
end = time.perf_counter()
times.append((end - start) * 1000) # ms
gc.enable()
# 剔除离群值 (IQR 方法)
times = np.array(times)
q1, q3 = np.percentile(times, [25, 75])
iqr = q3 - q1
lower = q1 - 1.5 * iqr
upper = q3 + 1.5 * iqr
clean_times = times[(times >= lower) & (times <= upper)]
# 稳定性检查
mean_time = np.mean(clean_times)
std_time = np.std(clean_times)
cv = std_time / mean_time
if cv > 0.1:
print(f"⚠️ 警告：性能不稳定 (CV={cv:.2f} > 0.1)")
print(f" 建议：增加测量次数到 50 次或检查系统负载")
return {
'mean': mean_time,
'std': std_time,
'cv': cv,
'min': np.min(clean_times),
'max': np.max(clean_times),
'runs': len(clean_times)
}
要求:
样本量:
数据规模:
3. 基线对比 (必须)
4. 测试覆盖 (必须)
测试代码示例:
# tests/test_correctness.py
import unittest
class TestCorrectness(unittest.TestCase):
def test_empty_input(self):
self.assertEqual(algorithm([]), [])
def test_single_element(self):
self.assertEqual(algorithm([1]), [1])
def test_sorted_input(self):
data = [1, 2, 3, 4, 5]
self.assertEqual(algorithm(data), sorted(data))
def test_reverse_input(self):
data = [5, 4, 3, 2, 1]
self.assertEqual(algorithm(data), sorted(data))
def test_random_input(self):
import random
random.seed(42)
data = [random.randint(0, 1000) for _ in range(100)]
self.assertEqual(algorithm(data), sorted(data))
def test_all_same_elements(self):
data = [5, 5, 5, 5, 5]
self.assertEqual(algorithm(data), sorted(data))
def test_with_duplicates(self):
data = [3, 1, 4, 1, 5, 9, 2, 6]
self.assertEqual(algorithm(data), sorted(data))
def test_consistency_with_stdlib(self):
import random
for _ in range(10):
data = [random.randint(0, 10000) for _ in range(1000)]
self.assertEqual(algorithm(data), sorted(data))
# tests/test_performance.py
import unittest
import time
import numpy as np
class TestPerformance(unittest.TestCase):
def test_stability_check(self):
"""稳定性检查 - CV < 0.1"""
times = []
for _ in range(20):
data = list(range(1000))
start = time.perf_counter()
algorithm(data)
elapsed = (time.perf_counter() - start) * 1000
times.append(elapsed)
cv = np.std(times) / np.mean(times)
if cv > 0.1:
print(f"⚠️ 警告：性能不稳定 (CV={cv:.2f})")
self.assertLess(cv, 0.15)
5. 实验报告
使用以下模板:
# [算法名称] 性能验证报告
## 1. 研究假设
H1: 新算法在 [特定场景] 下比 [基线算法] 快 X 倍
## 2. 算法描述
- 核心思想
- 时间复杂度：最好/平均/最坏
- 空间复杂度
- 创新点 (如果有)
## 3. 实验设计
- 基线算法：QuickSort, MergeSort, Timsort
- 数据规模：100 - 50000
- 数据模式：random, sorted, reverse, nearly_sorted
- 测量方法：预热 2 次，正式测量 20 次，剔除离群值
- 运行环境：CPU 型号、内存、Python 版本
## 4. 结果
### 4.1 正确性验证
所有测试通过，结果与标准库一致
### 4.2 性能对比
**Nearly-sorted 数据**(关键场景):
| Size | NewAlg (ms) | QuickSort (ms) | Speedup |
|------|-------------|----------------|---------|
| 1000 | 2.3 ± 0.1 | 15.2 ± 0.8 | 6.6x |
| 5000 | 12.1 ± 0.5 | 89.3 ± 3.2 | 7.4x |
**Random 数据**(对照):
| Size | NewAlg (ms) | QuickSort (ms) | Speedup |
|------|-------------|----------------|---------|
| 1000 | 45.2 ± 2.1 | 18.3 ± 0.9 | 0.4x |
### 4.3 复杂度拟合
```python
import numpy as np
from scipy.optimize import curve_fit
def fit_complexity(sizes, times):
"""拟合时间复杂度"""
sizes = np.array(sizes)
times = np.array(times)
# O(n) 拟合
p_linear, _ = curve_fit(lambda n, a: a * n, sizes, times)
r2_linear = 1 - (ss_res / ss_tot)
# O(n log n) 拟合
p_nlogn, _ = curve_fit(lambda n, a: a * n * np.log2(n), sizes, times)
r2_nlogn = 1 - (ss_res / ss_tot)
return {
'linear': (p_linear[0], r2_linear),
'nlogn': (p_nlogn[0], r2_nlogn)
}
fit_result = fit_complexity(sizes, times_adaptive)
print(f"O(n) 拟合：R² = {fit_result['linear'][1]:.4f}")
print(f"O(n log n) 拟合：R² = {fit_result['nlogn'][1]:.4f}")
**6. 可视化 (必须)**
必须生成以下图表:
1. ✅ **性能曲线图** (时间 vs 规模) - 所有算法对比
2. ✅ **Speedup 柱状图** (关键场景下 vs 基线)
3. ✅ **复杂度拟合曲线** (显示 O(n) 或 O(n log n) 拟合)
4. ✅ **箱线图** (显示时间分布和离群值)
---
## 类别 C: 管理软件系统
用于构建系统来演示软件工程概念 (如架构模式)。
### 项目结构
artifacts/management_system/ ├── README.md ├── requirements.txt ├── src/ │ ├── init.py │ ├── models.py # 数据模型 │ ├── database.py # 数据库操作 │ ├── services.py # 业务逻辑 │ └── views.py # 用户界面 ├── tests/ │ ├── test_models.py │ ├── test_services.py │ └── test_integration.py ├── data/ │ └── init_data.sql # 初始数据 ├── docs/ │ ├── system_design.md # 系统设计文档 │ ├── architecture.png # 架构图 │ └── comparison.md # 与传统架构的对比分析 ├── baseline/ # 对照组：传统架构实现 │ └── monolithic.py └── main.py
### 关键要求
**1. 架构实现质量**
- 分层清晰，职责单一
- 使用依赖注入，降低耦合
- 事务管理：数据库操作要有事务保护
- 错误处理：合理的异常处理和日志记录
**2. 数据库设计**
- 合理的表结构和外键约束
- 启用外键约束：`PRAGMA foreign_keys = ON` (SQLite)
- 使用参数化查询防止 SQL 注入
- 添加 CHECK 约束保证数据完整性
**3. 对照实验 (必须)**
**对照组实现** `baseline/monolithic.py`:
```python
"""
单体架构实现 - 反例
所有逻辑混在一个文件中，用于对比 MVC 架构的优势
"""
class MonolithicLibrary:
"""单体架构 - 职责不单一"""
def __init__(self):
self.books = {}
self.members = {}
self.loans = {}
def add_book(self, title, author, isbn):
book_id = len(self.books) + 1
self.books[book_id] = {
'id': book_id, 'title': title, 'author': author,
'isbn': isbn, 'available': True
}
return book_id
def loan_book(self, book_id, member_id):
# 混合了验证、业务逻辑、数据操作
if book_id not in self.books:
return None
if not self.books[book_id]['available']:
return None
loan_id = len(self.loans) + 1
self.loans[loan_id] = {...}
self.books[book_id]['available'] = False
return loan_id
维护成本对比实验 (必须，5 个场景):
| 场景 | 描述 | MVC 文件数 | 单体文件数 | MVC 行数 | 单体行数 |
|---|---|---|---|---|---|
| 1 | 添加"出版社"字段 | 3 | 1 | +15 | +8 |
| 2 | 借阅期限 14→30 天 | 1 | 1 | +2 | +2 |
| 3 | 添加图书预约功能 | 4 | 1 | +80 | +60 |
| 4 | 添加逾期罚款计算 | 3 | 1 | +50 | +40 |
| 5 | 支持多副本管理 | 4 | 1 | +100 | +80 |
量化对比指标:
运行分析:
# 计算圈复杂度
radon cc src/ -a -s
radon cc baseline/monolithic.py -a -s
# 计算维护性指数
radon mi src/ -s
radon mi baseline/monolithic.py -s
4. 测试覆盖 (必须)
测试代码示例:
# tests/test_models.py
import unittest
from src.models import Book, Member, Loan
from datetime import datetime, timedelta
class TestModels(unittest.TestCase):
def test_book_creation(self):
book = Book(id=1, title="Test", author="Author", isbn="123", year=2020)
self.assertEqual(book.title, "Test")
self.assertTrue(book.available)
def test_loan_overdue_detection(self):
loan = Loan(
id=1, book_id=1, member_id=1,
loan_date=datetime.now() - timedelta(days=20),
due_date=datetime.now() - timedelta(days=6)
)
self.assertTrue(loan.is_overdue)
self.assertEqual(loan.days_overdue, 6)
# tests/test_services.py
import unittest
from src.services import LibraryService
from src.database import Database
class TestServices(unittest.TestCase):
def setUp(self):
self.db = Database(":memory:")
self.service = LibraryService(self.db)
self.book_id = self.db.create_book(...)
self.member_id = self.db.create_member(...)
def test_delete_book_on_loan(self):
"""测试不能删除已借出的图书"""
self.service.loan_book(self.book_id, self.member_id)
with self.assertRaises(BookOnLoanError):
self.service.delete_book(self.book_id)
def test_member_with_overdue_cannot_borrow(self):
"""测试会员有逾期未还不能借书"""
with self.assertRaises(MemberHasOverdueError):
self.service.loan_book(another_book_id, self.member_id)
# tests/test_integration.py
import unittest
class TestIntegration(unittest.TestCase):
def test_full_loan_return_cycle(self):
"""测试完整的借阅 - 归还流程"""
pass
def test_concurrent_loans(self):
"""测试并发借阅"""
import threading
# ...
测试覆盖要求:
5. 系统设计文档
使用以下模板:
# [系统名称] 设计文档
## 1. 研究假设
H1: MVC 架构比单体架构在可维护性和可扩展性上更优
## 2. 系统架构
### 2.1 MVC 架构 (实验组)
[架构图]
- Model 层：数据模型和数据库操作
- View 层：用户界面
- Controller 层：业务逻辑
### 2.2 单体架构 (对照组)
[架构图]
- 所有逻辑混在一起
## 3. 实验设计
### 3.1 维护场景
**场景 1:添加"出版社"字段**
- MVC 架构:
- 修改文件：models.py, database.py, views.py
- 代码行数:+15 行
- 时间:10 分钟
- 单体架构:
- 修改文件：monolithic.py
- 代码行数:+8 行 (但需要修改多处)
- 时间:25 分钟 (需要理解整个文件)
**场景 2-5**: [类似分析]
### 3.2 代码质量指标
| 指标 | MVC | 单体 |
|------|-----|------|
| 代码行数 | 450 | 380 |
| 圈复杂度 (平均) | 3.2 | 8.5 |
| 模块耦合度 | 低 | 高 |
| 单元测试覆盖率 | 85% | 45% |
## 4. 结果分析
- MVC 架构在 5 个维护场景中，平均修改时间比单体架构少 40%
- MVC 架构的圈复杂度显著低于单体架构 (p < 0.01)
- MVC 架构更易于单元测试
## 5. 结论
假设得到支持：MVC 架构在可维护性上优于单体架构
## 6. 局限性
- 样本量小 (只有 5 个维护场景)
- 未测试大规模系统
- 未考虑学习曲线 (开发者熟悉度)
代码生成时间：
代码执行时间：
超时处理：
适用场景：
工作流程：
在开始写代码前，先评估：
如任何一项为"否"，进入研究流程。
使用 searxng 搜索相关技术文档：
搜索工具配置：
使用配置的搜索工具进行搜索（通过环境变量 RESEARCH_SEARCH_TOOL 配置）：
# 默认配置（在 .env 或 openclaw.json 中）
RESEARCH_SEARCH_TOOL="uv run scripts/searxng.py"
# 通用搜索命令格式
$RESEARCH_SEARCH_TOOL search "{query}" -n {result_count} --format json
搜索示例：
# 搜索统计模型
$RESEARCH_SEARCH_TOOL search "{模型名称} python implementation" -n 20 --format json
# 搜索算法实现
$RESEARCH_SEARCH_TOOL search "{算法名称} algorithm python tutorial" -n 20 --format json
# 搜索最佳实践
$RESEARCH_SEARCH_TOOL search "{分析方法} best practices assumptions" -n 10 --format json
# 搜索官方文档
$RESEARCH_SEARCH_TOOL search "site:scipy.org {模型名称}" -n 10 --format json
$RESEARCH_SEARCH_TOOL search "site:statsmodels.org {模型名称}" -n 10 --format json
支持的搜索工具：
uv run scripts/searxng.pypython scripts/bing_search.pycurl -X POST {api_endpoint}搜索关键词示例：
"Hierarchical Linear Modeling python statsmodels"
"Random Forest algorithm implementation python sklearn"
"Granger Causality Test assumptions python"
"Structural Equation Modeling python semopy tutorial"
"Random Forest algorithm implementation python sklearn"
"Granger Causality Test assumptions python"
"Structural Equation Modeling python semopy tutorial"
创建 research_notes.md 记录关键信息：
# 调研笔记：{方法/算法名称}
## 核心概念
- 定义：...
- 适用场景：...
- 统计假设：...
## Python 实现
- 推荐库：{库名称}
- 安装：pip install {库名称}
- 基本用法：
```python
from {库} import {类}
model = {类}()
model.fit(X, y)
[从搜索结果中提取可靠代码示例]
### 第四步：验证理解
在写完整代码前，先写一个最小可运行示例（MVP）：
```python
# mvp_test.py
"""最小可运行示例 - 验证对方法的理解"""
import numpy as np
from {库} import {模型}
# 生成简单测试数据
np.random.seed(42)
X = np.random.randn(100, 2)
y = X[:, 0] * 2 + X[:, 1] + np.random.randn(100) * 0.1
# 拟合并验证
model = {模型}()
model.fit(X, y)
print(f"R² = {model.score(X, y):.3f}")
# 预期：R² ≈ 0.9+（因为数据是线性生成的）
验证标准：
基于调研结果和 MVP 验证，编写完整分析代码。
## 方法论补充说明
**研究过程**：
本分析使用的 {方法名称} 在初始阶段不熟悉，通过以下步骤确保正确实现：
1. **文献调研**：搜索并阅读了 5 篇教程和官方文档
2. **MVP 验证**：先用简单数据验证理解正确
3. **代码审查**：对照官方示例检查实现
**参考资料**：
1. [库名称] 官方文档：{链接}
2. {教程名称}：{链接}
3. {论文名称}：{链接}
**置信度**：高（MVP 验证通过 + 对照官方示例）
创建 scripts/research_search.py：
#!/usr/bin/env python3
"""研究辅助搜索工具"""
import subprocess
import json
from pathlib import Path
def search_research_topic(topic: str, output_file: str = "research_notes.md"):
"""搜索并整理调研结果"""
queries = [
f"{topic} python implementation",
f"{topic} tutorial example",
f"{topic} assumptions requirements",
f"site:scipy.org {topic}",
f"site:statsmodels.org {topic}",
]
results = []
for query in queries:
cmd = [
"uv", "run", "scripts/searxng.py",
"search", query,
"-n", "10",
"--format", "json"
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
results.extend(json.loads(result.stdout))
# 整理结果
notes = generate_research_notes(topic, results[:20])
with open(output_file, 'w', encoding='utf-8') as f:
f.write(notes)
print(f"调研笔记已保存：{output_file}")
return output_file
def generate_research_notes(topic: str, results: list) -> str:
"""生成调研笔记"""
notes = f"# 调研笔记：{topic}\n\n"
notes += "## 搜索结果\n\n"
for i, result in enumerate(results, 1):
notes += f"{i}. [{result.get('title', '无标题')}]({result.get('url', '')})\n"
notes += f" {result.get('content', '无摘要')[:200]}...\n\n"
notes += "\n## 下一步\n"
notes += "1. 阅读官方文档\n2. 运行示例代码\n3. 编写 MVP 验证\n"
return notes
if __name__ == '__main__':
import sys
if len(sys.argv) > 1:
search_research_topic(sys.argv[1])
else:
print("用法：python research_search.py {搜索主题}")
# 搜索并生成调研笔记
python scripts/research_search.py "Hierarchical Linear Modeling"
# 输出：research_notes.md
在开始写代码前，确认：
如任何一项未完成，返回研究流程。
询问:
根据用户需求，选择对应类别（A/B/C）。
无论哪个类别，都遵循:
可读性优先
自包含
可复现
测试充分
交付前检查清单:
可复现性
统计严谨性(数据分析)
实验设计(算法/系统)
代码质量
报告完整性
向用户报告:
调研笔记存档位置：
~/.research-assistant/projects/{project_id}/
├── wiki/ # 知识库
├── drafts/ # 草稿
├── artifacts/ # 实验产物
├── research/ # ← 研究笔记目录
│ ├── {方法名称}_notes.md # 调研笔记
│ ├── mvp_tests/ # MVP 验证代码
│ │ └── {方法名称}_mvp_test.py
│ └── raw_search_results/ # 原始搜索结果（可选）
│ └── {方法名称}_search.json
└── .progress.json
存档规则：
research/{方法名称}_notes.mdresearch/mvp_tests/{方法名称}_mvp_test.pyresearch/raw_search_results/{方法名称}_search.json（可选，用于追溯）优点：
阶段 4.X：研究先行（Prose 层面控制）
if **coder 报告方法不熟悉**:
session "执行研究先行流程"
role: spark-science-coder
prompt: |
检测到分析方法（{方法名称}）不熟悉，执行研究先行流程：
1. 搜索调研（使用配置的搜索工具）
2. 生成 research_notes.md
3. 编写 MVP 验证
4. 验证通过后继续写完整代码
研究笔记存入：~/.research-assistant/projects/{{ project_id }}/research/{方法名称}_notes.md
let research_result = session "..."
# 验证 MVP 结果
if **research_result.mvp_passed == false**:
# MVP 验证失败，继续研究或通知用户
session "通知用户研究遇到困难"
prompt: |
MVP 验证失败（R² = {r_squared} < 0.5），可能原因：
1. 方法理解有误
2. 数据生成逻辑错误
3. 模型参数设置错误
建议：
- 继续研究（重新搜索、阅读官方文档）
- 更换方法（使用更熟悉的方法）
- 寻求人工帮助
在 .env 或 openclaw.json 中配置：
{
"env": {
"RESEARCH_SEARCH_TOOL": "uv run scripts/searxng.py",
"RESEARCH_MAX_TIME_MIN": "30",
"RESEARCH_MVP_THRESHOLD": "0.5"
}
}
配置说明：
RESEARCH_SEARCH_TOOL: 搜索工具命令RESEARCH_MAX_TIME_MIN: 研究超时时间（分钟）RESEARCH_MVP_THRESHOLD: MVP 验证通过阈值（R²）使用前验证：
#!/bin/bash
# scripts/verify_search_tool.sh
# 1. 检查环境变量是否设置
if [ -z "$RESEARCH_SEARCH_TOOL" ]; then
echo "❌ 错误：RESEARCH_SEARCH_TOOL 未配置"
echo "请在.env 中配置：RESEARCH_SEARCH_TOOL=\"uv run scripts/searxng.py\""
exit 1
fi
# 2. 检查命令是否可执行
if ! command -v $RESEARCH_SEARCH_TOOL &> /dev/null; then
echo "❌ 错误：搜索工具不可用：$RESEARCH_SEARCH_TOOL"
exit 1
fi
# 3. 测试搜索功能
echo "测试搜索功能..."
$RESEARCH_SEARCH_TOOL search "test" -n 1 --format json > /dev/null 2>&1
if [ $? -ne 0 ]; then
echo "❌ 错误：搜索工具测试失败"
exit 1
fi
echo "✅ 搜索工具验证通过"
降级方案：
RESEARCH_SEARCH_TOOL="uv run scripts/searxng.py"根据问题类型选择验证指标：
def validate_mvp(model, X_test, y_test, problem_type: str, domain: str = "general"):
"""
MVP 验证 - 根据问题类型和领域选择合适指标
参数：
model: 训练好的模型
X_test, y_test: 测试数据
problem_type: "regression" / "classification" / "clustering"
domain: "social_science" / "natural_science" / "general"
返回：
dict: {passed: bool, metric_name: str, metric_value: float, threshold: float}
"""
if problem_type == "regression":
from sklearn.metrics import r2_score
r_squared = r2_score(y_test, model.predict(X_test))
# 根据领域调整阈值
if domain == "social_science":
threshold = 0.3 # 社会科学 R² > 0.3 可接受
elif domain == "natural_science":
threshold = 0.7 # 自然科学要求更高
else:
threshold = 0.5
# 检测过于完美的数据
if r_squared > 0.99:
print("⚠️ 警告：R²过高（{:.3f}），可能数据生成有误".format(r_squared))
print("建议：检查数据生成逻辑，添加合理噪声")
return {
"passed": r_squared > threshold,
"metric_name": "R²",
"metric_value": r_squared,
"threshold": threshold
}
elif problem_type == "classification":
from sklearn.metrics import accuracy_score, f1_score
accuracy = accuracy_score(y_test, model.predict(X_test))
f1 = f1_score(y_test, model.predict(X_test), average='weighted')
# 使用 F1 分数（更稳健）
threshold = 0.5
passed = f1 > threshold
return {
"passed": passed,
"metric_name": "F1",
"metric_value": f1,
"threshold": threshold
}
elif problem_type == "clustering":
from sklearn.metrics import silhouette_score
labels = model.predict(X_test)
silhouette = silhouette_score(X_test, labels)
threshold = 0.3
passed = silhouette > threshold
return {
"passed": passed,
"metric_name": "轮廓系数",
"metric_value": silhouette,
"threshold": threshold
}
else:
raise ValueError(f"未知问题类型：{problem_type}")
使用示例：
# 回归问题（社会科学）
result = validate_mvp(model, X_test, y_test, "regression", "social_science")
print(f"MVP 验证：{result['metric_name']} = {result['metric_value']:.3f} > {result['threshold']}")
# 输出：MVP 验证：R² = 0.42 > 0.3 ✅ 通过
# 分类问题
result = validate_mvp(model, X_test, y_test, "classification")
print(f"MVP 验证：{result['metric_name']} = {result['metric_value']:.3f} > {result['threshold']}")
必填字段（Frontmatter）：
---
method_name: Hierarchical Linear Modeling
search_date: 2026-04-30
sources_count: 5
mvp_status: passed
mvp_metric: R² = 0.85
confidence: high
domain: social_science
tags: [multilevel, regression, statsmodels]
---
必填章节：
# 调研笔记：{方法名称}
## 1. 核心概念
- **定义**：...
- **适用场景**：...
- **统计假设**：...
- **数学公式**：（可选）
## 2. Python 实现
- **推荐库**：statsmodels
- **安装**：`pip install statsmodels`
- **导入**：`from statsmodels.regression.mixed_linear_model import MixedLM`
- **基本用法**：
```python
model = MixedLM(endog, exog, groups=groups)
result = model.fit()
print(result.summary())
| 参数 | 说明 | 默认值 | 示例 |
|---|---|---|---|
| endog | 因变量 | 必填 | y |
| exog | 固定效应自变量 | 必填 | X |
| groups | 分组变量 | 必填 | group_ids |
| exog_re | 随机效应自变量 | 可选 | None |
# 完整可运行示例
import numpy as np
from statsmodels.regression.mixed_linear_model import MixedLM
# 生成测试数据
np.random.seed(42)
n_groups = 10
n_per_group = 20
groups = np.repeat(range(n_groups), n_per_group)
X = np.random.randn(n_groups * n_per_group, 2)
y = X[:, 0] * 2 + np.random.randn(n_groups * n_per_group) * 0.1
# 拟合模型
model = MixedLM(y, X, groups=groups)
result = model.fit()
print(result.summary())
**质量检查清单**：
- [ ] Frontmatter 完整（方法名、日期、来源数、MVP 状态、置信度）
- [ ] 至少 3 个可靠来源
- [ ] 包含可运行代码示例
- [ ] 注明统计假设
- [ ] 注明关键参数及默认值
- [ ] MVP 验证通过（如适用）
- [ ] 置信度评估（high/medium/low）
**置信度评估标准**：
- **high**: MVP 验证通过 + 至少 2 个官方来源
- **medium**: MVP 验证通过 + 1 个官方来源
- **low**: MVP 未通过 或 无官方来源
---
## 附录 D：研究超时控制
**Prose 层面超时**：
```prose
session "执行研究先行流程"
role: spark-science-coder
timeout: {{ env.RESEARCH_MAX_TIME_MIN | default(30) | int * 60 }} # 转换为秒
prompt: |
执行研究先行流程...
on_timeout:
session "通知用户研究超时"
prompt: |
⚠️ 研究超时（{{ env.RESEARCH_MAX_TIME_MIN | default(30) }}分钟）
**已完成**：
- 搜索：{已完成步骤}
- 调研笔记：{是否生成}
- MVP: {是否完成}
**建议**：
1. 基于现有结果继续（可能不完整）
2. 寻求人工帮助
3. 更换更简单的方法
脚本层面超时：
#!/usr/bin/env python3
"""研究流程 - 带超时控制"""
import signal
import sys
from datetime import datetime
class TimeoutError(Exception):
pass
def timeout_handler(signum, frame):
raise TimeoutError(f"研究超时（{RESEARCH_MAX_TIME_MIN}分钟）")
def research_with_timeout(topic: str, max_time_min: int = 30):
"""带超时控制的研究流程"""
# 设置超时闹钟
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(max_time_min * 60)
start_time = datetime.now()
progress = {
"search": False,
"notes": False,
"mvp": False
}
try:
# 步骤 1：搜索
print(f"[{datetime.now().strftime('%H:%M:%S')}] 开始搜索...")
search_results = search_topic(topic)
progress["search"] = True
# 步骤 2：整理笔记
print(f"[{datetime.now().strftime('%H:%M:%S')}] 整理调研笔记...")
notes = generate_notes(search_results)
progress["notes"] = True
# 步骤 3：MVP 验证
print(f"[{datetime.now().strftime('%H:%M:%S')}] MVP 验证...")
mvp_result = validate_mvp()
progress["mvp"] = True
# 取消闹钟
signal.alarm(0)
elapsed = (datetime.now() - start_time).total_seconds() / 60
print(f"✅ 研究完成，耗时 {elapsed:.1f} 分钟")
return {
"success": True,
"elapsed_min": elapsed,
"progress": progress
}
except TimeoutError as e:
elapsed = (datetime.now() - start_time).total_seconds() / 60
print(f"\n⚠️ {str(e)}")
print(f"已完成：{progress}")
return {
"success": False,
"reason": "timeout",
"elapsed_min": elapsed,
"progress": progress
}
来源优先级排序：
| 优先级 | 来源类型 | 域名示例 | 可靠性 |
|---|---|---|---|
| 1 | 官方文档 | scipy.org, statsmodels.org, sklearn.org | ⭐⭐⭐ 高 |
| 2 | 学术论文 | arxiv.org, scholar.google.com | ⭐⭐⭐ 高 |
| 3 | 官方教程 | GitHub 官方仓库 | ⭐⭐ 中高 |
| 4 | 知名社区 | stackoverflow.com, crossvalidated.stats.stackexchange.com | ⭐⭐ 中 |
| 5 | 教育网站 | towardsdatascience.com, machinelearningmastery.com | ⭐⭐ 中 |
| 6 | 博客/论坛 | 个人博客、CSDN、知乎 | ⭐ 低 |
搜索策略：
#!/bin/bash
# scripts/search_with_reliability.sh
TOPIC="$1"
echo "=== 第一阶段：搜索官方文档 ==="
$RESEARCH_SEARCH_TOOL search "site:scipy.org $TOPIC" -n 5 --format json > official_results.json
$RESEARCH_SEARCH_TOOL search "site:statsmodels.org $TOPIC" -n 5 --format json >> official_results.json
$RESEARCH_SEARCH_TOOL search "site:sklearn.org $TOPIC" -n 5 --format json >> official_results.json
echo "=== 第二阶段：搜索学术论文 ==="
$RESEARCH_SEARCH_TOOL search "site:arxiv.org $TOPIC" -n 5 --format json > academic_results.json
echo "=== 第三阶段：搜索通用结果 ==="
$RESEARCH_SEARCH_TOOL search "$TOPIC python tutorial" -n 10 --format json > general_results.json
echo "=== 合并结果（按可靠性排序）=== "
cat official_results.json academic_results.json general_results.json > all_results.json
来源验证代码：
def verify_source_reliability(search_results: list) -> list:
"""
验证搜索结果可靠性并排序
参数：
search_results: 搜索结果列表
返回：
list: 按可靠性排序的结果，每个结果包含 reliability 字段
"""
# 可靠性域名映射
reliability_map = {
"scipy.org": "high",
"statsmodels.org": "high",
"sklearn.org": "high",
"arxiv.org": "high",
"scholar.google.com": "high",
"github.com": "medium", # 需进一步验证是否为官方仓库
"stackoverflow.com": "medium",
"crossvalidated.stats.stackexchange.com": "medium",
"towardsdatascience.com": "medium",
"machinelearningmastery.com": "medium",
}
verified_results = []
for result in search_results:
url = result.get("url", "")
domain = url.split("//")[-1].split("/")[0]
# 确定可靠性
reliability = reliability_map.get(domain, "low")
# 低可靠性来源需交叉验证
if reliability == "low":
# 检查是否有其他来源支持同一说法
cross_verified = False
for other in search_results:
if other["url"] != url:
other_domain = other["url"].split("//")[-1].split("/")[0]
if reliability_map.get(other_domain, "low") in ["high", "medium"]:
# 检查内容是否相似
if similar_content(result["content"], other["content"]):
cross_verified = True
reliability = "medium" # 交叉验证后提升为中等
break
if not cross_verified:
print(f"⚠️ 警告：低可靠性来源且未交叉验证：{url}")
result["reliability"] = reliability
verified_results.append(result)
# 按可靠性排序
reliability_order = {"high": 0, "medium": 1, "low": 2}
verified_results.sort(key=lambda x: reliability_order.get(x["reliability"], 2))
return verified_results
def similar_content(text1: str, text2: str, threshold: float = 0.6) -> bool:
"""简单内容相似度检查"""
# 可以使用更复杂的相似度算法（如 cosine similarity）
words1 = set(text1.lower().split())
words2 = set(text2.lower().split())
intersection = words1 & words2
union = words1 | words2
similarity = len(intersection) / len(union) if union else 0
return similarity > threshold
交叉验证规则：
### 交叉验证规则
**低可靠性来源必须交叉验证**：
- 至少 2 个中高可靠性来源确认同一用法
- 或 1 个官方文档 + 1 个社区确认
**代码示例验证**：
- 必须能运行通过
- 与官方文档示例对比
- 检查结果是否合理
**统计方法验证**：
- 检查统计假设是否一致
- 检查参数名称是否匹配官方文档
- 检查结果解释是否合理
回归问题数据生成：
def generate_regression_data(
n_samples: int = 100,
n_features: int = 2,
true_coef: np.ndarray = None,
noise_std: float = 0.1,
random_seed: int = 42
) -> tuple:
"""
生成回归测试数据
参数：
n_samples: 样本量（建议 ≥50）
n_features: 特征数
true_coef: 真实系数（None 则随机生成）
noise_std: 噪声标准差（建议 0.1-0.3）
random_seed: 随机种子
返回：
X, y, true_coef
"""
np.random.seed(random_seed)
X = np.random.randn(n_samples, n_features)
if true_coef is None:
true_coef = np.random.randn(n_features)
y = X @ true_coef + np.random.randn(n_samples) * noise_std
return X, y, true_coef
分类问题数据生成：
def generate_classification_data(
n_samples: int = 200,
n_features: int = 2,
n_classes: int = 2,
class_separation: float = 2.0,
random_seed: int = 42
) -> tuple:
"""
生成分类测试数据
参数：
n_samples: 样本量（建议 ≥100）
n_features: 特征数
n_classes: 类别数
class_separation: 类别分离度（1.0-3.0）
random_seed: 随机种子
返回：
X, y
"""
from sklearn.datasets import make_classification
X, y = make_classification(
n_samples=n_samples,
n_features=n_features,
n_classes=n_classes,
n_informative=n_features,
n_redundant=0,
class_sep=class_separation,
random_state=random_seed
)
return X, y
参数建议：
缓存策略：
import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta
def get_cache_key(topic: str, search_queries: list) -> str:
"""生成缓存键"""
content = f"{topic}:{':'.join(search_queries)}"
return hashlib.md5(content.encode()).hexdigest()
def load_from_cache(cache_key: str, cache_type: str = "search") -> dict | None:
"""
从缓存加载
参数：
cache_key: 缓存键
cache_type: search(7 天) / notes(30 天) / mvp(永久)
"""
cache_dir = Path("~/.research-assistant/cache/research").expanduser()
cache_file = cache_dir / f"{cache_type}_{cache_key}.json"
if not cache_file.exists():
return None
# 检查缓存是否过期
mtime = datetime.fromtimestamp(cache_file.stat().st_mtime)
max_age = {
"search": timedelta(days=7),
"notes": timedelta(days=30),
"mvp": timedelta(days=365)
}.get(cache_type, timedelta(days=7))
if datetime.now() - mtime > max_age:
print(f"⚠️ 缓存已过期（{max_age.days}天）")
return None
with open(cache_file) as f:
return json.load(f)
def save_to_cache(cache_key: str, data: dict, cache_type: str = "search"):
"""保存到缓存"""
cache_dir = Path("~/.research-assistant/cache/research").expanduser()
cache_dir.mkdir(parents=True, exist_ok=True)
cache_file = cache_dir / f"{cache_type}_{cache_key}.json"
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"✅ 缓存已保存：{cache_file}")
缓存内容：
缓存命中提示：
✅ 缓存命中：Hierarchical Linear Modeling
上次研究：2026-04-25
调研笔记：research/hlm/notes.md
MVP 状态：通过（R² = 0.85）
是否使用缓存结果？(y/n)
脚本位置: scripts/install_research_deps.py
用法:
# 自动安装方法所需依赖
python scripts/install_research_deps.py {方法名称}
# 示例
python scripts/install_research_deps.py hierarchical_linear_modeling
python scripts/install_research_deps.py "random forest"
功能:
支持的方法:
脚本位置: scripts/verify_search_tool.py
用法:
# 验证搜索工具配置
python scripts/verify_search_tool.py
验证步骤:
返回码:
降级方案: 如验证失败，可：
推荐流程:
# 1. 验证搜索工具
python scripts/verify_search_tool.py
# 2. 安装依赖（如需要）
python scripts/install_research_deps.py {方法名称}
# 3. 执行研究流程
# （通过 prose 工作流或直接调用 coder skill）