@dataclass(frozen=True)
class GraphFormatProfile:
"""Prompt, context serialization, and parser configuration for one LLM graph format."""
format: LLMGraphFormat
def context_fence_lang(self) -> str:
return "ttl" if self.format == LLMGraphFormat.TURTLE else "json"
def serialize_graph_for_prompt(self, graph: RDFGraph) -> str:
if self.format == LLMGraphFormat.TURTLE:
return graph.serialize_canonical_turtle()
return graph.serialize_compact_jsonld_for_prompt()
def format_ontology_chapter(
self,
graph: RDFGraph,
*,
suffix: str = "",
max_triples: int | None = None,
) -> str:
"""Serialize the ontology chapter, condensing it toward ``max_triples``.
This is the one point every ontology chapter passes through -- both unit
loops, render and critique, the shared snapshot and the ontology loop's
mutable working graph -- so it is where the prompt budget is enforced.
``suffix`` is built by the caller from the uncondensed graph, which stays
correct: the index only names terms by ``rdfs:label`` and their
domain/range, none of which condensing drops.
"""
condensed, _ = condense_graph_for_prompt(graph, max_triples)
body = self.serialize_graph_for_prompt(condensed)
chapter = f"\n\n# ONTOLOGY\n\n```{self.context_fence_lang()}\n{body}\n```\n"
return chapter + suffix
def format_facts_chapter(self, graph: RDFGraph) -> str:
body = self.serialize_graph_for_prompt(graph)
return (
"\n\n# SEMANTIC GRAPH OF FACTS\n"
"The following facts were extracted\n\n"
f"```{self.context_fence_lang()}\n{body}\n```\n"
)
def render_fresh_output_instruction(self, *, target: str = "facts") -> str:
if self.format == LLMGraphFormat.JSONLD:
return _OUTPUT_INSTRUCTION_JSONLD
if target == "ontology":
return _OUTPUT_INSTRUCTION_ONTOLOGY_TTL
return _OUTPUT_INSTRUCTION_FACTS_TTL
def render_update_output_instruction(self) -> str:
base = _OUTPUT_INSTRUCTION_GRAPH_UPDATE_BASE
if self.format == LLMGraphFormat.JSONLD:
return base + _OUTPUT_INSTRUCTION_GRAPH_UPDATE_JSONLD_GRAPH
return base + _OUTPUT_INSTRUCTION_GRAPH_UPDATE_TURTLE_GRAPH
def critique_graph_instruction(self) -> str:
if self.format == LLMGraphFormat.JSONLD:
return _OUTPUT_INSTRUCTION_CRITIQUE_JSONLD
return _OUTPUT_INSTRUCTION_CRITIQUE_TURTLE
def facts_operational_guidelines(
self,
*,
facts_namespace: str,
domain_ontologies_clause: str,
quantity_fallback_vocabulary: dict[str, str] | None = None,
search_guidelines: str = "",
) -> str:
return format_facts_operational_guidelines(
facts_namespace=facts_namespace,
domain_ontologies_clause=domain_ontologies_clause,
jsonld=self.format == LLMGraphFormat.JSONLD,
quantity_fallback_vocabulary=quantity_fallback_vocabulary,
search_guidelines=search_guidelines,
)
def format_instructions(
self,
report_cls: type[BaseModel],
*,
web_search_enabled: bool = True,
) -> str:
return format_instructions_for_model(
report_cls,
self.format,
web_search_enabled=web_search_enabled,
)
def parse_report(self, report_cls: type[T], text: str) -> T:
token = llm_graph_format_ctx.set(self.format)
try:
parser = PydanticOutputParser(pydantic_object=report_cls)
return parser.parse(text)
finally:
llm_graph_format_ctx.reset(token)
def llm_graph_format_context(self) -> AbstractContextManager[LLMGraphFormat]:
return _LLMGraphFormatContext(self.format)