154 lines
5.9 KiB
Python
154 lines
5.9 KiB
Python
# #region ContextAwarePromptBuilder [C:2] [TYPE Module] [SEMANTICS translate, prompt, context, dictionary]
|
|
# @BRIEF Pure-function prompt builder that enhances dictionary entries with context annotations.
|
|
# @LAYER Domain
|
|
# @RELATION DEPENDS_ON -> [DictionaryEntry]
|
|
# @RATIONALE Pure functions only — no I/O, no DB access. Separated from executor for testability.
|
|
# @REJECTED Embedding context inline in the executor would make it untestable without mocking DB.
|
|
|
|
#
|
|
# Typical workflow:
|
|
# 1. Call build_context_entries(dictionary_entries, row_context) to get annotated, prioritized entries
|
|
# 2. Each entry is rendered via render_entry() with optional priority flag
|
|
# 3. Jaccard similarity >= 0.5 triggers priority flagging
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
# #region ContextAwarePromptBuilder [C:2] [TYPE Class]
|
|
# @BRIEF Build LLM prompts with context-aware dictionary entries and similarity-based priority.
|
|
|
|
class ContextAwarePromptBuilder:
|
|
"""Build LLM prompts with context-aware dictionary entries.
|
|
|
|
Pure function — no I/O, no DB access.
|
|
"""
|
|
|
|
@staticmethod
|
|
def render_entry(entry: Any, priority: bool = False, row_context: dict | None = None) -> str:
|
|
"""Render a dictionary entry for the LLM prompt.
|
|
|
|
Args:
|
|
entry: DictionaryEntry-like object (must have source_term, target_term, has_context,
|
|
context_data, usage_notes attrs or dict keys).
|
|
priority: Whether this entry should be flagged as high priority.
|
|
row_context: Optional row context dict (unused in rendering, kept for API symmetry).
|
|
|
|
Returns:
|
|
Rendered prompt line string.
|
|
"""
|
|
# Support both object attrs and dict access
|
|
if isinstance(entry, dict):
|
|
source_term = entry.get("source_term", "")
|
|
target_term = entry.get("target_term", "")
|
|
has_context = entry.get("has_context", False)
|
|
context_data = entry.get("context_data")
|
|
usage_notes = entry.get("usage_notes")
|
|
else:
|
|
source_term = getattr(entry, "source_term", "")
|
|
target_term = getattr(entry, "target_term", "")
|
|
has_context = getattr(entry, "has_context", False)
|
|
context_data = getattr(entry, "context_data", None)
|
|
usage_notes = getattr(entry, "usage_notes", None)
|
|
|
|
# Build base line
|
|
line = f'"{source_term}" -> "{target_term}"'
|
|
|
|
# Add context annotation if present
|
|
if has_context and context_data:
|
|
context_items = []
|
|
if isinstance(context_data, dict):
|
|
context_items = [f"{k}={v}" for k, v in context_data.items()]
|
|
elif isinstance(context_data, str):
|
|
try:
|
|
parsed = json.loads(context_data)
|
|
if isinstance(parsed, dict):
|
|
context_items = [f"{k}={v}" for k, v in parsed.items()]
|
|
else:
|
|
context_items = [str(context_data)]
|
|
except (json.JSONDecodeError, TypeError):
|
|
context_items = [str(context_data)]
|
|
|
|
context_str = ", ".join(context_items)
|
|
|
|
# Truncate if too long (500 tokens ≈ 2000 chars)
|
|
if len(context_str) > 2000:
|
|
context_str = context_str[:1997] + "...[truncated]"
|
|
|
|
line = f'"{source_term}" (context: {context_str}) -> "{target_term}"'
|
|
|
|
# Add usage notes
|
|
if usage_notes:
|
|
notes = str(usage_notes)[:200] # cap at 200 chars
|
|
line += f" # Usage: {notes}"
|
|
|
|
# Add priority prefix
|
|
if priority:
|
|
line = f"# PRIORITY (context match) — {line}"
|
|
|
|
return line
|
|
|
|
@staticmethod
|
|
def compute_context_similarity(entry_context: dict | None, row_context: dict | None) -> float:
|
|
"""Jaccard similarity between entry context and row context. Returns 0.0-1.0.
|
|
|
|
Compares the sets of lowercased string values from both contexts.
|
|
Returns 1.0 for identical contexts, 0.0 for disjoint or missing.
|
|
"""
|
|
if not entry_context or not row_context:
|
|
return 0.0
|
|
|
|
entry_vals = set(str(v).lower() for v in entry_context.values() if v is not None)
|
|
row_vals = set(str(v).lower() for v in row_context.values() if v is not None)
|
|
|
|
if not entry_vals or not row_vals:
|
|
return 0.0
|
|
|
|
intersection = entry_vals & row_vals
|
|
union = entry_vals | row_vals
|
|
return len(intersection) / len(union)
|
|
|
|
@staticmethod
|
|
def build_context_entries(
|
|
dictionary_entries: list[Any],
|
|
row_context: dict | None = None,
|
|
) -> list[str]:
|
|
"""Build prioritized dictionary entry list with context annotations.
|
|
|
|
Args:
|
|
dictionary_entries: List of DictionaryEntry-like objects or dicts.
|
|
row_context: Optional dict of current row's context columns.
|
|
|
|
Returns:
|
|
List of rendered prompt strings, sorted with priority entries first.
|
|
"""
|
|
results: list[tuple[Any, bool]] = []
|
|
|
|
for entry in dictionary_entries:
|
|
priority = False
|
|
if row_context:
|
|
# Extract entry context_data
|
|
if isinstance(entry, dict):
|
|
entry_context = entry.get("context_data")
|
|
else:
|
|
entry_context = getattr(entry, "context_data", None)
|
|
|
|
if entry_context:
|
|
similarity = ContextAwarePromptBuilder.compute_context_similarity(
|
|
entry_context, row_context
|
|
)
|
|
priority = similarity >= 0.5
|
|
|
|
results.append((entry, priority))
|
|
|
|
# Sort: priority first, then non-priority
|
|
results.sort(key=lambda x: (not x[1]))
|
|
|
|
return [
|
|
ContextAwarePromptBuilder.render_entry(entry, priority, row_context)
|
|
for entry, priority in results
|
|
]
|
|
# #endregion ContextAwarePromptBuilder
|
|
|
|
|
|
# #endregion ContextAwarePromptBuilder
|