orchestrata_ultimate.agents.volks.analyst.agent

Reference › orchestrata_ultimate › agents › volks › analyst › agent

AnalystAgent (alias U:BA, role Business Analyst) turns the Concierge's cleaned-up requirements into a formal, numbered specification (specification 019). It extends BaseAgent (base), is re-exported by the package __init__ (init) and is imported by the kernel pipeline. Its prompt lives in prompts and its output parsing in parsers.

What it receives: a packet from U:Cx whose payload holds the requirements text, the conversation history, the task summary, the project directory and the Prolog facts. What it produces: a packet addressed to U:PM containing the structured analysis.

process routes the packet to _handle_analysis. That method sets the status to in progress, joins the conversation history and final requirements into one context string and calls _analyze_requirements, which sends it to the LLM with the analyst system prompt. The reply is kept as an output artifact and _parse_analysis extracts every structured section: project type, technology, spec metadata, user stories, requirements and non-functional requirements. _store_to_kg then writes the results to the knowledge graph, through a lazily initialised connection (_init_kg): _store_project_and_spec creates the project and spec nodes and _store_stories_and_reqs stores stories with their requirements. _build_escalation_packet finally builds the packet for the Project Manager (volk).

At a glance

Class diagram

Dependency diagram

API

Classes

AnalystAgent (pydantic)

Bases: BaseAgent · Decorators: traces(spec='019', story='US-001', reqs=['REQ-019-001', 'REQ-019-002']) · Line 32

Business Analyst agent — requirements elicitation.

Source

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

# mypy: ignore-errors
"""U:BA — Business Analyst volk for requirements elicitation.

Takes the raw user request from U:Cx, generates numbered user stories
and requirements via LLM, stores them in the knowledge graph, and
escalates to U:PM with structured requirements data.
"""

import logging
from typing import Any

from pydantic import Field

from orchestrata_ultimate.agents.volks.analyst.parsers import (
    parse_nfrs,
    parse_project_type,
    parse_requirements,
    parse_spec_metadata,
    parse_technology,
    parse_user_stories,
)
from orchestrata_ultimate.agents.volks.analyst.prompts import ANALYST_SYSTEM_PROMPT
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_directives import extract_graph_context
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)


@traces(spec="019", story="US-001", reqs=["REQ-019-001", "REQ-019-002"])
class AnalystAgent(BaseAgent):
    """Business Analyst agent — requirements elicitation.

    Receives raw user requests, classifies project type, generates
    structured user stories and requirements via LLM, stores results
    in the knowledge graph, and escalates to U:PM.
    """

    alias: str = Field(default="U:BA")
    role: str = "Business Analyst"

    def __init__(self, **kwargs) -> None:
        """Initialize AnalystAgent with empty analysis state."""
        super().__init__(**kwargs)
        self.escalation_target = "U:PM"
        self.task_summary = ""
        self.project_dir = ""
        self.project_type = ""
        self.technology: dict[str, str] = {}
        self.spec_metadata: dict[str, str] = {}
        self.user_stories: list[dict[str, Any]] = []
        self.requirements: list[dict[str, Any]] = []
        self.nfrs: list[dict[str, str]] = []
        self._kg_store: Any = None

    @traces(
        spec="019",
        story="US-001",
        reqs=["REQ-019-003", "REQ-019-004"],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packets to the appropriate handler.

        Args:
            packet: Incoming CXP packet to process.

        Returns:
            Response CXP packet, or None if unhandled subject.
        """
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject in ("implement", "analyze_requirements"):
            return await self._handle_analysis(packet)
        return None

    @traces(
        spec="019",
        story="US-002",
        reqs=[
            "REQ-019-005",
            "REQ-019-006",
            "REQ-019-007",
            "REQ-019-008",
            "REQ-019-009",
            "REQ-019-010",
            "REQ-019-011",
            "REQ-019-012",
        ],
    )
    async def _handle_analysis(self, packet: CXPPacket) -> CXPPacket:
        """Analyze requirements and generate user stories.

        Args:
            packet: CXP packet containing raw requirements from U:Cx.

        Returns:
            CXP packet addressed to U:PM with structured analysis.
        """
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        requirements_text = packet.payload.get("requirements", "")
        conversation_history = packet.payload.get("conversation_history", [])
        self.task_summary = packet.payload.get("task_summary", "")
        # Project root, when the pipeline has one (SPEC-003/W3.1 contract key
        # honored by architect/codedirector/junior/qa) -- used to resolve the
        # canonical KG project slug (_store_project_and_spec) instead of
        # slugifying task_summary, and forwarded to downstream stages so the
        # chain doesn't drop it (2026-07 audit).
        self.project_dir = packet.payload.get("project_dir", "")
        prolog_facts = packet.payload.get("prolog_facts", [])

        self.summary = "Analyzing requirements"
        _logger.info("Analyzing: %s", self.task_summary)

        # Include full Q&A history from U:Cx so no context is lost
        full_context = requirements_text
        if conversation_history:
            history_text = "\n".join(conversation_history)
            full_context = (
                f"CONVERSATION HISTORY (all Q&A rounds):\n"
                f"{history_text}\n\n"
                f"FINAL REQUIREMENTS:\n{requirements_text}"
            )

        response = await self._analyze_requirements(full_context)
        self.output_artifacts.append(f"=== REQUIREMENTS ANALYSIS ===\n{response}")
        self._parse_analysis(response)
        await self._store_to_kg()
        return self._build_escalation_packet(
            packet, requirements_text, prolog_facts, response
        )

    @traces(
        spec="019",
        story="US-002",
        reqs=["REQ-019-008", "REQ-019-009"],
    )
    async def _analyze_requirements(self, requirements_text: str) -> str:
        """Invoke LLM with the analyst system prompt.

        Named distinctly from ``_call_llm`` (was a same-named override with an
        incompatible single-arg signature — a Liskov violation that shadowed
        BaseAgent's ``(system_prompt, user_message, **kwargs)`` test-injection
        seam for anything treating this agent polymorphically). Delegates to
        ``self._call_llm`` — still the real seam ``agent._call_llm = fn`` /
        ``AnalystAgent(_call_llm=fn)`` patches — so injected LLM doubles keep
        working identically.

        Args:
            requirements_text: Raw intake details from U:Cx.

        Returns:
            LLM response text with structured analysis.
        """
        prompt = (
            "Analyze this user request and generate structured "
            "requirements.\n\n"
            f"User request: {self.task_summary}\n\n"
            f"Intake details:\n{requirements_text}"
        )
        return await self._call_llm(
            ANALYST_SYSTEM_PROMPT,
            prompt,
            volk="U:BA",
            graph_context=self._graph_context,
        )

    @traces(
        spec="019",
        story="US-002",
        reqs=["REQ-019-010"],
    )
    def _parse_analysis(self, response: str) -> None:
        """Parse all structured sections from LLM response.

        Args:
            response: Full LLM response text.
        """
        self.project_type = parse_project_type(response)
        self.technology = parse_technology(response)
        self.spec_metadata = parse_spec_metadata(response)
        self.user_stories = parse_user_stories(response)
        if response.strip() and not self.user_stories:
            # Dogfood F10: an LLM response without a well-formed USER_STORIES:
            # section used to flow through as [] — falsy, so the pipeline's
            # _extract_content wrote an EMPTY user_stories artifact with no
            # trace of the loss. Stories cannot be fabricated the way the
            # Concierge's task_summary fallback is, so the mitigation here is
            # loudness, not substitution.
            _logger.warning(
                "analyst: LLM response has no well-formed USER_STORIES: "
                "section — the user_stories artifact will be empty"
            )
        self.requirements = parse_requirements(response)
        self.nfrs = parse_nfrs(response)

    @traces(
        spec="019",
        story="US-003",
        reqs=[
            "REQ-019-035",
            "REQ-019-036",
            "REQ-019-037",
            "REQ-019-038",
            "REQ-019-039",
            "REQ-019-040",
        ],
    )
    def _build_escalation_packet(
        self,
        packet: CXPPacket,
        requirements_text: str,
        prolog_facts: list,
        response: str,
    ) -> CXPPacket:
        """Build the escalation packet to U:PM.

        Args:
            packet: Original incoming packet.
            requirements_text: Raw requirements from intake.
            prolog_facts: Prolog facts from governance.
            response: Raw LLM analysis response.

        Returns:
            CXP packet with structured analysis for U:PM.
        """
        story_count = len(self.user_stories)
        req_count = len(self.requirements)
        nfr_count = len(self.nfrs)
        _logger.info(
            "Analysis complete: %d stories, %d requirements, %d NFRs",
            story_count,
            req_count,
            nfr_count,
        )
        self.status = WorkStatus.RESOLVED
        self.summary = f"Requirements: {story_count} stories, {req_count} reqs"
        return CXPPacket(
            from_agent="U:BA",
            to_agent="U:PM",
            thread_id=packet.thread_id,
            subject="scope_review",
            payload={
                "requirements": requirements_text,
                "task_summary": self.task_summary,
                "prolog_facts": prolog_facts,
                "project_type": self.project_type,
                "technology": self.technology,
                "spec_metadata": self.spec_metadata,
                "user_stories": self.user_stories,
                "structured_requirements": self.requirements,
                "nfrs": self.nfrs,
                "analyst_output": response,
                "project_language": packet.payload.get("project_language", "python"),
                "project_dir": self.project_dir,
            },
            work_status=WorkStatus.PENDING,
        )

    @traces(
        spec="019",
        story="US-004",
        reqs=["REQ-019-030"],
    )
    def _init_kg(self) -> None:
        """Lazy-initialize knowledge graph connection."""
        if self._kg_store is not None:
            return
        try:
            store_cls = getattr(self, "_kg_store_cls", None)
            if store_cls is None:
                from orchestrata_ultimate.knowledge_graph.kg_design import KGDesignStore

                store_cls = KGDesignStore
            self._kg_store = store_cls()
        except Exception as exc:
            _logger.warning("KG init failed: %s", exc)
            self._kg_store = None

    @traces(
        spec="019",
        story="US-004",
        reqs=[
            "REQ-019-011",
            "REQ-019-031",
            "REQ-019-032",
            "REQ-019-033",
            "REQ-019-034",
        ],
    )
    async def _store_to_kg(self) -> None:
        """Store analysis results in the knowledge graph.

        Creates explotación project, spec, user stories, and requirements
        nodes. Non-fatal: logs warnings on failure.
        """
        self._init_kg()
        if not self._kg_store:
            _logger.warning("KG unavailable — skipping storage")
            return
        try:
            self._store_project_and_spec()
            self._store_stories_and_reqs()
        except Exception as exc:
            _logger.warning("KG storage error: %s", exc)

    def _store_project_and_spec(self) -> None:
        """Create project and spec nodes in the knowledge graph."""
        # Canonical KG project slug (project_resolver), NOT a slugified
        # task_summary: a free-text task/intent summary is neither stable
        # nor unique per project -- an empty task_summary produced
        # project_slug="" and a long one a silently-truncated-to-50-chars
        # slug, each MERGEing its own anonymous :Project hub in the live
        # graph that thousands of DEFINES edges then anchored onto (2026-07
        # audit). Falls back to the "unscoped" sentinel (never "") when no
        # project_dir is known, matching kernel/pipeline_kg.py's convention.
        from orchestrata_ultimate.knowledge_graph.project_resolver import (
            derive_project_id,
        )

        slug = derive_project_id(self.project_dir) or "unscoped"
        self._kg_store.ensure_project(
            slug=slug,
            name=self.task_summary,
            project_type=self.project_type,
        )
        spec_id = self.spec_metadata.get("spec_id", "001")
        spec_title = self.spec_metadata.get("title", self.task_summary)
        # store_spec returns the project-qualified Spec id (2026-07-12
        # multi-project identity migration) — kept, not the bare spec_id, so
        # _store_stories_and_reqs can pass it straight through to
        # store_user_stories/store_requirements without re-deriving the
        # project qualification itself.
        self._current_spec_id = self._kg_store.store_spec(
            project_slug=slug,
            spec_id=spec_id,
            title=spec_title,
        )

    @traces(spec="019", story="US-004", reqs=["REQ-019-031", "REQ-019-032"])
    def _store_stories_and_reqs(self) -> None:
        """Store user stories and their requirements in KG."""
        spec_id = getattr(self, "_current_spec_id", "001")
        story_id_map = self._kg_store.store_user_stories(
            spec_id=spec_id,
            stories=self.user_stories,
        )
        for story in self.user_stories:
            story_reqs = [
                r for r in self.requirements if r.get("story_id") == story["id"]
            ]
            qualified_story_id = story_id_map.get(story["id"])
            if story_reqs and qualified_story_id:
                self._kg_store.store_requirements(
                    story_id=qualified_story_id,
                    requirements=story_reqs,
                )
        _logger.info(
            "Stored to KG: %d stories, %d reqs",
            len(self.user_stories),
            len(self.requirements),
        )