diff --git a/samples/phase5-slice/output.docx b/samples/phase5-slice/output.docx new file mode 100644 index 0000000..fb293ec Binary files /dev/null and b/samples/phase5-slice/output.docx differ diff --git a/samples/phase5-slice/template.docx b/samples/phase5-slice/template.docx new file mode 100644 index 0000000..84629e7 Binary files /dev/null and b/samples/phase5-slice/template.docx differ diff --git a/scripts/run_phase5_slice.py b/scripts/run_phase5_slice.py new file mode 100644 index 0000000..ca63a0f --- /dev/null +++ b/scripts/run_phase5_slice.py @@ -0,0 +1,76 @@ +"""Phase 5 垂直切片运行脚本(CLI)。 + +用法: + python scripts/run_phase5_slice.py --template --output [--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() diff --git a/src/genesis/writer/orchestrator.py b/src/genesis/writer/orchestrator.py new file mode 100644 index 0000000..0124cff --- /dev/null +++ b/src/genesis/writer/orchestrator.py @@ -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 diff --git a/tests/test_phase5_writer_orchestrator.py b/tests/test_phase5_writer_orchestrator.py new file mode 100644 index 0000000..b0a93d1 --- /dev/null +++ b/tests/test_phase5_writer_orchestrator.py @@ -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