a2d46a7c6e
Ruff autofix removed pre-existing F401 unused imports across adapters, context, compiler, extractor, scheduler, web, and main. Also reordered models.py so its SQLAlchemy imports sit at the top of the file (E402 was triggered by the inline `utcnow` helper definition). Added a [dependency-groups] dev block (pytest, pytest-cov, ruff) so the zero-check test gauntlet — which runs `pytest --cov --cov-report=term-missing` — can resolve its plugins without a manual `uv pip install pytest-cov`.
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""
|
|
Context assembler — Phase 1 (thin).
|
|
|
|
Builds the user-turn prompt bundle that gets sent to the extraction engine.
|
|
Phase 1 bundle = domain template + optional focus directive + transcript text.
|
|
|
|
Future phases can layer in wiki context, related sources, etc.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
from second_brain.config import Config
|
|
from second_brain.models import Source
|
|
|
|
|
|
_FALLBACK_TEMPLATE = """\
|
|
You are extracting structured knowledge from the following source.
|
|
Domain: {domain}
|
|
{focus_section}
|
|
|
|
Return a JSON object with these keys:
|
|
- summary (string): 2-4 sentence overview
|
|
- key_points (list of strings): 5-10 takeaways
|
|
- entities (list of strings): notable people, tools, products, concepts
|
|
- claims (list of strings): specific factual assertions
|
|
- open_questions (list of strings): questions raised but unanswered
|
|
- contradictions (list of strings): points that contradict known facts
|
|
|
|
Source transcript / content:
|
|
{transcript}
|
|
"""
|
|
|
|
|
|
class ContextAssembler:
|
|
"""Assembles the prompt bundle for a single source."""
|
|
|
|
def __init__(self, config: Config) -> None:
|
|
self.config = config
|
|
self._template_cache: dict[str, str] = {}
|
|
|
|
def assemble(self, source: Source) -> str:
|
|
"""Return the full user-turn prompt string for the extraction engine."""
|
|
template = self._load_template(source.domain)
|
|
transcript = source.transcript_text or ""
|
|
|
|
focus_section = ""
|
|
if source.focus:
|
|
focus_section = f"Focus area: {source.focus}\n"
|
|
|
|
return template.format(
|
|
domain=source.domain,
|
|
focus_section=focus_section,
|
|
transcript=transcript,
|
|
)
|
|
|
|
def _load_template(self, domain: str) -> str:
|
|
"""Load and cache the domain prompt template from prompts/<domain>.md."""
|
|
if domain in self._template_cache:
|
|
return self._template_cache[domain]
|
|
|
|
template_path = self.config.prompts_dir / f"{domain}.md"
|
|
if template_path.exists():
|
|
raw = template_path.read_text(encoding="utf-8")
|
|
template = self._extract_template_body(raw)
|
|
else:
|
|
print(
|
|
f"[Assembler] Warning: no prompt template for domain {domain!r}, "
|
|
"using fallback"
|
|
)
|
|
template = _FALLBACK_TEMPLATE
|
|
|
|
self._template_cache[domain] = template
|
|
return template
|
|
|
|
@staticmethod
|
|
def _extract_template_body(raw: str) -> str:
|
|
"""Strip YAML front-matter if present, return body."""
|
|
if raw.startswith("---"):
|
|
parts = raw.split("---", 2)
|
|
if len(parts) >= 3:
|
|
return parts[2].strip()
|
|
return raw.strip()
|
|
|
|
|
|
__all__ = ["ContextAssembler"]
|