Spaces:
Sleeping
Sleeping
| """ | |
| ConsistencyValidator - cross-agent contradiction detection. | |
| Validates knowledge graph entities across all role-scoped sub-graphs | |
| to surface contradictions before SRS assembly. | |
| The validator is purely deterministic (no LLM calls). Output is | |
| append-only (merge_lists reducer); no code path modifies KG entities. | |
| """ | |
| import logging | |
| from typing import Any | |
| from .kg_service import KGService | |
| logger = logging.getLogger("consistency_validator") | |
| class ConsistencyValidator: | |
| """Deterministic cross-agent contradiction detection. | |
| Merges role-scoped KGs, delegates to KGService.find_contradictions(), | |
| deduplicates results, and provides markdown formatting for warnings. | |
| """ | |
| def __init__(self): | |
| self._kg = KGService() | |
| def validate_all(self, state: dict[str, Any]) -> list[dict]: | |
| """Validate consistency across all agent role graphs. | |
| Args: | |
| state: AgentState dict containing knowledge_graph key. | |
| Returns: | |
| Deduplicated list of contradiction dicts (empty if none). | |
| """ | |
| kg = state.get("knowledge_graph", {}) | |
| if not kg: | |
| return [] | |
| # Merge all role-scoped sub-graphs into one unified graph | |
| merged = self._kg.merge_role_graphs(state) | |
| if not merged.get("nodes"): | |
| return [] | |
| # Find contradictions in the merged graph | |
| contradictions = self._kg.find_contradictions(merged) | |
| # Deduplicate by entity_id pair (sorted tuple of IDs) | |
| seen: set[tuple[str, ...]] = set() | |
| deduplicated: list[dict] = [] | |
| for c in contradictions: | |
| key = tuple(sorted(c.get("entity_ids", []))) | |
| if key not in seen: | |
| seen.add(key) | |
| deduplicated.append(c) | |
| for c in deduplicated: | |
| logger.warning( | |
| "Contradiction [%s]: %s (roles: %s)", | |
| c.get("type", "unknown"), | |
| c.get("detail", ""), | |
| ", ".join(c.get("roles", [])), | |
| ) | |
| return deduplicated | |
| def format_warnings_section(contradictions: list[dict]) -> str: | |
| """Render contradictions as a markdown warnings section. | |
| Args: | |
| contradictions: List of contradiction dicts from validate_all(). | |
| Returns: | |
| Markdown string, empty string if no contradictions. | |
| """ | |
| if not contradictions: | |
| return "" | |
| lines = ["## Consistency Warnings"] | |
| lines.append( | |
| "\n> **Note:** The following inconsistencies were detected across " | |
| "agent outputs. These do not block SRS assembly but should be " | |
| "reviewed before final adoption.\n" | |
| ) | |
| for i, c in enumerate(contradictions, 1): | |
| sev = c.get("severity", "medium") | |
| # The severity already appears in the heading; this prefix just | |
| # makes high-severity rows scannable without relying on emoji. | |
| sev_marker = "[HIGH]" if sev == "high" else "[MED]" | |
| ctype = c.get("type", "unknown").replace("_", " ").title() | |
| detail = c.get("detail", "") | |
| roles = ", ".join(c.get("roles", [])) | |
| entity_ids = ", ".join(c.get("entity_ids", [])) | |
| lines.append(f"### {i}. {sev_marker} **{ctype}** *(Severity: {sev})*") | |
| lines.append(f"- **Detail:** {detail}") | |
| if entity_ids: | |
| lines.append(f"- **Entities:** {entity_ids}") | |
| if roles: | |
| lines.append(f"- **Agent Roles:** {roles}") | |
| conflicts = c.get("conflicts", []) | |
| if conflicts: | |
| for conflict in conflicts: | |
| prop = conflict.get("property", "unknown") | |
| vals = conflict.get("values", {}) | |
| if vals: | |
| val_str = "; ".join( | |
| f"{role}: {val}" for role, val in vals.items() | |
| ) | |
| lines.append(f"- **{prop}:** {val_str}") | |
| else: | |
| lines.append(f"- **{prop}:** conflicting values") | |
| lines.append("") | |
| return "\n".join(lines) | |