From 0644a68ecc135889cd9d404a937eeee7361f0245 Mon Sep 17 00:00:00 2001 From: kai Date: Wed, 6 May 2026 19:23:30 +0800 Subject: [PATCH] v0.21 alpha add research brief and compressed findings --- AGENTS.md | 18 +++-- PLAN.md | 15 ++++ README.md | 4 +- scripts/dr.py | 14 +++- scripts/runtime/assembly.py | 112 +++++++++++++++++++++++++- scripts/runtime/orchestrator.py | 16 +++- scripts/runtime/phase1.py | 132 ++++++++++++++++++++++++++++++ scripts/runtime/tasks.py | 138 ++++++++++++++++++++++++++++++-- scripts/v020_regression.py | 31 ++++++- tests/test_chapter_assembly.py | 14 ++++ tests/test_phase0_materials.py | 11 +++ tests/test_research_methods.py | 19 +++++ tests/test_v020_runtime.py | 42 ++++++++++ 13 files changed, 539 insertions(+), 27 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 37ddcd4..d153314 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -64,14 +64,16 @@ ### Phase 1：框架规划 - **驱动命令**：`uv run python scripts/dr.py init ` → `uv run python scripts/dr.py frame `（`/dr-init`、`/dr-frame` 只是薄封装） - **主导入口**：Python core 生成项目骨架与 framework；dr-plan 可作为表层访谈增强 -- **产出**：`projects//phase1/framework.md`（记录 research_method、8-15 章大纲，每 section 带研究思路与字数配额） -- **暂停点**：用户确认框架 +- **产出**：`phase1/material_brief.md` + `phase1/framework.md` + `phase1/research_brief.md/json` +- **暂停点**：用户确认材料解读、研究方法、任务切分、检索策略和章节框架 +- **硬规则**：clarification 不只是问范围；必须固化报告用途、基调、成功标准、任务拆分、每个任务轴的 prompt/skills/search route，让低成本 worker 也能独立执行 ### Phase 2：深度研究 - **驱动命令**：`uv run python scripts/dr.py research --workers 6` - **主导入口**：Python core 生成 task cards 并控制并发 -- **产出**：`projects//phase2/task_cards.json` + `packets/*.json` + `drafts/chXX.md` + `evidence/chXX-evidence.md` + `sources.jsonl` +- **产出**：`phase2/task_cards.json` + `packets/*.json` + `sources.jsonl` + `chapter_briefs/*.json` + `compressed_findings/*.json` + `drafts/chXX.md` - **不暂停**：全自动跑完 +- **防碎片化规则**：并发 worker 只写 evidence packet；`--build-briefs` 必须先收束为 chapter brief 和 compressed finding；章节正文必须从 compressed finding 写，不得把 packet 按顺序拼贴成报告 ### Phase 3：总编审校 - **驱动命令**：`uv run python scripts/dr.py review `（`/dr-review` 只是薄封装） @@ -190,6 +192,7 @@ uv run python scripts/dr.py methods list - 不把平台 subagent 当默认并发机制。 - 真实并发由 `scripts/runtime/workers.py` 的 worker pool 执行。 - 真实模型选择由 `configs/models.yaml` 和 `scripts/runtime/roles.py` 执行。 +- 信息检索默认走 `scripts/search.py` / `SearchClient` / `search-gateway` skill；不得把 Tavily MCP、browser MCP 或平台 web search 作为默认路径，除非用户明确授权。 --- @@ -215,10 +218,11 @@ uv run python scripts/dr.py methods list 1. `uv run python scripts/dr.py status ` 能看到 phase 状态。 2. Phase 2 存在 `phase2/task_cards.json`。 -3. `--execute-packets` 后存在 `phase2/packets/*.json` 和必要时的 `phase2/packet_errors/*.json`。 -4. `--build-briefs` 后存在 `phase2/chapter_briefs/*.json`。 -5. `--assemble-chapters` 后存在 `phase2/drafts/chXX.md` 和必要时的 `phase2/chapter_errors/*.json`。 -6. `scripts/v020_regression.py` 输出 `v0.20 regression PASS`。 +3. Phase 1 存在 `phase1/research_brief.md` 和 `phase1/research_brief.json`。 +4. `--execute-packets` 后存在 `phase2/packets/*.json` 和必要时的 `phase2/packet_errors/*.json`。 +5. `--build-briefs` 后存在 `phase2/chapter_briefs/*.json` 与 `phase2/compressed_findings/*.json`。 +6. `--assemble-chapters` 后存在 `phase2/drafts/chXX.md` 和必要时的 `phase2/chapter_errors/*.json`。 +7. `scripts/v020_regression.py` 输出 `v0.20 regression PASS`。 --- diff --git a/PLAN.md b/PLAN.md index abdcfa5..afb60be 100644 --- a/PLAN.md +++ b/PLAN.md @@ -686,3 +686,18 @@ OpenCode 的坑：如果只是在主会话里装样子地写"让 X agent 做"， - fixture 项目必须覆盖：material brief -> research brief -> task cards -> enrichment round -> compressed findings -> chapter draft -> Phase3 score gate。 - 搜索测试必须验证 subagent prompt 中包含 `search-gateway`，且不会提及 Tavily MCP 作为默认路径。 - 质量测试必须能让“泛泛咨询腔但有引用”的章节失败，让“具体审计发现+法规条款+整改动作+待补证据”的章节通过。 + +- 2026-05-06 v0.21-alpha implementation：**Research Brief 与压缩发现先行落地** + + **已落地**： + - `scripts/runtime/phase1.py` 新增 `phase1/research_brief.md` 与 `phase1/research_brief.json`，在 `frame` 阶段把材料简报、研究方法、工作语言、写作基调、成功标准、任务切分原则、每个任务轴的 prompt brief / search route / required skills / stop conditions 固化为文件。 + - `scripts/runtime/orchestrator.py` 生成 Phase2 task cards 时优先读取 `research_brief.json`，不再只依赖章节标题和 method axes。 + - `scripts/runtime/tasks.py` 扩展 `TaskCard` schema：`research_goal`、`research_method`、`prompt_brief`、`required_skills`、`allowed_materials`、`expected_evidence`、`stop_conditions`、`model_hint`；旧 task card 会自动补默认字段，保持 fixture 兼容。 + - `scripts/runtime/assembly.py` 新增 `build_compressed_findings()` 与 `validate_compressed_finding()`，`--build-briefs` 会同步写入 `phase2/compressed_findings/chXX.json`。 + - `--assemble-chapters` 改为从 `compressed_findings` 写中文章节，减少并发 packet 直接拼接造成的碎片化。 + - `AGENTS.md` 已同步更新 Phase1/2 真实产物、search-gateway 默认路径、Python core 验证锚点。 + + **仍未完成**： + - `phase2/enrichment_rounds/roundXX/coverage_gap.json` 还未实现；下一步应先做 deterministic coverage evaluator，再让补充 task cards 从 gap 生成。 + - Phase3 evaluator rubrics 仍是计划项；当前 deterministic review 已能抓部分 draft 质量问题，但还没有分维度评分与 finalize gate。 + - DOCX/PPTX/图片批量 OCR、表格抽取、材料 source registry 仍放入后续资料导入增强。 diff --git a/README.md b/README.md index f97e260..67107d4 100644 --- a/README.md +++ b/README.md @@ -183,7 +183,7 @@ uv run python scripts/search.py "dual-target siRNA GalNAc" --route patents ``` deep_research/ ├── PLAN.md # 完整方案（中断续接从此读起） -├── AGENTS.md # 研究方法论与规则（OpenCode 自动加载） +├── AGENTS.md # 跨平台研究方法论与运行规则 ├── README.md # 本文件 ├── secrets.env.example # 密钥模板 ├── secrets.env # 你的密钥（gitignore） @@ -224,7 +224,7 @@ deep_research/ ### 1. Python core 防止上下文污染 -OpenCode/Codex/Claude Code/Antigravity/Gemini CLI 的 subagent 或 agent thread 都可能把大上下文带入子会话。v0.20 起，默认并发由 Python runtime 的任务卡执行层控制：先生成 `phase2/task_cards.json`，再按任务卡产出 `phase2/packets/*.json`，最后组装中文章节。平台 agent 只负责调用 CLI 和展示状态。 +OpenCode/Codex/Claude Code/Antigravity/Gemini CLI 的 subagent 或 agent thread 都可能把大上下文带入子会话。v0.20 起，默认并发由 Python runtime 的任务卡执行层控制：先生成 `phase1/research_brief.json` 固化研究意图和任务切分，再生成 `phase2/task_cards.json`，按任务卡产出 `phase2/packets/*.json`，经 `phase2/compressed_findings/*.json` 收束后再组装中文章节。平台 agent 只负责调用 CLI 和展示状态。 ### 2. 信源分级（Tier 1-4 + 黑名单） diff --git a/scripts/dr.py b/scripts/dr.py index b68d19d..67f22ac 100644 --- a/scripts/dr.py +++ b/scripts/dr.py @@ -26,7 +26,7 @@ from scripts.lib.model_config import ( parse_model_overrides, resolve_model_profile, ) -from scripts.runtime.assembly import build_chapter_briefs, run_chapter_assembly_workers +from scripts.runtime.assembly import build_chapter_briefs, build_compressed_findings, run_chapter_assembly_workers from scripts.runtime.orchestrator import create_phase2_task_cards, write_placeholder_packets from scripts.runtime.methods import ResearchMethodRegistry from scripts.runtime.phase1 import create_project, render_framework, write_material_brief @@ -258,11 +258,14 @@ def cmd_research(args: argparse.Namespace) -> int: packet_count = len(list((project_root / "phase2" / "packets").glob("*.json"))) brief_count = 0 chapter_count = 0 + compressed_count = 0 source_count = None if args.build_briefs or args.assemble_chapters: source_count = rebuild_sources_from_packets(project_root) briefs = build_chapter_briefs(project_root) brief_count = len(briefs) + compressed_findings = build_compressed_findings(project_root) + compressed_count = len(compressed_findings) if args.assemble_chapters: from scripts.lib.zenmux_client import ZenMuxClient, load_secrets @@ -273,7 +276,7 @@ def cmd_research(args: argparse.Namespace) -> int: chapter_count = run_chapter_assembly_workers( project_root=project_root, - briefs=briefs, + briefs=compressed_findings, runtime=runtime, client_factory=chapter_client_factory, workers=args.workers, @@ -288,6 +291,7 @@ def cmd_research(args: argparse.Namespace) -> int: print(f"Sources rebuilt: {source_count}") if args.build_briefs or args.assemble_chapters: print(f"Chapter briefs: {brief_count}") + print(f"Compressed findings: {compressed_count}") if args.assemble_chapters: print(f"Chapter drafts: {chapter_count}") if args.dry_run: @@ -298,7 +302,7 @@ def cmd_research(args: argparse.Namespace) -> int: print("Wrote: phase2/task_cards.json and validated phase2/packets/*.json") print("Next: rerun with --build-briefs to aggregate packets into chapter briefs.") elif args.build_briefs: - print("Wrote: phase2/chapter_briefs/*.json") + print("Wrote: phase2/chapter_briefs/*.json and phase2/compressed_findings/*.json") print("Next: rerun with --assemble-chapters to write Chinese chapter drafts.") else: print("Wrote: phase2/task_cards.json and phase2/packets/*.json") @@ -373,6 +377,8 @@ def cmd_status(args: argparse.Namespace) -> int: drafts = sorted((project_root / "phase2" / "drafts").glob("ch*.md")) evidence = sorted((project_root / "phase2" / "evidence").glob("ch*-evidence.md")) task_cards = project_root / "phase2" / "task_cards.json" + research_brief = project_root / "phase1" / "research_brief.json" + compressed_findings = sorted((project_root / "phase2" / "compressed_findings").glob("ch*.json")) packet_counts = packet_state_counts(project_root) sources = project_root / "phase2" / "sources.jsonl" final_en = project_root / "phase4" / "final_en.md" @@ -399,6 +405,7 @@ def cmd_status(args: argparse.Namespace) -> int: print() print("Artifacts:") print(f" framework: {file_state(project_root / 'phase1' / 'framework.md')}") + print(f" research_brief.json: {file_state(research_brief)}") print(f" task_cards.json: {file_state(task_cards)}") print( " packets: " @@ -410,6 +417,7 @@ def cmd_status(args: argparse.Namespace) -> int: f"total={packet_counts['total']}" ) print(f" drafts: {len(drafts)}") + print(f" compressed findings: {len(compressed_findings)}") print(f" evidence files: {len(evidence)}") print(f" sources: {source_count}") print(f" final_en.md: {file_state(final_en)} ({en_words:,} words)") diff --git a/scripts/runtime/assembly.py b/scripts/runtime/assembly.py index da77078..4bf9e40 100644 --- a/scripts/runtime/assembly.py +++ b/scripts/runtime/assembly.py @@ -41,8 +41,41 @@ def validate_chapter_brief(brief: dict) -> None: raise ValueError("chapter brief requires counter_evidence") +def validate_compressed_finding(finding: dict) -> None: + required = { + "chapter_id", + "chapter_title", + "packet_ids", + "chapter_thesis", + "key_findings", + "evidence_landings", + "counter_evidence", + "source_ids", + "open_questions", + "writing_plan", + } + missing = sorted(required - set(finding)) + if missing: + raise ValueError(f"compressed finding missing fields: {missing}") + if not finding["chapter_id"]: + raise ValueError("chapter_id required") + if not finding["packet_ids"]: + raise ValueError("compressed finding requires packet_ids") + if not finding["chapter_thesis"]: + raise ValueError("compressed finding requires chapter_thesis") + if not finding["key_findings"]: + raise ValueError("compressed finding requires key_findings") + if not finding["evidence_landings"]: + raise ValueError("compressed finding requires evidence_landings") + if not finding["counter_evidence"]: + raise ValueError("compressed finding requires counter_evidence") + + def validate_chapter_markdown_citations(markdown: str, brief: dict) -> None: - validate_chapter_brief(brief) + if "key_findings" in brief: + validate_compressed_finding(brief) + else: + validate_chapter_brief(brief) cited = set(re.findall(r"\[(src_[A-Za-z0-9_-]+)\]", markdown)) allowed = set(brief.get("source_ids") or []) unknown = sorted(cited - allowed) @@ -103,9 +136,79 @@ def build_chapter_briefs(project_root: Path) -> list[dict]: return briefs +def _source_ids_from_item(item: dict) -> list[str]: + if item.get("source_ids"): + return list(item.get("source_ids") or []) + if item.get("source_id"): + return [item["source_id"]] + return [] + + +def build_compressed_findings(project_root: Path) -> list[dict]: + """Compress packet-level evidence into chapter-level writing inputs. + + This is intentionally deterministic: it does not invent a better narrative, + but it forces a chapter-level evidence map before any model writes prose. + """ + brief_dir = project_root / "phase2" / "chapter_briefs" + if not brief_dir.exists() or not list(brief_dir.glob("ch*.json")): + briefs = build_chapter_briefs(project_root) + else: + briefs = [ + json.loads(path.read_text(encoding="utf-8")) + for path in sorted(brief_dir.glob("ch*.json")) + ] + out_dir = project_root / "phase2" / "compressed_findings" + out_dir.mkdir(parents=True, exist_ok=True) + findings: list[dict] = [] + for brief in briefs: + validate_chapter_brief(brief) + core_claims = brief.get("core_claims") or [] + evidence_items = brief.get("evidence_items") or [] + first_claim = core_claims[0] if core_claims else {} + chapter_thesis = first_claim.get("claim") or f"{brief['chapter_title']} 需要以证据为中心重写。" + finding = { + "chapter_id": brief["chapter_id"], + "chapter_title": brief["chapter_title"], + "packet_ids": brief["packet_ids"], + "chapter_thesis": chapter_thesis, + "key_findings": [ + { + "finding": claim.get("claim") or claim.get("summary") or str(claim), + "source_ids": _source_ids_from_item(claim), + "confidence": claim.get("confidence", "medium"), + } + for claim in core_claims + ], + "evidence_landings": [ + { + "evidence": item.get("summary") or item.get("finding") or item.get("quote") or str(item), + "source_ids": _source_ids_from_item(item), + "landing_hint": item.get("landing_hint", "用于支撑本章关键判断或整改动作。"), + } + for item in evidence_items + ], + "counter_evidence": brief["counter_evidence"], + "source_ids": brief["source_ids"], + "open_questions": brief["open_questions"], + "writing_plan": [ + "先写本章判断，不按 packet 顺序堆砌。", + "每个二级小节至少落下具体审计发现、法规要求、记录/参数或整改证据。", + "正文末尾必须保留“证据落点与待补证据”表。", + ], + } + validate_compressed_finding(finding) + (out_dir / f"{brief['chapter_id']}.json").write_text( + json.dumps(finding, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + findings.append(finding) + return findings + + def build_chapter_user_prompt(brief: dict) -> str: return ( - "请根据以下 chapter brief 写一章正式中文 Markdown 正文。\n" + "请根据以下 compressed finding / chapter brief 写一章正式中文 Markdown 正文。\n" "目标是形成一个完整章节，而不是 packet 摘要。避免碎片化，按金字塔结构组织：章首先给结论，再用证据支撑。\n" "要求：标题必须是观点型判断；每个数字和事实保留 [src_xxx]；纳入反方证据；不要出现调度元数据。\n" "禁止写空泛咨询腔。每个二级小节都必须至少落下 2 个具体审计发现、法规要求、SOP/记录/参数/现场观察或整改证据；不要只写原则。\n" @@ -142,7 +245,10 @@ class ChapterAssemblyWorker: ) def write_chapter(self, *, project_root: Path, brief: dict) -> Path: - validate_chapter_brief(brief) + if "key_findings" in brief: + validate_compressed_finding(brief) + else: + validate_chapter_brief(brief) markdown = self.client.chat_complete( model=self.role.model, system=self._system_prompt(), diff --git a/scripts/runtime/orchestrator.py b/scripts/runtime/orchestrator.py index 001b8d8..8e622c9 100644 --- a/scripts/runtime/orchestrator.py +++ b/scripts/runtime/orchestrator.py @@ -8,7 +8,7 @@ from pathlib import Path from scripts.runtime.artifacts import ensure_phase_dirs, load_manifest, write_manifest from scripts.runtime.methods import ResearchMethodRegistry -from scripts.runtime.tasks import generate_task_cards, write_task_cards +from scripts.runtime.tasks import generate_task_cards, generate_task_cards_from_research_brief, write_task_cards def utc_now_iso() -> str: @@ -28,7 +28,19 @@ def create_phase2_task_cards( if (project_root / "manifest.json").exists(): method_key = load_manifest(project_root).get("research_method") method = ResearchMethodRegistry().get(method_key) - cards = generate_task_cards(project_root.name, framework.read_text(encoding="utf-8"), axes=axes, method=method) + research_brief_path = project_root / "phase1" / "research_brief.json" + framework_text = framework.read_text(encoding="utf-8") + if research_brief_path.exists(): + research_brief = json.loads(research_brief_path.read_text(encoding="utf-8")) + cards = generate_task_cards_from_research_brief( + project_root.name, + framework_text, + research_brief, + axes=axes, + method=method, + ) + else: + cards = generate_task_cards(project_root.name, framework_text, axes=axes, method=method) if not dry_run: ensure_phase_dirs(project_root) write_task_cards(project_root / "phase2" / "task_cards.json", cards) diff --git a/scripts/runtime/phase1.py b/scripts/runtime/phase1.py index 8e5ccfb..56c9fed 100644 --- a/scripts/runtime/phase1.py +++ b/scripts/runtime/phase1.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +import json import re from datetime import datetime, timezone from pathlib import Path @@ -11,6 +12,7 @@ from typing import Any from scripts.runtime.artifacts import PROJECTS_DIR, ensure_phase_dirs, load_manifest, write_manifest from scripts.runtime.materials import ingest_input_materials, render_material_inventory from scripts.runtime.methods import ResearchMethod, ResearchMethodRegistry +from scripts.runtime.tasks import AXIS_ROUTES def utc_now_iso() -> str: @@ -187,6 +189,132 @@ def write_material_brief( return out +def _axis_prompt_brief(axis: str, method: ResearchMethod) -> str: + prompts = { + "input_material_findings": "从用户材料中提取现场事实、审计发现、复盘记录和内部答复，并标注原始材料位置。", + "nmpa_fda_ema_ich_who_baseline": "把 NMPA、FDA、EMA、ICH、WHO、药典或 Annex 1 等要求转化为可核验的法规基线。", + "quality_system_gap": "把现场发现映射到质量体系流程缺口，覆盖偏差、变更、CAPA、文件、培训和数据完整性。", + "manufacturing_process_risk": "围绕生产工艺、设施、公用系统、CPP/CQA、验证和无菌保障识别系统性风险。", + "operations_management_gap": "诊断运营管理、跨部门协同、会议机制、指标体系和交付节奏的结构性问题。", + "team_capability": "识别人员能力、岗位职责、质量文化和管理梯队方面的缺口与建设路径。", + "capa_roadmap": "把差距转化为短中长期 CAPA 组合，要求绑定 owner、期限、优先级、关闭证据和复核机制。", + "verification_evidence": "定义整改完成后可被审计接受的验证证据，包括记录、报告、趋势和管理评审输入。", + "counter": "主动寻找反方证据、限制条件和可能降低严重度或改变优先级的解释，避免单向论证。", + } + return prompts.get(axis, f"按照 `{method.key}` 方法，对 {axis} 轴进行证据收集、证伪和结构化归纳。") + + +def _material_paths(manifest: dict[str, Any]) -> list[dict[str, str]]: + materials: list[dict[str, str]] = [] + for item in manifest.get("material_inventory") or []: + rel = item.get("ocr_extracted_to") or item.get("extracted_to") or item.get("copied_to") + if rel: + materials.append({"path": rel, "role": "input_material"}) + return materials + + +def build_research_brief_payload( + project_root: Path, + manifest: dict[str, Any], + method: ResearchMethod, +) -> dict[str, Any]: + """Create the file-backed Phase 1 research brief used by task-card generation.""" + axes = list(method.task_axes) + return { + "version": "0.21-alpha", + "topic": manifest.get("topic", project_root.name), + "research_method": method.key, + "method_name": method.name, + "work_language": "zh", + "tone": "事实型、整改导向、面向管理层和质量/生产负责人；避免空泛咨询腔。", + "central_question": f"如何基于已提供材料和权威法规/最佳实践，系统诊断“{manifest.get('topic', project_root.name)}”并形成可执行整改路线图？", + "success_criteria": [ + "每个核心判断都能回到用户材料、权威法规、最佳实践或反方证据。", + "短中长期整改建议必须绑定优先级、责任、关闭证据和复核机制。", + "章节写作必须先收束主线，再使用 evidence packet；不得按 packet 机械拼贴。", + ], + "phase2_inputs": { + "material_brief_path": "phase1/material_brief.md", + "framework_path": "phase1/framework.md", + "research_brief_path": "phase1/research_brief.json", + }, + "materials": _material_paths(manifest), + "task_planning": { + "chapter_source": "phase1/framework.md", + "axes": axes, + "required_skills": [ + "deep-research", + "search-gateway", + "search-strategy", + "source-quality", + "evidence-table", + "citation-manager", + ], + "search_routes_by_axis": {axis: AXIS_ROUTES.get(axis, ["general"]) for axis in axes}, + "axis_prompt_briefs": {axis: _axis_prompt_brief(axis, method) for axis in axes}, + "stop_conditions": [ + "每张任务卡至少形成 3 条可追溯 evidence_items，且不得编造 candidate_sources 以外来源。", + "关键 claim 不足 2 个独立 Tier 1-2 信源时，必须写入 open_questions 和证据缺口。", + "必须包含 counter_evidence；找不到反方证据时记录检索路径和限制。", + ], + "fragmentation_guard": "并发 worker 只生产 evidence packet；章节主线由 compressed_findings 收束，禁止直接把 packet 堆成正文。", + }, + "clarification_notes": { + "requires_user_review": True, + "questions_source": "phase1/material_brief.md", + "decision_items": [ + "确认报告用途、受众和脱敏边界。", + "确认研究方法是否适配当前场景；MECE 只是可选方法之一。", + "确认任务切分和检索策略是否足以让低成本模型独立执行。", + ], + }, + } + + +def write_research_brief( + project_root: Path, + manifest: dict[str, Any] | None = None, + method: ResearchMethod | None = None, +) -> tuple[Path, Path]: + manifest = manifest or load_manifest(project_root) + method = method or ResearchMethodRegistry().get(manifest.get("research_method")) + payload = build_research_brief_payload(project_root, manifest, method) + json_path = project_root / "phase1" / "research_brief.json" + md_path = project_root / "phase1" / "research_brief.md" + json_path.parent.mkdir(parents=True, exist_ok=True) + json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + lines = [ + f"# Phase 1 Research Brief：{payload['topic']}", + "", + f"- research_method: {payload['research_method']}", + f"- work_language: {payload['work_language']}", + f"- tone: {payload['tone']}", + "", + "## 中心问题", + "", + payload["central_question"], + "", + "## 成功标准", + "", + ] + lines.extend(f"- {item}" for item in payload["success_criteria"]) + lines.extend(["", "## 任务切分原则", ""]) + planning = payload["task_planning"] + lines.append(planning["fragmentation_guard"]) + lines.append("") + for axis in planning["axes"]: + routes = "、".join(planning["search_routes_by_axis"].get(axis, [])) + prompt = planning["axis_prompt_briefs"].get(axis, "") + lines.append(f"- `{axis}`：{prompt} 检索路径：{routes}") + lines.extend(["", "## 必读 Skills", ""]) + lines.extend(f"- {name}" for name in planning["required_skills"]) + lines.extend(["", "## 停止条件", ""]) + lines.extend(f"- {item}" for item in planning["stop_conditions"]) + lines.append("") + md_path.write_text("\n".join(lines), encoding="utf-8") + return md_path, json_path + + CHAPTER_TEMPLATES: dict[str, list[str]] = { "mckinsey_market": [ "核心结论先行界定市场机会与约束", @@ -329,10 +457,14 @@ def render_framework(project_root: Path, *, method_key: str | None = None, chapt out = project_root / "phase1" / "framework.md" out.parent.mkdir(parents=True, exist_ok=True) out.write_text("\n".join(lines), encoding="utf-8") + research_brief_md, research_brief_json = write_research_brief(project_root, manifest, method) manifest["phase1"] = { "status": "completed", "approved": False, "framework_path": "phase1/framework.md", + "research_brief_path": str(research_brief_md.relative_to(project_root)), + "research_brief_json_path": str(research_brief_json.relative_to(project_root)), + "requires_user_interview": True, "research_method": method.key, "updated_at": utc_now_iso(), } diff --git a/scripts/runtime/tasks.py b/scripts/runtime/tasks.py index 3746417..20524aa 100644 --- a/scripts/runtime/tasks.py +++ b/scripts/runtime/tasks.py @@ -63,6 +63,14 @@ class TaskCard: preferred_model_role: str = "dr_analyst" status: str = "pending" dependencies: list[str] = field(default_factory=list) + research_goal: str = "" + research_method: str = "" + prompt_brief: str = "" + required_skills: list[str] = field(default_factory=list) + allowed_materials: list[str] = field(default_factory=list) + expected_evidence: dict[str, Any] = field(default_factory=dict) + stop_conditions: list[str] = field(default_factory=list) + model_hint: str = "" def to_dict(self) -> dict[str, Any]: return asdict(self) @@ -105,6 +113,66 @@ def _questions_for_axis(chapter: Chapter, axis: str) -> list[str]: ] +def _default_required_skills(axis: str) -> list[str]: + skills = ["search-gateway", "search-strategy", "source-quality", "evidence-table"] + if axis == "counter": + skills.append("mckinsey-method") + return skills + + +def _default_expected_evidence(axis: str) -> dict[str, Any]: + return { + "min_tier_1_2_sources": 2, + "must_include_counter_evidence": True, + "must_include_source_metadata": True, + "preferred_evidence_types": [ + "regulatory_or_best_practice_requirement", + "site_or_material_finding", + "quantitative_fact_or_record", + "implementation_or_verification_evidence", + ], + "axis": axis, + } + + +def _default_stop_conditions() -> list[str]: + return [ + "已形成至少 3 条可追溯 evidence_items，且每条关键 claim 有 source_id。", + "已主动记录 counter_evidence 或明确说明未找到反方证据的检索路径。", + "candidate_sources 不足以支撑结论时停止写作，并把缺口写入 open_questions。", + ] + + +def _task_card_for_chapter_axis( + *, + chapter: Chapter, + axis: str, + routes: list[str], + method_key: str, + required_skills: list[str] | None = None, + allowed_materials: list[str] | None = None, + prompt_brief: str | None = None, + stop_conditions: list[str] | None = None, +) -> TaskCard: + return TaskCard( + task_id=f"{chapter.chapter_id}-{axis}", + chapter_ids=[chapter.chapter_id], + topic_axis=axis, + questions=_questions_for_axis(chapter, axis), + search_routes=routes, + output_packet=f"phase2/packets/{chapter.chapter_id}-{axis}.json", + preferred_model_role="dr_verifier" if axis == "counter" else "dr_analyst", + research_goal=f"为《{chapter.title}》收集并验证 {axis} 轴证据，形成可写入章节的具体判断与证据落点。", + research_method=method_key, + prompt_brief=prompt_brief or f"围绕《{chapter.title}》的 {axis} 轴，优先形成可证伪、可引用、可落地的证据包。", + required_skills=required_skills or _default_required_skills(axis), + allowed_materials=allowed_materials or [], + expected_evidence=_default_expected_evidence(axis), + stop_conditions=stop_conditions or _default_stop_conditions(), + model_hint="use_cross_model_verifier" if axis == "counter" else "use_cost_effective_research_worker", + ) + + def generate_task_cards( slug: str, framework_text: str, @@ -120,14 +188,56 @@ def generate_task_cards( for axis in selected_axes: routes = AXIS_ROUTES.get(axis, ["general"]) cards.append( - TaskCard( - task_id=f"{chapter.chapter_id}-{axis}", - chapter_ids=[chapter.chapter_id], - topic_axis=axis, - questions=_questions_for_axis(chapter, axis), - search_routes=routes, - output_packet=f"phase2/packets/{chapter.chapter_id}-{axis}.json", - preferred_model_role="dr_verifier" if axis == "counter" else "dr_analyst", + _task_card_for_chapter_axis( + chapter=chapter, + axis=axis, + routes=routes, + method_key=method.key if method else "", + ) + ) + validate_task_cards(cards) + return cards + + +def generate_task_cards_from_research_brief( + slug: str, + framework_text: str, + research_brief: dict[str, Any], + *, + axes: list[str] | None = None, + method: ResearchMethod | None = None, +) -> list[TaskCard]: + del slug + chapters = parse_framework_chapters(framework_text) + planning = research_brief.get("task_planning") or {} + method_key = research_brief.get("research_method") or (method.key if method else "") + selected_axes = axes or (method.task_axes if method else None) or list(planning.get("search_routes_by_axis") or []) or DEFAULT_AXES + routes_by_axis = planning.get("search_routes_by_axis") or {} + prompt_by_axis = planning.get("axis_prompt_briefs") or {} + base_skills = list(planning.get("required_skills") or []) + stop_conditions = list(planning.get("stop_conditions") or []) + allowed_materials = [ + str(item.get("path")) + for item in research_brief.get("materials", []) + if item.get("path") + ] + cards: list[TaskCard] = [] + for chapter in chapters: + for axis in selected_axes: + routes = list(routes_by_axis.get(axis) or AXIS_ROUTES.get(axis, ["general"])) + skills = base_skills or _default_required_skills(axis) + if "search-gateway" not in skills: + skills = ["search-gateway", *skills] + cards.append( + _task_card_for_chapter_axis( + chapter=chapter, + axis=axis, + routes=routes, + method_key=method_key, + required_skills=skills, + allowed_materials=allowed_materials, + prompt_brief=prompt_by_axis.get(axis), + stop_conditions=stop_conditions or None, ) ) validate_task_cards(cards) @@ -157,6 +267,18 @@ def detect_dependency_cycles(cards: list[TaskCard]) -> None: def validate_task_cards(cards: list[TaskCard]) -> None: seen: set[str] = set() for card in cards: + if not card.research_goal: + card.research_goal = f"围绕 {card.topic_axis} 轴收集并验证结构化证据。" + if not card.prompt_brief: + card.prompt_brief = f"按 {card.topic_axis} 轴形成证据包，避免泛泛结论。" + if not card.required_skills: + card.required_skills = _default_required_skills(card.topic_axis) + if "search-gateway" not in card.required_skills: + card.required_skills = ["search-gateway", *card.required_skills] + if not card.expected_evidence: + card.expected_evidence = _default_expected_evidence(card.topic_axis) + if not card.stop_conditions: + card.stop_conditions = _default_stop_conditions() if card.task_id in seen: raise ValueError(f"duplicate task_id: {card.task_id}") seen.add(card.task_id) diff --git a/scripts/v020_regression.py b/scripts/v020_regression.py index b7226ca..c81c817 100644 --- a/scripts/v020_regression.py +++ b/scripts/v020_regression.py @@ -34,6 +34,29 @@ def make_fixture(root: Path) -> Path: return project +def write_fixture_packet(project: Path) -> None: + cards = json.loads((project / "phase2" / "task_cards.json").read_text(encoding="utf-8")) + for index, card in enumerate(cards, start=1): + source_a = f"src_{index:03d}_a" + source_b = f"src_{index:03d}_b" + packet = { + "task_id": card["task_id"], + "claims": [{"claim": f"{card['task_id']} 回归证据支持章节主线", "source_ids": [source_a]}], + "evidence_items": [{"source_id": source_a, "summary": "权威来源支持该判断"}], + "counter_evidence": [{"claim": "证据仍需更多来源交叉验证", "source_ids": [source_b]}], + "source_ids": [source_a, source_b], + "source_quality_notes": [f"{source_a} Tier 1", f"{source_b} Tier 2"], + "open_questions": ["需要在真实项目中补充更多来源。"], + "raw_quotes_or_notes": ["English raw note can remain here."], + "sources": [ + {"id": source_a, "title": f"Regression Source {index}A", "url": f"https://example.com/source-{index}-a", "tier": 1}, + {"id": source_b, "title": f"Regression Source {index}B", "url": f"https://example.com/source-{index}-b", "tier": 2}, + ], + } + packet_path = project / card["output_packet"] + packet_path.write_text(json.dumps(packet, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + def main() -> int: python = sys.executable run([python, "scripts/dr.py", "skills", "validate"]) @@ -56,9 +79,13 @@ def main() -> int: ) project = make_fixture(tmp_root) run([python, "scripts/dr.py", "frame", str(project)]) + assert (project / "phase1" / "research_brief.json").exists() run([python, "scripts/dr.py", "approve", str(project)]) - run([python, "scripts/dr.py", "research", str(project), "--workers", "2", "--dry-run"]) - run([python, "scripts/dr.py", "research", str(project), "--workers", "2"]) + run([python, "scripts/dr.py", "research", str(project), "--workers", "2", "--axis", "literature", "--dry-run"]) + run([python, "scripts/dr.py", "research", str(project), "--workers", "2", "--axis", "literature"]) + write_fixture_packet(project) + run([python, "scripts/dr.py", "research", str(project), "--workers", "2", "--axis", "literature", "--build-briefs"]) + assert (project / "phase2" / "compressed_findings" / "ch01.json").exists() run([python, "scripts/dr.py", "review", str(project)]) run([python, "scripts/dr.py", "finalize", str(project), "--dry-run"]) print("v0.20 regression PASS") diff --git a/tests/test_chapter_assembly.py b/tests/test_chapter_assembly.py index 60b18f8..a8e18f3 100644 --- a/tests/test_chapter_assembly.py +++ b/tests/test_chapter_assembly.py @@ -11,10 +11,12 @@ if str(REPO_ROOT) not in sys.path: from scripts.runtime.assembly import ( ChapterAssemblyWorker, build_chapter_briefs, + build_compressed_findings, build_chapter_user_prompt, run_chapter_assembly_workers, validate_chapter_markdown_citations, validate_chapter_brief, + validate_compressed_finding, ) from scripts.runtime.roles import resolve_runtime_profile @@ -104,6 +106,17 @@ def test_build_chapter_briefs_aggregates_packets_by_chapter(tmp_path: Path) -> N assert "src_002" in brief["source_ids"] assert (project / "phase2/chapter_briefs/ch01.json").exists() + compressed = build_compressed_findings(project) + + assert len(compressed) == 1 + finding = compressed[0] + validate_compressed_finding(finding) + assert finding["chapter_id"] == "ch01" + assert "chapter_thesis" in finding + assert "evidence_landings" in finding + assert "src_001" in finding["source_ids"] + assert (project / "phase2/compressed_findings/ch01.json").exists() + def test_chapter_prompt_contains_brief_and_fragmentation_guard() -> None: brief = { @@ -122,6 +135,7 @@ def test_chapter_prompt_contains_brief_and_fragmentation_guard() -> None: assert "临床证据正在重塑需求判断" in prompt assert "避免碎片化" in prompt + assert "compressed finding" in prompt assert "只输出 Markdown" in prompt diff --git a/tests/test_phase0_materials.py b/tests/test_phase0_materials.py index c8cab30..77626a8 100644 --- a/tests/test_phase0_materials.py +++ b/tests/test_phase0_materials.py @@ -62,9 +62,20 @@ def test_framework_mentions_ingested_materials(tmp_path: Path) -> None: render_framework(project, method_key="gmp_quality_operations_diagnosis") framework = (project / "phase1/framework.md").read_text(encoding="utf-8") + research_brief_md = project / "phase1" / "research_brief.md" + research_brief_json = project / "phase1" / "research_brief.json" + brief = json.loads(research_brief_json.read_text(encoding="utf-8")) + assert "phase0/extracted/audit.md" in framework assert "NMPA、FDA、EMA、ICH、WHO" in framework assert "请先确认 `phase1/material_brief.md`" in framework + assert research_brief_md.exists() + assert "任务切分原则" in research_brief_md.read_text(encoding="utf-8") + assert brief["research_method"] == "gmp_quality_operations_diagnosis" + assert brief["work_language"] == "zh" + assert brief["task_planning"]["required_skills"] + assert brief["task_planning"]["search_routes_by_axis"]["counter"] == ["scholar", "general"] + assert brief["phase2_inputs"]["framework_path"] == "phase1/framework.md" def test_pdf_requiring_ocr_uses_firered_and_records_result(tmp_path: Path, monkeypatch) -> None: diff --git a/tests/test_research_methods.py b/tests/test_research_methods.py index c502dbc..3549c63 100644 --- a/tests/test_research_methods.py +++ b/tests/test_research_methods.py @@ -45,6 +45,23 @@ def test_orchestrator_reads_research_method_from_manifest(tmp_path: Path) -> Non "## 第1章 GMP 体系差距需要按法规和风险双轴定位\n\n研究思路。", encoding="utf-8", ) + (project / "phase1" / "research_brief.json").write_text( + json.dumps( + { + "research_method": "gmp_gap_assessment", + "work_language": "zh", + "task_planning": { + "required_skills": ["search-gateway", "evidence-table"], + "search_routes_by_axis": {"regulatory_gap": ["general", "news"]}, + "axis_prompt_briefs": {"regulatory_gap": "逐条映射法规基线与审计差距。"}, + "stop_conditions": ["证据不足则回炉补证。"], + }, + "materials": [{"path": "phase0/extracted/audit.md"}], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) cards = create_phase2_task_cards(project, dry_run=True) @@ -53,3 +70,5 @@ def test_orchestrator_reads_research_method_from_manifest(tmp_path: Path) -> Non "risk_classification", "capa_design", ] + assert cards[0]["prompt_brief"] == "逐条映射法规基线与审计差距。" + assert cards[0]["allowed_materials"] == ["phase0/extracted/audit.md"] diff --git a/tests/test_v020_runtime.py b/tests/test_v020_runtime.py index 4b69986..2585636 100644 --- a/tests/test_v020_runtime.py +++ b/tests/test_v020_runtime.py @@ -17,6 +17,7 @@ from scripts.runtime.tasks import ( TaskCard, detect_dependency_cycles, generate_task_cards, + generate_task_cards_from_research_brief, validate_packet, validate_task_cards, ) @@ -77,6 +78,47 @@ def test_generate_task_cards_from_chinese_framework() -> None: "ch02-regulatory", ] assert cards[0].output_packet == "phase2/packets/ch01-clinical.json" + assert cards[0].research_goal + assert "search-gateway" in cards[0].required_skills + assert cards[0].expected_evidence["min_tier_1_2_sources"] == 2 + assert cards[0].stop_conditions + + +def test_generate_task_cards_from_research_brief_carries_prompt_and_skills() -> None: + brief = { + "research_method": "gmp_quality_operations_diagnosis", + "work_language": "zh", + "tone": "面向管理层的事实型整改诊断", + "task_planning": { + "required_skills": ["search-gateway", "evidence-table", "source-quality"], + "search_routes_by_axis": { + "quality_system_gap": ["general", "news"], + "counter": ["scholar", "general"], + }, + "axis_prompt_briefs": { + "quality_system_gap": "把现场发现映射到质量体系流程缺口和法规要求。", + "counter": "主动寻找能削弱或限定结论的反方证据。", + }, + "stop_conditions": ["每张卡至少形成 3 条可追溯证据。"], + }, + "materials": [{"path": "phase0/extracted/audit.md", "role": "site_evidence"}], + } + framework = "## 第1章 质量体系闭环能力决定整改可信度\n\n研究思路。" + + cards = generate_task_cards_from_research_brief( + "baifan-test", + framework, + brief, + axes=["quality_system_gap", "counter"], + ) + + assert [card.task_id for card in cards] == ["ch01-quality_system_gap", "ch01-counter"] + assert cards[0].prompt_brief == "把现场发现映射到质量体系流程缺口和法规要求。" + assert cards[0].research_method == "gmp_quality_operations_diagnosis" + assert cards[0].allowed_materials == ["phase0/extracted/audit.md"] + assert cards[0].preferred_model_role == "dr_analyst" + assert cards[1].preferred_model_role == "dr_verifier" + assert "search-gateway" in cards[0].required_skills def test_task_card_validation_rejects_duplicates_and_cycles() -> None: