|
|
@@ -0,0 +1,259 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""
|
|
|
+ContextManager - assemble context packs with weighted priorities.
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import json
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any, Dict, List, Optional
|
|
|
+
|
|
|
+from .config import get_config
|
|
|
+from .index_manager import IndexManager
|
|
|
+from .snapshot_manager import SnapshotManager, SnapshotVersionMismatch
|
|
|
+
|
|
|
+
|
|
|
+class ContextManager:
|
|
|
+ DEFAULT_TEMPLATE = "plot"
|
|
|
+ TEMPLATE_WEIGHTS = {
|
|
|
+ "plot": {"core": 0.40, "scene": 0.35, "global": 0.25},
|
|
|
+ "battle": {"core": 0.35, "scene": 0.45, "global": 0.20},
|
|
|
+ "emotion": {"core": 0.45, "scene": 0.35, "global": 0.20},
|
|
|
+ "transition": {"core": 0.50, "scene": 0.25, "global": 0.25},
|
|
|
+ }
|
|
|
+
|
|
|
+ def __init__(self, config=None, snapshot_manager: Optional[SnapshotManager] = None):
|
|
|
+ self.config = config or get_config()
|
|
|
+ self.snapshot_manager = snapshot_manager or SnapshotManager(self.config)
|
|
|
+ self.index_manager = IndexManager(self.config)
|
|
|
+
|
|
|
+ def build_context(
|
|
|
+ self,
|
|
|
+ chapter: int,
|
|
|
+ template: str | None = None,
|
|
|
+ use_snapshot: bool = True,
|
|
|
+ save_snapshot: bool = True,
|
|
|
+ max_chars: Optional[int] = None,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ template = template or self.DEFAULT_TEMPLATE
|
|
|
+ if template not in self.TEMPLATE_WEIGHTS:
|
|
|
+ template = self.DEFAULT_TEMPLATE
|
|
|
+
|
|
|
+ if use_snapshot:
|
|
|
+ try:
|
|
|
+ cached = self.snapshot_manager.load_snapshot(chapter)
|
|
|
+ if cached:
|
|
|
+ return cached.get("payload", cached)
|
|
|
+ except SnapshotVersionMismatch:
|
|
|
+ # Snapshot incompatible; rebuild below.
|
|
|
+ pass
|
|
|
+
|
|
|
+ pack = self._build_pack(chapter)
|
|
|
+ assembled = self.assemble_context(pack, template=template, max_chars=max_chars)
|
|
|
+
|
|
|
+ if save_snapshot:
|
|
|
+ meta = {"template": template}
|
|
|
+ self.snapshot_manager.save_snapshot(chapter, assembled, meta=meta)
|
|
|
+
|
|
|
+ return assembled
|
|
|
+
|
|
|
+ def assemble_context(
|
|
|
+ self,
|
|
|
+ pack: Dict[str, Any],
|
|
|
+ template: str = DEFAULT_TEMPLATE,
|
|
|
+ max_chars: Optional[int] = None,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ weights = self.TEMPLATE_WEIGHTS.get(template, self.TEMPLATE_WEIGHTS[self.DEFAULT_TEMPLATE])
|
|
|
+ max_chars = max_chars or 8000
|
|
|
+
|
|
|
+ sections = {}
|
|
|
+ for section_name in ["core", "scene", "global", "memory", "preferences", "alerts"]:
|
|
|
+ if section_name in pack:
|
|
|
+ sections[section_name] = pack[section_name]
|
|
|
+
|
|
|
+ assembled: Dict[str, Any] = {"meta": pack.get("meta", {}), "sections": {}}
|
|
|
+ for name, content in sections.items():
|
|
|
+ weight = weights.get(name, 0.0)
|
|
|
+ budget = int(max_chars * weight) if weight > 0 else None
|
|
|
+ text = json.dumps(content, ensure_ascii=False)
|
|
|
+ if budget is not None and len(text) > budget:
|
|
|
+ text = text[:budget]
|
|
|
+ assembled["sections"][name] = {"content": content, "text": text, "budget": budget}
|
|
|
+
|
|
|
+ assembled["template"] = template
|
|
|
+ assembled["weights"] = weights
|
|
|
+ return assembled
|
|
|
+
|
|
|
+ def filter_invalid_items(self, items: List[Dict[str, Any]], source_type: str, id_key: str) -> List[Dict[str, Any]]:
|
|
|
+ confirmed = self.index_manager.get_invalid_ids(source_type, status="confirmed")
|
|
|
+ pending = self.index_manager.get_invalid_ids(source_type, status="pending")
|
|
|
+ result = []
|
|
|
+ for item in items:
|
|
|
+ item_id = str(item.get(id_key, ""))
|
|
|
+ if item_id in confirmed:
|
|
|
+ continue
|
|
|
+ if item_id in pending:
|
|
|
+ item = dict(item)
|
|
|
+ item["warning"] = "pending_invalid"
|
|
|
+ result.append(item)
|
|
|
+ return result
|
|
|
+
|
|
|
+ def apply_confidence_filter(self, items: List[Dict[str, Any]], min_confidence: float) -> List[Dict[str, Any]]:
|
|
|
+ filtered: List[Dict[str, Any]] = []
|
|
|
+ for item in items:
|
|
|
+ conf = item.get("confidence")
|
|
|
+ if conf is None or conf >= min_confidence:
|
|
|
+ filtered.append(item)
|
|
|
+ return filtered
|
|
|
+
|
|
|
+ def _build_pack(self, chapter: int) -> Dict[str, Any]:
|
|
|
+ state = self._load_state()
|
|
|
+ core = {
|
|
|
+ "chapter_outline": self._load_outline(chapter),
|
|
|
+ "protagonist_snapshot": state.get("protagonist_state", {}),
|
|
|
+ "recent_summaries": self._load_recent_summaries(chapter, window=3),
|
|
|
+ "recent_meta": self._load_recent_meta(state, chapter, window=3),
|
|
|
+ }
|
|
|
+
|
|
|
+ scene = {
|
|
|
+ "location_context": state.get("protagonist_state", {}).get("location", {}),
|
|
|
+ "appearing_characters": self._load_recent_appearances(),
|
|
|
+ }
|
|
|
+ scene["appearing_characters"] = self.filter_invalid_items(
|
|
|
+ scene["appearing_characters"], source_type="entity", id_key="entity_id"
|
|
|
+ )
|
|
|
+
|
|
|
+ global_ctx = {
|
|
|
+ "worldview_skeleton": self._load_setting("世界观"),
|
|
|
+ "power_system_skeleton": self._load_setting("力量体系"),
|
|
|
+ "style_contract_ref": self._load_setting("风格契约"),
|
|
|
+ }
|
|
|
+
|
|
|
+ preferences = self._load_json_optional(self.config.webnovel_dir / "preferences.json")
|
|
|
+ memory = self._load_json_optional(self.config.webnovel_dir / "project_memory.json")
|
|
|
+
|
|
|
+ return {
|
|
|
+ "meta": {"chapter": chapter},
|
|
|
+ "core": core,
|
|
|
+ "scene": scene,
|
|
|
+ "global": global_ctx,
|
|
|
+ "preferences": preferences,
|
|
|
+ "memory": memory,
|
|
|
+ "alerts": {
|
|
|
+ "disambiguation_warnings": state.get("disambiguation_warnings", [])[-10:],
|
|
|
+ "disambiguation_pending": state.get("disambiguation_pending", [])[-10:],
|
|
|
+ },
|
|
|
+ }
|
|
|
+
|
|
|
+ def _load_state(self) -> Dict[str, Any]:
|
|
|
+ path = self.config.state_file
|
|
|
+ if not path.exists():
|
|
|
+ return {}
|
|
|
+ return json.loads(path.read_text(encoding="utf-8"))
|
|
|
+
|
|
|
+ def _load_outline(self, chapter: int) -> str:
|
|
|
+ outline_dir = self.config.outline_dir
|
|
|
+ patterns = [
|
|
|
+ f"第{chapter}章*.md",
|
|
|
+ f"第{chapter:02d}章*.md",
|
|
|
+ f"第{chapter:03d}章*.md",
|
|
|
+ f"第{chapter:04d}章*.md",
|
|
|
+ ]
|
|
|
+ for pattern in patterns:
|
|
|
+ matches = list(outline_dir.glob(pattern))
|
|
|
+ if matches:
|
|
|
+ return matches[0].read_text(encoding="utf-8")
|
|
|
+ return f"[大纲未找到: 第{chapter}章]"
|
|
|
+
|
|
|
+ def _load_recent_summaries(self, chapter: int, window: int = 3) -> List[Dict[str, Any]]:
|
|
|
+ summaries = []
|
|
|
+ summaries_dir = self.config.webnovel_dir / "summaries"
|
|
|
+ for ch in range(max(1, chapter - window), chapter):
|
|
|
+ path = summaries_dir / f"ch{ch:04d}.md"
|
|
|
+ if path.exists():
|
|
|
+ summaries.append({"chapter": ch, "summary": path.read_text(encoding="utf-8")})
|
|
|
+ return summaries
|
|
|
+
|
|
|
+ def _load_recent_meta(self, state: Dict[str, Any], chapter: int, window: int = 3) -> List[Dict[str, Any]]:
|
|
|
+ meta = state.get("chapter_meta", {}) or {}
|
|
|
+ results = []
|
|
|
+ for ch in range(max(1, chapter - window), chapter):
|
|
|
+ for key in (f"{ch:04d}", str(ch)):
|
|
|
+ if key in meta:
|
|
|
+ results.append({"chapter": ch, **meta.get(key, {})})
|
|
|
+ break
|
|
|
+ return results
|
|
|
+
|
|
|
+ def _load_recent_appearances(self) -> List[Dict[str, Any]]:
|
|
|
+ appearances = self.index_manager.get_recent_appearances()
|
|
|
+ return appearances or []
|
|
|
+
|
|
|
+ def _load_setting(self, keyword: str) -> str:
|
|
|
+ settings_dir = self.config.settings_dir
|
|
|
+ candidates = [
|
|
|
+ settings_dir / f"{keyword}.md",
|
|
|
+ ]
|
|
|
+ for path in candidates:
|
|
|
+ if path.exists():
|
|
|
+ return path.read_text(encoding="utf-8")
|
|
|
+ # fallback: any file containing keyword
|
|
|
+ matches = list(settings_dir.glob(f"*{keyword}*.md"))
|
|
|
+ if matches:
|
|
|
+ return matches[0].read_text(encoding="utf-8")
|
|
|
+ return f"[{keyword}设定未找到]"
|
|
|
+
|
|
|
+ def _load_json_optional(self, path: Path) -> Dict[str, Any]:
|
|
|
+ if not path.exists():
|
|
|
+ return {}
|
|
|
+ try:
|
|
|
+ return json.loads(path.read_text(encoding="utf-8"))
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ return {}
|
|
|
+
|
|
|
+
|
|
|
+def main():
|
|
|
+ import argparse
|
|
|
+ from .cli_output import print_success, print_error
|
|
|
+
|
|
|
+ parser = argparse.ArgumentParser(description="Context Manager CLI")
|
|
|
+ parser.add_argument("--project-root", type=str, help="项目根目录")
|
|
|
+ parser.add_argument("--chapter", type=int, required=True)
|
|
|
+ parser.add_argument("--template", type=str, default=ContextManager.DEFAULT_TEMPLATE)
|
|
|
+ parser.add_argument("--no-snapshot", action="store_true")
|
|
|
+ parser.add_argument("--max-chars", type=int, default=8000)
|
|
|
+
|
|
|
+ args = parser.parse_args()
|
|
|
+
|
|
|
+ config = None
|
|
|
+ if args.project_root:
|
|
|
+ from .config import DataModulesConfig
|
|
|
+
|
|
|
+ config = DataModulesConfig.from_project_root(args.project_root)
|
|
|
+
|
|
|
+ manager = ContextManager(config)
|
|
|
+ try:
|
|
|
+ payload = manager.build_context(
|
|
|
+ chapter=args.chapter,
|
|
|
+ template=args.template,
|
|
|
+ use_snapshot=not args.no_snapshot,
|
|
|
+ save_snapshot=True,
|
|
|
+ max_chars=args.max_chars,
|
|
|
+ )
|
|
|
+ print_success(payload, message="context_built")
|
|
|
+ try:
|
|
|
+ manager.index_manager.log_tool_call("context_manager:build", True, chapter=args.chapter)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ except Exception as exc:
|
|
|
+ print_error("CONTEXT_BUILD_FAILED", str(exc), suggestion="请检查项目结构与依赖文件")
|
|
|
+ try:
|
|
|
+ manager.index_manager.log_tool_call(
|
|
|
+ "context_manager:build", False, error_code="CONTEXT_BUILD_FAILED", error_message=str(exc), chapter=args.chapter
|
|
|
+ )
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ main()
|
lingfengQAQ