From 24d3aa3b8557c25c054450545975d7c02cb44516 Mon Sep 17 00:00:00 2001 From: lhl Date: Thu, 13 Aug 2026 11:12:22 +0800 Subject: [PATCH] feat(qa): add QALoop (closed loop, regenerate only failed chapters) --- src/genesis/qa/qa_loop.py | 70 ++++++++++++++++++++++++++++++++++++ tests/test_phase5_qa_loop.py | 56 +++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 src/genesis/qa/qa_loop.py create mode 100644 tests/test_phase5_qa_loop.py diff --git a/src/genesis/qa/qa_loop.py b/src/genesis/qa/qa_loop.py new file mode 100644 index 0000000..df5cefa --- /dev/null +++ b/src/genesis/qa/qa_loop.py @@ -0,0 +1,70 @@ +"""QA 闭环：生成 → 校验 → 仅重生成失败章 → 复校验（Phase 5）。""" +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +from genesis.inference.engine import InferenceEngine +from genesis.inference.prompt_registry import PromptRegistry +from genesis.qa.guardrails import DEFAULT_MAX_QA_ROUNDS, QALoopController +from genesis.qa.report import QAReport +from genesis.qa.validator import QAValidator +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.orchestrator import _section_id_of +from genesis.writer.renderer import render_chapter_blocks +from genesis.writer.writer_agent import WriterAgent +from genesis.writer.writer_state import WriterState + + +class QALoop: + def __init__(self, max_rounds: int = DEFAULT_MAX_QA_ROUNDS) -> None: + self.controller = QALoopController(max_rounds=max_rounds) + + def _build(self, structured_source, samples_dir, engine, prompt_registry, template_path, output_path, session_id, only_ids=None, prev=None): + 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_map = dict(prev) if prev else {} + order = [c.chapter_id for c in ctxs] + sections: dict[str, list[Block]] = {} + for ctx in ctxs: + if only_ids is not None and ctx.chapter_id not in only_ids and ctx.chapter_id in contents_map: + content = contents_map[ctx.chapter_id] + else: + content = agent.generate_chapter(ctx) + contents_map[ctx.chapter_id] = 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 必须提供") + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + doc = DocxInjector(tpl).inject(sections, meta={}) + doc.save(output_path) + return [contents_map[cid] for cid in order] + + def run(self, structured_source, output_path, session_id="writer", samples_dir="samples", engine=None, prompt_registry=None, template_path=None) -> QAReport: + engine = engine or InferenceEngine() + prompt_registry = prompt_registry or PromptRegistry() + validator = QAValidator() + contents = self._build(structured_source, samples_dir, engine, prompt_registry, template_path, output_path, session_id) + report = validator.validate_doc(contents, structured_source) + while self.controller.can_continue() and report.failed_chapters: + self.controller.advance() + contents = self._build( + structured_source, + samples_dir, + engine, + prompt_registry, + template_path, + output_path, + session_id, + only_ids=set(report.failed_chapters), + prev={c.chapter_id: c for c in contents}, + ) + report = validator.validate_doc(contents, structured_source) + return report diff --git a/tests/test_phase5_qa_loop.py b/tests/test_phase5_qa_loop.py new file mode 100644 index 0000000..3b0a5ed --- /dev/null +++ b/tests/test_phase5_qa_loop.py @@ -0,0 +1,56 @@ +import pytest +from types import SimpleNamespace +from docx import Document + +from genesis.data_models import ParsedTemplate, ChapterMarker +from genesis.qa.qa_loop import QALoop +from genesis.writer.models import ChapterContent + + +class ImprovingEngine: + def __init__(self): + self.calls: dict[str, int] = {} + + def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + cid = variables["title"] + self.calls[cid] = self.calls.get(cid, 0) + 1 + n = self.calls[cid] + # "a" 首次即通过；"b" 首次太短失败，第二次（n>=2）充分 + text = "充分且规范的说明内容，满足写入规则要求。" if (cid == "A" or n >= 2) else "x" + return SimpleNamespace(data={"title": cid, "blocks": [{"type": "paragraph", "text": text}]}, status="ok") + + +def _make_template(path): + doc = Document() + doc.add_paragraph("A", style="Heading 1") + doc.add_paragraph("{{section:a}}") + doc.add_paragraph("B", style="Heading 1") + doc.add_paragraph("{{section:b}}") + doc.save(path) + + +def _ss(template_path): + parsed = ParsedTemplate( + file_name=template_path, + sections=[ + ChapterMarker(type="heading", name="A", level=1), + ChapterMarker(type="placeholder", name="section:a", level=0), + ChapterMarker(type="heading", name="B", level=1), + ChapterMarker(type="placeholder", name="section:b", level=0), + ], + placeholders={}, + styles={"defined": ["Heading 1"], "used": ["Heading 1"]}, + ) + return SimpleNamespace(template=parsed) + + +def test_qa_loop_regenerates_only_failed(tmp_path): + tpl = tmp_path / "tpl.docx" + out = tmp_path / "out.docx" + _make_template(str(tpl)) + engine = ImprovingEngine() + loop = QALoop(max_rounds=3) + report = loop.run(_ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=engine) + assert report.passed is True + assert engine.calls["A"] == 1 # A 首次即通过，未被重生成 + assert engine.calls["B"] == 2 # B 失败后被重生成一次