teacher-to-skills

83 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• SKILL.md13 KB683ae699f480upstream ↗

  ---
  name: teacher-to-skills
  description: Distill a teacher into a coaching Skill. Turn their courses, books, video, audio, PDFs, slides, transcripts, OCR, notes, and long-form materials into source-grounded learning Skills that teach unseen concepts progressively with reliable terminal ASCII or SVG visuals, ask two end-of-lesson questions together by default, give focused feedback, schedule review, test transfer, and support independent work. Also use for CoursePackage migration, teacher-method extraction, learning paths, practice and assessment libraries, private learner-state workflows, reusable personal methods, and graduation evaluation.
  ---
  
  # Teacher-to-Skills
  
  Distill a teacher's source material into a mentor Skill that can cite the source, demonstrate the teacher's method, guide practice, correct mistakes, and gradually reduce the learner's dependence on help.
  
  Use this method:
  
  **Capture → Cite → Compress → Connect → Codify → Coach → Practice → Consolidate → Transfer → Graduate**
  
  Preserve evidence before synthesis. Require observable learner work before mastery. Keep immutable teacher assets separate from private learner state.
  
  ## Read when needed
  
  - Pipeline, providers, schemas, recovery, and output layouts: [references/runtime.md](references/runtime.md)
  - Session routing, attempt-first assessment, feedback, and fading: [references/apprenticeship-protocol.md](references/apprenticeship-protocol.md)
  - Progressive concept teaching, SVG/ASCII explanation, and the two-question check: [references/micro-lesson-protocol.md](references/micro-lesson-protocol.md)
  - Claim and feedback provenance: [references/provenance-policy.md](references/provenance-policy.md)
  - Mastery transitions and review scheduling: [references/mastery-policy.md](references/mastery-policy.md)
  - Graduation evidence: [references/graduation-policy.md](references/graduation-policy.md)
  - External learner-state boundary: [references/external-learner-store-contract.md](references/external-learner-store-contract.md)
  
  ## Product invariants
  
  - Trace teacher claims to source, lesson, timestamp, chunk, card, keyframe, or page.
  - Distinguish `direct_source`, `source_grounded_synthesis`, `cross_source_synthesis`, `mentor_inference`, learner evidence, external knowledge, and unsupported content.
  - Preserve conflicting sources and their conditions; never flatten disagreement into false consensus.
  - Treat TeacherModel as source-supported domain behavior, not a personality or consciousness clone.
  - Treat recognition, fluency, lesson completion, and one success as insufficient mastery evidence.
  - Require an observable attempt before updating capability state.
  - Teach one unseen capability with a plain-language model, precise explanation, useful ASCII/SVG visual, example, counterexample, and exactly two end-of-lesson questions presented together by default.
  - Advance at most one mastery state per successful episode.
  - Require changed-context H0 performance for transfer and delayed parallel-form success for retention.
  - Keep PracticeEpisodes append-only and rebuild MasteryState from events.
  - Store real learner data only in a host-provided external private directory.
  - Require explicit learner approval before promoting or installing a Personal Skill.
  - Keep high-risk learning educational, source-bounded, and distinct from professional qualification.
  
  ## Detect source state
  
  Choose the smallest workflow that preserves the requested evidence:
  
  1. Raw video/audio: run capture, visual analysis for video, model-selected keyframes, distillation, audit, compilation, generation, and validation.
  2. Video plus PDFs: include document parsing or reuse existing OCR.
  3. Markdown/TXT/books/notes/OCR: skip media capture; build text chunks and evidence cards before compilation.
  4. Existing transcripts, analyses, documents, or distillation: resume from the narrowest missing stage.
  5. Existing CoursePackage 0.x: migrate to 1.0 before building teacher/runtime assets.
  6. Existing CoursePackage 1.0: compile missing TeacherModel, graph, banks, MentorPackage, and Skill.
  7. Multiple CoursePackages: migrate each, merge with namespaced IDs, preserve conflicts, then compile.
  
  Do not rerun expensive capture when stable artifacts already exist and inputs have not changed.
  
  ## Configure providers
  
  Separate capability from configuration. Report missing configuration and the smallest viable fallback.
  
  - Audio transcription: `AUDIO_TRANSCRIBE_API_KEY`, `AUDIO_TRANSCRIBE_BASE_URL`, `AUDIO_TRANSCRIBE_MODEL`.
  - Video/vision: `TEACHER_TO_SKILLS_VISION_API_KEY`, `TEACHER_TO_SKILLS_VISION_BASE_URL`, `TEACHER_TO_SKILLS_VISION_MODEL`.
  - Text distillation: `TEACHER_TO_SKILLS_TEXT_API_KEY`, `TEACHER_TO_SKILLS_TEXT_BASE_URL`, `TEACHER_TO_SKILLS_TEXT_MODEL` when LLM distillation is enabled.
  - PDF/OCR submission: `MINERU_API_TOKEN`, unless existing OCR is reused.
  - Local media: `ffmpeg` and `ffprobe`.
  
  If raw media needs an unavailable provider, stop before that capture stage. If transcripts, OCR, notes, or packages already exist, continue from them and report excluded modalities. Never store secrets in the repository or commit `.env`.
  
  ## Run the full compiler
  
  For media and documents:
  
  ```bash
  python scripts/run_course_pipeline.py \
    --input-dir <course-media> \
    --documents-input <pdf-or-directory> \
    --notes-input <notes> \
    --course-name <course-name> \
    --skill-name <skill-name> \
    --mode mentor \
    --apprenticeship full \
    --practice-depth deep \
    --learner-state external \
    --output-dir ./dist
  ```
  
  For text or books:
  
  ```bash
  python scripts/run_course_pipeline.py \
    --text-input <markdown-txt-or-directory> \
    --course-name <course-name> \
    --mode mentor \
    --apprenticeship full \
    --output-dir ./dist
  ```
  
  Use `--text-no-llm` for deterministic explicit-label extraction. Treat implicit teacher cognition from ordinary exposition as medium/low confidence and require review.
  
  ## Compile existing materials
  
  Build or migrate CoursePackage:
  
  ```bash
  python scripts/build_course_package.py \
    --course-name <course-name> \
    --source-dir <course-workspace>
  
  python scripts/migrate_course_package.py \
    <legacy-course-package.json>
  ```
  
  Compile the apprenticeship assets in dependency order:
  
  ```bash
  python scripts/build_teacher_model.py --source-dir <course-workspace>
  python scripts/build_capability_graph.py --source-dir <course-workspace>
  python scripts/build_practice_bank.py --source-dir <course-workspace> --practice-depth deep
  python scripts/build_assessment_bank.py --source-dir <course-workspace>
  python scripts/build_mentor_package.py --source-dir <course-workspace> --apprenticeship full
  python scripts/build_mentor_readiness_audit.py --source-dir <course-workspace>
  ```
  
  Then generate and validate:
  
  ```bash
  python scripts/build_course_skill.py \
    --course-name <course-name> \
    --skill-name <skill-name> \
    --mode mentor \
    --apprenticeship full \
    --learner-state external \
    --source-dir <course-workspace> \
    --output-dir ./dist
  ```
  
  The builder uses a temporary directory and installs the target only after validation.
  
  ## Merge courses
  
  ```bash
  python scripts/build_multi_course_package.py \
    --course <course-a-package-or-dir> \
    --course <course-b-package-or-dir> \
    --combined-name <combined-name> \
    --output-dir <combined-workspace>
  ```
  
  Migrate each package to an in-memory 1.0 form, namespace colliding IDs, keep source-course identity, record semantic conflicts, and route those conflicts into comparison rather than consensus.
  
  ## Choose role and apprenticeship
  
  - Default to `mentor` when the user wants capability formation.
  - Use `expert` for source lookup, course Q&A, explanation, and citations.
  - Use `consultant` for course-grounded diagnosis, option comparison, and advice.
  - Use `practitioner` for playbooks, checklists, templates, workflows, and artifacts.
  - Use `custom` for a user-defined source-bounded role.
  
  Treat role, scope, evidence strategy, apprenticeship mode, and learner-state strategy as separate dimensions. Mentor requests default to full apprenticeship; downgrade to guided or none when readiness is insufficient and report every blocker.
  
  ## Validate readiness
  
  Require the following for full apprenticeship:
  
  - valid TeacherModel with source-supported cues, decisions, demonstrations, feedback, and graduation signals;
  - acyclic CapabilityGraph with complete references and prerequisites;
  - at least one practice task and one assessment for each target capability;
  - behaviorally anchored rubrics and ordered H0–H4 hints;
  - retrieval, transfer, boundary, production, and graduation assessment coverage;
  - a complete Mentor protocol and graduation policy;
  - source audit that does not block core capabilities;
  - no duplicate IDs, dangling references, private learner data, or empty runtime files.
  
  Use guided mode when useful training assets exist but full teacher evidence does not. Do not hide gaps.
  
  ## Run Mentor Runtime
  
  First route the request as source lookup, direct explanation, diagnostic learning, guided practice, artifact feedback, real-world application, retrieval review, transfer test, or graduation test.
  
  - Answer explicit lookup directly; do not update mastery.
  - Honor explicit quick answers but label them as non-mastery evidence.
  - For an unseen capability, teach one progressive micro-lesson first, then present exactly two numbered formative questions together and wait. Use one-at-a-time pacing only when requested or needed for cognitive load.
  - For retrieval, review, assessment, transfer, and graduation, collect a prediction or attempt first.
  - Evaluate criterion-level rubric evidence, name one primary bottleneck, give the minimum effective hint, and request a concrete revision when the task or critical gap calls for it.
  - Record prediction, confidence, attempts, hints, rubric results, source evidence, inference, errors, outcome, reflection, mastery events, and next actions.
  - Explain why the next task was selected.
  - Fade only one support dimension after repeated success.
  
  Read the detailed protocol and policies before operating a generated Mentor Skill.
  
  ## Keep learner state external
  
  Resolve the host-provided store as:
  
  ```text
  {learner_store_root}/apprenticeships/{mentor_package_id}/
  ```
  
  Initialize, append, rebuild, schedule, select, validate, and build candidates with the generated runtime scripts. If the host cannot write, return a complete state patch and say it was not persisted. Never generate real `learner_progress.json` or apprenticeship state under Skill `references/`.
  
  ## Build Personal Skills and graduate
  
  Generate a candidate only after:
  
  - three successful practices;
  - two distinct contexts;
  - one H0 execution;
  - one failure, error, or counterexample;
  - clear trigger, preconditions, procedure, output, evaluator, and failure modes;
  - separable teacher lineage and personal adaptation.
  
  Promote only after regression tests, a new-context success, no material capability regression, copyright/privacy review, Skill validation, and explicit learner approval.
  
  Graduate only with delayed retention, unseen-case diagnosis, independent production, changed-context transfer, boundary recognition, critical comparison, personal adaptation evidence, and Personal Skill regression. After graduation, reduce Mentor behavior to source consultant, counterexample provider, advanced sparring partner, and source-update notifier.
  
  ## Verify outputs
  
  Run:
  
  ```bash
  python scripts/validate_teacher_to_skills_package.py --package <course_package.json>
  python scripts/validate_mentor_package.py --package <mentor_package.json>
  python scripts/validate_generated_skill.py --skill-dir <generated-skill>
  python "${CODEX_HOME:-$HOME/.codex}/skills/.system/skill-creator/scripts/quick_validate.py" .
  git diff --check
  ```
  
  Report source state, workflow, generated path, visual evidence mode, source readiness, mentor readiness, runtime readiness, apprenticeship downgrade, human-review needs, validation results, and remaining risks.
  
  ## Complete the user handoff
  
  Generation is not complete while the learner still does not know how to use the new Skill.
  
  After validation:
  
  1. Give a plain-language inventory of the material that was processed and anything that was missing or unreadable.
  2. State whether the result supports full coaching, guided learning, or source lookup only, and explain any downgrade without internal jargon.
  3. State whether raw source bodies or private learner records entered the generated Skill. Default to neither.
  4. When the user asked to enable the generated Skill, place the validated Skill in the current host's active Skill location only after checking for an unrelated name collision. Do not overwrite an unrelated Skill. If the host cannot reload it immediately, tell the user that one restart is required.
  5. End with one ready-to-send first-session request that collects the learner's goal, current level, available time, and real application before selecting one baseline task.
  6. For later updates, preserve external learner history and rebuild only the stages affected by new or changed source material.
  
  Do not expose compiler commands, schema names, runtime object names, or validation internals in the normal user handoff unless the user explicitly asks for developer details.
  
  Do not commit private transcripts, images, OCR, learner state, or copyrighted bodies unless the user explicitly authorizes publication.
  

• README.md9.8 KB850665891d13upstream ↗

  <div align="center">
  
  <img src="./docs/img/hero-zh.svg" alt="Teacher-to-Skills：把老师蒸馏成 Skill。提炼老师怎么看、怎么判断、怎么纠错，变成一位会示范、会纠错、最终放手的 AI 老师" width="100%">
  
  # Teacher-to-Skills
  
  **蒸馏你的老师，变成一个随时能问、会带你练的 Skill**
  
  [![License](https://img.shields.io/badge/license-Apache%202.0-2F8F83?style=flat-square)](./LICENSE)
  [![Works with](https://img.shields.io/badge/works%20with-Claude%20Code%20%C2%B7%20Codex%20%C2%B7%20OpenClaw-4C6FB8?style=flat-square)](./docs/install.md)
  [![Stars](https://img.shields.io/github/stars/JuneYaooo/teacher-to-skills?style=flat-square&color=7B5BB5)](https://github.com/JuneYaooo/teacher-to-skills/stargazers)
  
  **中文** · [English](./README.en.md) · [三步上手](#三步上手) · [能蒸馏哪些材料](#能蒸馏哪些材料) · [工作原理](#工作原理) · [安装说明](./docs/install.md) · [更新记录](./CHANGELOG.md)
  
  </div>
  
  <br>
  
  把老师的课程、书、录音、讲义和批注交给它，它会提炼老师**怎么看问题、怎么判断、怎么纠错**，打包成一位有出处的 AI 老师：先示范，再让你动手，按老师的标准给反馈，直到你不再需要它。
  
  | 你交给它 | 它蒸馏出 | 你得到 |
  | --- | --- | --- |
  | 老师的课程视频、书与讲义、录音访谈、文章笔记、批注反馈 | 出处、观察点、判断规则、示范、常见错误与纠偏、适用边界 | 一位随时能问、会出题、按老师标准点评、最终放手的 AI 老师 |
  
  ## 为什么是“蒸馏老师”，而不是“总结课程”
  
  一位好老师最值钱的，往往不是讲义上的结论，而是没写下来的那部分：先看哪里、凭什么下判断、学生最常错在哪、什么情况下这套方法不能用。
  
  普通摘要只留下“老师讲了什么”。Teacher-to-Skills 想留下的是“老师会怎么做”，并且把它变成你自己也能做到的能力。
  
  <img src="./docs/img/distill-zh.svg" alt="老师的课程视频、书与讲义、录音访谈、文章笔记和批注反馈，经过转录识别、标注出处、提炼方法，蒸馏出出处、观察点、判断规则、示范、常见错误与纠偏、适用边界六类内容" width="100%">
  
  | | 课程摘要 | 蒸馏出来的老师 Skill |
  | --- | --- | --- |
  | 留下什么 | 知识点和结论 | 老师的观察点、判断规则、示范和纠错方式 |
  | 怎么用 | 你自己去读 | 它出题、你先做，它按老师的标准点评 |
  | 可信度 | 难以分辨是原话还是改写 | 重要结论都能回查原文、原话或关键画面 |
  | 最终结果 | 读完就结束 | 隔一段时间、换个场景你也能独立做出来 |
  
  它**不会**声称复制老师的人格、意识或全部经验。材料里没有的东西，它会标成“证据不足”，不会编造“老师大概会这么说”。
  
  ## 蒸馏出来的老师会怎样带你
  
  <img src="./docs/img/journey-zh.svg" alt="八个阶段：定目标、看示范、跟着做、被纠偏、独立做、换场景、出师检验、持续精进。老师的支持从完整示范加提示，逐步减少到只给最终反馈" width="100%">
  
  - **先摸底，再讲课**：了解你的目标、基础和真实应用场景，从你最需要的知识点开始。
  - **先让你做**：每个知识点讲完，一起给出两道题，一道检查理解，一道检查应用，回答后分别点评。
  - **按老师的标准纠错**：每轮只抓最关键的一个问题，给刚好够用的提示，并让你自己修订。
  - **记得你错过什么**：保存你的尝试、错误和修订，按你的薄弱点安排复习，不重复原题。
  - **逐步放手**：随着你进步减少模板和提示，最后用没见过的场景、不给提示地检验你。
  - **知道什么时候算出师**：只有隔一段时间还会做、换场景也能用、并且知道方法边界时，才认为你真正学会。
  
  <img src="./docs/img/lesson-zh.svg" alt="一节课：选一个知识点，讲清楚（直觉、准确定义、例子、反例、三点小结），用终端 ASCII 或 SVG 画出来，一起出两道题；之后分别点评两道题，记录错误与修订，进入下一个知识点" width="100%">
  
  目标不是让你一直依赖这位 AI 老师，而是让你学到老师的方法，最后形成自己的方法。
  
  ## 三步上手
  
  你不需要运行脚本，也不需要理解内部文件结构，全程用自然语言和你的 AI 助手说话即可。
  
  **① 安装**
  
  把[安装说明](./docs/install.md)发给你的 AI 助手（Claude Code、Codex、OpenClaw 等），让它代你安装，装好后按提示重启。
  
  **② 交给它老师的材料，开始蒸馏**
  
  上传文件或告诉它材料所在的文件夹，然后直接说：
  
  > 请使用 Teacher-to-Skills 蒸馏这位老师的材料，做成一个能带我真正学会的老师 Skill。我的目标是把老师的方法用到真实工作中。重要结论要能回查出处，原始材料和个人学习记录保持私有。请先告诉我识别到了哪些材料、还缺什么；如果没有需要我决定的风险，就继续完成蒸馏、检查并启用新 Skill，最后告诉我怎样开始第一课。
  
  它会盘点材料，完成转录、识别、出处整理、老师方法提炼、练习设计和质量检查，中途中断了也能接着做。交付时它应该清楚告诉你：
  
  - 老师 Skill 是否已经生成并启用；
  - 哪些视频、文档、章节和笔记已经处理；
  - 哪些内容证据不足、识别不清或需要你确认；
  - 当前能做到完整带教、引导学习，还是只适合查资料；
  - 原始材料和个人学习记录保存在哪里，有没有被打包进 Skill。
  
  如果这些没有交代清楚，直接让它补做交付检查。
  
  **③ 跟着老师学**
  
  > 用刚刚生成的老师 Skill 开始带我学习。先问清楚我的目标、基础、每周可投入时间和一个真实应用场景；做一次简短摸底后，从一个知识点开始深入浅出地讲。讲完一起给我两道题，等我回答后分别点评。
  
  之后每次回来，一句话就够：
  
  | 你想做什么 | 可以直接说 |
  | --- | --- |
  | 继续学习 | “继续上次进度，给我今天最应该做的一项练习。” |
  | 问老师会怎么看 | “按老师的方法，这个问题应该先看什么？给出原话出处，这次不用练习。” |
  | 提交真实作品 | “这是我按老师方法完成的作品，先按老师的标准评价，再指出最关键的一个问题。” |
  | 复习巩固 | “根据我以前的错误安排一次复习，不要重复原题。” |
  | 检查迁移 | “换一个我没见过的场景，检查我能不能独立用这个方法。” |
  | 解决现实问题 | “先让我说明自己的判断和方案，再用老师的方法帮我纠偏。” |
  | 追加新材料 | “这是老师的新课，更新进老师 Skill，保留我原来的学习进度。” |
  | 申请出师 | “不要给提示，用新场景检验我是否真的掌握，并说明还缺什么证据。” |
  
  ## 能蒸馏哪些材料
  
  | 你手上的材料 | 它会怎么处理 |
  | --- | --- |
  | 🎬 课程视频、讲座录像 | 结合讲解声音、画面、幻灯片、板书和演示进行整理 |
  | 🎙️ 录音、播客、访谈 | 转成可检索的文字，并按章节和主题组织 |
  | 📚 书籍、PDF、扫描讲义 | 识别正文、结构、表格和重要页面 |
  | 📝 文章、笔记、批注、文字稿 | 保留原文位置，提炼方法、案例和练习 |
  | ♻️ 已有的转录或课程整理 | 跳过重复处理，在已有成果上继续 |
  | 👥 多位老师的材料 | 分别保留每位老师的观点、适用条件和分歧，不强行拼成一个答案 |
  
  材料不完整也可以开始。它会标出缺失的证据和需要人工确认的地方。材料只够解释概念时，它会如实退回到资料查询或引导学习，不会假装已经具备完整的带教能力。
  
  ## 工作原理
  
  <img src="./docs/img/layers-zh.svg" alt="四个层次：证据层、老师层、练习层组成老师 Skill，生成后不改动、随时回查出处；个人层是你的学习记录，私有并保存在 Skill 之外" width="100%">
  
  老师的材料和你的学习记录分开保存：老师那部分生成后不再改动，你的进度保存在 Skill 之外的私有目录，重新蒸馏或更新材料都不会丢失学习记录。
  
  <details>
  <summary>四个层次的文字说明</summary>
  
  | 层次 | 做什么 |
  | --- | --- |
  | 证据层 | 保存原文、原话、讲义和关键画面，每条结论都能回查 |
  | 老师层 | 提炼老师的观察点、判断规则、示范、纠错方式和边界 |
  | 练习层 | 按学习顺序设计练习、评分标准、复习和迁移检验 |
  | 个人层 | 保存你的尝试和真实结果，经你同意后沉淀成你自己的方法 |
  
  </details>
  
  ## 边界、隐私与版权
  
  - 只蒸馏有出处的方法和判断，不复制老师的人格、意识或全部经验。
  - 老师原意、基于材料的整理、AI 的推断和你自己的实践证据会明确区分。
  - 原始课程、转录、截图、扫描内容和个人学习记录默认不会打包进生成的 Skill。
  - 只有你明确同意后，经过多次实践验证的做法才会被建议为你的个人长期方法。
  - 医学、法律、金融、投资等高风险领域仅用于有来源的学习，不能替代专业资质或现实责任。
  - 请只蒸馏你有权使用的材料，并遵守课程版权和分发限制。
  
  ## 致谢
  
  - [Datawhale](https://github.com/datawhalechina) — AI 教育、开源课程与学习者社区实践。
  - [rfeng1016](https://github.com/rfeng1016) — 对渐进知识组织方向的建议。
  - [LINUX DO](https://linux.do/) — 中文开发者社区的讨论与反馈。
  
  ## License
  
  本项目采用 [Apache License 2.0](./LICENSE)。
  
  <div align="center">
  <sub>插图由 <code>docs/img/src/build_images.py</code> 生成，修改文案后重新运行即可更新。</sub>
  </div>
  

• scripts/advance_micro_lesson.py7.3 KB5ba51b3a29b5upstream ↗

  #!/usr/bin/env python3
  """Persist together-by-default or one-at-a-time micro-lesson pacing."""
  
  from __future__ import annotations
  
  import argparse
  import json
  from pathlib import Path
  from typing import Any
  
  from runtime_state import read_json, utc_now, write_json
  
  
  PHASES = [
      "teaching",
      "awaiting_answers",
      "feedback_batch",
      "awaiting_answer_1",
      "feedback_1",
      "ready_question_2",
      "awaiting_answer_2",
      "feedback_2",
      "complete",
  ]
  
  
  def state_path(state_dir: Path) -> Path:
      return state_dir.expanduser().resolve() / "apprenticeship_state.json"
  
  
  def load_state(state_dir: Path) -> tuple[Path, dict[str, Any]]:
      path = state_path(state_dir)
      if not path.exists():
          raise ValueError(f"missing apprenticeship state: {path}")
      return path, read_json(path, {})
  
  
  def start(
      state: dict[str, Any],
      *,
      unit_id: str,
      capability_id: str,
      question_ids: list[str],
      pacing: str = "together",
  ) -> None:
      if len(question_ids) != 2 or len(set(question_ids)) != 2:
          raise ValueError("a micro-lesson requires exactly two distinct question IDs")
      if pacing not in {"together", "one-at-a-time"}:
          raise ValueError(f"unsupported question pacing: {pacing}")
      active = state.get("active_learning_unit")
      if isinstance(active, dict) and active.get("phase") not in {"complete"}:
          raise ValueError(f"cannot replace active micro-lesson in phase {active.get('phase')}")
      state["active_learning_unit"] = {
          "unit_id": unit_id,
          "capability_id": capability_id,
          "phase": "teaching",
          "pacing": pacing,
          "question_ids": question_ids,
          "current_question_index": 0,
          "answered_question_ids": [],
          "last_answered_question_ids": [],
          "diagram_refs": [],
          "started_at": utc_now(),
          "updated_at": utc_now(),
      }
  
  
  def active_unit(state: dict[str, Any]) -> dict[str, Any]:
      active = state.get("active_learning_unit")
      if not isinstance(active, dict):
          raise ValueError("no active micro-lesson; start one first")
      if active.get("phase") not in PHASES:
          raise ValueError(f"invalid micro-lesson phase: {active.get('phase')}")
      if len(active.get("question_ids") or []) != 2:
          raise ValueError("active micro-lesson must contain exactly two questions")
      return active
  
  
  def normalize_indexes(question_index: int | list[int] | None) -> list[int]:
      values = question_index if isinstance(question_index, list) else [] if question_index is None else [question_index]
      if any(value not in {1, 2} for value in values) or len(values) != len(set(values)):
          raise ValueError(f"question indexes must be distinct values from 1 and 2: {values}")
      return values
  
  
  def transition(
      state: dict[str, Any],
      *,
      event: str,
      question_index: int | list[int] | None,
      diagram_ref: str | None,
  ) -> None:
      active = active_unit(state)
      phase = active["phase"]
      indexes = normalize_indexes(question_index)
      pacing = active.get("pacing") or "one-at-a-time"
      if pacing == "together":
          if event == "present":
              if phase != "teaching" or indexes:
                  raise ValueError(f"cannot present the two-question set from phase {phase}")
              active["current_question_index"] = 0
              active["phase"] = "awaiting_answers"
          elif event == "answer":
              if phase != "awaiting_answers" or not indexes:
                  raise ValueError(f"cannot record answers {indexes} from phase {phase}")
              question_ids = [active["question_ids"][index - 1] for index in indexes]
              new_ids = [question_id for question_id in question_ids if question_id not in active["answered_question_ids"]]
              if not new_ids:
                  raise ValueError("all supplied question answers were already recorded")
              active["answered_question_ids"].extend(new_ids)
              active["last_answered_question_ids"] = new_ids
              active["phase"] = "feedback_batch"
          elif event == "finish-feedback":
              if phase != "feedback_batch" or indexes:
                  raise ValueError(f"cannot finish batch feedback from phase {phase}")
              active["last_answered_question_ids"] = []
              if len(active["answered_question_ids"]) == 2:
                  active["phase"] = "complete"
                  active["completed_at"] = utc_now()
              else:
                  active["phase"] = "awaiting_answers"
          else:
              raise ValueError(f"unsupported event: {event}")
      else:
          index = indexes[0] if len(indexes) == 1 else None
          if event == "present":
              expected = 1 if phase == "teaching" else 2 if phase == "ready_question_2" else None
              if index != expected:
                  raise ValueError(f"cannot present question {index} from phase {phase}")
              active["current_question_index"] = index
              active["phase"] = f"awaiting_answer_{index}"
          elif event == "answer":
              if index not in {1, 2} or phase != f"awaiting_answer_{index}":
                  raise ValueError(f"cannot record answer {index} from phase {phase}")
              question_id = active["question_ids"][index - 1]
              if question_id not in active["answered_question_ids"]:
                  active["answered_question_ids"].append(question_id)
              active["last_answered_question_ids"] = [question_id]
              active["phase"] = f"feedback_{index}"
          elif event == "finish-feedback":
              if index not in {1, 2} or phase != f"feedback_{index}":
                  raise ValueError(f"cannot finish feedback {index} from phase {phase}")
              active["last_answered_question_ids"] = []
              active["phase"] = "ready_question_2" if index == 1 else "complete"
              if index == 2:
                  active["completed_at"] = utc_now()
          else:
              raise ValueError(f"unsupported event: {event}")
      if diagram_ref and diagram_ref not in active["diagram_refs"]:
          active["diagram_refs"].append(diagram_ref)
      active["updated_at"] = utc_now()
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Advance a two-question micro-lesson state machine.")
      parser.add_argument("--state-dir", required=True)
      parser.add_argument("--event", choices=["start", "present", "answer", "finish-feedback"], required=True)
      parser.add_argument("--unit-id")
      parser.add_argument("--capability-id")
      parser.add_argument("--question-id", action="append", default=[])
      parser.add_argument("--question-index", type=int, choices=[1, 2], action="append")
      parser.add_argument("--pacing", choices=["together", "one-at-a-time"], default="together")
      parser.add_argument("--diagram-ref")
      args = parser.parse_args()
  
      path, state = load_state(Path(args.state_dir))
      if args.event == "start":
          if not args.unit_id or not args.capability_id:
              raise SystemExit("start requires --unit-id and --capability-id")
          start(state, unit_id=args.unit_id, capability_id=args.capability_id, question_ids=args.question_id, pacing=args.pacing)
      else:
          transition(state, event=args.event, question_index=args.question_index, diagram_ref=args.diagram_ref)
      state["updated_at"] = utc_now()
      write_json(path, state)
      print(json.dumps({"active_learning_unit": state["active_learning_unit"], "state": str(path)}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/analyze_videos.py16 KB8d458e60c7d2upstream ↗

  #!/usr/bin/env python3
  # -*- coding: utf-8 -*-
  """Batch course visual analysis with optional supplemental screenshot markers."""
  
  import os
  import re
  import subprocess
  import argparse
  import tempfile
  import shutil
  from pathlib import Path
  from datetime import datetime
  
  from dotenv import load_dotenv
  from llm_client import analyze_video_with_vision_model
  
  load_dotenv()
  
  BASE_DIR = str(Path(__file__).resolve().parents[1])
  FFMPEG = os.getenv("FFMPEG") or shutil.which("ffmpeg") or "ffmpeg"
  FFPROBE = os.getenv("FFPROBE") or shutil.which("ffprobe") or "ffprobe"
  
  
  COURSE_ANALYSIS_PROMPT = """请观看这段课程视频，提取教学内容干货。
  
  ## 分析要求
  
  ### 一、核心知识点
  逐条列出本片段涉及的具体概念、方法论、模型、技巧：
  1. ...
  2. ...
  
  ### 二、补充截图标记
  
  **【截图标记规则 —— 只标实质教学画面】**
  
  只有画面满屏都是实质性知识内容时，才用 `[SCREENSHOT MM:SS]` 标记：
  
  ✅ **应该标记**：满屏文字列表（≥3要点）、数据表格/图表、思维导图、公式/模型图、软件操作界面、实物展示细节
  
  ❌ **不应标记**：人脸特写、标题/封面页、空白桌面、暖场聊天画面、模糊内容
  
  格式（描述越具体越好）：
  ```
  [SCREENSHOT 05:23] 满屏PPT，标题"核心方法论"，含数据曲线图+三个要点列表
  [SCREENSHOT 12:45] 板书手写公式及推导过程，标注了关键变量定义
  [SCREENSHOT 28:10] 软件后台数据看板，展示关键指标的变化趋势和漏斗
  ```
  
  如果完全没有值得截图的内容，写「本段无干货画面」
  
  注意：这些截图标记只是视频分析阶段的补充线索。最终关键帧证据由后续模型精选关键帧流程从候选帧 contact sheet 中选择。
  
  ### 三、教学结构
  - 本段讲了什么主题
  - 内容组织逻辑（怎么展开的）
  - 有没有总结
  
  ### 四、案例与举例
  讲者提到的具体案例、数据、故事（越具体越好）
  
  ### 五、金句与核心观点
  讲者的重要原话或关键论断
  
  注意：只关注教学内容，不要描述讲者外貌、讲课风格、语速等。基于真实内容，不要虚构。"""
  
  
  def get_video_duration(video_path: str) -> float:
      cmd = [FFPROBE, "-v", "error", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", video_path]
      r = subprocess.run(cmd, capture_output=True, text=True)
      return float(r.stdout.strip()) if r.returncode == 0 else 0
  
  
  def compress_720p(input_path: str, output_path: str) -> bool:
      """压缩到 720p，适合 API 上传"""
      cmd = [
          FFMPEG, "-i", input_path,
          "-vf", "scale=-2:720",
          "-c:v", "libx264", "-preset", "fast", "-crf", "28",
          "-c:a", "aac", "-b:a", "64k",
          "-y", output_path,
      ]
      try:
          subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
          return os.path.exists(output_path) and os.path.getsize(output_path) > 0
      except Exception as e:
          print(f"  压缩失败: {e}")
          return False
  
  
  def compress_480p(input_path: str, output_path: str) -> bool:
      """激进压缩到 480p，用于超大视频分片"""
      cmd = [
          FFMPEG, "-i", input_path,
          "-vf", "scale=-2:480",
          "-c:v", "libx264", "-preset", "fast", "-crf", "32",
          "-c:a", "aac", "-b:a", "48k",
          "-y", output_path,
      ]
      try:
          subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
          return os.path.exists(output_path) and os.path.getsize(output_path) > 0
      except Exception as e:
          print(f"  480p 压缩失败: {e}")
          return False
  
  
  def split_video_chunks(video_path: str, chunk_secs: int, output_dir: str, video_name: str) -> list:
      """按时间切分视频，返回 [(chunk_path, start_offset_sec), ...]"""
      duration = get_video_duration(video_path)
      if duration == 0 or duration <= chunk_secs * 1.1:
          return [(video_path, 0.0)]
  
      num = max(1, int(duration / chunk_secs))
      # 最后一片太短就合并
      if duration - (num - 1) * chunk_secs < 120 and num > 1:
          num -= 1
      actual_dur = duration / num
  
      chunks = []
      for i in range(num):
          start = i * actual_dur
          dur = actual_dur if i < num - 1 else duration - start
          cp = os.path.join(output_dir, f"{video_name}_chunk_{i+1:02d}.mp4")
          cmd = [FFMPEG, "-y", "-ss", str(start), "-i", video_path,
                 "-t", str(dur), "-c", "copy",
                 "-avoid_negative_ts", "make_zero", cp]
          try:
              subprocess.run(cmd, capture_output=True, text=True, timeout=300)
              if os.path.exists(cp) and os.path.getsize(cp) > 0:
                  chunks.append((cp, start))
                  print(f"  片段 {i+1}/{num}: offset={start/60:.0f}min, {os.path.getsize(cp)/(1024*1024):.0f}MB")
          except Exception as e:
              print(f"  切分失败: {e}")
  
      return chunks if chunks else [(video_path, 0.0)]
  
  
  def analyze_chunk(chunk_path: str, chunk_idx: int, total: int, offset_sec: float,
                    tmp_dir: str) -> dict:
      """分析单个视频片段，超大时自动二次压缩"""
      label = f"片段 {chunk_idx+1}/{total}" if total > 1 else "完整视频"
      chunk_dur = get_video_duration(chunk_path)
      raw_mb = os.path.getsize(chunk_path) / (1024 * 1024)
  
      work_path = chunk_path
      temp_result = None
  
      # 如果原始文件 > 35MB（base64 后约 47MB，可能超 API 限制），二次压缩到 480p
      if raw_mb > 35:
          compressed = os.path.join(tmp_dir, f"chunk_{chunk_idx+1}_480p.mp4")
          print(f"  📦 片段太大 ({raw_mb:.0f}MB)，二次压缩到 480p ...")
          if compress_480p(chunk_path, compressed):
              work_path = compressed
              compressed_mb = os.path.getsize(compressed) / (1024 * 1024)
              print(f"     ✅ {compressed_mb:.0f}MB")
          else:
              print(f"     ⚠️ 二次压缩失败，尝试直接发送")
  
      prompt = COURSE_ANALYSIS_PROMPT
      if total > 1:
          prompt += f"\n\n（第 {chunk_idx+1}/{total} 片段，约 {chunk_dur/60:.0f} 分钟。SCREENSHOT 时间点为片段内相对时间）"
      else:
          prompt += "\n\n注意：这是完整的视频内容，请尽可能详细地分析。"
  
      final_mb = os.path.getsize(work_path) / (1024 * 1024)
      print(f"  🤖 {label} ({chunk_dur/60:.0f}min, {final_mb:.0f}MB) ...")
  
      result = analyze_video_with_vision_model(work_path, prompt)
  
      content = result.get("content", "")
      if result.get("error"):
          content = f"**错误**: {result['error']}\n\n{content}"
  
      # 清理临时压缩文件
      if work_path != chunk_path and os.path.exists(work_path):
          try:
              os.remove(work_path)
          except Exception:
              pass
  
      return {
          "chunk_index": chunk_idx + 1,
          "offset_sec": offset_sec,
          "content": content,
      }
  
  
  def image_hash(img_path: str) -> str:
      """计算图片的感知哈希（aHash），用于去重"""
      from PIL import Image
      img = Image.open(img_path).convert("L").resize((16, 16), Image.LANCZOS)
      pixels = list(img.getdata())
      avg = sum(pixels) / len(pixels)
      return "".join("1" if p > avg else "0" for p in pixels)
  
  
  def hamming_distance(h1: str, h2: str) -> int:
      return sum(c1 != c2 for c1, c2 in zip(h1, h2))
  
  
  def deduplicate_screenshots(screenshots: list, screenshot_dir: str,
                              threshold: int = 8) -> list:
      """基于感知哈希去重，过滤相似画面（Hamming 距离 < threshold 视为重复）"""
      if len(screenshots) <= 1:
          return screenshots
  
      # 先提取所有截图的哈希
      hashes = {}
      for s in screenshots:
          # 找到对应的文件
          fname = f"{s['ts'].replace(':', '_')}_{s['desc'][:40].replace('/', '_').replace(' ', '_')}.jpg"
          spath = os.path.join(screenshot_dir, fname)
          if os.path.exists(spath):
              hashes[spath] = image_hash(spath)
  
      # 贪心去重：按时间顺序保留第一个，移除后续相似的
      kept = []
      for s in screenshots:
          fname = f"{s['ts'].replace(':', '_')}_{s['desc'][:40].replace('/', '_').replace(' ', '_')}.jpg"
          spath = os.path.join(screenshot_dir, fname)
          if spath not in hashes:
              kept.append(s)
              continue
  
          h = hashes[spath]
          is_dup = False
          for k in kept:
              kfname = f"{k['ts'].replace(':', '_')}_{k['desc'][:40].replace('/', '_').replace(' ', '_')}.jpg"
              kspath = os.path.join(screenshot_dir, kfname)
              if kspath in hashes and hamming_distance(h, hashes[kspath]) < threshold:
                  # 删除重复文件
                  try:
                      os.remove(spath)
                  except Exception:
                      pass
                  is_dup = True
                  break
  
          if not is_dup:
              kept.append(s)
  
      removed = len(screenshots) - len(kept)
      if removed > 0:
          print(f"    🧹 去重: 移除 {removed} 张重复截图")
  
      return kept
  
  
  def parse_screenshots(content: str, offset_sec: float = 0) -> list:
      """从分析内容中提取 [SCREENSHOT MM:SS] 时间点，转换为全局时间"""
      pattern = r'\[SCREENSHOT\s+(\d{1,2}):(\d{2})(?::(\d{2}))?\]\s*(.+?)(?=\n|$)'
      screenshots = []
      for m in re.finditer(pattern, content):
          mm = int(m.group(1))
          ss = int(m.group(2))
          rel_sec = mm * 60 + ss
          global_sec = rel_sec + offset_sec
          desc = m.group(4).strip()
          ts_label = f"{int(global_sec//60):02d}:{int(global_sec%60):02d}"
          screenshots.append({"global_sec": global_sec, "rel_sec": rel_sec, "ts": ts_label, "desc": desc[:80]})
      return screenshots
  
  
  def extract_frame(video_path: str, time_sec: float, output_path: str) -> bool:
      """从视频指定时间提取一帧"""
      cmd = [FFMPEG, "-y", "-ss", str(time_sec), "-i", video_path,
             "-vframes", "1", "-q:v", "2", output_path]
      try:
          subprocess.run(cmd, capture_output=True, text=True, timeout=60)
          return os.path.exists(output_path) and os.path.getsize(output_path) > 0
      except Exception:
          return False
  
  
  def process_video(video_path: str, output_dir: str, tmp_dir: str,
                    chunk_minutes: int = 12, force: bool = False) -> bool:
      """处理单个视频：压缩 → 分片 → 视觉分析 → 抽取补充截图"""
      video_name = Path(video_path).stem
      analysis_path = os.path.join(output_dir, f"{video_name}_analysis.md")
      screenshot_dir = os.path.join(output_dir, "screenshots", video_name)
  
      if os.path.exists(analysis_path) and not force:
          print(f"  ⏭ 跳过: {video_name}")
          return True
  
      print(f"\n{'='*60}")
      print(f"🔍 {video_name}")
  
      file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
      duration = get_video_duration(video_path)
      print(f"   {file_size_mb:.0f}MB, {duration/60:.1f} 分钟")
  
      chunk_secs = chunk_minutes * 60
      need_compress = file_size_mb > 400  # 超过 400MB 先压缩
      need_split = duration > chunk_secs * 1.1
      temp_files = []
      work_video = video_path
  
      # 压缩
      if need_compress:
          print(f"  📦 压缩到 720p ...")
          compressed = os.path.join(tmp_dir, f"{video_name}_720p.mp4")
          if compress_720p(video_path, compressed):
              temp_files.append(compressed)
              work_video = compressed
              new_size = os.path.getsize(compressed) / (1024 * 1024)
              print(f"  ✅ {new_size:.0f}MB")
          else:
              print(f"  ⚠️ 压缩失败，用原视频")
  
      # 分片
      if need_split:
          print(f"  ✂️ 分片（每段 {chunk_minutes} 分钟）...")
          chunks = split_video_chunks(work_video, chunk_secs, tmp_dir, video_name)
          for cp, _ in chunks:
              if cp != work_video:
                  temp_files.append(cp)
      else:
          chunks = [(work_video, 0.0)]
  
      print(f"  📊 {len(chunks)} 个片段待分析")
  
      # 分析每个片段
      all_analyses = []
      all_screenshots = []
      for i, (cp, offset) in enumerate(chunks):
          analysis = analyze_chunk(cp, i, len(chunks), offset, tmp_dir)
          all_analyses.append(analysis)
  
          # 解析截图标记
          shots = parse_screenshots(analysis["content"], offset)
          if shots:
              print(f"    📸 发现 {len(shots)} 个补充截图标记")
              all_screenshots.extend(shots)
  
      # 去重截图（基于时间戳，相同秒数只保留一个）
      seen_ts = set()
      all_shots_deduped = []
      for s in all_screenshots:
          ts_key = s["ts"]
          if ts_key not in seen_ts:
              seen_ts.add(ts_key)
              all_shots_deduped.append(s)
  
      # 先提取截图（从原视频，保证画质）
      if all_shots_deduped:
          os.makedirs(screenshot_dir, exist_ok=True)
          print(f"  📸 提取 {len(all_shots_deduped)} 张补充截图 ...")
          extracted = 0
          for s in all_shots_deduped:
              fname = f"{s['ts'].replace(':', '_')}_{s['desc'][:40].replace('/', '_').replace(' ', '_')}.jpg"
              spath = os.path.join(screenshot_dir, fname)
              if extract_frame(video_path, s["global_sec"], spath):
                  extracted += 1
  
          # 感知哈希去重
          all_shots_deduped = deduplicate_screenshots(all_shots_deduped, screenshot_dir)
          final_count = len([f for f in os.listdir(screenshot_dir) if f.endswith('.jpg')])
          print(f"  ✅ {final_count} 张补充截图")
      else:
          final_count = 0
  
      # 合并分析结果（截图数量用去重后的）
      parts = [
          f"# {video_name} — 视频内容分析\n",
          f"**时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n",
          f"**大小**: {file_size_mb:.0f}MB | **时长**: {duration/60:.1f} 分钟 | **片段**: {len(chunks)}\n",
      ]
      if final_count > 0:
          parts.append(f"**补充截图**: {final_count} 张\n")
      parts.append("\n---\n")
  
      for a in all_analyses:
          if len(all_analyses) > 1:
              offset_min = a["offset_sec"] / 60
              parts.append(f"\n## 片段 {a['chunk_index']}（偏移 {offset_min:.0f}min）\n")
          parts.append(a["content"] + "\n")
  
      if all_shots_deduped:
          parts.append("\n---\n\n## 补充截图索引\n\n")
          for s in all_shots_deduped:
              parts.append(f"- `[{s['ts']}]` {s['desc']}\n")
  
      with open(analysis_path, "w", encoding="utf-8") as f:
          f.write("\n".join(parts))
      print(f"  💾 {analysis_path}")
  
      # 清理临时文件
      for tf in temp_files:
          try:
              os.remove(tf)
          except Exception:
              pass
  
      return True
  
  
  def main():
      parser = argparse.ArgumentParser(description="批量视频画面分析 + 补充截图标记")
      parser.add_argument("--input-dir", required=True, help="视频目录")
      parser.add_argument("--course-name", required=True, help="课程名称")
      parser.add_argument("--base-dir", default=BASE_DIR, help="课程输出根目录，默认当前项目目录")
      parser.add_argument("--chunk-minutes", type=int, default=12, help="分片时长（分钟）")
      parser.add_argument("--force", action="store_true")
      parser.add_argument("--limit", type=int, default=0)
      parser.add_argument("--start-at", type=int, default=0)
      args = parser.parse_args()
  
      output_dir = os.path.join(args.base_dir, args.course_name, "analysis")
      os.makedirs(output_dir, exist_ok=True)
  
      tmp_dir = tempfile.mkdtemp(prefix="video_analysis_")
  
      video_files = sorted(
          str(path)
          for path in Path(args.input_dir).rglob("*")
          if path.is_file() and path.suffix.lower() == ".mp4"
      )
  
      if not video_files:
          print("ℹ️ 未找到 .mp4，跳过视频画面分析")
          return
  
      if args.start_at > 0:
          video_files = video_files[args.start_at:]
      if args.limit > 0:
          video_files = video_files[:args.limit]
  
      print(f"📂 {len(video_files)} 个视频 → {output_dir}")
      print(f"⏱️  分片: {args.chunk_minutes} 分钟/段")
  
      success = failed = 0
      try:
          for vf in video_files:
              if process_video(vf, output_dir, tmp_dir, args.chunk_minutes, args.force):
                  success += 1
              else:
                  failed += 1
      finally:
          shutil.rmtree(tmp_dir, ignore_errors=True)
  
      print(f"\n✅ {success} / ❌ {failed} / 共 {len(video_files)}")
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_assessment_bank.py5.6 KB98533d2d0a29upstream ↗

  #!/usr/bin/env python3
  """Build retrieval, transfer, boundary, production, and graduation assessments."""
  
  from __future__ import annotations
  
  import argparse
  import json
  from pathlib import Path
  from typing import Any
  
  from schema_utils import load_json, write_json
  from stable_ids import stable_id
  
  
  def assessment_item(node: dict[str, Any], rubric_ids: list[str], course_id: str, assessment_type: str) -> dict[str, Any]:
      prompts = {
          "retrieval": f"Without opening the course notes, reconstruct {node['name']}: cues, decision/action sequence, self-check, and one limit.",
          "transfer": f"Use {node['name']} in an unseen case where domain, scale, resources, time, audience, or risk differs. State structural similarities, differences, and a counterexample.",
          "boundary": f"Given a plausible case, decide whether {node['name']} should not be used. Name the decisive evidence, contraindication, and safe alternative.",
          "production": f"Independently produce a real artifact that demonstrates {node['name']}; include inputs, decisions, output, self-evaluation, and real-world result when available.",
          "graduation": f"Independently diagnose, choose, execute, evaluate, and defend a real-world use of {node['name']} with no method hints.",
      }
      novelty = {
          "retrieval": ["delayed", "new surface form"],
          "transfer": ["domain", "constraints", "risk"],
          "boundary": ["counterexample", "contraindication"],
          "production": ["real artifact", "independent execution"],
          "graduation": ["unseen case", "no hints", "real-world evidence"],
      }
      support = "none" if assessment_type in {"retrieval", "transfer", "graduation"} else "source-index"
      return {
          "id": stable_id("assessment", course_id, node["id"], assessment_type),
          "capability_ids": [node["id"]],
          "assessment_type": assessment_type,
          "prompt": prompts[assessment_type],
          "novelty_dimensions": novelty[assessment_type],
          "allowed_support": support,
          "time_condition": "Run after a delay for retention evidence." if assessment_type == "retrieval" else None,
          "rubric_ids": rubric_ids,
          "blindness_rule": "Do not show the teacher demonstration, answer evidence, method name, or rubric anchors before the first submission.",
          "evidence_answer": {
              "provenance": "source_grounded_synthesis",
              "evidence": list(node.get("source_evidence") or []),
              "evaluator_note": "Score observable behavior and boundary recognition, not lexical similarity to source wording.",
          },
          "parallel_form_group": stable_id("assessment", course_id, node["id"], "parallel"),
      }
  
  
  def build_assessment_bank(package: dict[str, Any], graph: dict[str, Any]) -> dict[str, Any]:
      course_id = package["manifest"]["course_id"]
      items: list[dict[str, Any]] = []
      for node in graph.get("nodes", []):
          types = ["retrieval", "boundary"]
          if node["type"] in {"procedure", "production", "evaluation", "transfer", "decision", "diagnosis"}:
              types.extend(["production", "transfer", "graduation"])
          else:
              types.append("transfer")
          node_items = [assessment_item(node, list(node.get("success_rubrics") or []), course_id, item_type) for item_type in types]
          items.extend(node_items)
          node["assessment_items"] = [item["id"] for item in node_items]
      required_types = ["retrieval", "transfer", "production", "boundary", "graduation"]
      present = {item["assessment_type"] for item in items}
      return {
          "schema_version": "1.0",
          "course_package_id": package["manifest"]["package_id"],
          "items": items,
          "graduation_matrix": {
              "required_assessment_types": required_types,
              "requirements": [
                  "Delayed no-hint retrieval succeeds on a parallel form.",
                  "An unseen case is diagnosed and completed independently.",
                  "A real artifact satisfies every critical rubric criterion.",
                  "The learner identifies boundaries, contraindications, and a counterexample.",
                  "The learner compares the teacher method with a personal adaptation and cites evidence for the change.",
                  "A Personal Skill candidate passes regression tests and receives explicit learner approval before promotion.",
              ],
              "single_success_is_sufficient": False,
              "requires_retention_and_transfer": True,
          },
          "quality": {
              "status": "ready" if set(required_types).issubset(present) and all(item["rubric_ids"] for item in items) else "blocked",
              "assessment_count": len(items),
              "types": sorted(present),
              "missing_types": sorted(set(required_types) - present),
              "capability_coverage": len({cap for item in items for cap in item["capability_ids"]}),
          },
      }
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build AssessmentBank 1.0.")
      parser.add_argument("--source-dir", required=True)
      parser.add_argument("--output")
      args = parser.parse_args()
      source_dir = Path(args.source_dir).expanduser().resolve()
      package = load_json(source_dir / "course_package.json")
      graph = load_json(source_dir / "capability_graph.json")
      bank = build_assessment_bank(package, graph)
      output = Path(args.output).expanduser().resolve() if args.output else source_dir / "assessment_bank.json"
      write_json(output, bank)
      write_json(source_dir / "capability_graph.json", graph)
      print(json.dumps({"output": str(output), "quality": bank["quality"]}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_capability_graph.py6.0 KBa7a2c46d4e39upstream ↗

  #!/usr/bin/env python3
  """Build a stable, validated capability graph from CoursePackage 1.0."""
  
  from __future__ import annotations
  
  import argparse
  import json
  from pathlib import Path
  from typing import Any
  
  from build_teacher_model import load_course_package
  from schema_utils import find_prerequisite_cycles, write_json
  from stable_ids import stable_id
  
  
  FIELD_TYPES = {
      "concepts": "concept",
      "diagnostics": "diagnosis",
      "methods": "decision",
      "workflows": "procedure",
      "templates": "production",
      "rubrics": "evaluation",
      "transfer_rules": "transfer",
  }
  
  
  def output_for(node_type: str, title: str) -> str:
      return {
          "concept": f"Explain and distinguish {title} without source cues.",
          "diagnosis": f"Detect the relevant signals and diagnose a case using {title}.",
          "decision": f"Select or reject {title} with explicit conditions and alternatives.",
          "procedure": f"Execute {title} and produce an inspectable artifact.",
          "production": f"Produce a usable artifact with {title}.",
          "evaluation": f"Evaluate an artifact with observable criteria from {title}.",
          "transfer": f"Transfer {title} to a changed context and name its limits.",
      }[node_type]
  
  
  def build_capability_graph(package: dict[str, Any]) -> dict[str, Any]:
      course_id = package["manifest"]["course_id"]
      nodes: list[dict[str, Any]] = []
      asset_to_capability: dict[str, str] = {}
      previous_by_type: dict[str, str] = {}
      for field, node_type in FIELD_TYPES.items():
          for index, asset in enumerate(package.get(field, [])):
              title = str(asset.get("title") or asset.get("summary") or f"{node_type}-{index + 1}")
              node_id = stable_id("capability", course_id, node_type, asset.get("id"), title)
              prerequisites: list[str] = []
              explicit = [str(item) for item in asset.get("related_capabilities", []) if item in asset_to_capability]
              prerequisites.extend(asset_to_capability[item] for item in explicit)
              if node_type != "concept" and previous_by_type.get("concept"):
                  prerequisites.append(previous_by_type["concept"])
              if node_type in {"procedure", "production", "evaluation", "transfer"} and previous_by_type.get("decision"):
                  prerequisites.append(previous_by_type["decision"])
              if node_type == "transfer" and previous_by_type.get("procedure"):
                  prerequisites.append(previous_by_type["procedure"])
              prerequisites = list(dict.fromkeys(item for item in prerequisites if item != node_id))
              nodes.append(
                  {
                      "id": node_id,
                      "name": title,
                      "type": node_type,
                      "description": str(asset.get("summary") or title),
                      "observable_outputs": list(asset.get("outputs") or [output_for(node_type, title)]),
                      "prerequisites": prerequisites,
                      "success_rubrics": [],
                      "common_errors": list(asset.get("failure_modes") or []),
                      "practice_tasks": [],
                      "assessment_items": [],
                      "source_evidence": list(asset.get("evidence") or []),
                      "difficulty_band": min(5, 1 + list(FIELD_TYPES.values()).index(node_type) // 2),
                      "transfer_dimensions": ["domain", "scale", "resources", "risk"] if node_type == "transfer" else [],
                      "human_review": asset.get("human_review", "recommended"),
                      "source_asset_ids": [asset["id"]],
                  }
              )
              asset_to_capability[asset["id"]] = node_id
              previous_by_type[node_type] = node_id
      edges: list[dict[str, Any]] = []
      for node in nodes:
          for prerequisite in node["prerequisites"]:
              edges.append(
                  {
                      "id": stable_id("edge", course_id, prerequisite, node["id"], "prerequisite_of"),
                      "from": prerequisite,
                      "to": node["id"],
                      "type": "prerequisite_of",
                      "confidence": "medium",
                      "human_review": "recommended",
                  }
              )
      cycles = find_prerequisite_cycles(nodes)
      node_ids = {node["id"] for node in nodes}
      dangling = [edge for edge in edges if edge["from"] not in node_ids or edge["to"] not in node_ids]
      uncovered = [node["id"] for node in nodes if not node["source_evidence"]]
      return {
          "schema_version": "1.0",
          "course_package_id": package["manifest"]["package_id"],
          "nodes": nodes,
          "edges": edges,
          "quality": {
              "status": "ready" if nodes and not cycles and not dangling else "blocked",
              "node_count": len(nodes),
              "edge_count": len(edges),
              "cycle_count": len(cycles),
              "cycles": cycles,
              "dangling_reference_count": len(dangling),
              "low_evidence_nodes": uncovered,
              "human_review_required": [node["id"] for node in nodes if node["human_review"] == "required"],
          },
      }
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build CapabilityGraph 1.0.")
      parser.add_argument("--course-package")
      parser.add_argument("--source-dir")
      parser.add_argument("--output")
      args = parser.parse_args()
      source_dir = Path(args.source_dir).expanduser().resolve() if args.source_dir else None
      package_path = Path(args.course_package).expanduser().resolve() if args.course_package else (source_dir / "course_package.json" if source_dir else None)
      if not package_path or not package_path.exists():
          raise SystemExit("provide --course-package or --source-dir")
      output = Path(args.output).expanduser().resolve() if args.output else package_path.parent / "capability_graph.json"
      graph = build_capability_graph(load_course_package(package_path))
      write_json(output, graph)
      print(json.dumps({"output": str(output), "quality": graph["quality"]}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_course_catalog.py7.7 KB4ffe38855916upstream ↗

  #!/usr/bin/env python3
  """Build a catalog for multiple distilled course workspaces."""
  
  from __future__ import annotations
  
  import argparse
  import datetime as dt
  import json
  from pathlib import Path
  from typing import Any
  
  
  ROOT = Path(__file__).resolve().parents[1]
  DEFAULT_BASE_DIR = ROOT / ".teacher-to-skills" / "courses"
  DEFAULT_OUTPUT_DIR = ROOT / "dist"
  
  
  def load_json(path: Path) -> dict[str, Any]:
      if not path.exists():
          return {}
      try:
          data = json.loads(path.read_text(encoding="utf-8"))
      except Exception:
          return {}
      return data if isinstance(data, dict) else {}
  
  
  def latest(patterns: list[str], root: Path) -> str:
      matches = []
      for pattern in patterns:
          matches.extend(root.glob(pattern))
      matches = sorted((path for path in matches if path.is_file()), key=lambda path: path.stat().st_mtime, reverse=True)
      return str(matches[0].relative_to(root)) if matches else ""
  
  
  def artifact_counts(course_dir: Path) -> dict[str, int]:
      return {
          "transcripts": len(list(course_dir.glob("transcripts/*_transcript.json"))),
          "visual_analyses": len(list(course_dir.glob("analysis/*_analysis.md"))),
          "screenshots": len([path for path in course_dir.glob("analysis/screenshots/**/*") if path.is_file()]),
          "keyframe_candidates": len([path for path in course_dir.glob("keyframe_candidates/**/*.jpg") if path.is_file()]),
          "keyframe_manifests": len(list(course_dir.glob("keyframe_selection/*_model_keyframes_manifest.json"))),
          "keyframe_summaries": len(list(course_dir.glob("keyframe_selection/model_keyframe_summary.md"))),
          "model_selected_keyframes": len([path for path in course_dir.glob("keyframes_model_selected/**/*.jpg") if path.is_file()]),
          "documents": len([path for path in course_dir.glob("documents/**/*") if path.is_file()]),
          "text_source_chunks": len(list(course_dir.glob("text_sources/chunks.jsonl"))),
          "text_evidence_cards": len(list(course_dir.glob("text_distillation/evidence_cards.jsonl"))),
          "text_synthesis": int((course_dir / "text_distillation" / "text_course_synthesis.md").exists()),
          "distillations": len(list(course_dir.glob("course_distillation_*.*"))),
      }
  
  
  def generated_skills_for_course(output_dir: Path, course_name: str, course_dir: Path) -> list[dict[str, Any]]:
      rows = []
      if not output_dir.exists():
          return rows
      for skill_dir in sorted(path for path in output_dir.iterdir() if path.is_dir()):
          manifest = load_json(skill_dir / "teacher_to_skills_manifest.json")
          source_dir = manifest.get("source", {}).get("source_dir") or manifest.get("source_dir")
          manifest_course = manifest.get("course", {}).get("name") or manifest.get("course_name")
          if manifest_course != course_name and source_dir != str(course_dir):
              continue
          roles = manifest.get("roles") or manifest.get("course", {}).get("roles") or manifest.get("modes") or manifest.get("course", {}).get("modes") or []
          rows.append(
              {
                  "skill_name": skill_dir.name,
                  "path": str(skill_dir),
                  "roles": roles,
                  "modes": roles,
                  "scope": manifest.get("scope", ""),
                  "evidence_strategy": manifest.get("evidence_strategy", ""),
                  "progress_strategy": manifest.get("progress_strategy", ""),
                  "generated_at": manifest.get("generated_at", ""),
              }
          )
      return rows
  
  
  def build_course_row(course_dir: Path, output_dir: Path) -> dict[str, Any]:
      progress = load_json(course_dir / "teacher_to_skills_progress.json")
      package = load_json(course_dir / "course_package.json")
      manifest = package.get("manifest") if isinstance(package.get("manifest"), dict) else {}
      course_name = progress.get("course_name") or manifest.get("course_name") or course_dir.name
      quality = package.get("quality") if isinstance(package.get("quality"), dict) else {}
      teacher = load_json(course_dir / "teacher_model.json")
      graph = load_json(course_dir / "capability_graph.json")
      practice = load_json(course_dir / "practice_bank.json")
      assessment = load_json(course_dir / "assessment_bank.json")
      mentor = load_json(course_dir / "mentor_package.json")
      readiness = load_json(course_dir / "mentor_readiness_audit.json")
      return {
          "course_name": course_name,
          "course_dir": str(course_dir),
          "status": {
              "next_stage": progress.get("next_stage"),
              "artifacts": progress.get("artifacts") or artifact_counts(course_dir),
              "quality": quality,
              "source_readiness": readiness.get("source_readiness", {}).get("status", "unknown"),
              "mentor_readiness": readiness.get("status", "missing"),
              "runtime_readiness": mentor.get("quality", {}).get("runtime_readiness", "missing"),
          },
          "schemas": {
              "course_package": package.get("schema_version"),
              "teacher_model": teacher.get("schema_version"),
              "capability_graph": graph.get("schema_version"),
              "practice_bank": practice.get("schema_version"),
              "assessment_bank": assessment.get("schema_version"),
              "mentor_package": mentor.get("schema_version"),
          },
          "apprenticeship_mode": mentor.get("manifest", {}).get("apprenticeship_mode", "none"),
          "capability_counts": {
              "capabilities": len(graph.get("nodes", [])),
              "tasks": len(practice.get("tasks", [])),
              "rubrics": len(practice.get("rubrics", [])),
              "assessments": len(assessment.get("items", [])),
          },
          "latest": {
              "distillation": latest(["course_distillation_*.md"], course_dir),
              "text_synthesis": latest(["text_distillation/text_course_synthesis.md"], course_dir),
              "course_package": "course_package.json" if (course_dir / "course_package.json").exists() else "",
              "progress": "teacher_to_skills_progress.json" if (course_dir / "teacher_to_skills_progress.json").exists() else "",
          },
          "generated_skills": generated_skills_for_course(output_dir, course_name, course_dir),
      }
  
  
  def build_catalog(base_dir: Path, output_dir: Path) -> dict[str, Any]:
      course_dirs = sorted(path for path in base_dir.iterdir() if path.is_dir()) if base_dir.exists() else []
      rows = [build_course_row(course_dir, output_dir) for course_dir in course_dirs]
      return {
          "schema_version": "1.0",
          "generated_at": dt.datetime.now().isoformat(timespec="seconds"),
          "base_dir": str(base_dir),
          "output_dir": str(output_dir),
          "course_count": len(rows),
          "courses": rows,
      }
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build a catalog for multiple distilled course workspaces.")
      parser.add_argument("--base-dir", default=str(DEFAULT_BASE_DIR), help="Course workspace root. Defaults to ./.teacher-to-skills/courses.")
      parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR), help="Generated skill output directory. Defaults to ./dist.")
      parser.add_argument("--catalog", help="Output catalog path. Defaults to <base-dir>/course_catalog.json.")
      args = parser.parse_args()
  
      base_dir = Path(args.base_dir).expanduser().resolve()
      output_dir = Path(args.output_dir).expanduser().resolve()
      catalog_path = Path(args.catalog).expanduser().resolve() if args.catalog else base_dir / "course_catalog.json"
      catalog_path.parent.mkdir(parents=True, exist_ok=True)
      catalog = build_catalog(base_dir, output_dir)
      catalog_path.write_text(json.dumps(catalog, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
      print(f"wrote {catalog_path}")
      print(f"courses: {catalog['course_count']}")
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_course_package.py17 KBdad81651526cupstream ↗

  #!/usr/bin/env python3
  """Normalize distilled course outputs into a generic CoursePackage JSON."""
  
  from __future__ import annotations
  
  import argparse
  import datetime as dt
  import json
  import re
  from pathlib import Path
  
  from migrate_course_package import migrate_package
  from schema_utils import issues_payload, load_json as load_schema_json, validate_schema, write_json
  
  
  SECTION_ALIASES = {
      "concepts": ["关键概念", "概念词汇", "词汇表", "术语"],
      "topics": ["跨课程主题", "主题图谱", "课程体系", "体系图"],
      "cases": ["完整示范", "讲评案例", "案例示范", "案例"],
      "methods": ["方法", "框架", "行动清单", "可执行行动"],
      "diagnostics": ["老师首先关注", "问题定性与诊断", "诊断判断", "诊断", "判断标准", "问题定位"],
      "workflows": ["执行流程", "工作流", "操作流程", "作业流程"],
      "rubrics": ["反馈纠错", "进阶与出师", "质量标准", "评价标准", "评分标准", "质检规则"],
      "templates": ["模板资产", "模板", "话术", "表格"],
      "transfer_rules": ["迁移规则", "应用迁移", "场景迁移"],
      "failure_modes": ["失效与误用", "失效", "误用", "反例"],
      "quotes": ["核心金句", "金句", "重要原话"],
      "study_paths": ["进阶与出师", "学习路径", "复习路径", "行动清单", "可执行行动"],
      "boundaries": ["不可复制背景", "边界", "风险", "注意事项", "限制"],
  }
  
  CARD_TYPE_TO_PACKAGE_FIELD = {
      "concept": "concepts",
      "method": "methods",
      "diagnostic": "diagnostics",
      "workflow": "workflows",
      "rubric": "rubrics",
      "template": "templates",
      "transfer": "transfer_rules",
      "failure_mode": "failure_modes",
      "quote": "quotes",
      "boundary": "boundaries",
      "attention_cue": "diagnostics",
      "problem_frame": "diagnostics",
      "decision_rule": "methods",
      "demonstration": "cases",
      "feedback_pattern": "rubrics",
      "progression_rule": "study_paths",
      "graduation_signal": "rubrics",
      "non_copyable_context": "boundaries",
  }
  
  
  class AssetRecord(dict):
      """Structured 1.0 asset with legacy substring-membership compatibility."""
  
      def __contains__(self, key: object) -> bool:
          if super().__contains__(key):
              return True
          if isinstance(key, str):
              return any(key in str(self.get(field) or "") for field in ["title", "summary", "legacy_text", "value"])
          return False
  
  
  def read_text(path: Path | None) -> str:
      if not path or not path.exists():
          return ""
      return path.read_text(encoding="utf-8", errors="ignore")
  
  
  def load_json(path: Path | None) -> object | None:
      if not path or not path.exists():
          return None
      return json.loads(path.read_text(encoding="utf-8"))
  
  
  def load_jsonl(path: Path) -> list[dict]:
      if not path.exists():
          return []
      rows = []
      for line in path.read_text(encoding="utf-8").splitlines():
          if not line.strip():
              continue
          item = json.loads(line)
          if isinstance(item, dict):
              rows.append(item)
      return rows
  
  
  def newest(source_dir: Path, pattern: str) -> Path | None:
      matches = sorted(source_dir.glob(pattern), key=lambda path: path.stat().st_mtime, reverse=True)
      return matches[0] if matches else None
  
  
  def section(text: str, aliases: list[str]) -> str:
      for alias in aliases:
          pattern = rf"(^##+\s*[^\n]*{re.escape(alias)}[^\n]*\n)([\s\S]*?)(?=^##+\s+|\Z)"
          match = re.search(pattern, text, flags=re.M)
          if match:
              return match.group(0).strip()
      return ""
  
  
  def bullets(text: str, limit: int = 80) -> list[str]:
      rows = []
      for line in text.splitlines():
          stripped = line.strip()
          if stripped.startswith(("- ", "* ")):
              value = stripped[2:].strip()
          elif re.match(r"^\d+[.)]\s+", stripped):
              value = re.sub(r"^\d+[.)]\s+", "", stripped).strip()
          else:
              continue
          if value:
              rows.append(value)
          if len(rows) >= limit:
              break
      return rows
  
  
  def add_unique(rows: list[str], value: str, limit: int = 120) -> None:
      value = value.strip()
      if not value or value in rows or len(rows) >= limit:
          return
      rows.append(value)
  
  
  def card_value(card: dict) -> str:
      title = str(card.get("title") or "").strip()
      body = str(card.get("quote") or card.get("summary") or "").strip()
      source = str(card.get("source_ref") or card.get("source_path") or "").strip()
      chunk = card.get("chunk_index")
      if title and body and title not in body:
          value = f"{title}：{body}"
      else:
          value = body or title
      if source:
          suffix = f"{source}#{chunk}" if chunk is not None else source
          value = f"{value}（{suffix}）"
      return value
  
  
  def load_text_cards(source_dir: Path) -> list[dict]:
      return load_jsonl(source_dir / "text_distillation" / "evidence_cards.jsonl")
  
  
  def merge_text_cards(package: dict, cards: list[dict]) -> None:
      for card in cards:
          card_type = card.get("card_type")
          value = card_value(card)
          target = CARD_TYPE_TO_PACKAGE_FIELD.get(card_type)
          if target:
              row = dict(card)
              row.setdefault("summary", row.get("quote") or value)
              row.setdefault("provenance", "direct_source" if card_type in {"quote", "demonstration"} else "source_grounded_synthesis")
              signature = (row.get("card_id"), row.get("title"), row.get("summary"))
              if not any(
                  isinstance(existing, dict)
                  and (existing.get("card_id"), existing.get("title"), existing.get("summary")) == signature
                  for existing in package[target]
              ):
                  package[target].append(row)
          elif card_type == "case":
              package["cases"].append(card)
          elif card_type in {"task", "open_question"}:
              package["learning_checks"].append(card)
  
  
  def normalize_lessons(data: object | None) -> list[dict]:
      if isinstance(data, list):
          items = data
      elif isinstance(data, dict):
          items = data.get("lesson_summaries") or data.get("lessons") or []
      else:
          items = []
      lessons = []
      for idx, item in enumerate(items, 1):
          if not isinstance(item, dict):
              lessons.append({"id": f"lesson-{idx:03d}", "title": str(item), "summary": "", "topics": [], "source": ""})
              continue
          title = item.get("title") or item.get("lesson_name") or item.get("video") or item.get("name") or f"Lesson {idx}"
          lessons.append(
              {
                  "id": item.get("id") or f"lesson-{idx:03d}",
                  "title": title,
                  "duration_minutes": item.get("duration_minutes"),
                  "summary": item.get("summary") or item.get("abstract") or "",
                  "topics": item.get("topics") or item.get("keywords") or [],
                  "source": item.get("source") or item.get("file") or "",
              }
          )
      return lessons
  
  
  def build_evidence(source_dir: Path) -> list[dict]:
      rows = []
      for path in sorted(source_dir.glob("transcripts/**/*.json")):
          rows.append({"type": "transcript", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("analysis/**/*_analysis.md")):
          rows.append({"type": "visual_analysis", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("analysis/screenshots/**/*")):
          if path.is_file():
              rows.append({"type": "screenshot", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("keyframe_selection/*_model_keyframes_manifest.json")):
          rows.append({"type": "model_keyframe_manifest", "path": str(path.relative_to(source_dir)), "granularity": "media"})
      for path in sorted(source_dir.glob("keyframe_selection/model_keyframe_summary.md")):
          rows.append({"type": "model_keyframe_summary", "path": str(path.relative_to(source_dir)), "granularity": "course"})
      for path in sorted(source_dir.glob("keyframes_model_selected/**/*")):
          if path.is_file():
              rows.append({"type": "model_selected_keyframe", "path": str(path.relative_to(source_dir)), "granularity": "frame"})
      for path in sorted(source_dir.glob("course_distillation_*.*")):
          rows.append({"type": "distillation", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("documents/**/*.md")):
          rows.append({"type": "document_ocr", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("documents/**/*.json")):
          rows.append({"type": "document_manifest", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("documents/**/*.html")) + sorted(source_dir.glob("documents/**/*.htm")):
          rows.append({"type": "document_raw_html", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("documents/**/*.txt")):
          rows.append({"type": "document_raw_text", "path": str(path.relative_to(source_dir)), "granularity": "file"})
      for path in sorted(source_dir.glob("text_sources/source_manifest.json")):
          rows.append({"type": "text_source_manifest", "path": str(path.relative_to(source_dir)), "granularity": "course"})
      for path in sorted(source_dir.glob("text_sources/chunks.jsonl")):
          rows.append({"type": "text_source_chunks", "path": str(path.relative_to(source_dir)), "granularity": "chunk"})
      for path in sorted(source_dir.glob("text_distillation/evidence_cards.jsonl")):
          rows.append({"type": "text_evidence_card", "path": str(path.relative_to(source_dir)), "granularity": "card"})
          for card in load_jsonl(path):
              rows.append(
                  {
                      "id": card.get("card_id"),
                      "type": "text",
                      "path": card.get("source_ref") or card.get("source_path") or str(path.relative_to(source_dir)),
                      "granularity": "card",
                      "source_id": card.get("source_id"),
                      "chunk_id": card.get("chunk_id"),
                      "card_id": card.get("card_id"),
                      "quote_summary": card.get("quote") or card.get("summary") or "",
                      "confidence": card.get("confidence", "medium"),
                  }
              )
      for path in sorted(source_dir.glob("text_distillation/text_course_synthesis.md")):
          rows.append({"type": "text_course_synthesis", "path": str(path.relative_to(source_dir)), "granularity": "course"})
      for path in sorted(source_dir.glob("text_distillation/text_distillation_quality.json")):
          rows.append({"type": "text_distillation_quality", "path": str(path.relative_to(source_dir)), "granularity": "course"})
      return rows
  
  
  def package_quality(package: dict) -> dict:
      counts = {
          "lessons": len(package["lessons"]),
          "concepts": len(package["concepts"]),
          "topics": len(package["topics"]),
          "methods": len(package["methods"]),
          "diagnostics": len(package["diagnostics"]),
          "workflows": len(package["workflows"]),
          "rubrics": len(package["rubrics"]),
          "templates": len(package["templates"]),
          "transfer_rules": len(package["transfer_rules"]),
          "failure_modes": len(package["failure_modes"]),
          "quotes": len(package["quotes"]),
          "evidence": len(package["evidence"]),
          "study_paths": len(package["study_paths"]),
          "boundaries": len(package["boundaries"]),
      }
      missing = [key for key, value in counts.items() if value == 0 and key not in {"boundaries"}]
      return {
          "counts": counts,
          "missing_recommended_fields": missing,
          "status": "usable" if counts["lessons"] or counts["evidence"] else "thin",
      }
  
  
  def distillation_audit_summary(source_dir: Path) -> dict | None:
      path = source_dir / "distillation_audit.json"
      if not path.exists():
          return None
      data = load_json(path)
      if not isinstance(data, dict):
          return None
      return {
          "json_path": "distillation_audit.json",
          "markdown_path": "distillation_audit.md" if (source_dir / "distillation_audit.md").exists() else "",
          "lesson_count": len(data.get("lessons") or []),
          "manual_review_required": (data.get("cross_validation_summary") or {}).get("manual_review_required", 0),
          "coverage_summary": data.get("coverage_summary") or {},
          "cross_validation_summary": data.get("cross_validation_summary") or {},
      }
  
  
  def build_legacy_package(course_name: str, source_dir: Path) -> dict:
      distillation_md = newest(source_dir, "course_distillation_*.md")
      distillation_json = newest(source_dir, "course_distillation_*.json")
      lesson_json = source_dir / "lesson_summaries.json"
      digest_text = read_text(distillation_md or source_dir / "course_digest.md")
      distillation_data = load_json(distillation_json)
      lessons = normalize_lessons(load_json(lesson_json) or distillation_data)
  
      sections = {key: section(digest_text, aliases) for key, aliases in SECTION_ALIASES.items()}
      package = {
          "schema_version": "0.1",
          "manifest": {
              "course_name": course_name,
              "source_dir": str(source_dir),
              "generated_at": dt.datetime.now().isoformat(timespec="seconds"),
              "distillation_markdown": str(distillation_md.relative_to(source_dir)) if distillation_md else "",
              "distillation_json": str(distillation_json.relative_to(source_dir)) if distillation_json else "",
          },
          "lessons": lessons,
          "concepts": bullets(sections["concepts"]),
          "topics": bullets(sections["topics"]),
          "cases": [],
          "methods": bullets(sections["methods"]),
          "diagnostics": bullets(sections["diagnostics"]),
          "workflows": bullets(sections["workflows"]),
          "rubrics": bullets(sections["rubrics"]),
          "templates": bullets(sections["templates"]),
          "transfer_rules": bullets(sections["transfer_rules"]),
          "failure_modes": bullets(sections["failure_modes"]),
          "learning_checks": [],
          "quotes": bullets(sections["quotes"]),
          "evidence": build_evidence(source_dir),
          "study_paths": bullets(sections["study_paths"]),
          "boundaries": bullets(sections["boundaries"]),
      }
      merge_text_cards(package, load_text_cards(source_dir))
      package["quality"] = package_quality(package)
      audit_summary = distillation_audit_summary(source_dir)
      if audit_summary:
          package["quality"]["distillation_audit"] = audit_summary
      return package
  
  
  def build_package(course_name: str, source_dir: Path) -> dict:
      """Build the canonical CoursePackage 1.0 while retaining legacy inputs."""
      legacy = build_legacy_package(course_name, source_dir)
      package, report = migrate_package(legacy, source_path=source_dir / "course_package.json")
      audit_summary = distillation_audit_summary(source_dir)
      if audit_summary:
          package["quality"]["distillation_audit"] = audit_summary
      package["manifest"]["build_report"] = {
          "migrator_version": report["target_schema_version"],
          "legacy_text_count": report["legacy_text_count"],
          "human_review_count": report["human_review_count"],
      }
      for field in ["concepts", "topics", "cases", "methods", "diagnostics", "workflows", "rubrics", "templates", "transfer_rules", "failure_modes", "boundaries", "quotes", "study_paths"]:
          package[field] = [AssetRecord(item) if isinstance(item, dict) else item for item in package[field]]
      return package
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build a generic CoursePackage from distilled course outputs.")
      parser.add_argument("--course-name", required=True)
      parser.add_argument("--source-dir", required=True)
      parser.add_argument("--output", help="Output JSON path. Defaults to <source-dir>/course_package.json.")
      args = parser.parse_args()
  
      source_dir = Path(args.source_dir).expanduser().resolve()
      if not source_dir.is_dir():
          raise SystemExit(f"source dir does not exist: {source_dir}")
      output = Path(args.output).expanduser().resolve() if args.output else source_dir / "course_package.json"
      package = build_package(args.course_name, source_dir)
      schema = load_schema_json(Path(__file__).resolve().parents[1] / "references" / "schemas" / "course_package.schema.json")
      validation = issues_payload(validate_schema(package, schema))
      if not validation["valid"]:
          raise SystemExit(json.dumps(validation, ensure_ascii=False, indent=2))
      write_json(output, package)
      write_json(source_dir / "course_package_build_report.json", {"schema_version": "1.0", "validation": validation})
      print(f"wrote {output}")
      print(json.dumps(package["quality"], ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_course_skill.py57 KBf0f0fce6e949upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/build_distillation_audit.py23 KBe4920219c07fupstream ↗

  #!/usr/bin/env python3
  """Build final quality and cross-validation audit files for a course workspace."""
  
  from __future__ import annotations
  
  import argparse
  import datetime as dt
  import json
  import re
  from pathlib import Path
  from typing import Any
  
  
  SCREENSHOT_RE = re.compile(r"\[SCREENSHOT\s+[^\]]+\]", flags=re.I)
  SLIDE_TERMS = ("ppt", "slide", "slides", "课件", "幻灯", "讲义", "courseware")
  BOARD_TERMS = ("板书", "白板", "blackboard", "whiteboard")
  DEMO_TERMS = ("演示", "demo", "操作", "software", "screen", "screens", "软件")
  DIAGRAM_TERMS = ("图表", "diagram", "chart", "流程图", "表格", "table")
  TERM_REVIEW_TERMS = ("术语", "专名", "错字", "校对", "同音", "ocr", "asr", "Teacher-to-Skills", "Skill")
  AUDIT_MODES = {"auto", "strict", "off"}
  
  
  def read_text(path: Path) -> str:
      return path.read_text(encoding="utf-8", errors="ignore")
  
  
  def load_json(path: Path) -> Any:
      return json.loads(read_text(path))
  
  
  def load_jsonl(path: Path) -> list[dict[str, Any]]:
      if not path.exists():
          return []
      rows: list[dict[str, Any]] = []
      for line in read_text(path).splitlines():
          if not line.strip():
              continue
          item = json.loads(line)
          if isinstance(item, dict):
              rows.append(item)
      return rows
  
  
  def rel(path: Path, root: Path) -> str:
      return str(path.relative_to(root))
  
  
  def normalize_stem(value: str) -> str:
      text = Path(value).stem if value else ""
      text = re.sub(r"(_transcript|_analysis|_model_keyframes_manifest)$", "", text)
      text = re.sub(r"[^0-9A-Za-z一-鿿]+", "", text).lower()
      return text
  
  
  def media_lesson_stem(value: str) -> str:
      """Return the actual media stem when capture names include a parent-folder prefix.
  
      The media capture stage prefixes transcript/keyframe names with the parent
      directory to avoid collisions, e.g. `4、需求采集与挖掘_需求挖掘...`.
      Visual analysis files are named from the MP4 stem only. For audit purposes
      these must resolve to the same lesson id.
      """
      text = Path(value).stem if value else ""
      text = re.sub(r"(_transcript|_analysis|_model_keyframes_manifest)$", "", text)
      if "_" in text:
          prefix, suffix = text.rsplit("_", 1)
          if "、" in prefix or "课" in prefix or re.match(r"^\d+", prefix):
              return suffix
      return text
  
  
  def lesson_id_from_media(value: str) -> str:
      return normalize_stem(media_lesson_stem(value))
  
  
  def inventory(paths: list[Path], root: Path) -> dict[str, Any]:
      return {"count": len(paths), "paths": [rel(path, root) for path in paths]}
  
  
  def transcript_id(path: Path, data: dict[str, Any]) -> str:
      return lesson_id_from_media(str(data.get("video_name") or data.get("video") or data.get("media") or path.name))
  
  
  def analysis_id(path: Path) -> str:
      return lesson_id_from_media(path.name)
  
  
  def manifest_id(path: Path, data: dict[str, Any]) -> str:
      return lesson_id_from_media(str(data.get("media") or path.name))
  
  
  def document_id(path: Path) -> str:
      return normalize_stem(path.name)
  
  
  def collect_transcripts(source_dir: Path) -> dict[str, dict[str, Any]]:
      rows: dict[str, dict[str, Any]] = {}
      for path in sorted(source_dir.glob("transcripts/**/*.json")):
          try:
              data = load_json(path)
          except Exception:
              data = {}
          if not isinstance(data, dict):
              data = {}
          lesson_id = transcript_id(path, data) or normalize_stem(path.name)
          rows[lesson_id] = {"path": path, "data": data}
      return rows
  
  
  def collect_analyses(source_dir: Path) -> dict[str, dict[str, Any]]:
      rows: dict[str, dict[str, Any]] = {}
      for path in sorted(source_dir.glob("analysis/**/*_analysis.md")):
          lesson_id = analysis_id(path)
          rows[lesson_id] = {"path": path, "text": read_text(path)}
      return rows
  
  
  def collect_manifests(source_dir: Path) -> dict[str, dict[str, Any]]:
      rows: dict[str, dict[str, Any]] = {}
      for path in sorted(source_dir.glob("keyframe_selection/*_model_keyframes_manifest.json")):
          try:
              data = load_json(path)
          except Exception:
              data = {}
          if not isinstance(data, dict):
              data = {}
          lesson_id = manifest_id(path, data)
          rows[lesson_id] = {"path": path, "data": data}
      return rows
  
  
  def collect_documents(source_dir: Path) -> dict[str, list[dict[str, Any]]]:
      rows: dict[str, list[dict[str, Any]]] = {}
      for pattern in ("documents/**/*.md", "documents/**/*.txt"):
          for path in sorted(source_dir.glob(pattern)):
              lesson_id = document_id(path)
              rows.setdefault(lesson_id, []).append({"path": path, "text": read_text(path)})
      return rows
  
  
  def collect_lesson_summaries(source_dir: Path) -> dict[str, dict[str, Any]]:
      path = source_dir / "lesson_summaries.json"
      if not path.exists():
          return {}
      try:
          data = load_json(path)
      except Exception:
          return {}
      items = data if isinstance(data, list) else data.get("lesson_summaries", []) if isinstance(data, dict) else []
      rows: dict[str, dict[str, Any]] = {}
      for idx, item in enumerate(items, 1):
          if not isinstance(item, dict):
              continue
          raw_id = item.get("video") or item.get("source") or item.get("file") or item.get("id") or f"lesson-{idx:03d}"
          rows[lesson_id_from_media(str(raw_id))] = item
      return rows
  
  
  def contains_any(text: str, terms: tuple[str, ...]) -> bool:
      lowered = text.lower()
      return any(term.lower() in lowered for term in terms)
  
  
  def numeric_duration_seconds(value: Any) -> float | None:
      if isinstance(value, (int, float)):
          return float(value)
      if isinstance(value, str) and re.fullmatch(r"\d+(\.\d+)?", value.strip()):
          return float(value)
      return None
  
  
  def matching_documents(lesson_id: str, documents: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]:
      matches: list[dict[str, Any]] = []
      for doc_id, rows in documents.items():
          if lesson_id and (lesson_id in doc_id or doc_id in lesson_id):
              matches.extend(rows)
      return matches
  
  
  def cross_validation_policy(audit_mode: str, source_count: int) -> dict[str, Any]:
      if audit_mode == "off":
          return {
              "mode": audit_mode,
              "required": False,
              "reason": "cross validation disabled by audit mode",
          }
      if audit_mode == "strict":
          return {
              "mode": audit_mode,
              "required": True,
              "reason": "strict audit mode requires review of missing or conflicting sources",
          }
      return {
          "mode": audit_mode,
          "required": source_count >= 2,
          "reason": "auto mode validates across sources only when comparable sources are available",
      }
  
  
  def related_text_trace(source_dir: Path, lesson_id: str) -> dict[str, list[str]]:
      chunks_path = source_dir / "text_sources" / "chunks.jsonl"
      cards_path = source_dir / "text_distillation" / "evidence_cards.jsonl"
      chunks = load_jsonl(chunks_path)
      cards = load_jsonl(cards_path)
      matched_chunks = []
      for chunk in chunks:
          source_ref = str(chunk.get("source_ref") or chunk.get("source_path") or "")
          if lesson_id and lesson_id in normalize_stem(source_ref):
              matched_chunks.append(rel(chunks_path, source_dir))
              break
      matched_cards = []
      for card in cards:
          source_ref = str(card.get("source_ref") or card.get("source_path") or "")
          title = str(card.get("title") or "")
          if lesson_id and (lesson_id in normalize_stem(source_ref) or lesson_id in normalize_stem(title)):
              matched_cards.append(rel(cards_path, source_dir))
              break
      return {"text_chunks": matched_chunks, "evidence_cards": matched_cards}
  
  
  def source_inventory(source_dir: Path) -> dict[str, Any]:
      transcripts = sorted(source_dir.glob("transcripts/**/*.json"))
      analyses = sorted(source_dir.glob("analysis/**/*_analysis.md"))
      screenshots = sorted(path for path in source_dir.glob("analysis/screenshots/**/*") if path.is_file())
      manifests = sorted(source_dir.glob("keyframe_selection/*_model_keyframes_manifest.json"))
      keyframes = sorted(path for path in source_dir.glob("keyframes_model_selected/**/*") if path.is_file())
      documents = sorted(source_dir.glob("documents/**/*.md")) + sorted(source_dir.glob("documents/**/*.txt"))
      text_sources = sorted(source_dir.glob("text_sources/**/*.json*"))
      text_distillation = sorted(source_dir.glob("text_distillation/**/*"))
      lesson_summaries = [source_dir / "lesson_summaries.json"] if (source_dir / "lesson_summaries.json").exists() else []
      packages = [source_dir / "course_package.json"] if (source_dir / "course_package.json").exists() else []
      return {
          "transcripts": inventory(transcripts, source_dir),
          "visual_analyses": inventory(analyses, source_dir),
          "screenshots": inventory(screenshots, source_dir),
          "keyframe_manifests": inventory(manifests, source_dir),
          "model_selected_keyframes": inventory(keyframes, source_dir),
          "documents": inventory(documents, source_dir),
          "text_sources": inventory([path for path in text_sources if path.is_file()], source_dir),
          "text_distillation": inventory([path for path in text_distillation if path.is_file()], source_dir),
          "lesson_summaries": inventory(lesson_summaries, source_dir),
          "packages": inventory(packages, source_dir),
      }
  
  
  def build_lesson(
      lesson_id: str,
      source_dir: Path,
      *,
      transcript: dict[str, Any] | None,
      analysis: dict[str, Any] | None,
      manifest: dict[str, Any] | None,
      documents: list[dict[str, Any]],
      summary: dict[str, Any] | None,
      audit_mode: str,
  ) -> dict[str, Any]:
      transcript_data = transcript.get("data", {}) if transcript else {}
      transcript_text = str(transcript_data.get("full_text") or transcript_data.get("text") or "")
      transcript_len = len(transcript_text.strip())
      duration = transcript_data.get("duration") or transcript_data.get("duration_seconds")
      numeric_duration = numeric_duration_seconds(duration)
      analysis_text = analysis.get("text", "") if analysis else ""
      document_text_len = sum(len(item["text"].strip()) for item in documents)
      keyframe_count = 0
      if manifest:
          data = manifest.get("data", {})
          keyframe_count = int(data.get("selected_count") or len(data.get("selected") or data.get("keyframes") or []))
      source_count = sum([bool(transcript), bool(analysis), bool(documents)])
      policy = cross_validation_policy(audit_mode, source_count)
  
      flags: list[str] = []
      if source_count >= 2:
          flags.append("supported_by_multiple_sources")
      elif source_count == 1:
          flags.append("single_source_available")
      if policy["required"] and not transcript:
          flags.append("missing_transcript")
      if policy["required"] and not analysis:
          flags.append("missing_visual_analysis")
      if policy["required"] and analysis and not transcript:
          flags.append("visual_only_claims")
      if policy["required"] and transcript and not analysis:
          flags.append("spoken_but_not_in_slides")
      if policy["required"] and analysis and contains_any(analysis_text, SLIDE_TERMS + DIAGRAM_TERMS + DEMO_TERMS) and not transcript_text:
          flags.append("slides_not_mentioned_in_transcript")
      if policy["required"] and transcript and analysis and contains_any(analysis_text, SLIDE_TERMS + DIAGRAM_TERMS + DEMO_TERMS):
          if not contains_any(transcript_text, ("图", "表", "课件", "ppt", "演示", "软件", "流程")):
              flags.append("transcript_visual_mismatch")
      if transcript_len < 20:
          flags.append("manual_review_required")
      if contains_any(transcript_text + "\n" + analysis_text + "\n" + "\n".join(item["text"] for item in documents), TERM_REVIEW_TERMS):
          flags.append("terminology_needs_review")
      if "terminology_needs_review" in flags or "missing_transcript" in flags or "missing_visual_analysis" in flags:
          flags.append("manual_review_required")
  
      flags = list(dict.fromkeys(flags))
      review_notes = []
      if "manual_review_required" in flags:
          review_notes.append("请人工复核本课的转录、课件/OCR、画面分析和术语一致性。")
      if "terminology_needs_review" in flags:
          review_notes.append("请重点检查专业术语、专名、同音误识别、ASR/OCR 错字。")
  
      trace = {
          "transcripts": [rel(transcript["path"], source_dir)] if transcript else [],
          "full_transcript": ["full_transcript.md"] if (source_dir / "full_transcript.md").exists() else [],
          "visual_analyses": [rel(analysis["path"], source_dir)] if analysis else [],
          "keyframe_manifests": [rel(manifest["path"], source_dir)] if manifest else [],
          "documents": [rel(item["path"], source_dir) for item in documents],
          "text_chunks": [],
          "evidence_cards": [],
          "lesson_summaries": ["lesson_summaries.json"] if summary else [],
      }
      trace.update(related_text_trace(source_dir, lesson_id))
  
      title = ""
      if summary:
          title = str(summary.get("title") or summary.get("lesson_name") or summary.get("video") or "")
      if not title:
          title = str(transcript_data.get("title") or transcript_data.get("video_name") or lesson_id)
  
      return {
          "lesson_id": lesson_id,
          "title": title,
          "media_ref": str(transcript_data.get("video_name") or lesson_id),
          "transcript_status": {
              "present": bool(transcript),
              "duration_seconds": duration,
              "text_length": transcript_len,
              "empty": bool(transcript and transcript_len == 0),
              "short_text": bool(transcript and transcript_len < 20),
              "probable_truncation": bool(numeric_duration and transcript_len and numeric_duration > 900 and transcript_len < 100),
              "terms_need_review": "terminology_needs_review" in flags,
          },
          "visual_status": {
              "present": bool(analysis),
              "analysis_length": len(analysis_text.strip()),
              "keyframe_count": keyframe_count,
              "screenshot_marker_count": len(SCREENSHOT_RE.findall(analysis_text)),
              "contains_slides_or_courseware": contains_any(analysis_text, SLIDE_TERMS),
              "contains_board_writing": contains_any(analysis_text, BOARD_TERMS),
              "contains_demo_or_software": contains_any(analysis_text, DEMO_TERMS),
              "contains_diagrams_or_tables": contains_any(analysis_text, DIAGRAM_TERMS),
          },
          "document_status": {
              "present": bool(documents),
              "paths": [rel(item["path"], source_dir) for item in documents],
              "text_length": document_text_len,
              "empty_or_short_text": bool(documents and document_text_len < 20),
          },
          "cross_validation": {
              "policy": policy,
              "source_count": source_count,
              "flags": flags,
              "review_recommendations": review_notes,
          },
          "traceability": trace,
      }
  
  
  def build_audit(course_name: str, source_dir: Path, audit_mode: str = "auto") -> dict[str, Any]:
      source_dir = source_dir.expanduser().resolve()
      if audit_mode not in AUDIT_MODES:
          raise ValueError(f"audit_mode must be one of {sorted(AUDIT_MODES)}")
      transcripts = collect_transcripts(source_dir)
      analyses = collect_analyses(source_dir)
      manifests = collect_manifests(source_dir)
      documents = collect_documents(source_dir)
      summaries = collect_lesson_summaries(source_dir)
  
      lesson_ids = sorted(set(transcripts) | set(analyses) | set(manifests) | set(summaries))
      lessons = []
      matched_doc_ids: set[str] = set()
      for lesson_id in lesson_ids:
          docs = matching_documents(lesson_id, documents)
          for doc in docs:
              matched_doc_ids.add(document_id(doc["path"]))
          lessons.append(
              build_lesson(
                  lesson_id,
                  source_dir,
                  transcript=transcripts.get(lesson_id),
                  analysis=analyses.get(lesson_id),
                  manifest=manifests.get(lesson_id),
                  documents=docs,
                  summary=summaries.get(lesson_id),
                  audit_mode=audit_mode,
              )
          )
  
      unmatched_documents = sum(1 for doc_id in documents if doc_id not in matched_doc_ids)
      manual_review_lessons = [lesson for lesson in lessons if "manual_review_required" in lesson["cross_validation"]["flags"]]
      multi_source_lessons = [lesson for lesson in lessons if "supported_by_multiple_sources" in lesson["cross_validation"]["flags"]]
      source_conflicts = [
          lesson
          for lesson in lessons
          if {"transcript_visual_mismatch", "slides_not_mentioned_in_transcript"} & set(lesson["cross_validation"]["flags"])
      ]
      evidence_cards = load_jsonl(source_dir / "text_distillation" / "evidence_cards.jsonl")
      teacher_card_types = ["attention_cue", "problem_frame", "decision_rule", "demonstration", "feedback_pattern", "progression_rule", "graduation_signal", "non_copyable_context"]
      teacher_counts = {kind: sum(card.get("card_type") == kind for card in evidence_cards) for kind in teacher_card_types}
      mentor_evidence_gaps = [kind for kind, count in teacher_counts.items() if count == 0]
  
      audit = {
          "schema_version": "1.0",
          "generated_at": dt.datetime.now().isoformat(timespec="seconds"),
          "course_name": course_name,
          "source_dir": str(source_dir),
          "audit_mode": audit_mode,
          "source_inventory": source_inventory(source_dir),
          "coverage_summary": {
              "expected_lessons": len(lesson_ids),
              "available_transcripts": len(transcripts),
              "available_visual_analyses": len(analyses),
              "available_document_groups": len(documents),
              "available_text_sources": len(load_jsonl(source_dir / "text_sources" / "chunks.jsonl")),
              "available_evidence_cards": len(load_jsonl(source_dir / "text_distillation" / "evidence_cards.jsonl")),
              "missing_transcripts": sum(1 for lesson in lessons if not lesson["transcript_status"]["present"]),
              "missing_visual_analyses": sum(1 for lesson in lessons if not lesson["visual_status"]["present"]),
              "unmatched_documents": unmatched_documents,
          },
          "mentor_evidence": {
              "card_counts": teacher_counts,
              "missing_teacher_evidence": mentor_evidence_gaps,
              "status": "ready" if not mentor_evidence_gaps else "partial",
              "note": "A mentor-evidence gap is distinct from source-capture failure. Missing teacher behavior must not be synthesized as teacher fact.",
          },
          "cross_validation_summary": {
              "multi_source_lessons": len(multi_source_lessons),
              "transcript_only_lessons": sum(
                  1
                  for lesson in lessons
                  if lesson["transcript_status"]["present"]
                  and not lesson["visual_status"]["present"]
                  and not lesson["document_status"]["present"]
              ),
              "visual_only_lessons": sum(
                  1 for lesson in lessons if lesson["visual_status"]["present"] and not lesson["transcript_status"]["present"]
              ),
              "document_only_groups": unmatched_documents,
              "source_conflicts": len(source_conflicts),
              "manual_review_required": len(manual_review_lessons) + unmatched_documents,
              "terminology_review_lessons": sum(
                  1 for lesson in lessons if "terminology_needs_review" in lesson["cross_validation"]["flags"]
              ),
          },
          "manual_review": [
              "人工校对建议：请复核专业术语、专名、同音误识别、OCR 错字，以及转录与课件不一致处。",
              "本报告为自动审计结果，不能替代对原始课程材料的人工核对。",
          ],
          "lessons": lessons,
      }
      if unmatched_documents:
          audit["manual_review"].append(f"发现 {unmatched_documents} 组未匹配到具体课程的文档/文字稿，请人工确认归属。")
      if mentor_evidence_gaps:
          audit["manual_review"].append("师承证据缺口：" + "、".join(mentor_evidence_gaps) + "。这些字段不得伪装为老师原意。")
      return audit
  
  
  def render_markdown(audit: dict[str, Any]) -> str:
      coverage = audit["coverage_summary"]
      cross = audit["cross_validation_summary"]
      lines = [
          f"# {audit['course_name']} Distillation Audit",
          "",
          "## Summary",
          "",
          f"- Expected lessons: {coverage['expected_lessons']}",
          f"- Transcripts: {coverage['available_transcripts']}",
          f"- Visual analyses: {coverage['available_visual_analyses']}",
          f"- Document groups: {coverage['available_document_groups']}",
          f"- Multi-source lessons: {cross['multi_source_lessons']}",
          f"- Manual review required: {cross['manual_review_required']}",
          f"- Mentor evidence status: {audit.get('mentor_evidence', {}).get('status', 'unknown')}",
          "",
          "## Manual Review",
          "",
      ]
      lines.extend(f"- {note}" for note in audit.get("manual_review", []))
      lines.extend(["", "## Lessons", ""])
      for lesson in audit.get("lessons", []):
          flags = ", ".join(lesson["cross_validation"]["flags"]) or "none"
          lines.extend(
              [
                  f"### {lesson['title'] or lesson['lesson_id']}",
                  "",
                  f"- Lesson id: `{lesson['lesson_id']}`",
                  f"- Transcript: {'present' if lesson['transcript_status']['present'] else 'missing'}; text length {lesson['transcript_status']['text_length']}",
                  f"- Visual analysis: {'present' if lesson['visual_status']['present'] else 'missing'}; keyframes {lesson['visual_status']['keyframe_count']}",
                  f"- Documents: {'present' if lesson['document_status']['present'] else 'missing'}",
                  f"- Cross-validation flags: {flags}",
              ]
          )
          for note in lesson["cross_validation"].get("review_recommendations", []):
              lines.append(f"- Review: {note}")
          lines.append("")
      return "\n".join(lines).rstrip() + "\n"
  
  
  def write_audit(course_name: str, source_dir: Path, audit_mode: str = "auto") -> dict[str, str]:
      audit = build_audit(course_name, source_dir, audit_mode=audit_mode)
      json_path = source_dir / "distillation_audit.json"
      markdown_path = source_dir / "distillation_audit.md"
      json_path.write_text(json.dumps(audit, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
      markdown_path.write_text(render_markdown(audit), encoding="utf-8")
      return {"json_path": json_path.name, "markdown_path": markdown_path.name}
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build final distillation quality and cross-validation audit files.")
      parser.add_argument("--course-name", required=True)
      parser.add_argument("--source-dir", required=True)
      parser.add_argument(
          "--audit-mode",
          choices=sorted(AUDIT_MODES),
          default="auto",
          help="Cross-validation policy: auto validates only when comparable sources exist, strict requires review, off records inventory only.",
      )
      args = parser.parse_args()
  
      source_dir = Path(args.source_dir).expanduser().resolve()
      if not source_dir.is_dir():
          raise SystemExit(f"source dir does not exist: {source_dir}")
      result = write_audit(args.course_name, source_dir, audit_mode=args.audit_mode)
      print(json.dumps(result, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_mentor_package.py6.2 KBdd2b54147e09upstream ↗

  #!/usr/bin/env python3
  """Build the stable MentorPackage runtime contract and protocol artifacts."""
  
  from __future__ import annotations
  
  import argparse
  import datetime as dt
  import json
  import shutil
  from pathlib import Path
  from typing import Any
  
  from build_mentor_readiness_audit import audit_markdown, build_readiness_audit
  from schema_utils import load_json, write_json
  from stable_ids import content_hash, stable_id
  
  
  ROOT = Path(__file__).resolve().parents[1]
  
  
  def graduation_policy(assessment_bank: dict[str, Any]) -> dict[str, Any]:
      return {
          "schema_version": "1.0",
          "policy_id": stable_id("assessment", assessment_bank.get("course_package_id"), "graduation-policy"),
          "required_assessment_types": assessment_bank.get("graduation_matrix", {}).get("required_assessment_types", []),
          "requirements": assessment_bank.get("graduation_matrix", {}).get("requirements", []),
          "evidence_requirements": {
              "independent_execution": True,
              "novel_context_transfer": True,
              "delayed_retention": True,
              "boundary_recognition": True,
              "real_world_artifact": True,
              "personal_skill_regression": True,
              "learner_approval": True,
          },
          "anti_shortcuts": [
              "Lesson completion, streaks, self-reported understanding, and one successful attempt are insufficient.",
              "Practice scores do not replace blind assessment results.",
              "Graduation from a course method is not professional licensure or certification.",
          ],
          "post_graduation_role": ["source consultant", "counterexample provider", "advanced sparring partner", "source update notifier"],
      }
  
  
  def build_mentor_package(source_dir: Path, *, requested_mode: str = "full", evidence_strategy: str = "standard") -> dict[str, Any]:
      package = load_json(source_dir / "course_package.json")
      teacher = load_json(source_dir / "teacher_model.json")
      assessment = load_json(source_dir / "assessment_bank.json")
      protocol_source = ROOT / "references" / "apprenticeship-protocol.md"
      shutil.copy2(protocol_source, source_dir / "mentor_protocol.md")
      micro_lesson_source = ROOT / "references" / "micro-lesson-protocol.md"
      shutil.copy2(micro_lesson_source, source_dir / "micro_lesson_protocol.md")
      policy = graduation_policy(assessment)
      write_json(source_dir / "graduation_policy.json", policy)
      audit = build_readiness_audit(source_dir)
      write_json(source_dir / "mentor_readiness_audit.json", audit)
      (source_dir / "mentor_readiness_audit.md").write_text(audit_markdown(audit), encoding="utf-8")
      allowed = audit["apprenticeship_mode_allowed"]
      actual_mode = requested_mode if requested_mode == "none" else ("full" if requested_mode == "full" and allowed == "full" else "guided" if allowed in {"full", "guided"} else "none")
      mentor_id = stable_id("mentor_package", package["manifest"]["package_id"], teacher["teacher_id"], actual_mode, evidence_strategy)
      payload = {
          "schema_version": "1.0",
          "manifest": {
              "mentor_package_id": mentor_id,
              "course_package_id": package["manifest"]["package_id"],
              "teacher_model_id": teacher["teacher_id"],
              "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
              "generator_version": "teacher-to-skills-1.0",
              "evidence_strategy": evidence_strategy,
              "apprenticeship_mode": actual_mode,
              "requested_apprenticeship_mode": requested_mode,
              "downgraded": requested_mode != actual_mode,
              "package_hashes": {
                  name: content_hash(load_json(source_dir / name))
                  for name in ["course_package.json", "teacher_model.json", "capability_graph.json", "practice_bank.json", "assessment_bank.json"]
              },
          },
          "learning_contract_template": {
              "desired_capabilities": [],
              "real_project": "",
              "constraints": {"time": "", "tools": [], "risk": [], "access": []},
              "teaching_preferences": {
                  "explanation_density": "normal",
                  "visual_mode": "auto",
                  "svg": "when-useful",
                  "formative_question_count": 2,
                  "question_pacing": "together",
              },
              "expected_artifacts": [],
              "graduation_target": policy["evidence_requirements"],
              "baseline_diagnostic_required": True,
          },
          "capability_graph_ref": "capability_graph.json",
          "teacher_model_ref": "teacher_model.json",
          "practice_bank_ref": "practice_bank.json",
          "assessment_bank_ref": "assessment_bank.json",
          "mentor_protocol_ref": "mentor_protocol.md",
          "micro_lesson_protocol_ref": "micro_lesson_protocol.md",
          "graduation_policy_ref": "graduation_policy.json",
          "learner_state_schema_ref": "schemas/apprenticeship_state.schema.json",
          "episode_schema_ref": "schemas/practice_episode.schema.json",
          "personal_skill_schema_ref": "schemas/personal_skill_candidate.schema.json",
          "quality": {
              "mentor_readiness": audit["status"],
              "runtime_readiness": "ready" if actual_mode != "none" else "blocked",
              "blockers": audit["blockers"],
              "human_review_required": audit["human_review_required"],
          },
      }
      write_json(source_dir / "mentor_package.json", payload)
      return payload
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build MentorPackage 1.0.")
      parser.add_argument("--source-dir", required=True)
      parser.add_argument("--apprenticeship", choices=["none", "guided", "full"], default="full")
      parser.add_argument("--evidence", choices=["standard", "strict"], default="standard")
      parser.add_argument("--output")
      args = parser.parse_args()
      source_dir = Path(args.source_dir).expanduser().resolve()
      payload = build_mentor_package(source_dir, requested_mode=args.apprenticeship, evidence_strategy=args.evidence)
      if args.output:
          write_json(Path(args.output).expanduser().resolve(), payload)
      print(json.dumps({"output": str(source_dir / "mentor_package.json"), "manifest": payload["manifest"], "quality": payload["quality"]}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_mentor_readiness_audit.py9.1 KB26dfee82b0e5upstream ↗

  #!/usr/bin/env python3
  """Audit whether a workspace can truthfully support full apprenticeship."""
  
  from __future__ import annotations
  
  import argparse
  import json
  import re
  from pathlib import Path
  from typing import Any
  
  from schema_utils import load_json, write_json
  
  
  PLACEHOLDER_PATTERNS = [
      re.compile(pattern, re.I)
      for pattern in [r"\bTODO\b", r"add .+ here", r"placeholder", r"implement here", r"rest of code", r"similar to above"]
  ]
  
  
  def placeholder_hits(paths: list[Path]) -> list[str]:
      hits: list[str] = []
      for path in paths:
          if not path.exists() or not path.is_file():
              continue
          text = path.read_text(encoding="utf-8", errors="ignore")
          if not text.strip() or any(pattern.search(text) for pattern in PLACEHOLDER_PATTERNS):
              hits.append(str(path))
      return hits
  
  
  def build_readiness_audit(source_dir: Path) -> dict[str, Any]:
      required = {
          "course_package": source_dir / "course_package.json",
          "teacher_model": source_dir / "teacher_model.json",
          "capability_graph": source_dir / "capability_graph.json",
          "practice_bank": source_dir / "practice_bank.json",
          "assessment_bank": source_dir / "assessment_bank.json",
          "mentor_protocol": source_dir / "mentor_protocol.md",
          "micro_lesson_protocol": source_dir / "micro_lesson_protocol.md",
          "graduation_policy": source_dir / "graduation_policy.json",
      }
      missing_files = [name for name, path in required.items() if not path.exists()]
      data = {
          name: load_json(path)
          for name, path in required.items()
          if path.exists() and path.suffix == ".json" and name not in {"graduation_policy"}
      }
      teacher_quality = (data.get("teacher_model") or {}).get("quality", {})
      graph_quality = (data.get("capability_graph") or {}).get("quality", {})
      practice_quality = (data.get("practice_bank") or {}).get("quality", {})
      assessment_quality = (data.get("assessment_bank") or {}).get("quality", {})
      graph_nodes = (data.get("capability_graph") or {}).get("nodes", [])
      practice_tasks = (data.get("practice_bank") or {}).get("tasks", [])
      rubrics = (data.get("practice_bank") or {}).get("rubrics", [])
      learning_units = (data.get("practice_bank") or {}).get("learning_units", [])
      assessments = (data.get("assessment_bank") or {}).get("items", [])
      blockers: list[str] = []
      blockers.extend(f"missing artifact: {name}" for name in missing_files)
      if teacher_quality.get("status") != "ready":
          blockers.append("TeacherModel lacks complete source-grounded cues, decisions, demonstrations, feedback, or graduation signals")
      if graph_quality.get("status") != "ready" or graph_quality.get("cycle_count") or graph_quality.get("dangling_reference_count"):
          blockers.append("CapabilityGraph is incomplete, cyclic, or contains dangling references")
      if graph_nodes and any(not node.get("practice_tasks") for node in graph_nodes):
          blockers.append("one or more capability nodes have no practice task")
      if graph_nodes and any(not node.get("assessment_items") for node in graph_nodes):
          blockers.append("one or more capability nodes have no assessment")
      if practice_quality.get("status") != "ready" or not practice_tasks or not rubrics:
          blockers.append("PracticeBank lacks complete tasks, rubrics, or hint ladders")
      if graph_nodes and (len(learning_units) != len(graph_nodes) or any(len(unit.get("questions") or []) != 2 for unit in learning_units)):
          blockers.append("PracticeBank lacks one complete two-question micro-lesson for every capability")
      if assessment_quality.get("status") != "ready" or not assessments:
          blockers.append("AssessmentBank does not cover the graduation matrix")
      placeholder_files = placeholder_hits([source_dir / "mentor_protocol.md", source_dir / "micro_lesson_protocol.md", source_dir / "graduation_policy.json"])
      if placeholder_files:
          blockers.append("mentor protocol or graduation policy contains placeholder content")
      package_integrity = ((data.get("course_package") or {}).get("quality") or {}).get("integrity", {})
      if package_integrity.get("dangling_reference_count", 0):
          blockers.append("CoursePackage contains dangling references")
  
      status = "ready" if not blockers else "partial" if data.get("teacher_model") and graph_nodes else "blocked"
      inferred_records = max(0, int(teacher_quality.get("record_count", 0)) - int(teacher_quality.get("source_grounded_record_count", 0)))
      audit = {
          "schema_version": "1.0",
          "status": status,
          "source_readiness": {
              "status": "ready" if data.get("course_package") and (data["course_package"].get("sources") or data["course_package"].get("evidence")) else "partial",
              "evidence_count": len((data.get("course_package") or {}).get("evidence", [])),
              "unsupported_claim_count": package_integrity.get("unsupported_claim_count", 0),
          },
          "teacher_model": {
              "status": teacher_quality.get("status", "missing"),
              "source_grounded_records": teacher_quality.get("source_grounded_record_count", 0),
              "inferred_records": inferred_records,
              "grounded_ratio": teacher_quality.get("grounded_ratio", 0),
              "missing_teacher_evidence": teacher_quality.get("missing_teacher_evidence", []),
          },
          "capability_graph": {
              "nodes": len(graph_nodes),
              "edges": len((data.get("capability_graph") or {}).get("edges", [])),
              "cycles": graph_quality.get("cycle_count", 0),
              "dangling_references": graph_quality.get("dangling_reference_count", 0),
          },
          "practice": {"tasks": len(practice_tasks), "rubrics": len(rubrics), "learning_units": len(learning_units), "two_question_units": sum(len(unit.get("questions") or []) == 2 for unit in learning_units), "capability_coverage": practice_quality.get("capability_coverage", 0)},
          "assessment": {"items": len(assessments), "types": assessment_quality.get("types", []), "missing_types": assessment_quality.get("missing_types", [])},
          "integrity": {
              "placeholder_files": placeholder_files,
              "dangling_references": package_integrity.get("dangling_reference_count", 0) + graph_quality.get("dangling_reference_count", 0),
              "source_conflicts": package_integrity.get("source_conflict_count", 0),
          },
          "blockers": list(dict.fromkeys(blockers)),
          "human_review_required": sorted(set(teacher_quality.get("human_review_required", [])) | set(((data.get("course_package") or {}).get("quality") or {}).get("mentor_readiness", {}).get("human_review_required", []))),
          "apprenticeship_mode_allowed": "full" if status == "ready" else "guided" if status == "partial" else "none",
      }
      return audit
  
  
  def audit_markdown(audit: dict[str, Any]) -> str:
      lines = [
          "# Mentor Readiness Audit",
          "",
          f"- Status: **{audit['status']}**",
          f"- Allowed apprenticeship mode: **{audit['apprenticeship_mode_allowed']}**",
          f"- Source evidence: {audit['source_readiness']['evidence_count']}",
          f"- Teacher grounded ratio: {audit['teacher_model']['grounded_ratio']}",
          f"- Capability nodes: {audit['capability_graph']['nodes']}",
          f"- Practice tasks / rubrics: {audit['practice']['tasks']} / {audit['practice']['rubrics']}",
          f"- Two-question micro-lessons: {audit['practice']['two_question_units']}",
          f"- Assessments: {audit['assessment']['items']}",
          "",
          "## Blockers",
          "",
      ]
      lines.extend(f"- {item}" for item in audit["blockers"] or ["None."])
      lines.extend(["", "## Human review", ""])
      lines.extend(f"- `{item}`" for item in audit["human_review_required"] or ["No required items."])
      lines.extend(["", "Full apprenticeship is only emitted when this audit is ready. Guided mode must report the listed evidence gaps.", ""])
      return "\n".join(lines)
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build mentor readiness audit.")
      parser.add_argument("--source-dir", required=True)
      parser.add_argument("--output-json")
      parser.add_argument("--output-markdown")
      parser.add_argument(
          "--mode",
          choices=["auto", "strict"],
          default="auto",
          help="In strict mode, exit non-zero unless full apprenticeship is ready.",
      )
      args = parser.parse_args()
      source_dir = Path(args.source_dir).expanduser().resolve()
      audit = build_readiness_audit(source_dir)
      json_path = Path(args.output_json).expanduser().resolve() if args.output_json else source_dir / "mentor_readiness_audit.json"
      md_path = Path(args.output_markdown).expanduser().resolve() if args.output_markdown else source_dir / "mentor_readiness_audit.md"
      write_json(json_path, audit)
      md_path.write_text(audit_markdown(audit), encoding="utf-8")
      print(json.dumps({"json": str(json_path), "markdown": str(md_path), "status": audit["status"], "allowed": audit["apprenticeship_mode_allowed"]}, ensure_ascii=False, indent=2))
      if args.mode == "strict" and audit["status"] != "ready":
          raise SystemExit(2)
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_multi_course_package.py10 KB224396701e66upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/build_okf_bundle.py14 KBfa62145f852cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/build_personal_skill_candidate.py5.6 KB6ef87c968e40upstream ↗

  #!/usr/bin/env python3
  """Build a provenance-preserving Personal Skill candidate from repeated episodes."""
  
  from __future__ import annotations
  
  import argparse
  import hashlib
  import json
  import re
  from pathlib import Path
  
  from runtime_state import highest_hint, read_json, read_jsonl, write_json
  
  
  def slugify(value: str) -> str:
      return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")[:63] or "personal-practice"
  
  
  def eligible(episodes: list[dict], capability_id: str) -> tuple[bool, list[str], list[dict]]:
      relevant = [item for item in episodes if capability_id in item.get("capability_ids", [])]
      successes = [item for item in relevant if item.get("outcome", {}).get("task_result") == "pass"]
      contexts = {str(item.get("context", {}).get("project") or item.get("context", {}).get("environment") or "") for item in successes}
      failures = [item for item in relevant if item.get("outcome", {}).get("task_result") in {"revise", "fail"}]
      reasons = []
      if len(successes) < 3:
          reasons.append("requires at least three successful practices")
      if len({item for item in contexts if item}) < 2:
          reasons.append("requires at least two distinct contexts")
      if not any(highest_hint(item) == "H0" for item in successes):
          reasons.append("requires at least one no-hint execution")
      if not failures and not any(item.get("errors") for item in relevant):
          reasons.append("requires a failure, counterexample, or recorded error")
      return not reasons, reasons, relevant
  
  
  def build_candidate(mentor: dict, graph: dict, episodes: list[dict], capability_id: str, name: str) -> dict:
      ok, reasons, relevant = eligible(episodes, capability_id)
      if not ok:
          raise ValueError("; ".join(reasons))
      node = next(item for item in graph.get("nodes", []) if item["id"] == capability_id)
      successes = [item for item in relevant if item.get("outcome", {}).get("task_result") == "pass"]
      failures = [item for item in relevant if item.get("outcome", {}).get("task_result") != "pass" or item.get("errors")]
      candidate_name = slugify(name)
      episode_ids = [item["episode_id"] for item in relevant]
      return {
          "schema_version": "1.0",
          "candidate_id": "candidate_" + hashlib.sha256(
              "\x1f".join([mentor["manifest"]["mentor_package_id"], capability_id, *episode_ids]).encode("utf-8")
          ).hexdigest()[:16],
          "name": candidate_name,
          "purpose": node["description"],
          "triggers": [f"Use when the user needs to {output.lower()}" for output in node.get("observable_outputs", [])[:3]],
          "preconditions": [f"Capability evidence for {item}" for item in node.get("prerequisites", [])],
          "procedure": ["Inspect the real context and constraints", "Select and execute the learned capability", "Evaluate the output against observable criteria", "Record boundary conditions and real-world result"],
          "tools": [],
          "outputs": node.get("observable_outputs", []),
          "evaluator": {"critical_invariants": ["observable artifact", "source/personal provenance separation", "known failure handled"], "evidence_episodes": [item["episode_id"] for item in successes]},
          "failure_modes": [error for item in failures for error in item.get("errors", [])] or ["Fails when applicability conditions or required evidence are absent"],
          "source_lineage": {"mentor_package_id": mentor["manifest"]["mentor_package_id"], "teacher_rules_used": [], "evidence_pointers": node.get("source_evidence", [])},
          "personal_adaptations": {"changes": [], "rationale": "To be completed and approved by the learner before promotion.", "episode_evidence": episode_ids},
          "real_world_evidence": [item.get("outcome", {}).get("real_world_result") for item in successes if item.get("outcome", {}).get("real_world_result")],
          "regression_tests": [
              {"prompt": f"Apply {node['name']} in a familiar context without hints.", "expected_invariants": ["observable output", "all critical criteria pass"]},
              {"prompt": f"Apply {node['name']} after one major constraint changes.", "expected_invariants": ["transfer rationale", "boundary named", "no unsupported teacher claim"]},
              {"prompt": f"Identify a case where {node['name']} should not be used.", "expected_invariants": ["contraindication", "safe alternative"]},
          ],
          "confidence": "candidate",
          "promotion_status": "watch",
      }
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Build a Personal Skill candidate from external episodes.")
      parser.add_argument("--state-dir", required=True)
      parser.add_argument("--references-dir", default="../references")
      parser.add_argument("--capability-id", required=True)
      parser.add_argument("--name", required=True)
      args = parser.parse_args()
      state_dir = Path(args.state_dir).expanduser().resolve()
      references = (Path(__file__).resolve().parent / args.references_dir).resolve() if not Path(args.references_dir).is_absolute() else Path(args.references_dir)
      try:
          candidate = build_candidate(read_json(references / "mentor_package.json", {}), read_json(references / "capability_graph.json", {}), read_jsonl(state_dir / "practice_episodes.jsonl"), args.capability_id, args.name)
      except ValueError as exc:
          print(json.dumps({"eligible": False, "reason": str(exc)}, ensure_ascii=False, indent=2))
          raise SystemExit(1)
      output = state_dir / "personal_skill_candidates" / f"{candidate['name']}.json"
      write_json(output, candidate)
      print(json.dumps({"eligible": True, "output": str(output), "promotion_status": candidate["promotion_status"]}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/build_practice_bank.py19 KB9ccad5222eeaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/build_teacher_model.py11 KB9d6f6ce6387aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/distill_course.py26 KBe838261f0079upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/distill_text_course.py16 KB7dd4dc7b5e1cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/fetch_course_evidence.py6.7 KBa666a8d046bcupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/initialize_apprenticeship.py3.6 KB0610b4a0ec6dupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/llm_client.py7.8 KB6b78fbf56e44upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/migrate_course_package.py17 KB68a4cf6d587cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/parse_mineru_documents.py9.2 KB8529a65f29f7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/progress.py7.1 KB7f062d719a80upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/rebuild_mastery_state.py1.0 KBea57698c6aa5upstream ↗

  #!/usr/bin/env python3
  """Rebuild derived MasteryState from append-only PracticeEpisodes."""
  
  from __future__ import annotations
  
  import argparse
  import json
  from pathlib import Path
  
  from runtime_state import read_json, read_jsonl, rebuild_mastery, write_json
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Rebuild mastery_state.json from practice_episodes.jsonl.")
      parser.add_argument("--state-dir", required=True)
      args = parser.parse_args()
      state_dir = Path(args.state_dir).expanduser().resolve()
      state = read_json(state_dir / "apprenticeship_state.json", {})
      mastery = rebuild_mastery(
          read_jsonl(state_dir / "practice_episodes.jsonl"),
          mentor_package_id=state["mentor_package_id"],
          learner_id=state["learner_id"],
      )
      write_json(state_dir / "mastery_state.json", mastery)
      print(json.dumps({"output": str(state_dir / "mastery_state.json"), "capability_count": len(mastery["capabilities"])}, ensure_ascii=False, indent=2))
  
  
  if __name__ == "__main__":
      main()
  

• scripts/record_practice_episode.py2.2 KB72c82abf3bdfupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/render_learning_svg.py9.3 KB3c2da7977648upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/run_course_pipeline.py22 KBd097b4d6954bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/runtime_state.py7.0 KBdf484f9adfdaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/schedule_retrieval.py3.0 KB2e2a58841619upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/schema_utils.py6.1 KBb2fd70db575eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/select_next_practice.py4.0 KB38350478b790upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/select_video_keyframes.py22 KBa41bd7118080upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/stable_ids.py2.1 KBd09b50493db8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/text_sources.py8.3 KB8f7e34e0386eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/transcribe_video.py15 KBa810c7f203caupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/validate_generated_skill.py7.7 KB7b30bc68f7c3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/validate_learner_state.py2.9 KB7e2dfd606293upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/validate_mentor_package.py4.0 KB9a8309ee4a71upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• scripts/validate_teacher_to_skills_package.py2.8 KBf520991c2e20upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/apprenticeship-protocol.md3.8 KB505fa84088b6upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/external-learner-store-contract.md1.4 KBe5459b95bab0upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/graduation-policy.md731 B231441f34681upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/mastery-policy.md884 Bea58a5203a0bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/micro-lesson-protocol.md5.9 KB7d6850c02b09upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/provenance-policy.md1003 B08f161b7f115upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/runtime.md13 KB45975e0e4fa1upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/apprenticeship_state.schema.json2.2 KB650318c00e76upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/assessment_bank.schema.json859 B1a152afa5753upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/capability_graph.schema.json874 B72000ce4d47eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/course_package.schema.json1.8 KBd205210564c7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/mastery_state.schema.json767 Bc7e370cc9552upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/mentor_package.schema.json1.4 KBc0714b9ad689upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/personal_skill_candidate.schema.json1.5 KBabdc5485401eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/practice_bank.schema.json1.2 KB3fdbb55bd3bfupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/practice_episode.schema.json1.5 KB86d9a80279d7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/schemas/teacher_model.schema.json1.2 KBcb6389172fc0upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• .env.example1.1 KB5662e975d075upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• .gitignore5.3 KBfcfd14cbfbaeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• agents/openai.yaml403 B9bc612088b74upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• agents/openclaw.yaml1.1 KBe06da06ec90cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• CHANGELOG.md3.2 KB6f36133ac34fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/distill-en.svg8.9 KB29652f80db14upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/distill-zh.svg8.3 KB6b17ba2e549eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/hero-en.svg8.3 KBb6c57fc78bbaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/hero-zh.svg8.3 KB64ea1ef52294upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/journey-en.svg8.4 KB768f141a568bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/journey-zh.svg8.5 KB58fa863586cbupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/layers-en.svg9.7 KBe94e86338403upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/layers-zh.svg9.7 KB2d4dff13bf23upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/lesson-en.svg10 KB7ad129d21549upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/lesson-zh.svg10 KB3aabe79aa9d3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/lsgogroup-wechat-public-code.jpg40 KB4d51c2c6aceeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/img/src/build_images.py31 KB8ca8aeb43589upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/install.md4.3 KBadd254245d44upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• examples/command-line-flow.txt1.6 KB7fb9a5cfdcd5upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• examples/generated-skill-structure.txt1.7 KBbb5cdcd1afddupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• install_as_skill.sh6.2 KB73eb38627ca8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• LICENSE11 KBd64569567334upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.en.md11 KB771b119045fdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• requirements.txt67 Bb9fccea73fa4upstream ↗

  openai>=1.0.0
  python-dotenv>=1.0.0
  requests>=2.31.0
  pillow>=10.0.0
  

• tests/test_micro_lesson_runtime.py12 KBae5ab458e0cdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/juneyaooo-teacher-to-skills.md