ontocast.agent.common

call_llm_with_retry(llm_tool, prompt, parser, prompt_kwargs, max_retries=3, retry_error_feedback=True, llm_graph_format=None) async

Call LLM and parse response with automatic retry on parsing failures.

This utility function implements a common pattern across agent functions: 1. Call LLM with a prompt 2. Parse the response 3. Retry if parsing fails (up to max_retries times)

On retry, if retry_error_feedback is True, the error message from the previous attempt is included in the prompt to help the LLM correct its output format.

Parameters:

Name	Type	Description	Default
llm_tool	LLMTool	The LLM tool instance to use for generation.	required
prompt	BasePromptTemplate	The prompt template to format and send to the LLM.	required
parser	BaseOutputParser[T]	The output parser to parse the LLM response.	required
prompt_kwargs	dict[str, Any]	Keyword arguments to pass to prompt.format_prompt().	required
max_retries	int	Maximum number of retry attempts (default: 3).	3
retry_error_feedback	bool	Whether to include error feedback in retry prompts (default: True).	True
llm_graph_format	LLMGraphFormat | None	When set, passed explicitly to model_validate as context={"llm_graph_format": ...} so graph wire fields coerce correctly.	None

Returns:

Type	Description
T	The parsed output of type T.

Raises:

Type	Description
Exception	If parsing fails after all retry attempts, raises the last parsing error.

Source code in ontocast/agent/common.py

async def call_llm_with_retry(
    llm_tool: LLMTool,
    prompt: BasePromptTemplate,
    parser: BaseOutputParser[T],
    prompt_kwargs: dict[str, Any],
    max_retries: int = 3,
    retry_error_feedback: bool = True,
    llm_graph_format: LLMGraphFormat | None = None,
) -> T:
    """Call LLM and parse response with automatic retry on parsing failures.

    This utility function implements a common pattern across agent functions:
    1. Call LLM with a prompt
    2. Parse the response
    3. Retry if parsing fails (up to max_retries times)

    On retry, if retry_error_feedback is True, the error message from the previous
    attempt is included in the prompt to help the LLM correct its output format.

    Args:
        llm_tool: The LLM tool instance to use for generation.
        prompt: The prompt template to format and send to the LLM.
        parser: The output parser to parse the LLM response.
        prompt_kwargs: Keyword arguments to pass to prompt.format_prompt().
        max_retries: Maximum number of retry attempts (default: 3).
        retry_error_feedback: Whether to include error feedback in retry prompts (default: True).
        llm_graph_format: When set, passed explicitly to ``model_validate`` as
            ``context={"llm_graph_format": ...}`` so graph wire fields coerce correctly.

    Returns:
        The parsed output of type T.

    Raises:
        Exception: If parsing fails after all retry attempts, raises the last parsing error.
    """
    last_error: Exception | None = None
    last_sanitized_content: str | None = None
    original_format_instructions = prompt_kwargs.get("format_instructions", "")

    for attempt in range(max_retries):
        try:
            # Create a copy of prompt_kwargs for this attempt
            attempt_kwargs = prompt_kwargs.copy()

            # On retry, add error feedback to help LLM correct format
            if attempt > 0 and retry_error_feedback and last_error is not None:
                # Use sanitized content in error feedback for consistency
                feedback_content = (
                    last_sanitized_content if last_sanitized_content else ""
                )
                error_feedback = (
                    f"\n\nIMPORTANT: The previous attempt failed to parse the response. "
                    f"Error: {str(last_error)}\n"
                    f"Previous response (for reference):\n{feedback_content}\n\n"
                    f"Please ensure your response strictly follows the format instructions "
                    f"and does not contain any control characters or invalid syntax."
                )
                # Add error feedback to format_instructions if present
                if "format_instructions" in attempt_kwargs:
                    attempt_kwargs["format_instructions"] = (
                        original_format_instructions + error_feedback
                    )
                else:
                    # If no format_instructions, add as a new field
                    attempt_kwargs["parsing_error_feedback"] = error_feedback

            # Call LLM
            response = await llm_tool(prompt.format_prompt(**attempt_kwargs))
            content_to_parse = strip_trailing_commas(
                strip_json_comments(_content_to_str(response.content))
            )
            last_sanitized_content = content_to_parse

            if llm_graph_format is not None and isinstance(
                parser, PydanticOutputParser
            ):
                json_object = parse_json_markdown(content_to_parse)
                model_cls = cast(type[BaseModel], parser.pydantic_object)
                parsed = cast(
                    T,
                    model_cls.model_validate(
                        json_object,
                        context={"llm_graph_format": llm_graph_format},
                    ),
                )
            else:
                parsed = parser.parse(content_to_parse)
            logger.debug(
                f"Successfully parsed LLM response on attempt {attempt + 1}/{max_retries}"
            )
            return parsed

        except Exception as e:
            last_error = e
            logger.warning(
                f"Failed to parse LLM response on attempt {attempt + 1}/{max_retries}: {str(e)}"
            )

            # If this was the last attempt, raise the error
            if attempt == max_retries - 1:
                logger.error(
                    f"Failed to parse LLM response after {max_retries} attempts. "
                    f"Last error: {str(e)}"
                )
                raise

    raise RuntimeError("Unexpected error in call_llm_with_retry")

render_suggestions_prompt(suggestions, stage)

Generate prompt templates from the suggestions.

Returns:

Type	Description
str	Combined string with general and concrete templates.
str	Returns empty string if both fields are empty.

Source code in ontocast/agent/common.py

def render_suggestions_prompt(suggestions: Suggestions, stage: WorkflowNode) -> str:
    """Generate prompt templates from the suggestions.

    Returns:
        Combined string with general and concrete templates.
        Returns empty string if both fields are empty.
    """

    # Generate general template if systemic_critique_summary is not empty
    general_template = ""
    if suggestions.systemic_critique_summary.strip():
        general_template = suggestion_general_template.format(
            general_suggestion=suggestions.systemic_critique_summary
        )

    concrete_template = ""
    if suggestions.actionable_fixes:
        # Generate concrete template if actionable_fixes is not empty
        concrete_template = suggestion_concrete_template.format(
            suggestion_str=suggestions.to_markdown()
        )

    if stage == WorkflowNode.TEXT_TO_FACTS:
        template = facts_template
    elif stage == WorkflowNode.TEXT_TO_ONTOLOGY:
        template = ontology_template
    else:
        raise ValueError(f"Stage {stage} not supported")
    if general_template or concrete_template:
        final_prompt = template.format(
            suggestions_instruction=f"\n\n{general_template}\n\n{concrete_template}"
        )
    else:
        final_prompt = ""
    return final_prompt

strip_json_comments(text)

Remove single-line // comments from JSON-like text while preserving string literals.

The LLM occasionally emits JavaScript-style // comments inside JSON output, which are not valid JSON. This function strips them by scanning the text token by token: JSON string literals (which may contain '//') are kept intact, while bare // … sequences are dropped.

Source code in ontocast/agent/common.py

def strip_json_comments(text: str) -> str:
    """Remove single-line // comments from JSON-like text while preserving string literals.

    The LLM occasionally emits JavaScript-style // comments inside JSON output,
    which are not valid JSON.  This function strips them by scanning the text
    token by token: JSON string literals (which may contain '//') are kept
    intact, while bare // … sequences are dropped.
    """

    def _replace(m: re.Match) -> str:
        matched = m.group()
        return matched if matched.startswith('"') else ""

    return _JSON_COMMENT_RE.sub(_replace, text)

strip_trailing_commas(text)

Remove trailing commas before } or ] (invalid in strict JSON).

Source code in ontocast/agent/common.py

def strip_trailing_commas(text: str) -> str:
    """Remove trailing commas before ``}`` or ``]`` (invalid in strict JSON)."""
    prev = None
    while prev != text:
        prev = text
        text = _TRAILING_COMMA_RE.sub(r"\1", text)
    return text