ontocast.util

count_visits_conditional_success(state, current_node)

Track node visits and handle success/failure conditions.

This function increments the visit counter for a node and manages the state based on success/failure conditions and maximum visit limits.

Parameters:

Name	Type	Description	Default
state	AgentState	The current agent state.	required
current_node		The node being visited.	required

Returns:

Name	Type	Description
AgentState	AgentState	Updated agent state after processing visit conditions.

Source code in ontocast/util.py

def count_visits_conditional_success(state: AgentState, current_node) -> AgentState:
    """Track node visits and handle success/failure conditions.

    This function increments the visit counter for a node and manages the state
    based on success/failure conditions and maximum visit limits.

    Args:
        state: The current agent state.
        current_node: The node being visited.

    Returns:
        AgentState: Updated agent state after processing visit conditions.
    """
    state.node_visits[current_node] += 1
    if state.status == Status.SUCCESS:
        logger.info(f"For {current_node}: status is SUCCESS, proceeding to next node")
        state.clear_failure()
    elif state.node_visits[current_node] >= state.max_visits:
        logger.error(f"For {current_node}: maximum visits exceeded")
        state.set_failure(current_node, reason="Maximum visits exceeded")
        state.status = Status.SUCCESS
    return state

setup_logging(debug=False)

Set up logging configuration for the project.

This function configures the logging system with appropriate formatting and log level based on the debug flag.

Parameters:

Name	Type	Description	Default
debug	bool	If True, sets logging level to DEBUG, otherwise INFO.	False

Source code in ontocast/util.py

def setup_logging(debug: bool = False) -> None:
    """Set up logging configuration for the project.

    This function configures the logging system with appropriate formatting
    and log level based on the debug flag.

    Args:
        debug: If True, sets logging level to DEBUG, otherwise INFO.
    """
    level = logging.DEBUG if debug else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        stream=sys.stdout,
    )

wrap_with(func, node_name, post_func)

Add a visit counter to a function.

This function wraps a given function with logging and post-processing functionality, typically used for workflow node execution.

Parameters:

Name	Type	Description	Default
func		The function to wrap.	required
node_name		The name of the node.	required
post_func		Function to execute after the main function.	required

Returns:

Type	Description
tuple[WorkflowNode, Callable]	tuple[WorkflowNode, Callable]: A tuple containing the node name and the wrapped function.

Source code in ontocast/util.py

def wrap_with(func, node_name, post_func) -> tuple[WorkflowNode, Callable]:
    """Add a visit counter to a function.

    This function wraps a given function with logging and post-processing
    functionality, typically used for workflow node execution.

    Args:
        func: The function to wrap.
        node_name: The name of the node.
        post_func: Function to execute after the main function.

    Returns:
        tuple[WorkflowNode, Callable]: A tuple containing the node name and
            the wrapped function.
    """

    @wraps(func)
    def wrapper(state: AgentState):
        logger.info(f"Starting to execute {node_name}")
        state = func(state)
        state = post_func(state, node_name)
        return state

    return node_name, wrapper