"""Writer Agent：调用推理引擎生成单章内容（Phase 5）。 接入真实 InferenceEngine.chat_structured（session_id/prompt/variables/schema/retry_count）， 并对章节级失败做有限重试；token 估算分块（真实拼回留待后续并发实现）。 """ from __future__ import annotations import asyncio from genesis.inference.engine import InferenceEngine from genesis.inference.prompt_registry import PromptRegistry from genesis.inference.types import Prompt, StructuredResult from genesis.writer.models import ChapterContent, GenerationContext from genesis.writer.writer_state import WriterState from genesis.writer.exceptions import WriterGenerationError from genesis.writer.language import ( find_language_violations, resolve_expected_language, ) WRITER_PROMPT_TEMPLATE = ( "你是概要设计书撰写专家。\n" "章节: {{chapter_id}} {{title}}\n" "本章小节结构:\n{{sub_headings}}\n" "写入规则:\n{{write_rules}}\n" "设计规则:\n{{design_rules}}\n" "模板样式:\n{{template_styles}}\n" "影响调查上下文:\n{{impact}}\n" "参考资料（本章对应数据）:\n{{data}}\n" "请输出符合 schema 的章节内容 JSON。\n" "【小节约束】若上方「本章小节结构」非空，必须按该小节顺序组织内容，" "每个小节以 type=heading、level=2 的内容块开头（标题使用小节原文），随后为该小节的内容块；" "不得遗漏或新增小节。若「本章小节结构」为空，则不得输出任何 type=heading 的内容块" "（章节标题已由模板提供），仅以 paragraph/table/list/note 块组织内容。\n" "【主题约束】本章必须且仅围绕标题「{{title}}」所对应的主题撰写，" "严格以「参考资料（本章对应数据）」中的要件定义数据和「影响调查上下文」为核心依据；" "禁止输出与本章无关的系统整体架构、通用设计说明等内容，禁止套用其他章节的主题。" "若本章数据为空，则基于规则与影响调查上下文简要撰写，不得虚构数据。\n" "【语言约束】章节正文（所有 block 的 text 字段）所使用的自然语言：{{language_instruction}}\n" ) CHAPTER_OUTPUT_SCHEMA = { "type": "object", "properties": { "title": {"type": "string"}, "blocks": { "type": "array", "items": { "type": "object", "properties": { "type": {"type": "string"}, "text": {"type": "string"}, "level": {"type": "integer"}, "headers": {"type": "array", "items": {"type": "string"}}, "rows": {"type": "array", "items": {"type": "array", "items": {"type": "string"}}}, "items": {"type": "array", "items": {"type": "string"}}, "source_uris": {"type": "array", "items": {"type": "string"}}, }, "required": ["type"], }, }, }, "required": ["title", "blocks"], } class WriterAgent: def __init__( self, session_id, engine: InferenceEngine, prompt_registry: PromptRegistry, state: WriterState, max_retries: int = 2, ) -> None: self.session_id = session_id self.engine = engine self.prompt_registry = prompt_registry self.state = state self.max_retries = max_retries def _chunk_source(self, source) -> list[dict]: if source is None: return [{"index": 0, "text": ""}] src = source if isinstance(source, str) else str(source) n = max(1, len(src) // 1800 + 1) return [{"index": i, "text": src[i * 1800:(i + 1) * 1800]} for i in range(n)] def _resolve_prompt(self) -> Prompt: """取用/注册 writer.chapter 模板，并保证返回 Prompt 对象。 优先使用 get_or_create（与测试 Fake 兼容）；真实 PromptRegistry 无该方法时， 回退为 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: 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() result = self.engine.chat_structured( session_id=self.session_id, prompt=prompt, variables=context.to_vars(), schema=CHAPTER_OUTPUT_SCHEMA, retry_count=2, ) # 真实 InferenceEngine.chat_structured 为 async；测试用同步 FakeEngine 返回普通对象。 # 兼容两者：若返回协程则通过 asyncio.run 驱动（调用方 orchestrator/qa_loop/脚本均为同步上下文）。 if asyncio.iscoroutine(result): result = asyncio.run(result) if result.status not in ("ok", "fallback"): # 透传底层错误详情（如 401 鉴权失败原因），便于人工门禁定位 err = getattr(result, "error", None) code = getattr(result, "error_code", None) suffix = "" if err: suffix += f"; {err}" if code: suffix += f" (code={code})" raise WriterGenerationError(f"引擎返回异常状态: {result.status}{suffix}") return result.data def generate_chapter(self, context: GenerationContext) -> ChapterContent: self._chunk_source(context.structured_source) # 分块可用性验证（真实拼回留待后续） # 步骤 A：推导本章期望输出语言（仅依赖 context，循环前置，稳定） # fallback 用「规则文档」（write_rules/design_rules，RAG 自日文作成说明书/记入规则检索， # 含假名可判日文）。不能用 impact/data（源数据）：样本含中文元素名（止损风控等）、 # 影响标签为汉字无假名 → 会把日文文档误判为期望 zh（真实试运行暴露）。 expected = resolve_expected_language( explicit=context.output_language, title=context.title, fallback_texts=[ "\n".join(context.write_rules or []), "\n".join(context.design_rules or []), ], ) last_err: Exception | None = None for _ in range(max(1, self.max_retries)): try: data = self._call_llm(context) except Exception as e: # 引擎可能抛出任意异常，统一按章节级失败重试 last_err = e continue try: content = ChapterContent.from_llm(context.chapter_id, context.title, data) except (KeyError, TypeError, ValueError) as e: last_err = e continue # 模板结构为准（design §6.5）：无子节结构的章，剔除 LLM 自造的 # heading 块（prompt 约束为尽力而为，此处程序化强制） if not (getattr(context.template_marker, "sub_headings", None) or []): content.blocks = [b for b in content.blocks if b.type != "heading"] # 步骤 A：输出语言一致性强制（期望语言可推导时，违规按失败重试） if expected: viol = find_language_violations(content.blocks, expected) if viol: last_err = WriterGenerationError( f"章节 {context.chapter_id} 语言不一致（期望 {expected}，" f"发现 {len(viol)} 处违规正文）" ) continue self.state.record_success(content) return content raise WriterGenerationError(f"章节 {context.chapter_id} 重试耗尽: {last_err}")