ontocast.integrations.langgraph

Embed the OntoCast pipeline as a node in someone else's LangGraph.

AgentState declares no Annotated[..., reducer] channels and every node returns the whole state, so adding the compiled graph directly to a parent StateGraph only works when the parent's state literally has raw_input, docling_doc, aggregated_facts and the rest. input_schema and output_schema narrow which of AgentState's own keys cross the boundary but cannot rename them, so they do not bridge a foreign state either.

:func:make_ontocast_node therefore asks for the mapping explicitly. That is 30 lines of adapter instead of a reducer refactor, and it is honest about where the boundary is.

make_ontocast_node(tools, *, to_agent_state, from_agent_state, recursion_limit=None, graph=None)

Build a node that runs the OntoCast pipeline inside another graph.

to_state, from_state = text_in_turtle_out()
node = make_ontocast_node(tools, to_agent_state=to_state, from_agent_state=from_state)

builder = StateGraph(MyState)
builder.add_node("extract", node)

Parameters:

Name	Type	Description	Default
tools	'ToolBox'	The dependency container. The graph is compiled once here, not per invocation.	required
to_agent_state	Callable[[Any], AgentState]	Maps the parent state to a fresh AgentState.	required
from_agent_state	Callable[[AgentState, Any], dict[str, Any]]	Maps the finished AgentState and the original parent state to a parent-state delta.	required
recursion_limit	int | None	LangGraph recursion limit for the inner run. Defaults to a value derived from the configured chunk budget. Leaving this unset is safer than passing LangGraph's default of 25, which a multi-chunk document exceeds.	None
graph	CompiledStateGraph | None	A pre-compiled OntoCast graph to reuse instead of compiling one.	None

Returns:

Type	Description
Callable[[Any, RunnableConfig], Awaitable[dict[str, Any]]]	An async node callable suitable for StateGraph.add_node.

Source code in ontocast/integrations/langgraph.py

def make_ontocast_node(
    tools: "ToolBox",
    *,
    to_agent_state: Callable[[Any], AgentState],
    from_agent_state: Callable[[AgentState, Any], dict[str, Any]],
    recursion_limit: int | None = None,
    graph: CompiledStateGraph | None = None,
) -> Callable[[Any, RunnableConfig], Awaitable[dict[str, Any]]]:
    """Build a node that runs the OntoCast pipeline inside another graph.

    ```python
    to_state, from_state = text_in_turtle_out()
    node = make_ontocast_node(tools, to_agent_state=to_state, from_agent_state=from_state)

    builder = StateGraph(MyState)
    builder.add_node("extract", node)
    ```

    Args:
        tools: The dependency container. The graph is compiled once here, not
            per invocation.
        to_agent_state: Maps the parent state to a fresh ``AgentState``.
        from_agent_state: Maps the finished ``AgentState`` and the original
            parent state to a parent-state delta.
        recursion_limit: LangGraph recursion limit for the inner run. Defaults
            to a value derived from the configured chunk budget. **Leaving this
            unset is safer than passing LangGraph's default of 25**, which a
            multi-chunk document exceeds.
        graph: A pre-compiled OntoCast graph to reuse instead of compiling one.

    Returns:
        An async node callable suitable for ``StateGraph.add_node``.
    """
    compiled = graph or create_agent_graph(tools, name="ontocast")
    limit = recursion_limit if recursion_limit is not None else _default_limit(tools)

    async def ontocast_node(state: Any, config: RunnableConfig) -> dict[str, Any]:
        initial = to_agent_state(state)
        # Merge rather than replace: the caller's config carries callbacks,
        # tags and run metadata that tracing depends on.
        merged: RunnableConfig = {**(config or {})}
        merged["recursion_limit"] = merged.get("recursion_limit") or limit

        # `ainvoke`, not `astream`: the HTTP layer streams because it wants
        # intermediate node output, and an embedded node does not.
        result = await compiled.ainvoke(initial, merged)

        # LangGraph hands back a plain dict for a pydantic state schema.
        final = (
            result
            if isinstance(result, AgentState)
            else AgentState.model_validate(result)
        )
        return from_agent_state(final, state)

    return ontocast_node

text_in_turtle_out(*, text_key='input', ontology_key='ontology_ttl', facts_key='facts_ttl')

Return a ready-made mapping pair for the common text-to-Turtle case.

Reads a string off the parent state and writes back two Turtle strings, so a parent state needs only those three plain keys.

Parameters:

Name	Type	Description	Default
text_key	str	Parent-state key holding the source text.	'input'
ontology_key	str	Parent-state key to write the ontology Turtle to.	'ontology_ttl'
facts_key	str	Parent-state key to write the facts Turtle to.	'facts_ttl'

Returns:

Type	Description
tuple[Callable[[Any], AgentState], Callable[[AgentState, Any], dict[str, Any]]]	The (to_agent_state, from_agent_state) pair.

Source code in ontocast/integrations/langgraph.py

def text_in_turtle_out(
    *,
    text_key: str = "input",
    ontology_key: str = "ontology_ttl",
    facts_key: str = "facts_ttl",
) -> tuple[Callable[[Any], AgentState], Callable[[AgentState, Any], dict[str, Any]]]:
    """Return a ready-made mapping pair for the common text-to-Turtle case.

    Reads a string off the parent state and writes back two Turtle strings, so
    a parent state needs only those three plain keys.

    Args:
        text_key: Parent-state key holding the source text.
        ontology_key: Parent-state key to write the ontology Turtle to.
        facts_key: Parent-state key to write the facts Turtle to.

    Returns:
        The ``(to_agent_state, from_agent_state)`` pair.
    """

    def to_agent_state(state: Any) -> AgentState:
        text = _read_key(state, text_key)
        if not isinstance(text, str):
            raise TypeError(
                f"Expected a string at {text_key!r}, got {type(text).__name__}"
            )
        return AgentState(raw_input={f"{text_key}.txt": text.encode("utf-8")})

    def from_agent_state(final: AgentState, _parent: Any) -> dict[str, Any]:
        ontology_ttl = ""
        artifacts = [o for o in final.reduced_ontology_artifacts if not o.is_null()]
        if artifacts:
            ontology_ttl = artifacts[0].graph.serialize_canonical_turtle()
        return {
            ontology_key: ontology_ttl,
            facts_key: final.aggregated_facts.serialize_canonical_turtle(),
        }

    return to_agent_state, from_agent_state