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 
20
21
22
23
24
25
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 
23
24
25
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 
28
29
30
31
32
33
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 
31
32
33
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.

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

    def __init__(
        self,
        llm_provider: AtomicLLMProvider,
        search_provider: AtomicSearchProvider | None = None,
        web_search_config: WebSearchConfig | None = None,
        web_search_enabled: bool = False,
        web_search_top_k: int = 3,
        web_search_max_snippet_chars: int = 400,
        web_search_max_total_chars: int = 1800,
        web_search_for_ontology_render: bool = True,
        web_search_for_ontology_critic: bool = True,
        web_search_for_facts_render: bool = False,
        web_search_for_facts_critic: bool = False,
        web_search_planner_enabled: bool = True,
        web_search_planner_max_queries: int = 3,
        web_search_planner_min_query_chars: int = 12,
        web_search_planner_min_confidence: float = 0.35,
        web_search_reuse_evidence_across_attempt: bool = True,
        web_search_allowed_domains: tuple[str, ...] = (),
        web_search_blocked_domains: tuple[str, ...] = (),
        web_search_min_snippet_chars: int = 40,
    ):
        self.llm_provider = llm_provider
        self.search_provider = search_provider
        self.web_search_config = web_search_config

        if web_search_config is not None:
            self.web_search_enabled = web_search_config.enabled
            self.web_search_top_k = web_search_config.top_k
            self.web_search_max_snippet_chars = web_search_config.max_snippet_chars
            self.web_search_max_total_chars = web_search_config.max_total_chars
            self.web_search_for_ontology_render = (
                web_search_config.ontology_render_enabled
            )
            self.web_search_for_ontology_critic = (
                web_search_config.ontology_critic_enabled
            )
            self.web_search_for_facts_render = web_search_config.facts_render_enabled
            self.web_search_for_facts_critic = web_search_config.facts_critic_enabled
            self.web_search_planner_enabled = web_search_config.planner_enabled
            self.web_search_planner_max_queries = web_search_config.planner_max_queries
            self.web_search_planner_min_query_chars = (
                web_search_config.planner_min_query_chars
            )
            self.web_search_planner_min_confidence = (
                web_search_config.planner_min_confidence
            )
            self.web_search_reuse_evidence_across_attempt = (
                web_search_config.reuse_evidence_across_attempt
            )
            self.web_search_allowed_domains = {
                value.strip().lower()
                for value in web_search_config.allowed_domains
                if value.strip()
            }
            self.web_search_blocked_domains = {
                value.strip().lower()
                for value in web_search_config.blocked_domains
                if value.strip()
            }
            self.web_search_min_snippet_chars = web_search_config.min_snippet_chars
        else:
            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_for_ontology_render
            self.web_search_for_ontology_critic = web_search_for_ontology_critic
            self.web_search_for_facts_render = web_search_for_facts_render
            self.web_search_for_facts_critic = web_search_for_facts_critic
            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 = {
                value.strip().lower()
                for value in web_search_allowed_domains
                if value.strip()
            }
            self.web_search_blocked_domains = {
                value.strip().lower()
                for value in web_search_blocked_domains
                if value.strip()
            }
            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)

get_llm_tool(budget_tracker) async

Return a budget-aware LLM tool instance.

Source code in ontocast/tool/atomic.py 
128
129
130
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 
132
133
134
135
136
137
138
139
140
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 
142
143
144
145
146
147
148
149
150
151
152
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 
12
13
14
15
16
17
class SearchHit(BaseModel):
    """Single web-search hit used as optional grounding context."""

    title: str
    url: str
    snippet: str