13.2 AI-Assisted Requirement Extraction

1. LLM-Based Requirement Extraction

When requirements are extracted by hand, published case studies typically find that analysts miss 20% to 40% of the needs actually voiced in stakeholder interviews, and the error rate climbs with transcript length. Missed requirements surface late, during integration or user acceptance testing, where each one can cost 10 to 100 times more to address than it would have at elicitation time. Automating extraction is not about speed alone; it is about catching what human attention inevitably drops.

Imagine handing a two-hour stakeholder recording to a machine and receiving, thirty seconds later, a clean list of user stories. Each story carries a confidence score and links to the exact transcript passage it came from. The core task: given unstructured text (a transcript, ticket, or document), produce a list of structured UserStory objects. The real challenge is getting the LLM to produce consistent, well-formed output at scale, not creative variations that drift from the schema.

Structured output mode constrains an LLM to produce only valid JSON conforming to a predefined schema (such as a Pydantic model, where Pydantic is a Python validation library that defines data schemas as annotated classes and enforces type constraints at runtime), rather than free-form text that requires parsing and validation after the fact. It matters because unconstrained LLM output is inherently variable: the same prompt may produce a JSON object in one call, a Markdown table in the next, and a bulleted list in a third. That variability forces brittle post-processing that breaks at scale. The mechanism provides the model with a JSON Schema at inference time. The decoding process itself rejects any token sequence that would violate the schema, guaranteeing that every response parses without error. Use structured output mode whenever the downstream consumer is code (not a human reader) and the output must conform to a fixed contract. Use free-form generation when you need creative prose, explanation, or exploratory reasoning where rigid structure would limit the model's usefulness.

We use three techniques from Chapter 10 to ensure reliable extraction: In short: constrain the model's output format, show it what good looks like, and never feed it more text than it can handle well.

• Structured output mode: constrain the LLM to produce JSON conforming to a Pydantic schema, eliminating format parsing errors entirely.
• Few-shot examples (where a small number of input/output pairs are included in the prompt to demonstrate the desired behavior): provide 2 to 3 examples of well-formed extractions so the LLM learns the desired level of specificity.
• Chunked processing: split long transcripts into overlapping windows (with context carryover) to stay within context limits and maintain extraction quality.

from pydantic import BaseModel, Field
from anthropic import Anthropic


class ExtractedStory(BaseModel):
    """A user story extracted from unstructured text."""
    role: str = Field(description="The stakeholder role (e.g., researcher, lab technician)")
    capability: str = Field(description="What the user wants to do (10-200 chars)")
    benefit: str = Field(description="Why it matters (10-200 chars)")
    priority_hint: str = Field(description="Inferred priority: must/should/could/wont")
    source_quote: str = Field(description="The verbatim passage this was extracted from")
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="Extraction confidence: 1.0 = explicit statement, "
                    "0.5 = inferred from context, 0.0 = speculative"
    )


class ExtractionResult(BaseModel):
    """Container for all stories extracted from one text chunk."""
    stories: list[ExtractedStory]
    ambiguities: list[str] = Field(
        default_factory=list,
        description="Passages where the stakeholder intent was unclear"
    )


EXTRACTION_PROMPT = """You are a requirements engineer. Extract user stories from
the following stakeholder transcript.

For each requirement or feature request mentioned (explicitly or implicitly),
produce a user story in the standard format:
  - role: the stakeholder role (use the speaker's actual role if known)
  - capability: what they want the system to do (be specific, not vague)
  - benefit: why it matters to them (connect to their workflow or pain point)
  - priority_hint: infer from language cues (must/should/could/wont)
  - source_quote: the exact passage you extracted this from
  - confidence: how certain you are this is a genuine requirement

Rules:
1. Each story should describe ONE capability, not a compound feature.
2. Avoid ambiguous terms (fast, easy, user-friendly) without quantification.
3. If the speaker implies a requirement without stating it explicitly, extract
   it with confidence < 0.7 and note the ambiguity.
4. Do NOT invent requirements that are not supported by the transcript.

Example input:
"We really need the dashboard to show real-time experiment status. Right now
I have to SSH into each machine to check if the runs finished."

Example output:
{
  "stories": [{
    "role": "researcher",
    "capability": "view the status of all running experiments on a single dashboard without SSH access",
    "benefit": "I can monitor experiment progress from my desk instead of logging into each machine individually",
    "priority_hint": "must",
    "source_quote": "We really need the dashboard to show real-time experiment status. Right now I have to SSH into each machine to check if the runs finished.",
    "confidence": 0.95
  }],
  "ambiguities": []
}

Transcript to analyze:
{transcript}"""


def extract_requirements(
    transcript: str,
    client: Anthropic,
    model: str = "claude-sonnet-4-20250514",
    chunk_size: int = 3000,
    overlap: int = 500,
) -> list[ExtractionResult]:
    """Extract user stories from a transcript using an LLM.

    Splits long transcripts into overlapping chunks to maintain
    context while staying within quality bounds.

    Args:
        transcript: The raw transcript text.
        client: Anthropic API client.
        model: Model to use for extraction.
        chunk_size: Characters per chunk.
        overlap: Character overlap between chunks.

    Returns:
        List of ExtractionResult, one per chunk.
    """
    # Split into overlapping chunks
    chunks = []
    start = 0
    while start < len(transcript):
        end = min(start + chunk_size, len(transcript))
        chunks.append(transcript[start:end])
        start += chunk_size - overlap

    results = []
    for i, chunk in enumerate(chunks):
        prompt = EXTRACTION_PROMPT.format(transcript=chunk)

        response = client.messages.create(
            model=model,
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}],
            # Request structured JSON output
            response_format={
                "type": "json_schema",
                "json_schema": {
                    "name": "extraction_result",
                    "schema": ExtractionResult.model_json_schema(),
                },
            },
        )

        # Parse the structured response
        import json
        raw = json.loads(response.content[0].text)
        result = ExtractionResult.model_validate(raw)
        results.append(result)

    return results

2. Post-Extraction Validation

Raw LLM extraction produces user stories that may be well-formed syntactically but flawed semantically. A story might use ambiguous language, combine multiple capabilities into one, or describe a solution rather than a need. We apply the quality validators from Section 13.1 as an automated post-extraction filter.

from dataclasses import dataclass


@dataclass
class ValidationReport:
    """Report from validating an extracted story."""
    story: ExtractedStory
    invest_results: dict[str, bool]
    ambiguous_terms: list[str]
    issues: list[str]

    @property
    def is_clean(self) -> bool:
        """True if no validation issues were found."""
        return len(self.issues) == 0


def validate_extracted_story(story: ExtractedStory) -> ValidationReport:
    """Run all quality checks on an extracted story.

    Combines INVEST checks, ambiguity detection,
    and structural validation.
    """
    issues = []

    # Check for ambiguous terms in capability and benefit
    # check_ambiguity is defined in Section 13.1; it scans text for vague words like "fast", "easy", or "user-friendly"
    cap_ambiguous = check_ambiguity(story.capability)
    ben_ambiguous = check_ambiguity(story.benefit)
    all_ambiguous = cap_ambiguous + ben_ambiguous

    if all_ambiguous:
        issues.append(
            f"Ambiguous terms found: {', '.join(all_ambiguous)}. "
            "Replace with quantified definitions."
        )

    # Check for compound capabilities
    compound_signals = [" and ", " also ", " additionally ", " as well as "]
    if any(sig in story.capability.lower() for sig in compound_signals):
        issues.append(
            "Capability appears compound (contains 'and'/'also'). "
            "Consider splitting into separate stories."
        )

    # Check that benefit is distinct from capability
    # (a common LLM failure mode: restating the capability as the benefit)
    from difflib import SequenceMatcher  # computes a ratio (0 to 1) of matching subsequences between two strings
    similarity = SequenceMatcher(
        None,
        story.capability.lower(),
        story.benefit.lower(),
    ).ratio()
    if similarity > 0.6:
        issues.append(
            f"Benefit is too similar to capability (similarity={similarity:.2f}). "
            "The benefit should explain WHY, not restate WHAT."
        )

    # Check for solution language in what should be a need statement
    solution_signals = [
        "use a", "implement", "build a", "create a database",
        "add a button", "write a script", "use SQL", "deploy",
    ]
    if any(sig in story.capability.lower() for sig in solution_signals):
        issues.append(
            "Capability contains solution language. "
            "Rephrase as a need (what) rather than a solution (how)."
        )

    # Build a minimal UserStory for INVEST checking
    try:
        full_story = UserStory(
            id="TEMP",
            role=StakeholderRole(story.role) if story.role in [
                e.value for e in StakeholderRole
            ] else StakeholderRole.RESEARCHER,
            capability=story.capability,
            benefit=story.benefit,
            source="extraction",
        )
        invest = check_invest(full_story)
    except Exception:
        invest = {}
        issues.append("Could not construct a valid UserStory for INVEST checking.")

    return ValidationReport(
        story=story,
        invest_results=invest,
        ambiguous_terms=all_ambiguous,
        issues=issues,
    )


def filter_and_report(
    results: list[ExtractionResult],
    min_confidence: float = 0.5,
) -> tuple[list[ExtractedStory], list[ValidationReport]]:
    """Filter extracted stories by confidence and validation.

    Returns:
        Tuple of (clean_stories, all_reports).
    """
    clean = []
    reports = []

    for result in results:
        for story in result.stories:
            if story.confidence < min_confidence:
                continue  # Too speculative; save for follow-up questions

            report = validate_extracted_story(story)
            reports.append(report)

            if report.is_clean:
                clean.append(story)

    total = sum(len(r.stories) for r in results)
    passed = len(clean)
    print(f"Extracted: {total} | Above confidence: {len(reports)} | "
          f"Passed validation: {passed} ({100*passed/max(len(reports),1):.0f}%)")

    return clean, reports

3. Requirement Clustering and Deduplication

Validation ensures that each individual story is well-formed, but it cannot tell you whether two well-formed stories from different transcripts are actually requesting the same thing.

When requirements are extracted from multiple transcripts, interviews, and documents, the same need often appears in different words. "The system should export results as CSV" and "I need to download my data in spreadsheet format" share zero words in common, yet their embedding vectors sit just 0.08 cosine distance (a metric that measures the angular separation between two vectors, where 0 means identical direction and 1 means orthogonal) apart, closer than many exact-synonym pairs. We use embedding-based similarity to cluster related requirements and merge near-duplicates.

From Vectors to Clusters

The approach has three steps: embed each requirement as a vector using a sentence transformer (a neural network trained to map sentences to fixed-length vectors that capture semantic meaning), cluster the vectors with Density-Based Spatial Clustering of Applications with Noise (DBSCAN) (which needs no predefined cluster count), and select a representative from each cluster while merging metadata (sources, confidence scores, tags) from the duplicates.

import numpy as np
from sklearn.cluster import DBSCAN
from sklearn.metrics.pairwise import cosine_distances


def embed_stories(
    stories: list[ExtractedStory],
    client: Anthropic,
    model: str = "voyage-3",
) -> np.ndarray:
    """Compute embedding vectors for a list of stories.

    Uses the story's capability + benefit as the text to embed.
    """
    texts = [
        f"{s.role}: {s.capability}. Benefit: {s.benefit}"
        for s in stories
    ]

    # Batch embed (most APIs support batch requests)
    # Using a generic embedding call pattern
    embeddings = []
    batch_size = 64
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        # Placeholder for actual embedding API call
        # In production, use your preferred embedding provider
        response = client.embeddings.create(
            model=model,
            input=batch,
        )
        embeddings.extend([e.embedding for e in response.data])

    return np.array(embeddings)


def cluster_requirements(
    stories: list[ExtractedStory],
    embeddings: np.ndarray,
    eps: float = 0.15,
    min_samples: int = 2,
) -> dict[int, list[int]]:
    """Cluster requirements by semantic similarity using DBSCAN.

    Args:
        stories: The extracted stories.
        embeddings: Embedding vectors (one per story).
        eps: Maximum cosine distance for two stories to be neighbors.
        min_samples: Minimum cluster size.

    Returns:
        Dict mapping cluster_id -> list of story indices.
        Cluster -1 contains unclustered (unique) stories.
    """
    # DBSCAN on cosine distance
    distance_matrix = cosine_distances(embeddings)
    clustering = DBSCAN(
        eps=eps,
        min_samples=min_samples,
        metric="precomputed",
    ).fit(distance_matrix)

    # Group indices by cluster label
    clusters: dict[int, list[int]] = {}
    for idx, label in enumerate(clustering.labels_):
        clusters.setdefault(label, []).append(idx)

    n_clusters = len([k for k in clusters if k != -1])
    n_noise = len(clusters.get(-1, []))
    print(f"Found {n_clusters} clusters, {n_noise} unique stories")

    return clusters


def deduplicate_cluster(
    stories: list[ExtractedStory],
    indices: list[int],
) -> ExtractedStory:
    """Merge near-duplicate stories within a cluster.

    Selects the highest-confidence story as the representative
    and merges source quotes from all duplicates.
    """
    cluster_stories = [stories[i] for i in indices]

    # Pick the highest-confidence story as representative
    representative = max(cluster_stories, key=lambda s: s.confidence)

    # Merge source quotes for traceability
    all_quotes = [s.source_quote for s in cluster_stories]
    merged_quote = " | ".join(set(all_quotes))

    # Boost confidence when multiple sources agree
    merged_confidence = min(
        1.0,
        representative.confidence + 0.1 * (len(cluster_stories) - 1)
    )

    return ExtractedStory(
        role=representative.role,
        capability=representative.capability,
        benefit=representative.benefit,
        priority_hint=representative.priority_hint,
        source_quote=merged_quote,
        confidence=merged_confidence,
    )

4. Conflict Detection as Constraint Satisfiability

Clustering and deduplication reduce a sprawling list to a manageable set of unique requirements, but they say nothing about whether those surviving requirements are compatible with one another.

Two requirements conflict when satisfying both simultaneously is impossible or creates a logical contradiction. Conflicts are surprisingly common in multi-stakeholder projects because different departments optimize for different objectives. The security team wants mandatory two-factor authentication (2FA) on every action; the lab technician team wants one-click instrument control with no interruptions. The compliance officer requires a 7-year audit trail; the data engineer wants to purge old data quarterly to manage storage costs.

We can formalize conflict detection as follows. Treat each requirement \(r_i\) as a propositional variable. Define a conflict relation \(C(r_i, r_j)\) that holds when requirements \(r_i\) and \(r_j\) cannot both be satisfied. The requirement set is conflict-free if:

In the general case, determining whether a set of natural-language requirements is jointly satisfiable is undecidable, because requirements that specify arbitrary computational behavior can encode instances of the halting problem. In other words, no algorithm can guarantee a correct yes-or-no answer for every possible requirement set, so we must rely on heuristic methods that work well in practice even though they cannot cover all theoretical corner cases. In practice, we use a two-stage approach: an LLM identifies candidate conflicts from pairwise comparison, and then a graph analysis confirms structural conflicts through cycle detection.

5. Pairwise Conflict Identification with LLMs

Checking all \(\binom{n}{2}\) pairs of requirements for conflicts is \(O(n^2)\), which becomes expensive for large requirement sets. We reduce the search space using the same embedding-based similarity from the clustering step: conflicts are most likely between requirements that operate in the same domain (high similarity) but impose different constraints. We check only pairs with cosine similarity above a threshold.

class ConflictAssessment(BaseModel):
    """LLM assessment of whether two requirements conflict."""
    conflicts: bool = Field(description="True if the requirements conflict")
    conflict_type: str = Field(
        description="Type: 'logical' (cannot both be true), "
                    "'resource' (compete for same resource), "
                    "'temporal' (incompatible timing), 'none'"
    )
    explanation: str = Field(description="Why they conflict or why they do not")
    severity: str = Field(description="high/medium/low/none")
    resolution_hint: str = Field(
        description="Suggested approach to resolve the conflict"
    )


CONFLICT_PROMPT = """Analyze whether these two requirements conflict with each other.

Requirement A ({id_a}):
  Role: {role_a}
  Capability: {cap_a}
  Benefit: {benefit_a}

Requirement B ({id_b}):
  Role: {role_b}
  Capability: {cap_b}
  Benefit: {benefit_b}

A conflict exists when:
- Both requirements cannot be satisfied simultaneously (logical conflict)
- Both requirements compete for the same limited resource (resource conflict)
- Both requirements impose incompatible timing constraints (temporal conflict)

A conflict does NOT exist when:
- Requirements are about different aspects of the system
- Requirements can be satisfied by different design choices
- One requirement is a refinement or specialization of the other

Provide your assessment."""


def find_candidate_conflicts(
    stories: list[ExtractedStory],
    embeddings: np.ndarray,
    similarity_threshold: float = 0.5,
) -> list[tuple[int, int, float]]:
    """Find requirement pairs that might conflict.

    Uses embedding similarity to focus on requirements
    in the same domain. Returns (idx_a, idx_b, similarity) triples.
    """
    from sklearn.metrics.pairwise import cosine_similarity

    sim_matrix = cosine_similarity(embeddings)
    candidates = []

    for i in range(len(stories)):
        for j in range(i + 1, len(stories)):
            if sim_matrix[i, j] >= similarity_threshold:
                candidates.append((i, j, float(sim_matrix[i, j])))

    # Sort by similarity (most similar first, most likely to conflict)
    candidates.sort(key=lambda x: -x[2])

    print(f"Found {len(candidates)} candidate pairs "
          f"(from {len(stories) * (len(stories)-1) // 2} total pairs)")

    return candidates


def assess_conflict(
    story_a: ExtractedStory,
    story_b: ExtractedStory,
    id_a: str,
    id_b: str,
    client: Anthropic,
    model: str = "claude-sonnet-4-20250514",
) -> ConflictAssessment:
    """Use an LLM to assess whether two requirements conflict."""
    prompt = CONFLICT_PROMPT.format(
        id_a=id_a, role_a=story_a.role,
        cap_a=story_a.capability, benefit_a=story_a.benefit,
        id_b=id_b, role_b=story_b.role,
        cap_b=story_b.capability, benefit_b=story_b.benefit,
    )

    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "conflict_assessment",
                "schema": ConflictAssessment.model_json_schema(),
            },
        },
    )

    import json
    raw = json.loads(response.content[0].text)
    return ConflictAssessment.model_validate(raw)

6. Graph-Based Conflict Analysis

Pairwise conflict detection catches direct contradictions, but some conflicts are transitive: requirement A does not conflict with B, and B does not conflict with C, but A and C together create an impossible situation through their shared dependency on B. We catch these structural conflicts by building a conflict graph and analyzing its properties.

The conflict graph \(G_C = (V, E)\) has requirements as vertices and two types of edges: conflicts_with (undirected, red) and depends_on (directed, blue). A transitive conflict manifests as a cycle in this graph that passes through at least one conflict edge. We detect these using NetworkX's cycle detection algorithms.

import networkx as nx
from typing import NamedTuple


class Conflict(NamedTuple):
    """A detected conflict between requirements."""
    req_a: str
    req_b: str
    conflict_type: str
    severity: str
    explanation: str
    resolution_hint: str


class ConflictGraph:
    """A graph for detecting direct and transitive requirement conflicts."""

    def __init__(self):
        self.graph = nx.Graph()  # Undirected for conflict edges
        self.dep_graph = nx.DiGraph()  # Directed for dependency edges

    def add_requirement(self, req_id: str, **attrs) -> None:
        """Add a requirement node to both graphs."""
        self.graph.add_node(req_id, **attrs)
        self.dep_graph.add_node(req_id, **attrs)

    def add_conflict(self, req_a: str, req_b: str, **attrs) -> None:
        """Record a direct conflict between two requirements."""
        self.graph.add_edge(req_a, req_b, edge_type="conflict", **attrs)

    def add_dependency(self, from_req: str, to_req: str) -> None:
        """Record that from_req depends on to_req."""
        self.dep_graph.add_edge(from_req, to_req)

    def find_direct_conflicts(self) -> list[Conflict]:
        """Return all directly conflicting pairs."""
        conflicts = []
        for u, v, data in self.graph.edges(data=True):
            if data.get("edge_type") == "conflict":
                conflicts.append(Conflict(
                    req_a=u, req_b=v,
                    conflict_type=data.get("conflict_type", "unknown"),
                    severity=data.get("severity", "medium"),
                    explanation=data.get("explanation", ""),
                    resolution_hint=data.get("resolution_hint", ""),
                ))
        return conflicts

    def find_dependency_cycles(self) -> list[list[str]]:
        """Find circular dependencies (A depends on B depends on A).

        Circular dependencies are a structural conflict:
        no implementation order can satisfy them.
        """
        try:
            cycles = list(nx.simple_cycles(self.dep_graph))
            return [c for c in cycles if len(c) > 1]
        except nx.NetworkXError:
            return []

    def find_conflict_clusters(self) -> list[set[str]]:
        """Find groups of mutually conflicting requirements.

        A conflict cluster is a connected component in the
        conflict subgraph. Large clusters indicate systemic
        disagreements between stakeholder groups.
        """
        conflict_subgraph = nx.Graph()
        for u, v, data in self.graph.edges(data=True):
            if data.get("edge_type") == "conflict":
                conflict_subgraph.add_edge(u, v)

        components = list(nx.connected_components(conflict_subgraph))
        # Only return components with actual conflicts (size > 1)
        return [c for c in components if len(c) > 1]

    def impact_analysis(self, req_id: str) -> dict[str, list[str]]:
        """Analyze the impact of changing or removing a requirement.

        Returns all requirements that conflict with or depend on
        the given requirement, transitively.
        """
        # Direct conflicts
        conflicts = [
            v for v in self.graph.neighbors(req_id)
            if self.graph.edges[req_id, v].get("edge_type") == "conflict"
        ]

        # Transitive dependents (who depends on this requirement?)
        dependents = list(nx.ancestors(self.dep_graph, req_id))

        # Transitive dependencies (what does this requirement need?)
        dependencies = list(nx.descendants(self.dep_graph, req_id))

        return {
            "direct_conflicts": conflicts,
            "dependents": dependents,
            "dependencies": dependencies,
        }

    def summary(self) -> str:
        """Generate a human-readable conflict summary."""
        direct = self.find_direct_conflicts()
        cycles = self.find_dependency_cycles()
        clusters = self.find_conflict_clusters()

        lines = [
            f"Requirements: {self.graph.number_of_nodes()}",
            f"Direct conflicts: {len(direct)}",
            f"Dependency cycles: {len(cycles)}",
            f"Conflict clusters: {len(clusters)}",
        ]

        if direct:
            lines.append("\nDirect conflicts:")
            for c in direct:
                lines.append(
                    f"  {c.req_a} <-> {c.req_b} "
                    f"[{c.severity}] {c.conflict_type}: {c.explanation}"
                )

        if cycles:
            lines.append("\nDependency cycles:")
            for cycle in cycles:
                chain = " -> ".join(cycle + [cycle[0]])
                lines.append(f"  {chain}")

        return "\n".join(lines)


# Example: build a conflict graph
cg = ConflictGraph()
cg.add_requirement("US-010", text="Require 2FA on all actions")
cg.add_requirement("US-011", text="One-click instrument control")
cg.add_requirement("US-012", text="7-year audit trail retention")
cg.add_requirement("US-013", text="Quarterly data purge for storage")
cg.add_requirement("US-014", text="All data in a single database")

# Direct conflicts
cg.add_conflict("US-010", "US-011",
    conflict_type="logical",
    severity="high",
    explanation="2FA on every action prevents one-click workflows",
    resolution_hint="Define 'sensitive actions' that require 2FA vs. "
                    "routine actions that use session-based auth")

cg.add_conflict("US-012", "US-013",
    conflict_type="temporal",
    severity="high",
    explanation="Cannot retain data for 7 years and purge quarterly",
    resolution_hint="Distinguish audit data (retained) from working "
                    "data (purgeable); archive audit trail separately")

# Dependencies
cg.add_dependency("US-013", "US-014")  # Purge needs single DB
cg.add_dependency("US-012", "US-014")  # Retention needs single DB

print(cg.summary())

7. Putting It Together: The Extraction Pipeline

So far we have built each stage in isolation: extraction, validation, clustering, and conflict detection, each with its own inputs and outputs.

The complete extraction pipeline chains these components into a single function that takes raw transcripts and produces a validated, deduplicated, conflict-checked requirement set with a populated traceability matrix. Figure 13.2 shows the six stages and the data that flows between them. Figure 13.2.1 illustrates the end-to-end requirement extraction pipeline.

from pathlib import Path


def requirements_pipeline(
    transcript_dir: Path,
    client: Anthropic,
    similarity_threshold: float = 0.5,
    confidence_threshold: float = 0.5,
    cluster_eps: float = 0.15,
) -> tuple[list[ExtractedStory], ConflictGraph, TraceabilityMatrix]:
    """End-to-end requirement discovery pipeline.

    1. Read all transcripts from a directory
    2. Extract requirements from each using LLM
    3. Validate and filter
    4. Cluster and deduplicate
    5. Detect conflicts
    6. Build traceability matrix

    Args:
        transcript_dir: Directory containing .txt transcript files.
        client: Anthropic API client.
        similarity_threshold: Cosine similarity for conflict candidates.
        confidence_threshold: Minimum extraction confidence.
        cluster_eps: DBSCAN epsilon for deduplication clustering.

    Returns:
        Tuple of (final_stories, conflict_graph, traceability_matrix).
    """
    # Step 1: Read transcripts
    transcripts = {}
    for path in sorted(transcript_dir.glob("*.txt")):
        transcripts[path.stem] = path.read_text(encoding="utf-8")
    print(f"Loaded {len(transcripts)} transcripts")

    # Step 2: Extract from each transcript
    all_results = []
    for source_id, text in transcripts.items():
        results = extract_requirements(text, client)
        for r in results:
            for s in r.stories:
                s.source_quote = f"[{source_id}] {s.source_quote}"
        all_results.extend(results)

    # Step 3: Validate and filter
    clean_stories, reports = filter_and_report(
        all_results, min_confidence=confidence_threshold
    )
    print(f"After validation: {len(clean_stories)} clean stories")

    # Step 4: Cluster and deduplicate
    if len(clean_stories) >= 2:
        embeddings = embed_stories(clean_stories, client)
        clusters = cluster_requirements(
            clean_stories, embeddings, eps=cluster_eps
        )

        # Deduplicate within each cluster
        final_stories = []
        for label, indices in clusters.items():
            if label == -1:
                # Unclustered stories are unique
                final_stories.extend(
                    clean_stories[i] for i in indices
                )
            else:
                # Merge duplicates within cluster
                representative = deduplicate_cluster(
                    clean_stories, indices
                )
                final_stories.append(representative)
    else:
        final_stories = clean_stories
        embeddings = embed_stories(clean_stories, client)

    print(f"After deduplication: {len(final_stories)} stories")

    # Step 5: Conflict detection
    conflict_graph = ConflictGraph()
    for i, story in enumerate(final_stories):
        req_id = f"US-{i+1:03d}"
        conflict_graph.add_requirement(
            req_id,
            role=story.role,
            capability=story.capability,
            benefit=story.benefit,
        )

    # Re-embed final stories for conflict candidate search
    final_embeddings = embed_stories(final_stories, client)
    candidates = find_candidate_conflicts(
        final_stories, final_embeddings,
        similarity_threshold=similarity_threshold,
    )

    for idx_a, idx_b, sim in candidates:
        id_a = f"US-{idx_a+1:03d}"
        id_b = f"US-{idx_b+1:03d}"
        assessment = assess_conflict(
            final_stories[idx_a], final_stories[idx_b],
            id_a, id_b, client,
        )
        if assessment.conflicts:
            conflict_graph.add_conflict(
                id_a, id_b,
                conflict_type=assessment.conflict_type,
                severity=assessment.severity,
                explanation=assessment.explanation,
                resolution_hint=assessment.resolution_hint,
            )

    print(conflict_graph.summary())

    # Step 6: Build traceability matrix
    matrix = TraceabilityMatrix()
    for source_id in transcripts:
        matrix.add_source(source_id, f"Transcript: {source_id}")

    for i, story in enumerate(final_stories):
        req_id = f"US-{i+1:03d}"
        full_story = UserStory(
            id=req_id,
            role=StakeholderRole.RESEARCHER,
            capability=story.capability,
            benefit=story.benefit,
            priority=Priority(story.priority_hint),
            source=story.source_quote.split("]")[0].strip("["),
        )
        matrix.add_story(full_story)

        # Link to source transcript
        source_key = story.source_quote.split("]")[0].strip("[")
        if source_key in transcripts:
            matrix.link(source_key, req_id, LinkType.DERIVED_FROM)

    report = matrix.coverage_report()
    orphans = matrix.orphan_stories()
    print(f"\nTraceability: {len(report)} stories, "
          f"{len(orphans)} orphans")

    return final_stories, conflict_graph, matrix

Exercises

1. Conceptual: The pairwise conflict check has \(O(n^2)\) complexity in the worst case. Our embedding similarity filter reduces this, but the worst case remains quadratic. Propose a hierarchical approach that first checks conflicts between requirement clusters (from the deduplication step) and only examines individual pairs within conflicting clusters. What is the new complexity in terms of the number of clusters \(k\) and the average cluster size \(c\)?
2. Coding: Implement a generate_follow_up_questions function that takes the current requirement set and the list of ambiguities from extraction, and uses an LLM to produce targeted questions for the next stakeholder interview. The function should prioritize: (1) resolving detected conflicts, (2) quantifying ambiguous terms, (3) filling coverage gaps (stakeholder roles with few or no stories). Test with a mock requirement set containing 3 conflicts and 5 ambiguities.
3. Analysis: The confidence boosting in deduplicate_cluster adds 0.1 per additional source. Is this a reasonable heuristic? Consider a scenario where 5 stakeholders from the same department all request the same feature (high redundancy, same perspective) versus 3 stakeholders from different departments (lower redundancy, diverse perspectives). Design a confidence boosting formula that accounts for source diversity, not just source count.