"""Phase 1 project initialization and framework generation.""" from __future__ import annotations import hashlib import json import re from datetime import datetime, timezone from pathlib import Path 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: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def slugify_topic(topic: str) -> str: slug = re.sub(r"[^a-zA-Z0-9]+", "-", topic.lower()).strip("-") if slug: return slug[:80] digest = hashlib.sha1(topic.encode("utf-8")).hexdigest()[:8] return f"research-{digest}" def create_project( *, topic: str, slug: str | None = None, projects_dir: Path = PROJECTS_DIR, method_key: str | None = None, report_type: str = "research", model_profile: str = "medium", target_words: int = 30000, input_materials: list[str] | None = None, ) -> Path: method = ResearchMethodRegistry().get(method_key) project_slug = slug or slugify_topic(topic) project_root = projects_dir / project_slug if project_root.exists(): raise FileExistsError(f"project already exists: {project_root}") ensure_phase_dirs(project_root) (project_root / "phase0" / "inputs").mkdir(parents=True, exist_ok=True) (project_root / "phase0" / "extracted").mkdir(parents=True, exist_ok=True) material_inventory = ingest_input_materials(project_root, input_materials) now = utc_now_iso() manifest: dict[str, Any] = { "version": "0.20.0", "runtime": "python-core-v0.20", "topic": topic, "slug": project_slug, "report_title": topic, "type": report_type, "work_language": "zh", "model_profile": model_profile, "research_method": method.key, "target_words": target_words, "input_materials": input_materials or [], "material_inventory": material_inventory, "created_at": now, "updated_at": now, "phase1": { "status": "initialized", "approved": False, "requires_user_interview": True, "material_brief_path": "phase1/material_brief.md", }, "phase2": {"status": "pending"}, "phase3": {"status": "pending"}, "phase4": {"status": "pending"}, } write_manifest(project_root, manifest) _write_interview_seed(project_root, manifest, method) write_material_brief(project_root, manifest, method) return project_root def _write_interview_seed(project_root: Path, manifest: dict[str, Any], method: ResearchMethod) -> None: material_text = render_material_inventory(manifest.get("material_inventory") or []) text = ( f"# Phase 1 访谈记录\n\n" f"- 主题：{manifest['topic']}\n" f"- 研究方法：{method.key} - {method.name}\n" f"- 报告类型：{manifest['type']}\n" f"- 目标字数：{manifest['target_words']}\n" f"- 工作语言：中文主写作；检索关键词、证据摘录和来源笔记可保留英文。\n\n" f"## 已提供材料\n\n{material_text}\n\n" "## 后续访谈问题\n\n" "1. 本报告最重要的决策用途是什么？\n" "2. 是否有必须覆盖或必须排除的公司、产品、工艺、市场或法规范围？\n" "3. 结论偏好是战略建议、风险清单、投资判断，还是执行路线图？\n" ) (project_root / "phase1" / "interview.md").write_text(text, encoding="utf-8") def _material_excerpt(project_root: Path, rel_path: str, *, max_chars: int = 1200) -> str: path = project_root / rel_path if not path.exists(): return "（未找到抽取文本）" text = path.read_text(encoding="utf-8") compact = "\n".join(line.rstrip() for line in text.splitlines() if line.strip()) return compact[:max_chars] + ("..." if len(compact) > max_chars else "") def _derive_material_observations(project_root: Path, inventory: list[dict[str, Any]]) -> list[str]: combined_parts: list[str] = [] for item in inventory: rel = item.get("ocr_extracted_to") or item.get("extracted_to") if rel and (project_root / rel).exists(): combined_parts.append((project_root / rel).read_text(encoding="utf-8")) text = "\n".join(combined_parts) checks = [ ("审计范围覆盖生产管理、原液、制剂和无菌相关模块，报告需要同时处理 GMP 合规、工艺转移和运营协同，而不是只写质量体系。", ["生产管理", "原液", "制剂", "无菌"]), ("材料显示高风险项为 0、中风险项为 1，适合采用“商业化 readiness 与系统成熟度差距”而非“体系失控”作为初始假设。", ["高风险 0", "中风险1", "低风险7"]), ("商业化经验、无菌保障细节、文件要求与执行一致性是需要访谈确认的主线风险。", ["商业化经验不足", "无菌保障", "文件要求与执行一致性"]), ("工艺规程、批记录、CPP/CQA、VMPR/VMP、验证主计划等内容反复出现，说明工艺验证和商业化文件体系可能是 Phase 2 的重点证据轴。", ["CPP", "CQA", "VMPR", "VMP"]), ("温度、压差、WFI、冷却段微生物、RABS/ORABS、first air、APS 等无菌和设施细节需要映射到 EU Annex 1、NMPA GMP 和企业 SOP。", ["温度", "压差", "WFI", "APS"]), ("复盘材料包含责任人和局部答复，后续整改路线图应尽量回填 owner、期限、关闭证据和复核机制。", ["填写人", "是否已经回答完整", "整改"]), ] observations = [message for message, needles in checks if any(needle in text for needle in needles)] return observations or ["材料已导入但尚未形成足够结构化判断；需要先访谈确认研究用途、范围和优先级。"] def write_material_brief( project_root: Path, manifest: dict[str, Any] | None = None, method: ResearchMethod | None = None, ) -> Path: """Write a Phase 0/1 material brief that must be reviewed before Phase 2.""" manifest = manifest or load_manifest(project_root) method = method or ResearchMethodRegistry().get(manifest.get("research_method")) inventory = manifest.get("material_inventory") or [] lines = [ f"# Phase 0 材料简报：{manifest.get('topic', project_root.name)}", "", "status: 待用户确认", f"research_method: {method.key}", "", "## 已导入材料", "", render_material_inventory(inventory), "", "## 材料初步解读", "", "以下内容由 Python core 从已落盘材料抽样生成，只作为访谈起点；不得直接视为最终结论。", "", ] lines.extend(["## 初步问题聚类（待访谈确认）", ""]) for observation in _derive_material_observations(project_root, inventory): lines.append(f"- {observation}") lines.append("") lines.append("## 材料摘录") lines.append("") for item in inventory: rel = item.get("ocr_extracted_to") or item.get("extracted_to") if not rel: continue lines.extend( [ f"### {Path(rel).name}", "", _material_excerpt(project_root, rel), "", ] ) lines.extend( [ "## 建议访谈确认点", "", "1. 本报告的最重要用途是什么：内部整改、客户沟通、董事会决策，还是外部审计准备？", "2. 哪些审计发现最需要优先展开：无菌保障、工艺验证、数据完整性、质量体系闭环，还是运营协同？", "3. 是否存在必须排除或脱敏的项目、人员、客户、产品或工艺信息？", "4. 短中长期整改的时间边界如何定义，例如 30/90/180 天，还是按临床/商业化里程碑划分？", "5. 是否需要把 NMPA、FDA、EMA、ICH、WHO 的法规基线分别映射到整改责任人和证据包？", "", "## Gate", "", "请用户确认本材料简报与访谈问题后，再生成或批准 `phase1/framework.md` 并进入 Phase 2。", "", ] ) out = project_root / "phase1" / "material_brief.md" out.parent.mkdir(parents=True, exist_ok=True) out.write_text("\n".join(lines), encoding="utf-8") 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": [ "核心结论先行界定市场机会与约束", "临床与真实世界证据决定需求天花板", "监管路径和支付环境重塑商业化节奏", "竞争格局正在从单点产品转向组合能力", "专利与技术壁垒决定长期利润池", "中国市场的准入和供给能力形成独立变量", "资本市场预期与基本面之间存在可验证偏差", "反方证据限定结论边界并提示回撤风险", "战略选择应围绕资源约束排序", "执行路线图需要把证据缺口转化为行动清单", ], "gmp_gap_assessment": [ "监管基线决定整改范围而非企业主观偏好", "现状差距需要按法规条款和业务流程双重定位", "质量风险分级决定 CAPA 优先级", "根因分析质量决定整改能否闭环", "CAPA 设计必须绑定责任人、证据和期限", "验证计划决定整改是否可被审计接受", "供应商和外包管理常是系统性缺口放大器", "数据完整性风险需要独立成章处理", "实施路线图需要平衡停线风险与合规风险", "管理层治理机制决定整改能否持续", ], "cmc_process_risk": [ "工艺流程图是识别放大风险的起点", "CQA 与 CPP 的映射决定控制策略质量", "放大过程的失效模式集中在传质、混合和稳定性", "分析方法和放行标准决定证据可信度", "技术转移风险来自知识隐性化和现场差异", "供应链约束会改变工艺控制边界", "偏差和变更管理决定商业化后的韧性", "监管沟通策略需要提前固化关键假设", "反方证据限定平台工艺可复制性", "CMC 路线图需要把风险转化为验证实验", ], "rd_go_no_go": [ "科学假设强度决定项目是否值得进入下一阶段", "POC 证据需要同时证明有效性和可转化性", "安全性窗口决定适应症与人群选择", "IP 与 FTO 风险决定商业化自由度", "开发路径需要把关键不确定性前置验证", "竞争窗口决定速度是否仍有战略价值", "CMC 与临床运营能力影响真实可行性", "反方证据决定 go/no-go 阈值", "投资强度应与证据成熟度匹配", "决策门槛需要形成可执行检查表", ], "management_consulting": [ "现状诊断需要区分症状、根因和约束条件", "能力差距决定组织改进优先级", "流程断点揭示跨部门协作成本", "治理结构决定决策速度和责任清晰度", "运营模型需要匹配战略目标而非照搬标杆", "数字化工具只有嵌入流程才产生价值", "绩效指标需要避免局部最优", "变革阻力本身是方案设计输入", "路线图需要把 quick wins 与系统建设分层", "落地机制决定咨询建议能否转化为成果", ], "gmp_quality_operations_diagnosis": [ "现场审计发现需要先转化为可验证的系统性问题图谱", "法规基线决定质量体系差距的严重度与整改边界", "生产工艺体系风险来自流程、设施、公用系统和验证证据的耦合缺口", "偏差、变更、CAPA 和数据完整性决定质量系统能否闭环", "人员能力与质量文化决定制度是否真正落地", "运营管理问题需要区分组织、流程、会议机制和指标体系缺口", "跨部门协同断点会放大 GMP 风险和交付风险", "标杆实践应转化为短中长期整改组合而非口号", "整改路线图必须绑定责任、优先级、证据和复核机制", "管理层治理机制决定白帆能否从一次整改转向持续改进", ], } def render_framework(project_root: Path, *, method_key: str | None = None, chapter_count: int = 10) -> Path: manifest = load_manifest(project_root) registry = ResearchMethodRegistry() method = registry.get(method_key or manifest.get("research_method")) if method_key: manifest["research_method"] = method.key titles = CHAPTER_TEMPLATES.get(method.key) or CHAPTER_TEMPLATES["mckinsey_market"] chapter_count = max(8, min(15, chapter_count)) selected = titles[:chapter_count] quota = max(800, int(manifest.get("target_words", 30000)) // len(selected)) sections = "\n".join(f"- {item}" for item in method.framework_sections) axes = "、".join(method.task_axes) material_text = render_material_inventory(manifest.get("material_inventory") or []) lines = [ f"# {manifest.get('report_title') or manifest['topic']}：研究框架", "", f"research_method: {method.key}", f"method_name: {method.name}", f"work_language: 中文主写作；检索关键词、证据摘录、source title、raw notes 可保留英文。", f"target_words: {manifest.get('target_words', 30000)}", "", "## 方法选择", "", f"本项目采用 `{method.key}`，因为其结构原则是：{method.structure_principle}", "", "框架模块：", sections, "", "Phase 2 任务轴：", f"- {axes}", "", "## 输入材料与使用边界", "", material_text, "", "这些材料作为现场问题线索和内部事实起点使用；正式结论仍需结合 NMPA、FDA、EMA、ICH、WHO 等权威法规、指南和最佳实践进行验证。", "", "## 中心假设", "", f"围绕“{manifest['topic']}”形成可被证据支持或证伪的中文主线；所有核心判断必须绑定来源 ID。", "", ] for idx, title in enumerate(selected, start=1): lines.extend( [ f"## 第{idx}章 {title}", "", f"建议字数：约 {quota} 字。", f"研究思路：围绕 `{method.key}` 的方法框架，从 {axes} 等任务轴并发收集 evidence packet，再由 chapter assembly 收束为完整中文章节。", "证据要求：至少 2 个独立 Tier 1-2 信源；不足时在正文标注待验证；必须包含反方证据。", "", ] ) lines.extend( [ "## 暂停点", "", "请先确认 `phase1/material_brief.md` 的材料解读和访谈问题，再确认本框架后进入 Phase 2。若章节逻辑、方法框架或字数配额需要调整，应先修改本文件。", "", "确认后运行：`uv run python scripts/dr.py approve `；未批准时 `research` 默认会拒绝推进，可用 `--force` 临时覆盖。", "", ] ) 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(), } manifest["updated_at"] = utc_now_iso() write_manifest(project_root, manifest) return out