feat(writer): add WriteOrchestrator.generate (vertical slice, real DocxInjector.inject)

2026-08-13 10:32:35 +08:00
parent f175203a23
commit 08e311b0a1
5 changed files with 183 additions and 0 deletions
@@ -0,0 +1,76 @@
+"""Phase 5 垂直切片运行脚本（CLI）。
+
+用法:
+  python scripts/run_phase5_slice.py --template <PATH> --output <PATH> [--fake]
+
+- --fake: 使用内置 FakeEngine（无需 LLM 客户端/API Key），用于验证管线打通。
+- 不带 --fake: 使用真实 InferenceEngine()，需先配置 LLM 客户端（人工质量门禁）。
+
+管线: WordTemplateParser → StructuredSource → WriteOrchestrator.generate → docx。
+"""
+from __future__ import annotations
+
+import argparse
+from types import SimpleNamespace
+
+from docx import Document
+
+from genesis.data_models import StructuredSource
+from genesis.parsers.word_template_parser import WordTemplateParser
+from genesis.writer.orchestrator import WriteOrchestrator
+
+
+class FakeEngine:
+    """切片验证用假引擎：返回固定章节内容，避免真实 LLM 调用。"""
+
+    def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
+        return SimpleNamespace(
+            data={
+                "title": variables.get("title", "章"),
+                "blocks": [{"type": "paragraph", "text": f"【{variables.get('title','')}】自动生成的内容（fake 模式）"}],
+            },
+            status="ok",
+        )
+
+
+def _build_source(template_path: str) -> StructuredSource:
+    parsed = WordTemplateParser().parse(template_path)
+    return StructuredSource(
+        tables=[],
+        template=parsed,
+        rule_docs=[],
+        image_analyses=[],
+        existing_system=None,
+        comments=[],
+    )
+
+
+def _main() -> None:
+    parser = argparse.ArgumentParser(description="Phase 5 垂直切片运行脚本")
+    parser.add_argument("--template", required=True, help="概要设计模板 docx 路径")
+    parser.add_argument("--output", required=True, help="输出 docx 路径")
+    parser.add_argument("--fake", action="store_true", help="使用 FakeEngine（无 LLM）")
+    parser.add_argument("--samples-dir", default="samples", help="RAG 样本目录")
+    args = parser.parse_args()
+
+    source = _build_source(args.template)
+    engine = FakeEngine() if args.fake else None  # None → 真实 InferenceEngine()（需客户端配置）
+
+    orch = WriteOrchestrator()
+    contents = orch.generate(
+        source,
+        args.output,
+        samples_dir=args.samples_dir,
+        engine=engine,
+        template_path=args.template,
+    )
+    loaded = Document(args.output)
+    joined = "\n".join(p.text for p in loaded.paragraphs)
+    print(f"[slice] 生成章节数: {len(contents)}")
+    print(f"[slice] 输出路径: {args.output}")
+    print(f"[slice] 注入校验: {'OK' if any('自动生成' in p.text for p in loaded.paragraphs) else 'EMPTY'}")
+    print(f"[slice] 文本内容预览:\n{joined[:200]}")
+
+
+if __name__ == "__main__":
+    _main()
@@ -0,0 +1,58 @@
+"""Writer 编排：上下文装配 → 逐章生成 → 渲染 → docx 注入（Phase 5 垂直切片）。"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from docx import Document
+
+from genesis.data_models import StructuredSource
+from genesis.inference.engine import InferenceEngine
+from genesis.inference.prompt_registry import PromptRegistry
+from genesis.writer.context_builder import build_contexts
+from genesis.writer.docx_injector import Block, DocxInjector
+from genesis.writer.models import ChapterContent
+from genesis.writer.renderer import render_chapter_blocks
+from genesis.writer.writer_agent import WriterAgent
+from genesis.writer.writer_state import WriterState
+
+
+def _section_id_of(placeholder: str | None) -> str | None:
+    if not placeholder or not placeholder.startswith("section:"):
+        return None
+    return placeholder[len("section:"):]
+
+
+class WriteOrchestrator:
+    def generate(
+        self,
+        structured_source: StructuredSource,
+        output_path: str,
+        session_id: str = "writer",
+        samples_dir: str = "samples",
+        engine=None,
+        prompt_registry=None,
+        template_path: str | None = None,
+    ) -> list[ChapterContent]:
+        engine = engine or InferenceEngine()
+        prompt_registry = prompt_registry or PromptRegistry()
+        ctxs = build_contexts(structured_source, samples_dir)
+        state = WriterState([c.chapter_id for c in ctxs])
+        agent = WriterAgent(session_id=session_id, engine=engine, prompt_registry=prompt_registry, state=state)
+
+        contents: list[ChapterContent] = []
+        sections: dict[str, list[Block]] = {}
+        for ctx in ctxs:
+            content = agent.generate_chapter(ctx)
+            contents.append(content)
+            blocks = render_chapter_blocks(content)
+            sec_id = _section_id_of(ctx.template_marker.section_placeholder)
+            if sec_id:
+                sections[sec_id] = blocks
+
+        tpl = template_path or getattr(structured_source.template, "file_name", None)
+        if not tpl:
+            raise ValueError("template_path 必须提供（structured_source.template.file_name 为空）")
+        doc = DocxInjector(tpl).inject(sections, meta={})
+        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+        doc.save(output_path)
+        return contents
@@ -0,0 +1,49 @@
+import pytest
+from types import SimpleNamespace
+from docx import Document
+
+from genesis.data_models import ParsedTemplate, ChapterMarker
+from genesis.writer.orchestrator import WriteOrchestrator
+from genesis.writer.models import ChapterContent
+
+
+class FakeEngine:
+    def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
+        # 返回一个固定章节内容（title + 一个段落块）
+        return SimpleNamespace(
+            data={"title": variables["title"], "blocks": [{"type": "paragraph", "text": "自动生成的内容"}]},
+            status="ok",
+        )
+
+
+def _make_template(path):
+    doc = Document()
+    doc.add_paragraph("はじめに", style="Heading 1")
+    doc.add_paragraph("{{section:introduction}}")
+    doc.save(path)
+
+
+def _ss(template_path):
+    parsed = ParsedTemplate(
+        file_name=template_path,
+        sections=[
+            ChapterMarker(type="heading", name="はじめに", level=1),
+            ChapterMarker(type="placeholder", name="section:introduction", level=0),
+        ],
+        placeholders={},
+        styles={"defined": ["Heading 1"], "used": ["Heading 1"]},
+    )
+    return SimpleNamespace(template=parsed)
+
+
+def test_generate_produces_filled_docx(tmp_path):
+    tpl = tmp_path / "tpl.docx"
+    out = tmp_path / "out.docx"
+    _make_template(str(tpl))
+    orch = WriteOrchestrator()
+    contents = orch.generate(_ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=FakeEngine())
+    assert isinstance(contents, list) and len(contents) == 1
+    # 输出 docx 含注入文本且无残留异常
+    loaded = Document(str(out))
+    joined = "\n".join(p.text for p in loaded.paragraphs)
+    assert "自动生成的内容" in joined