Spaces:
Runtime error
Runtime error
File size: 6,107 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """In-place fact update on REFERENCE-tier baseline text — no scratch generation."""
from __future__ import annotations
import logging
from backend.config import settings
from backend.core.composition_output import (
accept_narrative_section_output,
sanitize_section_prose,
)
from backend.core.paragraph_merge import merge_observations_into_paragraph
from backend.core.paragraph_retriever import InterferenceLevel
from backend.llm import openai_client
from backend.models.schema import TemplateSchema
from backend.prompts.mapping_prompt import (
FACT_GROUNDING_RULES,
MAPPING_SYSTEM_BASE_MAXIMUM,
MAPPING_SYSTEM_BASE_MEDIUM,
MAPPING_SYSTEM_BASE_MINIMUM,
MAPPING_USER_TEMPLATE,
RATING_HINT_TEMPLATE,
RICS_DOMAIN_RULES,
_observations_bulleted,
)
logger = logging.getLogger(__name__)
# Relocated verbatim to backend/prompts/mapping_prompt.py (test-exempt). Aliased
# here to preserve the existing internal/private reference and import surface.
_RICS_DOMAIN_RULES = RICS_DOMAIN_RULES
def select_mapping_prompt(interference_level: str) -> str:
"""Return the tier-specific mapping system prompt base for the given level."""
level = (interference_level or "maximum").strip().lower()
if level == "minimum":
return MAPPING_SYSTEM_BASE_MINIMUM
if level == "medium":
return MAPPING_SYSTEM_BASE_MEDIUM
return MAPPING_SYSTEM_BASE_MAXIMUM
def _normalize_interference_level(
interference_level: str | InterferenceLevel | None,
) -> InterferenceLevel:
"""Map caller input to a supported composition tier; default maximum."""
raw = str(interference_level or "").strip().lower()
if raw in ("minimum", "medium", "maximum"):
return raw # type: ignore[return-value]
return "maximum"
def compose_mapping_system_prompt(
interference_level: str | InterferenceLevel | None,
) -> str:
"""Full mapping system prompt: tier base + shared RICS domain rules."""
level = _normalize_interference_level(interference_level)
return (
select_mapping_prompt(level).strip()
+ "\n\n"
+ FACT_GROUNDING_RULES.strip()
+ "\n\n"
+ _RICS_DOMAIN_RULES.strip()
)
def build_interference_messages(
interference_level: str | InterferenceLevel | None,
*,
observations: list[str],
baseline: str,
schema: TemplateSchema,
section_id: str = "",
section_title: str = "",
rating_value: str | None = None,
extra_references: list[str] | None = None,
) -> list[dict[str, str]]:
"""Select the prompt builder for minimum / medium / maximum AI involvement."""
level = _normalize_interference_level(interference_level)
baseline_text = baseline.strip()
if extra_references:
extras = "\n\n".join(item.strip() for item in extra_references if item.strip())
if extras:
baseline_text = f"{baseline_text}\n\n{extras}".strip()
rating_line = ""
if schema.rating_system.detected and rating_value:
rating_line = RATING_HINT_TEMPLATE.format(rating_value=rating_value)
system = compose_mapping_system_prompt(level)
user = MAPPING_USER_TEMPLATE.format(
section_id=section_id or "—",
section_label=section_title or section_id or "—",
rating_line=rating_line,
first_reference_baseline_paragraph=baseline_text or "(none)",
observations_bulleted=_observations_bulleted(observations),
)
return [
{"role": "system", "content": system.strip()},
{"role": "user", "content": user.strip()},
]
def map_inplace_baseline(
baseline_text: str,
observations: list[str],
schema: TemplateSchema,
*,
section_id: str = "",
section_title: str = "",
rating_value: str | None = None,
messages: list[dict[str, str]] | None = None,
) -> str:
"""Apply in-place fact updates to the retrieved past-report baseline only."""
baseline = (baseline_text or "").strip()
if not baseline:
return ""
if not observations:
return baseline
merged = merge_observations_into_paragraph(baseline, observations, schema)
if settings.use_llm_paragraph_mapping and openai_client.is_available():
llm_messages = messages or build_interference_messages(
"maximum",
observations=observations,
baseline=baseline,
schema=schema,
section_id=section_id,
section_title=section_title,
rating_value=rating_value,
)
try:
out = openai_client.chat_text(
llm_messages,
model=settings.mapping_model,
max_tokens=settings.max_tokens_mapping,
temperature=0.0,
)
text = sanitize_section_prose((out or "").strip())
if accept_narrative_section_output(text, observations):
return text
except Exception as exc: # noqa: BLE001
logger.warning(
"In-place LLM mapping failed (%s); using deterministic merge.", exc
)
return merged
def map_reference_paragraph(
reference_paragraph: str,
observations: list[str],
schema: TemplateSchema,
interference_level: InterferenceLevel,
*,
section_id: str = "",
section_title: str = "",
rating_value: str | None = None,
extra_references: list[str] | None = None,
) -> str:
"""Map notes onto the assembled REFERENCE baseline using tier-specific prompts."""
level = _normalize_interference_level(interference_level)
messages = build_interference_messages(
level,
observations=observations,
baseline=reference_paragraph,
schema=schema,
section_id=section_id,
section_title=section_title,
rating_value=rating_value,
extra_references=extra_references,
)
return map_inplace_baseline(
reference_paragraph,
observations,
schema,
section_id=section_id,
section_title=section_title,
rating_value=rating_value,
messages=messages,
)
|