fix(writer): 章节级数据注入 + 修复 prompt 从未渲染的关键缺陷

- models: CHAPTER_SHEET_TYPES/CHAPTER_IMPACT_ELEMENT 章节→数据映射(design §6.8 ①)
- models: _format_chapter_data 按章定向渲染 ExcelTable 为 Markdown;GENERIC 自由記述作通用背景
- models: _format_impact 支持按 ElementType 过滤(章节级影响上下文)
- writer_agent: {{source}}(全量repr) → {{data}}(章节数据);新增【主题约束】
- writer_agent: 关键修复——真实 PromptRegistry.get 返回模板字符串,
  引擎对 str 不做变量渲染,LLM 实际收到的是 {{占位符}} 原文;
  _resolve_prompt 统一包装为 Prompt 对象保证渲染
- 真实试运行验证:13 章主题全部正确、引用影响调查数据、语言漂移消除
This commit is contained in:
lhl
2026-08-24 11:50:59 +08:00
parent 6372bb17b9
commit 80ccc324fc
6 changed files with 410 additions and 27 deletions
+1
View File
@@ -114,3 +114,4 @@
| 2026-08-23 | Agent 实现 | Impact Agent MVP 实现(TDD RED→GREEN→全量 351 passed / 99.27%):data_models 新增 ChangeType/ChangeElement/ImpactWarning/ChangeAnalysis/ImpactReport + StructuredSource.impact_report;新增 src/genesis/impact/code_parser.py Java 解析 @RestController/@Service/@Entity/@TableName、existing_system_explorer.py 组装、impact_agent.py 变更点定位+impact_report_to_dict 序列化);source_aggregator 解除 existing_system=None 硬编码(新增 existing_system_path 门控);writer 集成——GenerationContext.impact_report + to_vars()[impact]、WRITER_PROMPT_TEMPLATE 加「影响调查上下文」、context_builder 透传、WriteOrchestrator.generate 门控自动跑 ImpactAgent + 回填 + 默认 metadoc_title/version/created_at);新造样本 要件定義_追加改修_股票.xlsx(对齐 sunOnly 真实类名);门禁(fake 模式)PASS:影响调查 total=16 new=5 modified=8 deleted=3 unchanged=50 warnings=0,概要设计书 13 章产出,影响调查书 JSON 可下载 | src/genesis/data_models.py; src/genesis/impact/code_parser.py; src/genesis/impact/existing_system_explorer.py; src/genesis/impact/impact_agent.py; src/genesis/parsers/source_aggregator.py; src/genesis/writer/models.py; src/genesis/writer/writer_agent.py; src/genesis/writer/context_builder.py; src/genesis/writer/orchestrator.py; tests/test_data_models.py; tests/test_code_parser.py; tests/test_existing_system_explorer.py; tests/test_impact_agent.py; tests/test_source_aggregator.py; tests/test_phase5_models.py; tests/test_phase5_writer_agent.py; tests/test_phase5_context_builder.py; tests/test_phase5_writer_orchestrator.py; samples/要件定義_追加改修_股票.xlsx; samples/existing-system/; docs/superpowers/specs/2026-08-23-impact-mvp.md; _AI_USAGE_LOG.md | deepseek-v4-flash |
| 2026-08-23 | Agent 实现 | CodeParser 多语言就绪预备性重构(行为不变):code_parser.py 重构为语言适配器注册表——BaseLanguageParser 抽象基类(extensions/source_files/parse_file 契约)+ JavaLanguageParser(迁移全部 Java 正则,输出与重构前逐字节一致:20Ctrl/25Svc/22Ent/113EP/13模块/219类/278imports+ LANGUAGE_PARSERS 注册表 + register_language_parser 扩展点 + CodeParser 分发器(parse(root, language=None) 自动按扩展名探测/显式语言覆盖/多语言合并 language 逗号连接/无源码与不支持语言抛 CodeParseError);source_aggregator 新增 existing_system_language 透传参数;config.py allowed_extensions 补 .py/.ts/.go/.cs;删除自动探测下不可达 continue 死分支;TDD REDImportError)→ GREENtest_code_parser 17 passed + test_source_aggregator 19 passed)→ 全量 362 passed / 99.28%(基线 351/99.27%+11 新测试);门禁(fake 模式)复跑 PASSsummary 与 MVP 基线一致(total=16 new=5 modified=8 deleted=3 unchanged=50 warnings=0),概要设计书 13 章;未实现任何新语言(仅 Java,后续加语言=新增适配器类+register_language_parser,下游零改动) | src/genesis/impact/code_parser.py; src/genesis/parsers/source_aggregator.py; src/genesis/config.py; tests/test_code_parser.py; tests/test_source_aggregator.py; _AI_USAGE_LOG.md | deepseek-v4-flash |
| 2026-08-24 | Agent 实现 | 试运行脚本与运行说明:新增 scripts/run_trial.py(库级端到端驱动——argparse 默认指向 samples/ 股票追加改修+sunOnly+真实模板;--fake 离线 FakeEngine / 默认 build_inference_engine 真实 LLM 读 .env keySourceParser→WriteOrchestrator 门控自动影响调查→output/output.docx + impact-report.json;暴露 main(argv) 供测试);新增 tests/test_run_trial.pyfake 模式断言 13 章 + summary 16/5/8/3/50/0 + 文件产出 + docx 注入);.gitignore 新增 output/(试运行产物不入库);新建 README.md(安装/配置 .env/离线与真实试运行命令/输出/测试);TDD REDModuleNotFoundError)→ GREEN2 passed)→ 全量 364 passed / 99.28%fake 试运行 PASS13 章、summary 与 MVP 基线一致、output.docx 含注入文本);真实 LLM 试运行待用户配置 .env API Key | scripts/run_trial.py; tests/test_run_trial.py; .gitignore; README.md; _AI_USAGE_LOG.md | deepseek-v4-flash |
| 2026-08-24 11:50 | Agent 实现 | Writer 章节级数据注入(design §6.8 ①落地):CHAPTER_SHEET_TYPES/CHAPTER_IMPACT_ELEMENT 映射、_format_chapter_data 按章定向注入 ExcelTable、_format_impact 按要素类型过滤;修复关键缺陷——真实 PromptRegistry.get 返回模板字符串导致引擎从不渲染变量(LLM 收到 {{占位符}} 原文),_resolve_prompt 统一包装为 Prompt 对象;prompt 增加【主题约束】。真实试运行验证:13 章主题全部正确(機能/画面/帳票/DB/IF/バッチ各写其义)、引用 F001-F004/TB001-004 影响数据、语言漂移顺带消除 | src/genesis/writer/models.py, src/genesis/writer/writer_agent.py, tests/test_phase5_models.py, tests/test_phase5_writer_agent.py, tests/test_phase5_context_builder.py | x-preview-f-free (opencode) |
+98 -9
View File
@@ -4,6 +4,37 @@ from __future__ import annotations
from dataclasses import dataclass, field
from typing import Literal
from genesis.data_models import ElementType, SheetType
# 单表渲染行数上限(防 token 爆炸;样本量小,通常不触发)
MAX_ROWS_PER_TABLE = 200
# design.md §6.8 ①「DataGate.load(structured_source, selector=该章数据)」的章节级选择器:
# 章节占位符 id → 本章对应的 Excel Sheet 类型
# - introduction 概览章注入全部类型表
# - 未登记的章节 id 缺省为空列表 → 仅 GENERIC 自由記述作背景
_ALL_SHEET_TYPES = list(SheetType)
CHAPTER_SHEET_TYPES: dict[str, list[SheetType]] = {
"introduction": _ALL_SHEET_TYPES,
"function_list": [SheetType.FUNCTION],
"screen_list": [SheetType.SCREEN],
"report_list": [SheetType.REPORT],
"db_design": [SheetType.DATABASE],
"if_definition": [SheetType.INTERFACE],
"batch_list": [SheetType.BATCH],
}
# 章节 id → 影响调查要素类型(None = 展示全部,用于概览章)
CHAPTER_IMPACT_ELEMENT: dict[str, ElementType | None] = {
"function_list": ElementType.FUNCTION,
"screen_list": ElementType.SCREEN,
"report_list": ElementType.REPORT,
"db_design": ElementType.DB,
"if_definition": ElementType.IF,
"batch_list": ElementType.BATCH,
"introduction": None,
}
@dataclass
class ContentBlock:
@@ -90,18 +121,73 @@ class GenerationContext:
"design_rules": "\n".join(self.design_rules),
"template_styles": ", ".join(sorted(self.template_styles)),
"prior_state": str(self.prior_state) if self.prior_state is not None else "",
"source": str(self.structured_source) if self.structured_source is not None else "",
"impact": _format_impact(self.impact_report),
"data": _format_chapter_data(
self.structured_source,
CHAPTER_SHEET_TYPES.get(self.chapter_id, []),
),
"impact": _format_impact(
self.impact_report,
CHAPTER_IMPACT_ELEMENT.get(self.chapter_id, None),
),
}
def _format_impact(report: object | None) -> str:
"""影响调查书格式化为 prompt 可读文本(无报告/无分析时为空串)。"""
def _render_table(tb) -> list[str]:
"""单张 ExcelTable 渲染为可读 Markdown 行(表名行 + 表头 + 分隔 + 数据行)。"""
lines = [f"### 表: {tb.name}{tb.detected_type.value}"]
headers = list(tb.headers)
lines.append("| " + " | ".join(headers) + " |")
lines.append("|" + "|".join([" --- "] * len(headers)) + "|")
for row in tb.rows[:MAX_ROWS_PER_TABLE]:
cells = []
for h in headers:
v = row.get(h)
value = getattr(v, "value", v)
cells.append("" if value is None else str(value))
lines.append("| " + " | ".join(cells) + " |")
return lines
def _format_chapter_data(structured_source: object | None, sheet_types: list[SheetType]) -> str:
"""按章节定向格式化要件定义数据(design.md §6.8 ① selector=该章数据)。
- 命中 sheet_types 的表全部注入(章节主题数据)
- GENERIC(自由記述)作为通用背景始终注入
- structured_source 为 None 或无任何可注入表时返回空串
"""
if structured_source is None:
return ""
tables = getattr(structured_source, "tables", None) or []
matched = [
t for t in tables
if t.detected_type in sheet_types and t.detected_type != SheetType.GENERIC
]
generic = [t for t in tables if t.detected_type == SheetType.GENERIC]
selected = matched + generic
if not selected:
return ""
lines: list[str] = []
for tb in selected:
lines.extend(_render_table(tb))
lines.append("")
return "\n".join(lines).rstrip()
def _format_impact(report: object | None, element_type: ElementType | None = None) -> str:
"""将影响调查书格式化为 prompt 可读文本(无报告/无分析时为空串)。
element_type 非 None 时仅保留该类型要素(章节级定向,design.md §6.8 ①);
警告始终保留(不依赖要素类型)。
"""
if report is None:
return ""
ca = getattr(report, "change_analysis", None)
if ca is None:
return ""
def keep(el) -> bool:
return element_type is None or el.element_type == element_type.value
summary = getattr(report, "summary", {}) or {}
lines = [f"project_type={getattr(ca, 'project_type', '')}"]
lines.append(
@@ -113,13 +199,16 @@ def _format_impact(report: object | None) -> str:
)
)
for el in getattr(ca, "new_elements", []) or []:
lines.append(f"[新規] {el.element_id} {el.element_type} {el.name}")
if keep(el):
lines.append(f"[新規] {el.element_id} {el.element_type} {el.name}")
for el in getattr(ca, "modified_elements", []) or []:
impacted = ", ".join(el.impacted_existing) or "-"
lines.append(f"[変更] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
if keep(el):
impacted = ", ".join(el.impacted_existing) or "-"
lines.append(f"[変更] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
for el in getattr(ca, "deleted_elements", []) or []:
impacted = ", ".join(el.impacted_existing) or "-"
lines.append(f"[削除] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
if keep(el):
impacted = ", ".join(el.impacted_existing) or "-"
lines.append(f"[削除] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
for w in getattr(ca, "warnings", []) or []:
lines.append(f"[警告] {w.element_id}: {w.issue}")
return "\n".join(lines)
+18 -6
View File
@@ -22,8 +22,12 @@ WRITER_PROMPT_TEMPLATE = (
"设计规则:\n{{design_rules}}\n"
"模板样式:\n{{template_styles}}\n"
"影响调查上下文:\n{{impact}}\n"
"参考资料:\n{{source}}\n"
"参考资料(本章对应数据):\n{{data}}\n"
"请输出符合 schema 的章节内容 JSON。\n"
"【主题约束】本章必须且仅围绕标题「{{title}}」所对应的主题撰写,"
"严格以「参考资料(本章对应数据)」中的要件定义数据和「影响调查上下文」为核心依据;"
"禁止输出与本章无关的系统整体架构、通用设计说明等内容,禁止套用其他章节的主题。"
"若本章数据为空,则基于规则与影响调查上下文简要撰写,不得虚构数据。\n"
"【语言约束】章节正文(所有 block 的 text 字段)所使用的自然语言,"
"必须与章节标题「{{title}}」所用语言保持一致:标题为日文则用日文撰写,"
"为中文则用中文撰写,依此类推。"
@@ -77,16 +81,24 @@ class WriterAgent:
return [{"index": i, "text": src[i * 1800:(i + 1) * 1800]} for i in range(n)]
def _resolve_prompt(self) -> Prompt:
"""取用/注册 writer.chapter 模板。
"""取用/注册 writer.chapter 模板,并保证返回 Prompt 对象
优先使用 get_or_create(与测试 Fake 兼容);真实 PromptRegistry 无该方法时,
回退为 register + get。
回退为 register + get。注意:真实 get() 返回模板字符串而非 Prompt,
而引擎 _render_prompt 仅对 Prompt 对象做变量渲染、str 原样发送
(否则 {{占位符}} 不被替换直接进 LLM)→ 此处统一包装为 Prompt。
"""
get_or_create = getattr(self.prompt_registry, "get_or_create", None)
if get_or_create is not None:
return get_or_create("writer.chapter", WRITER_PROMPT_TEMPLATE)
self.prompt_registry.register("writer.chapter", "1", WRITER_PROMPT_TEMPLATE)
return self.prompt_registry.get("writer.chapter", "1")
resolved = get_or_create("writer.chapter", WRITER_PROMPT_TEMPLATE)
else:
self.prompt_registry.register("writer.chapter", "1", WRITER_PROMPT_TEMPLATE)
resolved = self.prompt_registry.get("writer.chapter", "1")
if isinstance(resolved, Prompt):
return resolved
template = getattr(resolved, "template", None) or str(resolved)
version = str(getattr(resolved, "version", "1") or "1")
return Prompt(name="writer.chapter", version=version, template=template)
def _call_llm(self, context: GenerationContext) -> dict:
prompt = self._resolve_prompt()
+40
View File
@@ -45,3 +45,43 @@ def test_build_contexts_carries_impact_report():
ctxs = build_contexts(ss, samples_dir="nonexistent_dir_xyz")
assert ctxs[0].impact_report is report
assert "project_type" in ctxs[0].to_vars()["impact"] or ctxs[0].to_vars()["impact"] != ""
def test_build_contexts_vars_data_scoped_per_chapter():
"""集成:多章模板 + FUNCTION/DB 表 → 各章 to_vars()['data'] 按章节定向。"""
from genesis.data_models import (
CellValue, ExcelTable, Provenance, SheetType, StructuredSource,
)
template = ParsedTemplate(
file_name="t.docx",
sections=[
ChapterMarker(type="heading", name="機能一覧", level=1),
ChapterMarker(type="placeholder", name="section:function_list", level=0),
ChapterMarker(type="heading", name="DB設計", level=1),
ChapterMarker(type="placeholder", name="section:db_design", level=0),
],
placeholders={},
styles={"used": ["Heading1"]},
)
def cell(v):
return CellValue(value=v, provenance=Provenance("f.xlsx", "s", 1, "A", ""))
ss = StructuredSource(
tables=[
ExcelTable("機能一覧", SheetType.FUNCTION, "openpyxl",
["機能ID", "機能名"],
[{"機能ID": cell("F001"), "機能名": cell("止损风控")}]),
ExcelTable("DB定義", SheetType.DATABASE, "openpyxl",
["テーブルID", "テーブル名"],
[{"テーブルID": cell("T001"), "テーブル名": cell("trade_order")}]),
],
template=template, rule_docs=[], image_analyses=[],
existing_system=None, comments=[],
)
ctxs = build_contexts(ss, samples_dir="nonexistent_dir_xyz")
by_id = {c.chapter_id: c for c in ctxs}
fn = by_id["function_list"].to_vars()["data"]
db = by_id["db_design"].to_vars()["data"]
assert "F001" in fn and "trade_order" not in fn
assert "T001" in db and "止损风控" not in db
+143 -12
View File
@@ -1,6 +1,66 @@
from genesis.data_models import (
CellValue,
ExcelTable,
ParsedTemplate,
ChapterMarker,
Provenance,
SheetType,
StructuredSource,
)
from genesis.writer.models import ContentBlock, ChapterContent, GenerationContext, ChapterSpec
def _cell(v):
return CellValue(value=v, provenance=Provenance("要件.xlsx", "s1", 1, "A", ""))
def _table(name, sheet_type, headers, rows):
return ExcelTable(
name=name,
detected_type=sheet_type,
extraction_method="openpyxl",
headers=headers,
rows=[{h: _cell(v) for h, v in zip(headers, row)} for row in rows],
)
def _mk_source():
"""含 FUNCTION / DATABASE / GENERIC 三张表的 StructuredSource(模板仅占位)。"""
template = ParsedTemplate(
file_name="t.docx",
sections=[ChapterMarker(type="heading", name="x", level=1)],
placeholders={},
styles={"used": []},
)
return StructuredSource(
tables=[
_table("機能一覧", SheetType.FUNCTION, ["機能ID", "機能名"], [["F001", "止损风控"], ["F002", "订单登录"]]),
_table("DB定義", SheetType.DATABASE, ["テーブルID", "テーブル名"], [["T001", "trade_order"]]),
_table("自由記述", SheetType.GENERIC, ["text"], [["系统采用前后端分离架构"]]),
],
template=template,
rule_docs=[],
image_analyses=[],
existing_system=None,
comments=[],
)
_UNSET = object()
def _ctx(cid, title, source=_UNSET, report=None):
return GenerationContext(
chapter_id=cid, title=title,
template_marker=ChapterSpec(chapter_id=cid, title=title, section_placeholder=f"section:{cid}"),
structured_source=None if source is None else (
_mk_source() if source is _UNSET else source
),
write_rules=[], design_rules=[], template_styles=set(),
impact_report=report,
)
def test_content_block_defaults():
b = ContentBlock(block_id="b1", type="paragraph", text="你好")
assert b.level is None
@@ -37,24 +97,16 @@ def test_generation_context_impact_var_formats_report():
warnings=[],
)
report = ImpactReport(metadata={"version": "v1"}, change_analysis=ca, summary={"new": 1})
ctx = GenerationContext(
chapter_id="db_design", title="DB 设计",
template_marker=ChapterSpec(chapter_id="db_design", title="DB 设计", section_placeholder="{{section:db_design}}"),
structured_source=None, write_rules=["规则1"], design_rules=["规则2"],
template_styles={"Heading 1"}, impact_report=report,
)
# function_list 章节对应 ElementType=機能 → 该要素应保留
ctx = _ctx("function_list", "機能一覧", source=None, report=report)
impact = ctx.to_vars()["impact"]
assert "止损风控" in impact
assert "F001" in impact
def _impact_ctx(report):
return GenerationContext(
chapter_id="db_design", title="DB 设计",
template_marker=ChapterSpec(chapter_id="db_design", title="DB 设计", section_placeholder="{{section:db_design}}"),
structured_source=None, write_rules=[], design_rules=[],
template_styles=set(), impact_report=report,
)
# function_list 章节对应 ElementType=機能 → [削除] 機能要素可保留
return _ctx("function_list", "機能一覧", source=None, report=report)
def test_impact_var_empty_when_report_without_analysis():
@@ -82,3 +134,82 @@ def test_impact_var_formats_deleted_and_warnings():
def test_chapter_spec_placeholder_optional():
s = ChapterSpec(chapter_id="x", title="X", section_placeholder=None)
assert s.section_placeholder is None
# ---------- 章节级数据注入(design.md §6.8 ①:selector=该章数据) ----------
def test_to_vars_data_scoped_by_chapter():
"""db_design 章的 data 只含 DATABASE 表 + GENERIC 自由記述,不含 FUNCTION 表。"""
data = _ctx("db_design", "DB設計").to_vars()["data"]
assert "テーブルID" in data and "trade_order" in data # DB 表内容
assert "前后端分离" in data # GENERIC 通用背景
assert "止损风控" not in data and "機能ID" not in data # FUNCTION 表被排除
def test_to_vars_data_function_chapter_gets_function_table():
data = _ctx("function_list", "機能一覧").to_vars()["data"]
assert "F001" in data and "止损风控" in data
assert "trade_order" not in data # DB 表被排除
def test_to_vars_data_introduction_gets_all_tables():
"""introduction 概览章注入全部类型表。"""
data = _ctx("introduction", "はじめに").to_vars()["data"]
assert "F001" in data and "trade_order" in data
def test_to_vars_data_empty_when_no_matching_table():
"""无对应类型表(如 report_list 无 REPORT 表)→ 仅 GENERIC 背景,不报错。"""
data = _ctx("report_list", "帳票一覧").to_vars()["data"]
assert "帳票ID" not in data
assert "前后端分离" in data # GENERIC 始终保留
def test_to_vars_data_none_safe_and_source_key_removed():
"""structured_source=None → data 为空串;旧的全量 source 键不再提供。"""
vars_ = _ctx("db_design", "DB設計", source=None).to_vars()
assert vars_["data"] == ""
assert "source" not in vars_
def test_to_vars_data_renders_markdown_table_shape():
"""表渲染为可读 Markdown 形态(表名行 + 表头行 + 分隔行)。"""
data = _ctx("db_design", "DB設計").to_vars()["data"]
assert "DB定義" in data
assert "| テーブルID | テーブル名 |" in data
assert "---" in data
assert "| T001 | trade_order |" in data
def test_impact_filtered_by_chapter_element_type():
"""db_design 章 impact 只显示 DB 要素;機能要素被过滤。"""
from genesis.data_models import ChangeAnalysis, ChangeElement, ChangeType, ImpactReport
ca = ChangeAnalysis(
project_type="enhancement",
new_elements=[
ChangeElement("T001", "DB", "订单表", ChangeType.NEW),
ChangeElement("F001", "機能", "止损风控", ChangeType.NEW),
],
modified_elements=[],
deleted_elements=[],
unchanged_elements=[],
warnings=[],
)
report = ImpactReport(metadata={}, change_analysis=ca, summary={"new": 2})
impact = _ctx("db_design", "DB設計", source=None, report=report).to_vars()["impact"]
assert "T001" in impact and "订单表" in impact
assert "F001" not in impact and "止损风控" not in impact
def test_impact_introduction_shows_all_elements():
"""introduction 章(None 过滤)展示全部要素。"""
from genesis.data_models import ChangeAnalysis, ChangeElement, ChangeType, ImpactReport
ca = ChangeAnalysis(
project_type="enhancement",
new_elements=[ChangeElement("T001", "DB", "订单表", ChangeType.NEW),
ChangeElement("F001", "機能", "止损风控", ChangeType.NEW)],
modified_elements=[], deleted_elements=[], unchanged_elements=[], warnings=[],
)
report = ImpactReport(metadata={}, change_analysis=ca, summary={"new": 2})
impact = _ctx("introduction", "はじめに", source=None, report=report).to_vars()["impact"]
assert "T001" in impact and "F001" in impact
+110
View File
@@ -66,6 +66,116 @@ def test_prompt_template_has_impact_context_var():
assert "{{impact}}" in WRITER_PROMPT_TEMPLATE
def test_prompt_template_enforces_topic_and_chapter_data():
# 主题约束:正文必须围绕本章标题主题,依据章节数据,禁止套用系统整体架构
assert "主题约束" in WRITER_PROMPT_TEMPLATE
assert "参考资料(本章对应数据)" in WRITER_PROMPT_TEMPLATE
assert "{{data}}" in WRITER_PROMPT_TEMPLATE
# 旧的全量 source 变量不再使用
assert "{{source}}" not in WRITER_PROMPT_TEMPLATE
class RecordingEngine(FakeEngine):
"""记录 variables 以断言 prompt 渲染输入。"""
def __init__(self):
super().__init__()
self.last_vars = None
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
self.last_vars = dict(variables)
return super().chat_structured(
session_id=session_id, prompt=prompt,
variables=variables, schema=schema, retry_count=retry_count,
)
class RealStylePromptRegistry:
"""模拟真实 PromptRegistry:无 get_or_createget() 返回模板字符串(非 Prompt)。"""
def __init__(self):
self._tpl = {}
def register(self, name, version, template):
self._tpl[(name, version)] = template
def get(self, name, version=None):
return self._tpl[(name, version)]
class RenderingEngine:
"""模拟真实引擎渲染行为:仅当收到 Prompt 对象时才用 jinja2 渲染(engine._render_prompt 契约)。"""
def __init__(self):
self.last_rendered = None
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
from jinja2 import Template
from genesis.inference.types import Prompt as P
tpl = prompt.template if isinstance(prompt, P) else prompt # str 原样返回 → 不渲染
self.last_rendered = Template(tpl).render(**variables) if isinstance(prompt, P) else tpl
return SimpleNamespace(
data={"title": variables["title"], "blocks": [{"type": "paragraph", "text": "ok"}]},
status="ok",
)
def test_prompt_is_rendered_with_real_style_registry():
"""回归:真实 PromptRegistry.get 返回模板字符串 → 引擎按 str 原样发送,
{{data}}/{{title}} 等占位符从未被替换(真实试运行暴露)。
WriterAgent 必须保证交给引擎的是可渲染的 Prompt 对象。"""
agent = WriterAgent(
session_id="s",
engine=RenderingEngine(),
prompt_registry=RealStylePromptRegistry(),
state=WriterState(["db_design"]),
)
ctx = _ctx("db_design", "DB 設計")
agent.generate_chapter(ctx)
assert "{{" not in agent.engine.last_rendered # 无残留占位符
assert "DB 設計" in agent.engine.last_rendered # title 已渲染
assert "W1" in agent.engine.last_rendered # write_rules 已渲染
def test_generate_chapter_passes_chapter_scoped_data_var():
from genesis.data_models import (
CellValue, ExcelTable, ParsedTemplate, ChapterMarker,
Provenance, SheetType, StructuredSource,
)
def cell(v):
return CellValue(value=v, provenance=Provenance("f.xlsx", "s", 1, "A", ""))
src = StructuredSource(
tables=[
ExcelTable("DB定義", SheetType.DATABASE, "openpyxl",
["テーブルID", "テーブル名"],
[{"テーブルID": cell("T001"), "テーブル名": cell("trade_order")}]),
ExcelTable("機能一覧", SheetType.FUNCTION, "openpyxl",
["機能ID", "機能名"],
[{"機能ID": cell("F001"), "機能名": cell("止损风控")}]),
],
template=ParsedTemplate("t.docx",
[ChapterMarker(type="heading", name="x", level=1)],
{}, {"used": []}),
rule_docs=[], image_analyses=[], existing_system=None, comments=[],
)
ctx = GenerationContext(
chapter_id="db_design", title="DB 設計",
template_marker=ChapterSpec(chapter_id="db_design", title="DB 設計", section_placeholder=None),
structured_source=src, write_rules=["W1"], design_rules=["D1"],
template_styles={"Heading1"},
)
engine = RecordingEngine()
agent = WriterAgent(session_id="s", engine=engine,
prompt_registry=FakePromptRegistry(),
state=WriterState(["db_design"]))
agent.generate_chapter(ctx)
data = engine.last_vars["data"]
assert "T001" in data and "trade_order" in data # 本章对应数据
assert "止损风控" not in data # 其他章数据被排除
def test_generate_chapter_retries():
class Boom(FakeEngine):
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):