Skip to content

ontocast.tool.atomic

Minimal tool contracts for atomic render/critic loops.

AtomicLLMProvider

Bases: Protocol

Provides budget-aware LLM instances for atomic loop calls.

Source code in ontocast/tool/atomic.py 
21
22
23
24
25
26
class AtomicLLMProvider(Protocol):
    """Provides budget-aware LLM instances for atomic loop calls."""

    async def get_llm_tool(self, budget_tracker) -> LLMTool:
        """Return an LLM tool tied to the given budget tracker."""
        ...

get_llm_tool(budget_tracker) async

Return an LLM tool tied to the given budget tracker.

Source code in ontocast/tool/atomic.py 
24
25
26
async def get_llm_tool(self, budget_tracker) -> LLMTool:
    """Return an LLM tool tied to the given budget tracker."""
    ...

AtomicSearchProvider

Bases: Protocol

Provides optional web-search retrieval for ontology grounding.

Source code in ontocast/tool/atomic.py 
29
30
31
32
33
34
class AtomicSearchProvider(Protocol):
    """Provides optional web-search retrieval for ontology grounding."""

    async def search(self, query: str, max_results: int) -> list[SearchHit]:
        """Return web hits relevant to the query."""
        ...

search(query, max_results) async

Return web hits relevant to the query.

Source code in ontocast/tool/atomic.py 
32
33
34
async def search(self, query: str, max_results: int) -> list[SearchHit]:
    """Return web hits relevant to the query."""
    ...

AtomicToolBox

Small tool surface used by atomic render/critic paths.

Configuration arrives as config sections, never as unpacked scalars. An earlier signature accepted both a :class:WebSearchConfig and seventeen flat web_search_* parameters mirroring its fields, chosen between by an if/else; production passed the section and only tests took the flat branch, so the tested configuration path was not the one that shipped. Each default also existed three times -- here, in settings.py, and inline at the read sites. Now settings.py is the single source.

Source code in ontocast/tool/atomic.py 
 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
class AtomicToolBox:
    """Small tool surface used by atomic render/critic paths.

    Configuration arrives as config *sections*, never as unpacked scalars. An
    earlier signature accepted both a :class:`WebSearchConfig` and seventeen
    flat ``web_search_*`` parameters mirroring its fields, chosen between by an
    ``if/else``; production passed the section and only tests took the flat
    branch, so the tested configuration path was not the one that shipped. Each
    default also existed three times -- here, in ``settings.py``, and inline at
    the read sites. Now ``settings.py`` is the single source.
    """

    def __init__(
        self,
        llm_provider: AtomicLLMProvider,
        search_provider: AtomicSearchProvider | None = None,
        web_search_config: WebSearchConfig | None = None,
        facts_validation_config: FactsValidationConfig | None = None,
        citation_vocabulary: dict[str, str] | None = None,
    ):
        """Build the atomic tool surface.

        Args:
            llm_provider: Supplies budget-aware LLM tools.
            search_provider: Optional web-search backend. Without one, search
                returns no hits regardless of configuration.
            web_search_config: Web-grounding settings. Defaults to
                :class:`WebSearchConfig`, which is disabled unless configured.
            facts_validation_config: Facts-gate settings consumed by the render
                and repair paths. Defaults to :class:`FactsValidationConfig`.
            citation_vocabulary: Bibliographic terms for citation-metadata
                units. Configuration rather than retrieval: a reference list is
                not domain content, so its vocabulary never reaches the catalog.
        """
        web_search = web_search_config or WebSearchConfig()
        facts_validation = facts_validation_config or FactsValidationConfig()

        self.llm_provider = llm_provider
        self.search_provider = search_provider
        self.web_search_config = web_search

        self.object_property_literal_check = (
            facts_validation.object_property_literal_check
        )
        # Finding-driven repair renders: each one is a provider call.
        self.facts_llm_repair_visits = facts_validation.llm_repair_visits
        # Code predicates for the LLM-free code -> catalog IRI repair.
        self.code_predicates: tuple[str, ...] = tuple(facts_validation.code_predicates)
        self.property_alias_min_ratio = facts_validation.property_alias_min_ratio
        self.citation_vocabulary: dict[str, str] = dict(citation_vocabulary or {})
        # Fallback vocabulary the facts prompt names for bounded quantities when
        # retrieval supplied no suitable class. An explicitly empty mapping
        # forbids the fallback.
        self.quantity_fallback_vocabulary: dict[str, str] | None = dict(
            facts_validation.quantity_fallback_vocabulary
        )
        # Non-meta vocabularies a deployment shares across catalogs and does not
        # want reported as unknown terms.
        self.additional_standard_namespaces: tuple[str, ...] = tuple(
            facts_validation.additional_standard_namespaces
        )
        # Everything the deterministic term checks must treat as blessed, as
        # one object -- see ValidationPolicy.
        self.validation_policy = ValidationPolicy(
            additional_standard_namespaces=self.additional_standard_namespaces,
            quantity_fallback_vocabulary=self.quantity_fallback_vocabulary,
            code_predicates=self.code_predicates,
        )
        # A sibling of ValidationPolicy, deliberately not a field on it.
        # ValidationPolicy answers "what must never be flagged"; this answers
        # "what blocks a unit from leaving the loop". Both travel to the unit
        # loop, but the term checks and the catalog lint have no business
        # knowing about acceptance.
        self.acceptance_policy = FactsAcceptancePolicy(
            blocking_fix_severity=facts_validation.accept_blocking_severity,
        )

        self.web_search_enabled = web_search.enabled
        self.web_search_top_k = web_search.top_k
        self.web_search_max_snippet_chars = web_search.max_snippet_chars
        self.web_search_max_total_chars = web_search.max_total_chars
        self.web_search_for_ontology_render = web_search.ontology_render_enabled
        self.web_search_for_ontology_critic = web_search.ontology_critic_enabled
        self.web_search_for_facts_render = web_search.facts_render_enabled
        self.web_search_for_facts_critic = web_search.facts_critic_enabled
        self.web_search_planner_enabled = web_search.planner_enabled
        self.web_search_planner_max_queries = web_search.planner_max_queries
        self.web_search_planner_min_query_chars = web_search.planner_min_query_chars
        self.web_search_planner_min_confidence = web_search.planner_min_confidence
        self.web_search_reuse_evidence_across_attempt = (
            web_search.reuse_evidence_across_attempt
        )
        self.web_search_allowed_domains = _domain_set(web_search.allowed_domains)
        self.web_search_blocked_domains = _domain_set(web_search.blocked_domains)
        self.web_search_min_snippet_chars = web_search.min_snippet_chars

    async def get_llm_tool(self, budget_tracker) -> LLMTool:
        """Return a budget-aware LLM tool instance."""
        return await self.llm_provider.get_llm_tool(budget_tracker)

    async def search(
        self, query: str, max_results: int | None = None
    ) -> list[SearchHit]:
        """Run optional web search and return normalized hits."""
        if not self.web_search_enabled or self.search_provider is None:
            return []

        result_limit = max_results if max_results is not None else self.web_search_top_k
        return await self.search_provider.search(query=query, max_results=result_limit)

    def web_grounding_enabled_for_node(self, node: WorkflowNode) -> bool:
        """Return whether web grounding is enabled for a workflow node."""
        if not self.web_search_enabled:
            return False
        mapping = {
            WorkflowNode.TEXT_TO_ONTOLOGY: self.web_search_for_ontology_render,
            WorkflowNode.CRITICISE_ONTOLOGY: self.web_search_for_ontology_critic,
            WorkflowNode.TEXT_TO_FACTS: self.web_search_for_facts_render,
            WorkflowNode.CRITICISE_FACTS: self.web_search_for_facts_critic,
        }
        return mapping.get(node, False)

__init__(llm_provider, search_provider=None, web_search_config=None, facts_validation_config=None, citation_vocabulary=None)

Build the atomic tool surface.

Parameters:

Name Type Description Default
llm_provider AtomicLLMProvider

Supplies budget-aware LLM tools.

 required
search_provider AtomicSearchProvider | None

Optional web-search backend. Without one, search returns no hits regardless of configuration.

 None
web_search_config WebSearchConfig | None

Web-grounding settings. Defaults to :class:WebSearchConfig, which is disabled unless configured.

 None
facts_validation_config FactsValidationConfig | None

Facts-gate settings consumed by the render and repair paths. Defaults to :class:FactsValidationConfig.

 None
citation_vocabulary dict[str, str] | None

Bibliographic terms for citation-metadata units. Configuration rather than retrieval: a reference list is not domain content, so its vocabulary never reaches the catalog.

 None
Source code in ontocast/tool/atomic.py 
 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
def __init__(
    self,
    llm_provider: AtomicLLMProvider,
    search_provider: AtomicSearchProvider | None = None,
    web_search_config: WebSearchConfig | None = None,
    facts_validation_config: FactsValidationConfig | None = None,
    citation_vocabulary: dict[str, str] | None = None,
):
    """Build the atomic tool surface.

    Args:
        llm_provider: Supplies budget-aware LLM tools.
        search_provider: Optional web-search backend. Without one, search
            returns no hits regardless of configuration.
        web_search_config: Web-grounding settings. Defaults to
            :class:`WebSearchConfig`, which is disabled unless configured.
        facts_validation_config: Facts-gate settings consumed by the render
            and repair paths. Defaults to :class:`FactsValidationConfig`.
        citation_vocabulary: Bibliographic terms for citation-metadata
            units. Configuration rather than retrieval: a reference list is
            not domain content, so its vocabulary never reaches the catalog.
    """
    web_search = web_search_config or WebSearchConfig()
    facts_validation = facts_validation_config or FactsValidationConfig()

    self.llm_provider = llm_provider
    self.search_provider = search_provider
    self.web_search_config = web_search

    self.object_property_literal_check = (
        facts_validation.object_property_literal_check
    )
    # Finding-driven repair renders: each one is a provider call.
    self.facts_llm_repair_visits = facts_validation.llm_repair_visits
    # Code predicates for the LLM-free code -> catalog IRI repair.
    self.code_predicates: tuple[str, ...] = tuple(facts_validation.code_predicates)
    self.property_alias_min_ratio = facts_validation.property_alias_min_ratio
    self.citation_vocabulary: dict[str, str] = dict(citation_vocabulary or {})
    # Fallback vocabulary the facts prompt names for bounded quantities when
    # retrieval supplied no suitable class. An explicitly empty mapping
    # forbids the fallback.
    self.quantity_fallback_vocabulary: dict[str, str] | None = dict(
        facts_validation.quantity_fallback_vocabulary
    )
    # Non-meta vocabularies a deployment shares across catalogs and does not
    # want reported as unknown terms.
    self.additional_standard_namespaces: tuple[str, ...] = tuple(
        facts_validation.additional_standard_namespaces
    )
    # Everything the deterministic term checks must treat as blessed, as
    # one object -- see ValidationPolicy.
    self.validation_policy = ValidationPolicy(
        additional_standard_namespaces=self.additional_standard_namespaces,
        quantity_fallback_vocabulary=self.quantity_fallback_vocabulary,
        code_predicates=self.code_predicates,
    )
    # A sibling of ValidationPolicy, deliberately not a field on it.
    # ValidationPolicy answers "what must never be flagged"; this answers
    # "what blocks a unit from leaving the loop". Both travel to the unit
    # loop, but the term checks and the catalog lint have no business
    # knowing about acceptance.
    self.acceptance_policy = FactsAcceptancePolicy(
        blocking_fix_severity=facts_validation.accept_blocking_severity,
    )

    self.web_search_enabled = web_search.enabled
    self.web_search_top_k = web_search.top_k
    self.web_search_max_snippet_chars = web_search.max_snippet_chars
    self.web_search_max_total_chars = web_search.max_total_chars
    self.web_search_for_ontology_render = web_search.ontology_render_enabled
    self.web_search_for_ontology_critic = web_search.ontology_critic_enabled
    self.web_search_for_facts_render = web_search.facts_render_enabled
    self.web_search_for_facts_critic = web_search.facts_critic_enabled
    self.web_search_planner_enabled = web_search.planner_enabled
    self.web_search_planner_max_queries = web_search.planner_max_queries
    self.web_search_planner_min_query_chars = web_search.planner_min_query_chars
    self.web_search_planner_min_confidence = web_search.planner_min_confidence
    self.web_search_reuse_evidence_across_attempt = (
        web_search.reuse_evidence_across_attempt
    )
    self.web_search_allowed_domains = _domain_set(web_search.allowed_domains)
    self.web_search_blocked_domains = _domain_set(web_search.blocked_domains)
    self.web_search_min_snippet_chars = web_search.min_snippet_chars

get_llm_tool(budget_tracker) async

Return a budget-aware LLM tool instance.

Source code in ontocast/tool/atomic.py 
138
139
140
async def get_llm_tool(self, budget_tracker) -> LLMTool:
    """Return a budget-aware LLM tool instance."""
    return await self.llm_provider.get_llm_tool(budget_tracker)

search(query, max_results=None) async

Run optional web search and return normalized hits.

Source code in ontocast/tool/atomic.py 
142
143
144
145
146
147
148
149
150
async def search(
    self, query: str, max_results: int | None = None
) -> list[SearchHit]:
    """Run optional web search and return normalized hits."""
    if not self.web_search_enabled or self.search_provider is None:
        return []

    result_limit = max_results if max_results is not None else self.web_search_top_k
    return await self.search_provider.search(query=query, max_results=result_limit)

web_grounding_enabled_for_node(node)

Return whether web grounding is enabled for a workflow node.

Source code in ontocast/tool/atomic.py 
152
153
154
155
156
157
158
159
160
161
162
def web_grounding_enabled_for_node(self, node: WorkflowNode) -> bool:
    """Return whether web grounding is enabled for a workflow node."""
    if not self.web_search_enabled:
        return False
    mapping = {
        WorkflowNode.TEXT_TO_ONTOLOGY: self.web_search_for_ontology_render,
        WorkflowNode.CRITICISE_ONTOLOGY: self.web_search_for_ontology_critic,
        WorkflowNode.TEXT_TO_FACTS: self.web_search_for_facts_render,
        WorkflowNode.CRITICISE_FACTS: self.web_search_for_facts_critic,
    }
    return mapping.get(node, False)

SearchHit

Bases: BaseModel

Single web-search hit used as optional grounding context.

Source code in ontocast/tool/atomic.py 
13
14
15
16
17
18
class SearchHit(BaseModel):
    """Single web-search hit used as optional grounding context."""

    title: str
    url: str
    snippet: str