72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""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 []),
|
|
"disputes": dumps(cluster.get("disputed_attributes") 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)
|