diff --git a/Dockerfile b/Dockerfile index e87d122..8df508e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,8 @@ FROM python:3.12-slim-bookworm +LABEL org.opencontainers.image.title="Conflict Checker" \ + org.opencontainers.image.description="Classic and experimental scoped Agent pipelines" + RUN apt-get update \ && apt-get install -y --no-install-recommends poppler-utils \ && rm -rf /var/lib/apt/lists/* diff --git a/README.md b/README.md index 54e2f1c..21ef1a6 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,30 @@ python cli/run_check.py samples/your_set.pdf --out out/your_set # -> out/your_set/report.md + conflicts.json ``` +The Classic pipeline remains the recommended default. The experimental Agent +fork runs in the same image and can be selected in the web UI or from the CLI: + +```bash +python cli/run_check.py samples/your_set.pdf --mode agent --out out/agent-run +``` + +Agent mode uses OpenRouter for every model call and runs bounded specialist +waves: one-sheet extraction, sheet-index/jurisdiction orientation, semantic +linkers partitioned by level and object family, per-cluster conflict critics, +batched code review, cluster-scoped constructability, summary-only completeness, +central Brain consolidation, and one-finding RFI writers. It returns the same +`conflicts`, `validated_issues`, `rfis`, and `summary` fields as Classic. + +Agent artifacts are also written under `/agent/`, including wave +snapshots and the final Project Memory. `summary.agent_stats`, +`summary.cost_by_stage`, and `summary.models_used` are job-local, so concurrent +Agent jobs do not share accounting. + +Optional `AGENT_*_MODEL` variables select an OpenRouter model per specialist. +The `AGENT_*_CONCURRENCY` and scope-cap variables in `backend/.env.example` +bound fan-out and prompt size. Agent mode intentionally ignores the hybrid/local +text option in v1. + Web UI (upload + view): ```bash @@ -155,6 +179,11 @@ uvicorn backend.main:app --reload --port 8099 # open http://127.0.0.1:8099 Or use Docker: `docker compose up -d` (see **Setup** above). +The standard Docker image contains both pipelines; no additional queue, +database, or model service is required. Set `AI_API_KEY` in `backend/.env` and +choose Agent mode per request. Treat Agent output as experimental and compare it +against a reviewed golden set before using it for issuance decisions. + ## Conflict categories `dimensional_disagreement`, `elevation_disagreement`, `location_mismatch`, diff --git a/backend/.env.example b/backend/.env.example index 705c680..97fc051 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -4,6 +4,29 @@ AI_BASE_URL=https://openrouter.ai/api/v1 AI_API_KEY=sk-or-... MODEL=google/gemini-2.5-pro +# Optional Agent-mode OpenRouter model overrides (inherit MODEL/TEXT_MODEL when blank) +AGENT_EXTRACT_MODEL= +AGENT_INDEX_MODEL= +AGENT_JURISDICTION_MODEL= +AGENT_LINKER_MODEL= +AGENT_CONFLICT_MODEL= +AGENT_CODE_MODEL= +AGENT_CONSTRUCT_MODEL= +AGENT_COMPLETENESS_MODEL= +AGENT_BRAIN_MODEL= +AGENT_RFI_MODEL= + +# Agent-mode hard scope limits / concurrency +AGENT_LINK_MAX_ASSERTIONS=60 +AGENT_CLUSTER_MAX_ASSERTIONS=24 +AGENT_CONFLICT_MAX_IMAGES=6 +AGENT_CODE_BATCH_SIZE=60 +AGENT_BRAIN_MAX_TOKENS=16384 +AGENT_LINK_CONCURRENCY=4 +AGENT_CONFLICT_CONCURRENCY=4 +AGENT_SPECIALIST_CONCURRENCY=4 +AGENT_RFI_CONCURRENCY=4 + # Pipeline tuning PDF_DPI=100 MAX_PAGES=60 diff --git a/backend/agents/__init__.py b/backend/agents/__init__.py new file mode 100644 index 0000000..cb1355f --- /dev/null +++ b/backend/agents/__init__.py @@ -0,0 +1,10 @@ +"""Parallel, specialist-agent pipeline isolated from the Classic runner.""" + + +def run_agent_pipeline(*args, **kwargs): + """Lazy package-level entry point that avoids importing optional runtime deps.""" + from backend.agents.runner import run_agent_pipeline as _run + + return _run(*args, **kwargs) + +__all__ = ["run_agent_pipeline"] diff --git a/backend/agents/base.py b/backend/agents/base.py new file mode 100644 index 0000000..d4e464c --- /dev/null +++ b/backend/agents/base.py @@ -0,0 +1,89 @@ +"""Shared contracts and job-local accounting for Agent-mode workers.""" + +import threading +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Protocol + + +@dataclass(frozen=True) +class AgentScope: + """A bounded work package passed to exactly one specialist agent.""" + + scope_id: str + payload: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class AgentResult: + """Artifacts returned by a specialist for collection by the orchestrator.""" + + scope_id: str + artifacts: List[Dict[str, Any]] = field(default_factory=list) + error: str = "" + + +@dataclass +class AgentUsage: + """Thread-safe usage accounting owned by one Agent pipeline run.""" + + usd: float = 0.0 + calls: int = 0 + cached: int = 0 + by_stage: Dict[str, Dict[str, Any]] = field(default_factory=dict) + models: Dict[str, set] = field(default_factory=lambda: { + "vision": set(), + "text_cloud": set(), + }) + _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + def record( + self, + stage: str, + model: str, + usd: float = 0.0, + cached: bool = False, + has_images: bool = False, + ) -> None: + with self._lock: + bucket = self.by_stage.setdefault( + stage, {"usd": 0.0, "calls": 0, "cached": 0} + ) + if cached: + self.cached += 1 + bucket["cached"] += 1 + else: + self.calls += 1 + self.usd += usd + bucket["calls"] += 1 + bucket["usd"] += usd + family = "vision" if has_images else "text_cloud" + self.models[family].add(model) + + def snapshot(self) -> Dict[str, Any]: + with self._lock: + return { + "usd": self.usd, + "calls": self.calls, + "cached": self.cached, + "by_stage": {k: dict(v) for k, v in self.by_stage.items()}, + "models": { + "vision": sorted(self.models["vision"]), + "text_local": [], + "text_cloud": sorted(self.models["text_cloud"]), + "fallback_count": 0, + }, + } + + +class ScopedAgent(Protocol): + """Protocol implemented by each future specialist agent.""" + + name: str + + def run(self, scope: AgentScope) -> AgentResult: + ... + + +def failure(scope: AgentScope, error: Exception) -> AgentResult: + """Convert a worker exception into a non-fatal scoped result.""" + return AgentResult(scope_id=scope.scope_id, error=str(error)) diff --git a/backend/agents/brain.py b/backend/agents/brain.py new file mode 100644 index 0000000..5aef589 --- /dev/null +++ b/backend/agents/brain.py @@ -0,0 +1,129 @@ +"""Central merge, judge, and prioritization agent.""" + +import json +import re +from typing import Dict, List, Tuple + +from backend import config +from backend.agents.base import AgentUsage +from backend.agents.prompts import BRAIN_SYSTEM_PROMPT, BRAIN_USER_PROMPT +from backend.llm import call_json +from backend.pipeline._stage import collect_list, validate_issue + + +def _finding_ref(finding: Dict, index: int) -> str: + return ( + finding.get("issue_id") + or f"{finding.get('agent', 'agent')}:{finding.get('scope_id', '?')}:{index + 1}" + ) + + +def _signature(finding: Dict) -> Tuple[str, str, str]: + norm = lambda value: re.sub(r"[^a-z0-9]+", " ", str(value).lower()).strip() + description = " ".join(norm(finding.get("description")).split()[:12]) + return ( + norm(finding.get("category")), + norm(finding.get("location")), + description, + ) + + +def _fallback(findings: List[Dict]) -> Tuple[List[Dict], List[Dict]]: + """Conservative local consolidation when the Brain call fails.""" + kept: Dict[Tuple[str, str, str], Dict] = {} + refs: Dict[Tuple[str, str, str], List[str]] = {} + decisions: List[Dict] = [] + severity_rank = {"critical": 4, "high": 3, "medium": 2, "low": 1} + for index, finding in enumerate(findings): + ref = _finding_ref(finding, index) + supported = bool(finding.get("evidence")) or finding.get("agent") == "completeness" + if not supported or not finding.get("description"): + decisions.append({ + "finding_refs": [ref], + "action": "dropped", + "reason": "missing actionable support", + "kept_issue_id": None, + }) + continue + signature = _signature(finding) + if signature not in kept: + kept[signature] = dict(finding) + refs[signature] = [ref] + else: + refs[signature].append(ref) + existing = kept[signature] + if severity_rank.get(finding.get("severity"), 2) > severity_rank.get( + existing.get("severity"), 2 + ): + existing["severity"] = finding.get("severity") + existing["evidence"] = ( + existing.get("evidence") or [] + ) + (finding.get("evidence") or []) + + issues = list(kept.values()) + for index, (signature, issue) in enumerate(kept.items()): + issue["issue_id"] = issue.get("issue_id") or f"AGENT-{index + 1:04d}" + issue["risk_score"] = { + "critical": 95, "high": 75, "medium": 50, "low": 25 + }.get(issue.get("severity"), 50) + issue["recommended_priority"] = { + "critical": "immediate", + "high": "before_bid", + "medium": "before_construction", + "low": "track_only", + }.get(issue.get("severity"), "before_construction") + decisions.append({ + "finding_refs": refs[signature], + "action": "merged" if len(refs[signature]) > 1 else "kept", + "reason": "conservative deterministic fallback", + "kept_issue_id": issue["issue_id"], + }) + issues.sort(key=lambda item: -int(item.get("risk_score") or 0)) + return issues, decisions + + +class BrainAgent: + name = "brain" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run( + self, + findings: List[Dict], + sheet_index: Dict, + jurisdiction: Dict, + ) -> Tuple[List[Dict], List[Dict]]: + instruction = BRAIN_USER_PROMPT + for key, value in { + "sheet_index": sheet_index, + "jurisdiction": jurisdiction, + "findings": findings, + }.items(): + instruction = instruction.replace( + "{" + key + "}", json.dumps(value, ensure_ascii=True) + ) + parsed = call_json( + system_prompt=BRAIN_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.AGENT_BRAIN_MAX_TOKENS, + model=config.AGENT_BRAIN_MODEL, + usage_tracker=self.usage, + usage_stage="agent.brain", + ) + issues = collect_list( + parsed, "issues", lambda item: validate_issue(item, item.get("source_stage", "")) + ) + if not issues: + return _fallback(findings) + raw_issues = parsed.get("issues") if isinstance(parsed, dict) else [] + for index, issue in enumerate(issues): + raw = raw_issues[index] if index < len(raw_issues) else {} + issue["issue_id"] = issue.get("issue_id") or f"AGENT-{index + 1:04d}" + issue["risk_score"] = raw.get("risk_score") or issue.get("risk_score") or 50 + issue["recommended_priority"] = ( + raw.get("recommended_priority") or "before_construction" + ) + issues.sort(key=lambda item: -int(item.get("risk_score") or 0)) + decisions = parsed.get("decisions") or [] + return issues, [item for item in decisions if isinstance(item, dict)] diff --git a/backend/agents/code_agent.py b/backend/agents/code_agent.py new file mode 100644 index 0000000..83ff495 --- /dev/null +++ b/backend/agents/code_agent.py @@ -0,0 +1,95 @@ +"""Scoped code/accessibility review agents.""" + +from typing import Dict, List + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline import code_refs +from backend.pipeline._serialize import dumps, slim_sheets +from backend.pipeline._stage import collect_list, validate_issue +from backend.pipeline.jurisdiction import active_review_paths +from backend.prompts import CODE_REVIEW_SYSTEM_PROMPT, CODE_REVIEW_USER_INSTRUCTION + + +def build_code_scopes( + sheets: List[Dict], jurisdiction: Dict, sheet_index: Dict +) -> List[AgentScope]: + cap = max(1, config.AGENT_CODE_BATCH_SIZE) + fragments: List[Dict] = [] + for sheet in sheets: + assertions = sheet.get("assertions") or [] + if not assertions: + continue + for offset in range(0, len(assertions), cap): + fragments.append({ + **sheet, + "assertions": assertions[offset:offset + cap], + }) + batches: List[List[Dict]] = [] + current: List[Dict] = [] + count = 0 + for fragment in fragments: + size = len(fragment["assertions"]) + if current and count + size > cap: + batches.append(current) + current, count = [], 0 + current.append(fragment) + count += size + if current: + batches.append(current) + return [ + AgentScope( + scope_id=f"code:{index + 1}", + payload={ + "sheets": batch, + "jurisdiction": jurisdiction, + "sheet_index": sheet_index, + }, + ) + for index, batch in enumerate(batches) + ] + + +class CodeAgent: + name = "code_reviewer" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + sheets = scope.payload.get("sheets") or [] + jurisdiction = scope.payload.get("jurisdiction") or {} + sheet_index = scope.payload.get("sheet_index") or {} + assertions = [ + assertion for sheet in sheets + for assertion in sheet.get("assertions", []) + ] + excerpts = code_refs.retrieve( + active_review_paths(jurisdiction), assertions + ) + instruction = CODE_REVIEW_USER_INSTRUCTION + for key, value in { + "jurisdiction": dumps(jurisdiction), + "sheet_index": dumps(sheet_index), + "assertions": dumps(slim_sheets(sheets)), + "code_references": code_refs.format_excerpts(excerpts), + }.items(): + instruction = instruction.replace("{" + key + "}", value) + parsed = call_json( + system_prompt=CODE_REVIEW_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.CODE_MAX_TOKENS, + model=config.AGENT_CODE_MODEL, + usage_tracker=self.usage, + usage_stage="agent.code", + ) + findings = collect_list( + parsed, "issues", lambda item: validate_issue(item, "code") + ) + for finding in findings: + finding.update(agent=self.name, scope_id=scope.scope_id) + return AgentResult(scope_id=scope.scope_id, artifacts=findings) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/completeness.py b/backend/agents/completeness.py new file mode 100644 index 0000000..bb09e12 --- /dev/null +++ b/backend/agents/completeness.py @@ -0,0 +1,55 @@ +"""Summary-only drawing-set completeness agent.""" + +import json +from typing import Dict, List + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.agents.prompts import COMPLETENESS_SYSTEM_PROMPT, COMPLETENESS_USER_PROMPT +from backend.llm import call_json +from backend.pipeline._stage import collect_list, validate_issue + + +def build_sheet_summaries(sheets: List[Dict]) -> List[Dict]: + """Return counts and classifications only; never raw assertions.""" + return [{ + "sheet_number": sheet.get("sheet_number"), + "sheet_title": sheet.get("sheet_title"), + "discipline": sheet.get("discipline"), + "drawing_type": sheet.get("drawing_type"), + "level": sheet.get("level"), + "assertion_count": len(sheet.get("assertions") or []), + "unresolved_count": len(sheet.get("unresolved_items") or []), + } for sheet in sheets] + + +class CompletenessAgent: + name = "completeness" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + instruction = COMPLETENESS_USER_PROMPT + for key in ("sheet_index", "sheet_summaries", "cluster_summary"): + instruction = instruction.replace( + "{" + key + "}", + json.dumps(scope.payload.get(key) or {}, ensure_ascii=True), + ) + parsed = call_json( + system_prompt=COMPLETENESS_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.QAQC_MAX_TOKENS, + model=config.AGENT_COMPLETENESS_MODEL, + usage_tracker=self.usage, + usage_stage="agent.completeness", + ) + findings = collect_list( + parsed, "issues", lambda item: validate_issue(item, "qaqc") + ) + for finding in findings: + finding.update(agent=self.name, scope_id=scope.scope_id) + return AgentResult(scope_id=scope.scope_id, artifacts=findings) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/conflict_critic.py b/backend/agents/conflict_critic.py new file mode 100644 index 0000000..2ca7f17 --- /dev/null +++ b/backend/agents/conflict_critic.py @@ -0,0 +1,74 @@ +"""Per-cluster conflict critics with hard evidence and image caps.""" + +from typing import Dict, List + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline.conflict_checker import _evidence_block, _valid_conflict +from backend.prompts import CONFLICT_SYSTEM_PROMPT, CONFLICT_USER_INSTRUCTION + + +def _as_finding(conflict: Dict, scope_id: str) -> Dict: + return { + "issue_id": conflict.get("conflict_id") or "", + "source_stage": "conflict", + "category": conflict.get("category") or "uncategorized", + "severity": conflict.get("severity") or "medium", + "confidence": conflict.get("confidence") or "medium", + "location": conflict.get("location") or "", + "disciplines": conflict.get("disciplines") or [], + "sheets": conflict.get("sheets") or [], + "description": conflict.get("description") or "", + "evidence": conflict.get("evidence") or [], + "recommended_resolution": conflict.get("recommended_resolution") or "", + "code_reference": None, + "agent": "conflict_critic", + "scope_id": scope_id, + } + + +class ConflictCriticAgent: + name = "conflict_critic" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + cluster = dict(scope.payload["cluster"]) + cluster["assertions"] = ( + cluster.get("assertions") or [] + )[:config.AGENT_CLUSTER_MAX_ASSERTIONS] + page_to_b64: Dict[int, str] = scope.payload.get("page_to_b64") or {} + images: List[str] = [] + for page_number in ( + cluster.get("page_numbers") or [] + )[:config.AGENT_CONFLICT_MAX_IMAGES]: + if page_to_b64.get(page_number): + images.append(page_to_b64[page_number]) + instruction = ( + CONFLICT_USER_INSTRUCTION + .replace("{location}", cluster.get("location") or "") + .replace("{evidence}", _evidence_block(cluster)) + ) + parsed = call_json( + system_prompt=CONFLICT_SYSTEM_PROMPT, + user_text=instruction, + images_b64=images, + max_tokens=config.REASON_MAX_TOKENS, + model=config.AGENT_CONFLICT_MODEL, + usage_tracker=self.usage, + usage_stage="agent.conflict", + ) + candidates = parsed if isinstance(parsed, list) else ( + parsed.get("conflicts") if isinstance(parsed, dict) else [] + ) + findings = [] + for candidate in candidates or []: + conflict = _valid_conflict(candidate, cluster) + if conflict: + findings.append(_as_finding(conflict, scope.scope_id)) + return AgentResult(scope_id=scope.scope_id, artifacts=findings) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/construct_agent.py b/backend/agents/construct_agent.py new file mode 100644 index 0000000..5a935ec --- /dev/null +++ b/backend/agents/construct_agent.py @@ -0,0 +1,70 @@ +"""Zone/cluster-scoped constructability agents.""" + +from typing import Dict, List + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline._serialize import dumps, slim_clusters +from backend.pipeline._stage import collect_list, validate_issue +from backend.prompts import ( + CONSTRUCTABILITY_SYSTEM_PROMPT, + CONSTRUCTABILITY_USER_INSTRUCTION, +) + + +def build_construct_scopes( + clusters: List[Dict], conflict_findings: List[Dict] +) -> List[AgentScope]: + scopes = [] + for index, cluster in enumerate(clusters): + related = [ + finding for finding in conflict_findings + if finding.get("scope_id") == f"conflict:{cluster.get('key')}" + or finding.get("location") == cluster.get("location") + ] + scopes.append(AgentScope( + scope_id=f"construct:{index + 1}", + payload={"cluster": cluster, "conflicts": related}, + )) + return scopes + + +class ConstructabilityAgent: + name = "constructability" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + cluster = dict(scope.payload["cluster"]) + cluster["assertions"] = ( + cluster.get("assertions") or [] + )[:config.AGENT_CLUSTER_MAX_ASSERTIONS] + instruction = CONSTRUCTABILITY_USER_INSTRUCTION + substitutions = { + "assertions": dumps(cluster["assertions"]), + "clusters": dumps(slim_clusters([cluster])), + "conflicts": dumps(scope.payload.get("conflicts") or []), + } + for key, value in substitutions.items(): + instruction = instruction.replace("{" + key + "}", value) + parsed = call_json( + system_prompt=CONSTRUCTABILITY_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.CONSTRUCT_MAX_TOKENS, + model=config.AGENT_CONSTRUCT_MODEL, + usage_tracker=self.usage, + usage_stage="agent.constructability", + ) + findings = collect_list( + parsed, + "issues", + lambda item: validate_issue(item, "constructability"), + ) + for finding in findings: + finding.update(agent=self.name, scope_id=scope.scope_id) + return AgentResult(scope_id=scope.scope_id, artifacts=findings) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/extractors.py b/backend/agents/extractors.py new file mode 100644 index 0000000..2123a90 --- /dev/null +++ b/backend/agents/extractors.py @@ -0,0 +1,106 @@ +"""Scoped extraction and orientation agents.""" + +import json +from typing import Dict + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline.extractor import _normalize_sheet +from backend.pipeline.sheet_index import _index_input +from backend.prompts import ( + EXTRACTOR_SYSTEM_PROMPT, + EXTRACTOR_USER_INSTRUCTION, + JURISDICTION_SYSTEM_PROMPT, + JURISDICTION_USER_INSTRUCTION, + SHEET_INDEX_SYSTEM_PROMPT, + SHEET_INDEX_USER_INSTRUCTION, +) + + +class SheetExtractorAgent: + name = "sheet_extractor" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + page = scope.payload["page"] + instruction = EXTRACTOR_USER_INSTRUCTION.replace( + "{sheet_hint}", str(scope.payload.get("sheet_hint") or "") + ) + parsed = call_json( + system_prompt=EXTRACTOR_SYSTEM_PROMPT, + user_text=instruction, + images_b64=[page["base64"]], + max_tokens=config.EXTRACT_MAX_TOKENS, + model=config.AGENT_EXTRACT_MODEL, + usage_tracker=self.usage, + usage_stage="agent.extract", + ) + if not isinstance(parsed, dict): + raise ValueError("no structured extraction returned") + sheet = _normalize_sheet(parsed, page["page_number"]) + return AgentResult(scope_id=scope.scope_id, artifacts=[sheet]) + except Exception as exc: + return failure(scope, exc) + + +class SheetIndexAgent: + name = "sheet_index" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + sheets = scope.payload.get("sheets") or [] + instruction = SHEET_INDEX_USER_INSTRUCTION.replace( + "{sheet_index_input}", + json.dumps(_index_input(sheets), ensure_ascii=True), + ) + parsed = call_json( + system_prompt=SHEET_INDEX_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.SHEET_INDEX_MAX_TOKENS, + model=config.AGENT_INDEX_MODEL, + usage_tracker=self.usage, + usage_stage="agent.sheet_index", + ) + if isinstance(parsed, list): + parsed = {"sheet_index": parsed, "missing_expected_sheets": []} + if not isinstance(parsed, dict): + raise ValueError("no sheet index returned") + return AgentResult(scope_id=scope.scope_id, artifacts=[parsed]) + except Exception as exc: + return failure(scope, exc) + + +class JurisdictionAgent: + name = "jurisdiction" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + project_input: Dict = scope.payload.get("project_input") or {} + instruction = JURISDICTION_USER_INSTRUCTION.replace( + "{project_input}", json.dumps(project_input, ensure_ascii=True) + ) + parsed = call_json( + system_prompt=JURISDICTION_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.JURISDICTION_MAX_TOKENS, + model=config.AGENT_JURISDICTION_MODEL, + usage_tracker=self.usage, + usage_stage="agent.jurisdiction", + ) + if not isinstance(parsed, dict): + raise ValueError("no jurisdiction profile returned") + profile = parsed.get("project_code_profile") + artifact = profile if isinstance(profile, dict) else parsed + return AgentResult(scope_id=scope.scope_id, artifacts=[artifact]) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/linker.py b/backend/agents/linker.py new file mode 100644 index 0000000..184c8e0 --- /dev/null +++ b/backend/agents/linker.py @@ -0,0 +1,193 @@ +"""Bounded semantic linkers that build coordination clusters.""" + +import json +import re +from collections import defaultdict +from typing import Dict, Iterable, List, Tuple + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline.clusterer import cluster_by_location +from backend.pipeline.llm_clusterer import _location +from backend.prompts import CLUSTER_SYSTEM_PROMPT, CLUSTER_USER_INSTRUCTION + + +def _family(assertion: Dict) -> str: + location = assertion.get("location_key") or {} + if location.get("room"): + return "room" + if location.get("grid"): + return "grid" + if location.get("detail_reference"): + return "detail" + tag = str(location.get("tag") or "") + match = re.match(r"[A-Za-z]+", tag) + return ( + (match.group(0).lower() if match else "") + or (assertion.get("object_type") or "").lower() + or "general" + ) + + +def build_link_scopes(sheets: List[Dict]) -> List[AgentScope]: + """Partition facts by level and object/tag family, then enforce a hard cap.""" + buckets: Dict[Tuple[str, str], List[Dict]] = defaultdict(list) + for sheet in sheets: + for assertion in sheet.get("assertions", []): + enriched = { + **assertion, + "discipline": sheet.get("discipline") or "Unknown", + "sheet_number": sheet.get("sheet_number"), + "page_number": sheet.get("page_number"), + } + level = str((assertion.get("location_key") or {}).get("level") + or sheet.get("level") or "unknown").lower() + buckets[(level, _family(assertion))].append(enriched) + + scopes: List[AgentScope] = [] + cap = max(2, config.AGENT_LINK_MAX_ASSERTIONS) + for (level, family), assertions in sorted(buckets.items()): + for offset in range(0, len(assertions), cap): + chunk = assertions[offset:offset + cap] + if len(chunk) < 2: + continue + scopes.append(AgentScope( + scope_id=f"{level}:{family}:{offset // cap + 1}", + payload={"assertions": chunk, "level": level, "family": family}, + )) + return scopes + + +def _payload(assertions: Iterable[Dict]) -> List[Dict]: + return [{ + "assertion_id": item.get("id"), + "discipline": item.get("discipline"), + "sheet_number": item.get("sheet_number"), + "attribute": item.get("attribute"), + "value": item.get("value"), + "location_key": item.get("location_key"), + "source_text": item.get("source_text"), + } for item in assertions] + + +def _fallback(assertions: List[Dict]) -> List[Dict]: + """Use the deterministic linker within this scope when semantic linking fails.""" + by_sheet: Dict[Tuple, Dict] = {} + for item in assertions: + key = (item.get("sheet_number"), item.get("page_number")) + sheet = by_sheet.setdefault(key, { + "sheet_number": item.get("sheet_number"), + "page_number": item.get("page_number"), + "discipline": item.get("discipline"), + "assertions": [], + }) + sheet["assertions"].append(item) + return cluster_by_location(list(by_sheet.values())) + + +class LinkerAgent: + name = "linker" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + assertions = scope.payload.get("assertions") or [] + by_id = {item.get("id"): item for item in assertions if item.get("id")} + instruction = CLUSTER_USER_INSTRUCTION.replace( + "{normalized_assertions}", + json.dumps(_payload(assertions), ensure_ascii=True), + ) + parsed = call_json( + system_prompt=CLUSTER_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.CLUSTER_MAX_TOKENS, + model=config.AGENT_LINKER_MODEL, + usage_tracker=self.usage, + usage_stage="agent.link", + ) + raw = parsed if isinstance(parsed, list) else ( + parsed.get("clusters") if isinstance(parsed, dict) else [] + ) + clusters: List[Dict] = [] + for candidate in raw or []: + if not isinstance(candidate, dict): + continue + members = [ + by_id[item_id] + for item_id in candidate.get("assertion_ids") or [] + if item_id in by_id + ] + if len(members) < 2: + continue + allowed_sheets = [] + for member in members: + sheet = member.get("sheet_number") + if sheet not in allowed_sheets: + allowed_sheets.append(sheet) + allowed_sheets = allowed_sheets[:config.AGENT_CONFLICT_MAX_IMAGES] + members = [ + member for member in members + if member.get("sheet_number") in allowed_sheets + ][:config.AGENT_CLUSTER_MAX_ASSERTIONS] + primary = candidate.get("primary_location_key") or {} + clusters.append({ + "key": f"{scope.scope_id}:{candidate.get('cluster_id') or len(clusters) + 1}", + "location": _location(primary), + "disciplines": sorted({ + member.get("discipline") or "Unknown" for member in members + }), + "page_numbers": sorted({ + member["page_number"] for member in members + if member.get("page_number") + }), + "sheets": sorted({ + member["sheet_number"] for member in members + if member.get("sheet_number") + }), + "assertions": members, + "kind": "agent_semantic", + "scope_id": scope.scope_id, + }) + if not clusters: + clusters = _fallback(assertions) + for cluster in clusters: + cluster["scope_id"] = scope.scope_id + cluster["kind"] = "agent_deterministic" + return AgentResult(scope_id=scope.scope_id, artifacts=clusters) + except Exception as exc: + return failure(scope, exc) + + +def build_object_graph(clusters: List[Dict]) -> Dict: + """Build a deterministic graph view from linker output.""" + nodes = [] + edges = [] + seen = set() + for cluster in clusters: + cluster_id = cluster.get("key") + nodes.append({ + "id": cluster_id, + "type": "cluster", + "location": cluster.get("location"), + "sheets": cluster.get("sheets") or [], + }) + for assertion in cluster.get("assertions") or []: + assertion_id = assertion.get("id") + if not assertion_id: + continue + if assertion_id not in seen: + seen.add(assertion_id) + nodes.append({ + "id": assertion_id, + "type": assertion.get("object_type") or "assertion", + "sheet": assertion.get("sheet_number"), + }) + edges.append({ + "source": assertion_id, + "target": cluster_id, + "relationship": "member_of", + }) + return {"nodes": nodes, "edges": edges} diff --git a/backend/agents/memory.py b/backend/agents/memory.py new file mode 100644 index 0000000..bb04e17 --- /dev/null +++ b/backend/agents/memory.py @@ -0,0 +1,67 @@ +"""Thread-safe per-job blackboard for the Agent pipeline.""" + +import copy +import json +import os +import threading +from typing import Any, Dict, Iterable, Optional + + +_COLLECTION_KEYS = {"sheets", "clusters", "findings", "decisions", "rfis"} +_MAPPING_KEYS = {"sheet_index", "jurisdiction", "object_graph"} +_MEMORY_KEYS = _COLLECTION_KEYS | _MAPPING_KEYS + + +class ProjectMemory: + """Owns intermediate Agent-mode state and optional debug snapshots.""" + + def __init__(self, artifact_dir: Optional[str] = None) -> None: + self.artifact_dir = artifact_dir + self._lock = threading.RLock() + self._data: Dict[str, Any] = { + **{key: [] for key in _COLLECTION_KEYS}, + **{key: {} for key in _MAPPING_KEYS}, + } + if artifact_dir: + os.makedirs(artifact_dir, exist_ok=True) + + def replace(self, key: str, value: Any) -> None: + """Replace one named memory section.""" + self._validate_key(key) + with self._lock: + self._data[key] = copy.deepcopy(value) + + def append(self, key: str, value: Dict[str, Any]) -> None: + """Append one artifact to a list-backed memory section.""" + if key not in _COLLECTION_KEYS: + raise KeyError(f"{key!r} is not an appendable memory section") + with self._lock: + self._data[key].append(copy.deepcopy(value)) + + def extend(self, key: str, values: Iterable[Dict[str, Any]]) -> None: + """Append several artifacts under one lock.""" + if key not in _COLLECTION_KEYS: + raise KeyError(f"{key!r} is not an appendable memory section") + with self._lock: + self._data[key].extend(copy.deepcopy(list(values))) + + def snapshot(self) -> Dict[str, Any]: + """Return a detached, JSON-serializable view of current state.""" + with self._lock: + return copy.deepcopy(self._data) + + def dump(self, filename: str = "memory.json") -> Optional[str]: + """Persist a snapshot when this job has an artifact directory.""" + if not self.artifact_dir: + return None + path = os.path.join(self.artifact_dir, filename) + temp_path = f"{path}.tmp" + with open(temp_path, "w", encoding="utf-8") as f: + json.dump(self.snapshot(), f, indent=2) + os.replace(temp_path, path) + return path + + @staticmethod + def _validate_key(key: str) -> None: + if key not in _MEMORY_KEYS: + raise KeyError(f"Unknown project memory section: {key!r}") diff --git a/backend/agents/orchestrator.py b/backend/agents/orchestrator.py new file mode 100644 index 0000000..0efeb4b --- /dev/null +++ b/backend/agents/orchestrator.py @@ -0,0 +1,78 @@ +"""Wave scheduler for the Agent pipeline.""" + +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from typing import Callable, Dict, Iterable, List, Optional + +from backend.agents.base import AgentResult, AgentScope, ScopedAgent +from backend.agents.memory import ProjectMemory + + +@dataclass +class AgentStats: + """Job-local accounting; never shared across concurrent jobs.""" + + calls: int = 0 + scopes: int = 0 + merges: int = 0 + failed_scopes: List[str] = field(default_factory=list) + + def as_dict(self) -> Dict: + return { + "calls": self.calls, + "scopes": self.scopes, + "merges": self.merges, + "failed_scopes": list(self.failed_scopes), + } + + +class Orchestrator: + """Coordinates bounded fan-out/fan-in waves against one ProjectMemory.""" + + def __init__( + self, + memory: ProjectMemory, + on_stage: Optional[Callable[[str], None]] = None, + ) -> None: + self.memory = memory + self.on_stage = on_stage + self.stats = AgentStats() + + def stage(self, name: str) -> None: + print(f"\n=== {name} ===") + if self.on_stage: + self.on_stage(name) + + def initialize(self) -> Dict: + """Initialize the job-local artifact store.""" + self.stage("Initialize agent pipeline") + self.memory.dump() + return self.stats.as_dict() + + def run_scopes( + self, + agent: ScopedAgent, + scopes: Iterable[AgentScope], + concurrency: int, + ) -> List[AgentResult]: + """Run independent scopes; one failure never aborts the wave.""" + scope_list = list(scopes) + if not scope_list: + return [] + results: List[AgentResult] = [] + with ThreadPoolExecutor(max_workers=max(1, concurrency)) as pool: + futures = {pool.submit(agent.run, scope): scope for scope in scope_list} + for future in as_completed(futures): + scope = futures[future] + self.stats.scopes += 1 + try: + result = future.result() + except Exception as exc: + result = AgentResult(scope_id=scope.scope_id, error=str(exc)) + if result.error: + self.stats.failed_scopes.append( + f"{agent.name}:{scope.scope_id}: {result.error}" + ) + results.append(result) + results.sort(key=lambda result: result.scope_id) + return results diff --git a/backend/agents/prompts.py b/backend/agents/prompts.py new file mode 100644 index 0000000..c40ab53 --- /dev/null +++ b/backend/agents/prompts.py @@ -0,0 +1,25 @@ +"""Prompts unique to the scoped Agent pipeline.""" + +COMPLETENESS_SYSTEM_PROMPT = """You are a senior construction-document completeness reviewer. +Review only the supplied sheet index and aggregate counts. Identify missing sheets, +schedules, details, or clearly incomplete coverage. Do not infer drawing facts and do +not report direct design conflicts. A missing-information finding may use the sheet +index itself as evidence. Respond only with valid JSON.""" + +COMPLETENESS_USER_PROMPT = """Review this summarized drawing set for completeness. +Return {"issues":[{"issue_id":"string","source_stage":"qaqc","category":"missing_sheet | missing_schedule | missing_detail | missing_information | bid_readiness | permit_readiness | other","severity":"critical | high | medium | low","confidence":"high | medium | low","location":"sheet or drawing set","disciplines":["string"],"sheets":["string"],"description":"string","evidence":[],"recommended_resolution":"string","code_reference":null}]}. +Sheet index: {sheet_index} +Aggregate sheet summaries: {sheet_summaries} +Cluster summary: {cluster_summary}""" + +BRAIN_SYSTEM_PROMPT = """You are the central decision layer for a construction drawing +review. Merge duplicate specialist findings, reject vague or unsupported findings, +preserve verbatim evidence, and prioritize the kept issues. Do not create new issues. +Conflicts need drawing evidence; completeness findings may instead cite an explicit +missing item from the sheet index. Return only valid JSON.""" + +BRAIN_USER_PROMPT = """Judge and consolidate these scoped specialist findings. +Return {"issues":[{"issue_id":"string","source_stage":"conflict | qaqc | code | constructability","category":"string","severity":"critical | high | medium | low","confidence":"high | medium | low","location":"string","disciplines":["string"],"sheets":["string"],"description":"string","evidence":[{"discipline":"string","sheet":"string","source_text":"string","asserted_value":"string"}],"recommended_resolution":"string","code_reference":"string or null","risk_score":1,"recommended_priority":"immediate | before_bid | before_permit | before_construction | track_only"}],"decisions":[{"finding_refs":["string"],"action":"kept | merged | dropped","reason":"string","kept_issue_id":"string or null"}]}. +Sheet index: {sheet_index} +Jurisdiction summary: {jurisdiction} +Specialist findings: {findings}""" diff --git a/backend/agents/rfi_writer.py b/backend/agents/rfi_writer.py new file mode 100644 index 0000000..93fc352 --- /dev/null +++ b/backend/agents/rfi_writer.py @@ -0,0 +1,42 @@ +"""One-finding-per-call RFI writers.""" + +from backend import config +from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure +from backend.llm import call_json +from backend.pipeline._serialize import dumps +from backend.pipeline.rfi import _valid_rfi +from backend.prompts import RFI_SYSTEM_PROMPT, RFI_USER_INSTRUCTION + + +class RFIWriterAgent: + name = "rfi_writer" + + def __init__(self, usage: AgentUsage) -> None: + self.usage = usage + + def run(self, scope: AgentScope) -> AgentResult: + try: + finding = scope.payload["finding"] + instruction = RFI_USER_INSTRUCTION.replace( + "{prioritized_issues}", dumps([finding]) + ) + parsed = call_json( + system_prompt=RFI_SYSTEM_PROMPT, + user_text=instruction, + max_tokens=config.RFI_MAX_TOKENS, + model=config.AGENT_RFI_MODEL, + usage_tracker=self.usage, + usage_stage="agent.rfi", + ) + candidates = parsed if isinstance(parsed, list) else ( + parsed.get("rfi_comments") if isinstance(parsed, dict) else [] + ) + rfis = [] + for candidate in candidates or []: + rfi = _valid_rfi(candidate) + if rfi: + rfi["issue_id"] = rfi.get("issue_id") or finding.get("issue_id") + rfis.append(rfi) + return AgentResult(scope_id=scope.scope_id, artifacts=rfis[:1]) + except Exception as exc: + return failure(scope, exc) diff --git a/backend/agents/runner.py b/backend/agents/runner.py new file mode 100644 index 0000000..f13e33a --- /dev/null +++ b/backend/agents/runner.py @@ -0,0 +1,311 @@ +"""Public entry point for the scoped Agent-mode pipeline.""" + +import json +import os +from typing import Callable, Dict, Optional + +from backend import config +from backend.agents.base import AgentScope, AgentUsage +from backend.agents.brain import BrainAgent +from backend.agents.code_agent import CodeAgent, build_code_scopes +from backend.agents.completeness import CompletenessAgent, build_sheet_summaries +from backend.agents.conflict_critic import ConflictCriticAgent +from backend.agents.construct_agent import ConstructabilityAgent, build_construct_scopes +from backend.agents.extractors import ( + JurisdictionAgent, + SheetExtractorAgent, + SheetIndexAgent, +) +from backend.agents.linker import LinkerAgent, build_link_scopes, build_object_graph +from backend.agents.memory import ProjectMemory +from backend.agents.orchestrator import Orchestrator +from backend.agents.rfi_writer import RFIWriterAgent +from backend.pipeline.pdf_processor import convert_pdf_to_images +from backend.pipeline.report import build_report, to_markdown +from backend.pipeline.sheet_index import derive_project_meta_from_cover + + +def run_agent_pipeline( + pdf_path: str, + out_dir: Optional[str] = None, + on_stage: Optional[Callable[[str], None]] = None, + project_input: Optional[Dict] = None, + source_name: Optional[str] = None, +) -> Dict: + """Run all scoped specialist waves and return a Classic-compatible report.""" + if not os.path.isfile(pdf_path): + raise FileNotFoundError(pdf_path) + + agent_dir = os.path.join(out_dir, "agent") if out_dir else None + memory = ProjectMemory(artifact_dir=agent_dir) + orchestrator = Orchestrator(memory=memory, on_stage=on_stage) + usage = AgentUsage() + orchestrator.initialize() + + orchestrator.stage("Agent ingest: PDF -> images") + pages = convert_pdf_to_images(pdf_path) + page_to_b64 = {page["page_number"]: page["base64"] for page in pages} + + orchestrator.stage("Agent wave 1: extract sheets") + extract_scopes = [ + AgentScope( + scope_id=f"sheet:{page['page_number']}", + payload={"page": page}, + ) + for page in pages + ] + extract_results = orchestrator.run_scopes( + SheetExtractorAgent(usage), extract_scopes, config.EXTRACT_CONCURRENCY + ) + sheets = [ + artifact + for result in extract_results + for artifact in result.artifacts + ] + sheets.sort(key=lambda sheet: sheet.get("page_number") or 0) + memory.replace("sheets", sheets) + memory.dump("01-extract.json") + + cover_meta = derive_project_meta_from_cover( + sheets, source_name or os.path.basename(pdf_path) + ) + merged_input = {**cover_meta, **(project_input or {})} + orchestrator.stage("Agent wave 2: sheet index and jurisdiction") + index_results = orchestrator.run_scopes( + SheetIndexAgent(usage), + [AgentScope("sheet-index", {"sheets": sheets})], + 1, + ) + jurisdiction_results = orchestrator.run_scopes( + JurisdictionAgent(usage), + [AgentScope("jurisdiction", {"project_input": merged_input})], + 1, + ) + sheet_index = ( + index_results[0].artifacts[0] + if index_results and index_results[0].artifacts else {} + ) + jurisdiction = ( + jurisdiction_results[0].artifacts[0] + if jurisdiction_results and jurisdiction_results[0].artifacts else {} + ) + memory.replace("sheet_index", sheet_index) + memory.replace("jurisdiction", jurisdiction) + memory.dump("02-orient.json") + + orchestrator.stage("Agent wave 3: scoped semantic linking") + link_results = orchestrator.run_scopes( + LinkerAgent(usage), + build_link_scopes(sheets), + config.AGENT_LINK_CONCURRENCY, + ) + clusters = [ + artifact + for result in link_results + for artifact in result.artifacts + ][:config.CLUSTER_MAX] + object_graph = build_object_graph(clusters) + memory.replace("clusters", clusters) + memory.replace("object_graph", object_graph) + memory.dump("03-link.json") + + orchestrator.stage("Agent wave 4: per-cluster conflict critics") + conflict_scopes = [ + AgentScope( + scope_id=f"conflict:{cluster.get('key')}", + payload={"cluster": cluster, "page_to_b64": page_to_b64}, + ) + for cluster in clusters + ] + conflict_results = orchestrator.run_scopes( + ConflictCriticAgent(usage), + conflict_scopes, + config.AGENT_CONFLICT_CONCURRENCY, + ) + conflict_findings = [ + artifact + for result in conflict_results + for artifact in result.artifacts + ] + memory.extend("findings", conflict_findings) + + orchestrator.stage("Agent wave 5: scoped specialists") + code_results = orchestrator.run_scopes( + CodeAgent(usage), + build_code_scopes(sheets, jurisdiction, sheet_index), + config.AGENT_SPECIALIST_CONCURRENCY, + ) + construct_results = orchestrator.run_scopes( + ConstructabilityAgent(usage), + build_construct_scopes(clusters, conflict_findings), + config.AGENT_SPECIALIST_CONCURRENCY, + ) + completeness_scope = AgentScope("completeness", { + "sheet_index": sheet_index, + "sheet_summaries": build_sheet_summaries(sheets), + "cluster_summary": { + "count": len(clusters), + "by_kind": _counts(clusters, "kind"), + }, + }) + completeness_results = orchestrator.run_scopes( + CompletenessAgent(usage), [completeness_scope], 1 + ) + specialist_findings = [ + artifact + for result in code_results + construct_results + completeness_results + for artifact in result.artifacts + ] + memory.extend("findings", specialist_findings) + gap_findings = [ + { + "issue_id": f"AGENT-GAP-{index + 1:03d}", + "source_stage": "qaqc", + "category": "analysis_gap", + "severity": "low", + "confidence": "high", + "location": failed_scope.split(":", 2)[1] if ":" in failed_scope else "", + "disciplines": [], + "sheets": [], + "description": f"Agent analysis scope did not complete: {failed_scope}", + "evidence": [], + "recommended_resolution": "Review this scope manually or rerun the job.", + "code_reference": None, + "agent": "completeness", + "scope_id": "failed-scopes", + } + for index, failed_scope in enumerate(orchestrator.stats.failed_scopes) + ] + memory.extend("findings", gap_findings) + memory.dump("05-specialists.json") + + orchestrator.stage("Agent wave 6: Brain merge, judge, prioritize") + all_findings = memory.snapshot()["findings"] + if all_findings: + prioritized, decisions = BrainAgent(usage).run( + all_findings, sheet_index, jurisdiction + ) + else: + prioritized, decisions = [], [] + memory.extend("decisions", decisions) + orchestrator.stats.merges = sum( + 1 for decision in decisions if decision.get("action") == "merged" + ) + + orchestrator.stage("Agent wave 7: per-finding RFI writers") + rfi_scopes = [ + AgentScope( + scope_id=f"rfi:{finding.get('issue_id') or index + 1}", + payload={"finding": finding}, + ) + for index, finding in enumerate(prioritized) + ] + rfi_results = orchestrator.run_scopes( + RFIWriterAgent(usage), rfi_scopes, config.AGENT_RFI_CONCURRENCY + ) + rfis = [ + artifact for result in rfi_results for artifact in result.artifacts + ] + memory.extend("rfis", rfis) + memory.dump("memory.json") + + orchestrator.stage("Build agent report") + conflicts = [_finding_as_conflict(item) for item in conflict_findings] + report = build_report( + conflicts=conflicts, + sheets=sheets, + clusters=clusters, + source=source_name or os.path.basename(pdf_path), + ) + report.update({ + "project_input": merged_input, + "jurisdiction": jurisdiction, + "sheet_index": sheet_index, + "project_intelligence": object_graph, + "validated_issues": prioritized, + "rfis": rfis, + }) + cost = usage.snapshot() + orchestrator.stats.calls = cost["calls"] + stats = orchestrator.stats.as_dict() + report["summary"].update({ + "pipeline_mode": "agent", + "agent_status": "complete", + "agent_stats": stats, + "by_stage": { + "conflicts": len(conflict_findings), + "qaqc": sum( + 1 for item in specialist_findings + if item.get("source_stage") == "qaqc" + ), + "code": sum( + 1 for item in specialist_findings + if item.get("source_stage") == "code" + ), + "constructability": sum( + 1 for item in specialist_findings + if item.get("source_stage") == "constructability" + ), + "validated": len(prioritized), + "rfis": len(rfis), + }, + "cost_usd": round(cost["usd"], 4), + "llm_calls": cost["calls"], + "cached_calls": cost["cached"], + "cost_by_stage": cost["by_stage"], + "text_backend": "openrouter", + "models_used": cost["models"], + }) + + if out_dir: + os.makedirs(out_dir, exist_ok=True) + _dump(out_dir, "assertions.json", sheets) + _dump(out_dir, "clusters.json", [_without_base64(item) for item in clusters]) + _dump(out_dir, "sheet_index.json", sheet_index) + _dump(out_dir, "jurisdiction.json", jurisdiction) + _dump(out_dir, "project_intelligence.json", object_graph) + _dump(out_dir, "validated_issues.json", prioritized) + _dump(out_dir, "rfis.json", rfis) + _dump(out_dir, "conflicts.json", report) + with open(os.path.join(out_dir, "report.md"), "w", encoding="utf-8") as f: + f.write(to_markdown(report)) + + return report + + +def _dump(out_dir: str, name: str, value) -> None: + with open(os.path.join(out_dir, name), "w", encoding="utf-8") as f: + json.dump(value, f, indent=2) + + +def _counts(items, key: str) -> Dict[str, int]: + counts: Dict[str, int] = {} + for item in items: + value = str(item.get(key) or "unknown") + counts[value] = counts.get(value, 0) + 1 + return counts + + +def _finding_as_conflict(finding: Dict) -> Dict: + return { + "category": finding.get("category") or "uncategorized", + "severity": finding.get("severity") or "medium", + "disciplines": finding.get("disciplines") or [], + "location": finding.get("location") or "", + "sheets": finding.get("sheets") or [], + "description": finding.get("description") or "", + "evidence": finding.get("evidence") or [], + "recommended_resolution": finding.get("recommended_resolution") or "", + "confidence": finding.get("confidence") or "medium", + "cluster_key": finding.get("scope_id"), + } + + +def _without_base64(cluster: Dict) -> Dict: + return { + **cluster, + "assertions": [ + {key: value for key, value in assertion.items() if key != "base64"} + for assertion in cluster.get("assertions") or [] + ], + } diff --git a/backend/config.py b/backend/config.py index 5d7009d..62a6504 100644 --- a/backend/config.py +++ b/backend/config.py @@ -22,6 +22,31 @@ MODEL = os.getenv("MODEL", "google/gemini-2.5-pro") # MODEL when unset. TEXT_MODEL = os.getenv("TEXT_MODEL", "") or MODEL +# Agent-mode model overrides (OpenRouter IDs). Empty values inherit the +# matching general-purpose model so the skeleton requires no extra config. +AGENT_EXTRACT_MODEL = os.getenv("AGENT_EXTRACT_MODEL", "") or MODEL +AGENT_INDEX_MODEL = os.getenv("AGENT_INDEX_MODEL", "") or TEXT_MODEL +AGENT_JURISDICTION_MODEL = os.getenv("AGENT_JURISDICTION_MODEL", "") or TEXT_MODEL +AGENT_LINKER_MODEL = os.getenv("AGENT_LINKER_MODEL", "") or TEXT_MODEL +AGENT_CONFLICT_MODEL = os.getenv("AGENT_CONFLICT_MODEL", "") or MODEL +AGENT_CODE_MODEL = os.getenv("AGENT_CODE_MODEL", "") or TEXT_MODEL +AGENT_CONSTRUCT_MODEL = os.getenv("AGENT_CONSTRUCT_MODEL", "") or TEXT_MODEL +AGENT_COMPLETENESS_MODEL = os.getenv("AGENT_COMPLETENESS_MODEL", "") or TEXT_MODEL +AGENT_BRAIN_MODEL = os.getenv("AGENT_BRAIN_MODEL", "") or TEXT_MODEL +AGENT_RFI_MODEL = os.getenv("AGENT_RFI_MODEL", "") or TEXT_MODEL + +# Agent-mode hard scope limits. These are intentionally independent of Classic +# batching so Agent workers can never grow into whole-set reasoning calls. +AGENT_LINK_MAX_ASSERTIONS = int(os.getenv("AGENT_LINK_MAX_ASSERTIONS", "60")) +AGENT_CLUSTER_MAX_ASSERTIONS = int(os.getenv("AGENT_CLUSTER_MAX_ASSERTIONS", "24")) +AGENT_CONFLICT_MAX_IMAGES = int(os.getenv("AGENT_CONFLICT_MAX_IMAGES", "6")) +AGENT_CODE_BATCH_SIZE = int(os.getenv("AGENT_CODE_BATCH_SIZE", "60")) +AGENT_BRAIN_MAX_TOKENS = int(os.getenv("AGENT_BRAIN_MAX_TOKENS", "16384")) +AGENT_LINK_CONCURRENCY = int(os.getenv("AGENT_LINK_CONCURRENCY", "4")) +AGENT_CONFLICT_CONCURRENCY = int(os.getenv("AGENT_CONFLICT_CONCURRENCY", "4")) +AGENT_SPECIALIST_CONCURRENCY = int(os.getenv("AGENT_SPECIALIST_CONCURRENCY", "4")) +AGENT_RFI_CONCURRENCY = int(os.getenv("AGENT_RFI_CONCURRENCY", "4")) + # -- Hybrid (local text LLM) ---------------------------------------- # Optional OpenAI-compatible local endpoint (e.g. a vLLM box) for the text-only # QAQC stages. Vision stages ALWAYS use OpenRouter. The user picks hybrid per diff --git a/backend/jobs.py b/backend/jobs.py index b02fbe0..0d548ee 100644 --- a/backend/jobs.py +++ b/backend/jobs.py @@ -19,11 +19,13 @@ import threading from typing import Dict, Optional from backend import config +from backend.agents.runner import run_agent_pipeline from backend.pipeline.runner import run_pipeline from backend.email_sender import send_conflict_report _jobs: Dict[str, Dict] = {} _lock = threading.Lock() +PIPELINE_MODES = {"classic", "agent"} def _set(job_id: str, **fields) -> None: @@ -32,8 +34,14 @@ def _set(job_id: str, **fields) -> None: def create_job(pdf_path: str, source_filename: str, email: Optional[str] = None, - project_input: Optional[Dict] = None, text_local: bool = False) -> str: + project_input: Optional[Dict] = None, text_local: bool = False, + pipeline_mode: str = "classic") -> str: """Register a job and kick off its background thread. Returns the job_id.""" + pipeline_mode = pipeline_mode.strip().lower() + if pipeline_mode not in PIPELINE_MODES: + raise ValueError(f"Unsupported pipeline mode: {pipeline_mode!r}") + # Agent mode v1 is OpenRouter-only. + text_local = bool(text_local and pipeline_mode == "classic") job_id = uuid.uuid4().hex[:12] with _lock: _jobs[job_id] = { @@ -43,33 +51,39 @@ def create_job(pdf_path: str, source_filename: str, email: Optional[str] = None, "email": email or None, "project_input": project_input or {}, "text_local": text_local, + "pipeline_mode": pipeline_mode, "stage": None, "created_at": time.time(), "finished_at": None, "report": None, "error": None, } - threading.Thread(target=_run, args=(job_id, pdf_path, project_input, text_local), + threading.Thread(target=_run, args=( + job_id, pdf_path, project_input, text_local, pipeline_mode, + ), daemon=True).start() return job_id def _run(job_id: str, pdf_path: str, project_input: Optional[Dict] = None, - text_local: bool = False) -> None: + text_local: bool = False, pipeline_mode: str = "classic") -> None: out_dir = os.path.join(config.OUTPUT_DIR, job_id) try: _set(job_id, status="running") # Keep a copy of the source PDF so its sheets can be viewed later. os.makedirs(out_dir, exist_ok=True) shutil.copy2(pdf_path, os.path.join(out_dir, "source.pdf")) - report = run_pipeline( - pdf_path, - out_dir=out_dir, - on_stage=lambda name: _set(job_id, stage=name), - project_input=project_input, - source_name=_jobs[job_id].get("source"), - text_local=text_local, - ) + runner = run_agent_pipeline if pipeline_mode == "agent" else run_pipeline + runner_kwargs = { + "out_dir": out_dir, + "on_stage": lambda name: _set(job_id, stage=name), + "project_input": project_input, + "source_name": _jobs[job_id].get("source"), + } + if pipeline_mode == "classic": + runner_kwargs["text_local"] = text_local + report = runner(pdf_path, **runner_kwargs) + report.setdefault("summary", {})["pipeline_mode"] = pipeline_mode _set(job_id, status="done", report=report, finished_at=time.time(), stage=None) _notify(job_id, report, out_dir) except Exception as e: @@ -155,6 +169,7 @@ def get_job(job_id: str) -> Optional[Dict]: "email": None, "project_input": report.get("project_input", {}), "text_local": report.get("summary", {}).get("text_backend") == "local", + "pipeline_mode": report.get("summary", {}).get("pipeline_mode", "classic"), "stage": None, "created_at": os.path.getmtime(source_pdf) if os.path.isfile(source_pdf) else None, "finished_at": os.path.getmtime(report_path), diff --git a/backend/llm.py b/backend/llm.py index 83ff9b3..cf53cad 100644 --- a/backend/llm.py +++ b/backend/llm.py @@ -237,18 +237,17 @@ def _repair_truncated(raw: str) -> Optional[Dict[str, Any]]: return None -def _record_cost(response) -> None: +def _response_cost(response) -> Optional[float]: """Pull OpenRouter's per-call USD cost out of the usage object, if present.""" try: dump = response.model_dump() except Exception: - return + return None usage = dump.get("usage") or {} cost = usage.get("cost") if cost is None: cost = (usage.get("cost_details") or {}).get("upstream_inference_cost") - if isinstance(cost, (int, float)): - _add_cost(float(cost)) + return float(cost) if isinstance(cost, (int, float)) else None def _parse(raw: str) -> Optional[Dict[str, Any]]: @@ -268,6 +267,8 @@ def call_json( images_b64: Optional[List[str]] = None, max_tokens: int = 4096, model: Optional[str] = None, + usage_tracker: Optional[Any] = None, + usage_stage: str = "?", ) -> Optional[Dict[str, Any]]: """ Send one chat completion expecting a JSON object back. @@ -287,6 +288,10 @@ def call_json( hit = _cache_get(cache_key) if hit is not None: _add_cached() + if usage_tracker: + usage_tracker.record( + usage_stage, be["model"], cached=True, has_images=has_images + ) return hit content: List[Dict[str, Any]] = [] @@ -314,7 +319,13 @@ def call_json( if use_json_mode: kwargs["response_format"] = {"type": "json_object"} response = client.chat.completions.create(**kwargs) - _record_cost(response) + usd = _response_cost(response) + if usd is not None: + _add_cost(usd) + if usage_tracker: + usage_tracker.record( + usage_stage, be["model"], usd=usd or 0.0, has_images=has_images + ) raw = _strip_fences(response.choices[0].message.content or "") parsed = _parse(raw) if parsed is not None: diff --git a/backend/main.py b/backend/main.py index 3c52e11..e25e1c9 100644 --- a/backend/main.py +++ b/backend/main.py @@ -18,7 +18,7 @@ from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from backend import config -from backend.jobs import create_job, get_job +from backend.jobs import PIPELINE_MODES, create_job, get_job from backend.pipeline.pdf_processor import render_page_jpeg app = FastAPI(title=config.APP_TITLE, version=config.APP_VERSION) @@ -42,6 +42,7 @@ async def check( occupancy: Optional[str] = Form(None), work_type: Optional[str] = Form(None), text_local: bool = Form(False), + pipeline_mode: str = Form("classic"), ): """ Accept a PDF, start a background conflict check, and return a job_id @@ -53,6 +54,12 @@ async def check( """ if not file.filename.lower().endswith(".pdf"): raise HTTPException(status_code=400, detail="Please upload a PDF.") + pipeline_mode = pipeline_mode.strip().lower() + if pipeline_mode not in PIPELINE_MODES: + raise HTTPException( + status_code=400, + detail=f"pipeline_mode must be one of: {', '.join(sorted(PIPELINE_MODES))}", + ) os.makedirs(config.UPLOAD_DIR, exist_ok=True) suffix = "_" + os.path.basename(file.filename) @@ -70,8 +77,14 @@ async def check( if v and v.strip() } job_id = create_job(tmp_path, source_filename=file.filename, email=email, - project_input=project_input, text_local=text_local) - return JSONResponse({"job_id": job_id, "status": "queued", "email": email}) + project_input=project_input, text_local=text_local, + pipeline_mode=pipeline_mode) + return JSONResponse({ + "job_id": job_id, + "status": "queued", + "email": email, + "pipeline_mode": pipeline_mode, + }) @app.get("/jobs/{job_id}") diff --git a/backend/pipeline/runner.py b/backend/pipeline/runner.py index ae8035e..fdae24f 100644 --- a/backend/pipeline/runner.py +++ b/backend/pipeline/runner.py @@ -147,6 +147,7 @@ def run_pipeline( report["summary"]["cost_by_stage"] = cost.get("by_stage", {}) report["summary"]["text_backend"] = "local" if text_local else "openrouter" report["summary"]["models_used"] = cost.get("models", {}) + report["summary"]["pipeline_mode"] = "classic" print(f"[Runner] LLM cost: ${cost['usd']:.4f} over {cost['calls']} live calls" f" ({cost.get('cached', 0)} cached)") diff --git a/cli/run_check.py b/cli/run_check.py index 8bd7644..6a86239 100644 --- a/cli/run_check.py +++ b/cli/run_check.py @@ -17,6 +17,7 @@ _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) +from backend.agents.runner import run_agent_pipeline # noqa: E402 from backend.pipeline.runner import run_pipeline # noqa: E402 @@ -25,6 +26,8 @@ def main() -> int: parser.add_argument("pdf", help="Path to the PDF drawing set") parser.add_argument("--out", default=None, help="Directory for artifacts (default: out/)") + parser.add_argument("--mode", choices=("classic", "agent"), default="classic", + help="Pipeline implementation to run (default: classic)") parser.add_argument("--project-name", default=None) parser.add_argument("--address", default=None) parser.add_argument("--occupancy", default=None) @@ -43,7 +46,13 @@ def main() -> int: }.items() if v } out_dir = args.out or os.path.join("out", os.path.splitext(os.path.basename(args.pdf))[0]) - report = run_pipeline(args.pdf, out_dir=out_dir, project_input=project_input or None) + runner = run_agent_pipeline if args.mode == "agent" else run_pipeline + report = runner( + args.pdf, + out_dir=out_dir, + project_input=project_input or None, + source_name=os.path.basename(args.pdf), + ) s = report["summary"] print("\n" + "=" * 60) diff --git a/frontend/index.html b/frontend/index.html index 87249bf..c499570 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -97,6 +97,15 @@ +