orchestrata_ultimate.agents.volks.analyst.parsers

Reference › orchestrata_ultimate › agents › volks › analyst › parsers

parsers holds the regular-expression and line parsers that turn the Analyst's LLM reply into data. Only agent imports it, and the labelled sections it reads are defined by the template in prompts. _extract_section cuts the text between a start marker and the first end marker, and _parse_kv_lines reads key: value lines. The public parsers are parse_project_type, parse_technology, parse_spec_metadata, parse_user_stories (stories with their acceptance criteria, assembled line by line through _parse_stories_from_section and _process_story_line), parse_requirements (trimmed at the next section boundary by _trim_requirements_section, entries read by _extract_req_lines) and parse_nfrs (via _extract_nfr_lines). Most carry @traces markers for specification 019.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/analyst/parsers.py

"""LLM output parsers for the Analyst volk (U:BA).

Regex-based extraction of structured sections from LLM responses:
PROJECT_TYPE, TECHNOLOGY, SPEC_METADATA, USER_STORIES,
REQUIREMENTS, and NON_FUNCTIONAL.
"""

import re
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

# Compiled patterns for requirement/story/NFR extraction.
#
# REQ_PATTERN matches the CANONICAL story-scoped id only (RUL-031): ``US``, a
# hyphen, then three digits. It previously read ``US\d+`` — the legacy
# spelling — which meant this parser could not read a single id from the
# migrated corpus and silently returned an empty list for every requirement
# the analyst produced. Accepting the legacy form "just in case an LLM emits
# it" is what this parser must NOT do: its output flows into the corpus, so
# tolerating legacy here re-introduces exactly what RUL-031 eliminated.
REQ_PATTERN = re.compile(r"(REQ-\d{3}-US-\d{3}-\d{3}):\s*(.+)")
US_PATTERN = re.compile(r"(US-\d{3}):\s*(.+)")
NFR_PATTERN = re.compile(r"(NFR-\d{3}):\s*(.+)")
# The story segment of a canonical requirement id, for deriving ``story_id``.
REQ_STORY_PATTERN = re.compile(r"US-(\d{3})")


def _extract_section(
    response: str,
    start_marker: str,
    end_markers: list[str],
) -> str:
    """Extract text between a start marker and the first end marker.

    Args:
        response: Full LLM response text.
        start_marker: Section header to find (e.g. "TECHNOLOGY:").
        end_markers: List of subsequent section headers that end
            this section.

    Returns:
        The extracted section text, or empty string if not found.
    """
    if start_marker not in response:
        return ""
    section = response.split(start_marker)[1]
    for marker in end_markers:
        if marker in section:
            section = section.split(marker)[0]
            break
    return section.strip()


@traces(spec="019", story="US-002", reqs=["REQ-019-013", "REQ-019-014", "REQ-019-015"])
def parse_project_type(response: str) -> str:
    """Extract PROJECT_TYPE from LLM response.

    Args:
        response: Full LLM response text.

    Returns:
        Lowercased project type string, or empty string.
    """
    if "PROJECT_TYPE:" not in response:
        return ""
    line = response.split("PROJECT_TYPE:")[1].split("\n")[0].strip()
    return line.lower()


@traces(spec="019", story="US-002", reqs=["REQ-019-016", "REQ-019-017", "REQ-019-018"])
def parse_technology(response: str) -> dict[str, str]:
    """Extract TECHNOLOGY key-value pairs from LLM response.

    Args:
        response: Full LLM response text.

    Returns:
        Dictionary of technology field to value.
    """
    section = _extract_section(
        response,
        "TECHNOLOGY:",
        ["SPEC_METADATA:", "USER_STORIES:", "REQUIREMENTS:"],
    )
    if not section:
        return {}
    return _parse_kv_lines(section)


def _parse_kv_lines(section: str) -> dict[str, str]:
    """Parse key: value lines from a section.

    Args:
        section: Multi-line text with "- key: value" lines.

    Returns:
        Dictionary of parsed key-value pairs.
    """
    result: dict[str, str] = {}
    for line in section.split("\n"):
        line = line.strip().lstrip("- ")
        if ":" in line:
            key, val = line.split(":", 1)
            result[key.strip()] = val.strip()
    return result


@traces(spec="019", story="US-002", reqs=["REQ-019-019", "REQ-019-020", "REQ-019-021"])
def parse_spec_metadata(response: str) -> dict[str, str]:
    """Extract SPEC_METADATA key-value pairs from LLM response.

    Args:
        response: Full LLM response text.

    Returns:
        Dictionary of spec metadata fields.
    """
    section = _extract_section(
        response,
        "SPEC_METADATA:",
        ["USER_STORIES:", "REQUIREMENTS:", "NON_FUNCTIONAL:"],
    )
    if not section:
        return {}
    return _parse_kv_lines(section)


@traces(spec="019", story="US-002", reqs=["REQ-019-022", "REQ-019-023", "REQ-019-024"])
def parse_user_stories(
    response: str,
) -> list[dict[str, Any]]:
    """Extract USER_STORIES from LLM response.

    Parses story ID/title, role/want/benefit fields, and
    acceptance criteria lists.

    Args:
        response: Full LLM response text.

    Returns:
        List of story dicts with id, title, as_a, i_want,
        so_that, and acceptance_criteria keys.
    """
    section = _extract_section(
        response,
        "USER_STORIES:",
        ["REQUIREMENTS:", "NON_FUNCTIONAL:"],
    )
    if not section:
        return []
    return _parse_stories_from_section(section)


@traces(spec="019", story="US-002", reqs=["REQ-019-022", "REQ-019-023", "REQ-019-024"])
def _parse_stories_from_section(
    section: str,
) -> list[dict[str, Any]]:
    """Parse individual stories from a USER_STORIES section.

    Args:
        section: Text content of the USER_STORIES section.

    Returns:
        List of parsed story dictionaries.
    """
    stories: list[dict[str, Any]] = []
    current: dict[str, Any] | None = None
    in_criteria = False

    for line in section.split("\n"):
        stripped = line.strip()
        if not stripped:
            continue
        current, in_criteria = _process_story_line(
            stripped, current, in_criteria, stories
        )

    if current:
        stories.append(current)
    return stories


@traces(spec="019", story="US-002", reqs=["REQ-019-022", "REQ-019-023"])
def _process_story_line(
    stripped: str,
    current: dict[str, Any] | None,
    in_criteria: bool,
    stories: list[dict[str, Any]],
) -> tuple[dict[str, Any] | None, bool]:
    """Process a single line within the USER_STORIES section.

    Args:
        stripped: Stripped line text.
        current: Currently-being-built story dict, or None.
        in_criteria: Whether we are inside acceptance criteria.
        stories: Accumulator list for completed stories.

    Returns:
        Tuple of (current_story, in_criteria) after processing.
    """
    us_match = US_PATTERN.match(stripped)
    if us_match:
        if current:
            stories.append(current)
        new_story: dict[str, Any] = {
            "id": us_match.group(1),
            "title": us_match.group(2),
            "as_a": "",
            "i_want": "",
            "so_that": "",
            "acceptance_criteria": [],
        }
        return new_story, False

    if current is None:
        return current, in_criteria

    lower = stripped.lower()
    if lower.startswith("as a "):
        current["as_a"] = stripped[5:]
        return current, False
    if lower.startswith("i want "):
        current["i_want"] = stripped[7:]
        return current, False
    if lower.startswith("so that "):
        current["so_that"] = stripped[8:]
        return current, False
    if "acceptance criteria" in lower:
        return current, True
    if in_criteria and stripped.startswith("- "):
        current["acceptance_criteria"].append(stripped[2:])
    return current, in_criteria


@traces(spec="019", story="US-002", reqs=["REQ-019-025", "REQ-019-026"])
def parse_requirements(
    response: str,
) -> list[dict[str, Any]]:
    """Extract REQUIREMENTS from LLM response.

    Args:
        response: Full LLM response text.

    Returns:
        List of requirement dicts with id, description,
        story_id, priority, and category keys.
    """
    if "REQUIREMENTS:" not in response:
        return []
    section = response.split("REQUIREMENTS:")[1]
    section = _trim_requirements_section(section)
    return _extract_req_lines(section)


@traces(spec="019", story="US-002", reqs=["REQ-019-025"])
def _trim_requirements_section(section: str) -> str:
    """Trim the requirements section at the next section boundary.

    Args:
        section: Raw text after "REQUIREMENTS:".

    Returns:
        Trimmed section text.
    """
    end_markers = ["NON_FUNCTIONAL:", "NFR-"]
    for marker in end_markers:
        if marker in section:
            idx = section.index(marker)
            if marker == "NFR-":
                nl = section.rfind("\n", 0, idx)
                if nl >= 0:
                    section = section[:nl]
            else:
                section = section.split(marker)[0]
            break
    return section


@traces(spec="019", story="US-002", reqs=["REQ-019-025", "REQ-019-026"])
def _extract_req_lines(
    section: str,
) -> list[dict[str, Any]]:
    """Extract requirement entries from section lines.

    Args:
        section: Trimmed requirements section text.

    Returns:
        List of requirement dictionaries.
    """
    requirements: list[dict[str, Any]] = []
    for line in section.strip().split("\n"):
        stripped = line.strip()
        match = REQ_PATTERN.match(stripped)
        if not match:
            continue
        req_id = match.group(1)
        description = match.group(2)
        us_match = REQ_STORY_PATTERN.search(req_id)
        story_id = f"US-{us_match.group(1)}" if us_match else ""
        requirements.append(
            {
                "id": req_id,
                "description": description,
                "story_id": story_id,
                "priority": "NORMAL",
                "category": "",
            }
        )
    return requirements


@traces(spec="019", story="US-002", reqs=["REQ-019-027", "REQ-019-028", "REQ-019-029"])
def parse_nfrs(response: str) -> list[dict[str, str]]:
    """Extract NON_FUNCTIONAL requirements from LLM response.

    Args:
        response: Full LLM response text.

    Returns:
        List of NFR dicts with id, description, and category.
    """
    if "NON_FUNCTIONAL:" not in response:
        return []
    section = response.split("NON_FUNCTIONAL:")[1]
    return _extract_nfr_lines(section)


@traces(spec="019", story="US-002", reqs=["REQ-019-027", "REQ-019-028", "REQ-019-029"])
def _extract_nfr_lines(
    section: str,
) -> list[dict[str, str]]:
    """Extract NFR entries from section lines.

    Args:
        section: Raw text after "NON_FUNCTIONAL:".

    Returns:
        List of NFR dictionaries.
    """
    nfrs: list[dict[str, str]] = []
    for line in section.strip().split("\n"):
        stripped = line.strip()
        match = NFR_PATTERN.match(stripped)
        if not match:
            continue
        nfr_id = match.group(1)
        rest = match.group(2)
        category = ""
        cat_match = re.search(r"\(category:\s*(\w+)\)", rest)
        if cat_match:
            category = cat_match.group(1)
            rest = rest[: cat_match.start()].strip()
        nfrs.append(
            {
                "id": nfr_id,
                "description": rest,
                "category": category,
            }
        )
    return nfrs