Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 136 additions & 10 deletions backend/app/agents/norma.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,81 @@
from app.core.config import settings
from app.core.llm import get_language_name

QUESTION_LABELS = {
"q1": "Does the system infer from inputs to generate outputs?",
"q2": "Does the system operate with some level of autonomy?",
"q3": "Does the system perform any prohibited practices?",
"q4": "Is the system covered by EU product safety legislation?",
"q5": "Does the product require third-party conformity assessment?",
"q6": "Primary domain of the system's intended use",
"q7": "Does the system evaluate individual natural persons?",
"q8": "Does the system perform profiling of personal data?",
"q9": "Does the system exclusively perform narrow/procedural tasks?",
"q10": "Can persons under 18 access the system or be affected by its outputs?",
"q11": "Does the training data include special categories of personal data?",
"q12": "What level of human oversight does the system support?",
"q13": "Does the system involve remote biometric identification?",
"q14": "Is the system deployed by or on behalf of a public authority?",
"q15": "Does the system interact directly with people?",
"q16": "Does the system generate synthetic content?",
"q17": "Can the system be used to create deepfakes?",
}

AREA_LABELS = {
"cybersecurity": "Cybersecurity",
"documentation": "Documentation",
"qms": "Quality Management System",
"incidents": "Incidents",
"risk-management": "Risk Management",
"data-governance": "Data Governance",
"accuracy": "Accuracy",
"records": "Records",
"robustness": "Robustness",
"oversight": "Oversight",
"transparency": "Transparency",
"surveillance": "Surveillance",
"undp-zero-question": "UNDP Zero Question",
"undp-org-readiness": "UNDP Organisational Readiness",
"undp-planning": "UNDP Planning",
"undp-rights-mapping": "UNDP Rights Mapping",
"undp-data-diligence": "UNDP Data Diligence",
"undp-risk-analysis": "UNDP Risk Analysis",
"undp-risk-management": "UNDP Risk Management",
"undp-monitoring": "UNDP Monitoring",
"undp-framework-alignment": "UNDP Framework Alignment",
"env-energy-carbon": "Energy & Carbon",
"env-hardware": "Hardware",
"env-data-management": "Data Management",
"env-model-efficiency": "Model Efficiency",
"env-lifecycle": "Lifecycle",
"env-monitoring": "Environmental Monitoring",
}


def _downshift_headings(text: str) -> str:
import re

text = re.sub(r"^-{3,}\s*$", "", text, flags=re.MULTILINE)
return re.sub(r"^(#{1,5}) ", lambda m: "#" + m.group(1) + " ", text, flags=re.MULTILINE)


def _format_evidence_key(item_key: str) -> str:
import re

from app.data.evidence_questions import EVIDENCE_QUESTIONS

match = re.match(r"^(.+)-([A-Z]+-\d+)-(\d+)$", item_key)
if not match:
return item_key
area_id, item_code, q_idx = match.groups()
area_name = AREA_LABELS.get(area_id, area_id.replace("-", " ").title())
question = EVIDENCE_QUESTIONS.get(item_key, "")
label = f"{area_name} — {item_code} (Q{int(q_idx) + 1})"
if question:
label += f": {question}"
return label


SYSTEM_INSTRUCTION_TEMPLATE = """\
You are Norma, an AI compliance assistant. You help users understand and comply with \
regulatory frameworks, human rights standards, and environmental requirements for AI systems.
Expand Down Expand Up @@ -47,12 +122,45 @@ def _language_rule(language: str) -> str:
)


def _group_evidence_by_framework(
evidence: list[dict],
) -> list[tuple[str, int, list[dict]]]:
from app.data.evidence_questions import EVIDENCE_ORDER, EVIDENCE_QUESTIONS

totals: dict[str, int] = {"EU AI Act": 0, "UNDP Human Rights Due Diligence": 0, "Environmental Impact": 0}
for key in EVIDENCE_QUESTIONS:
if key.startswith("undp-"):
totals["UNDP Human Rights Due Diligence"] += 1
elif key.startswith("env-"):
totals["Environmental Impact"] += 1
else:
totals["EU AI Act"] += 1

groups: dict[str, list[dict]] = {}
for ev in evidence:
key = ev["item_key"]
if key.startswith("undp-"):
framework = "UNDP Human Rights Due Diligence"
elif key.startswith("env-"):
framework = "Environmental Impact"
else:
framework = "EU AI Act"
groups.setdefault(framework, []).append(ev)

for items in groups.values():
items.sort(key=lambda ev: EVIDENCE_ORDER.get(ev["item_key"], 999999))

order = ["EU AI Act", "UNDP Human Rights Due Diligence", "Environmental Impact"]
return [(title, totals[title], groups.get(title, [])) for title in order]


def build_system_prompt(
*,
framework_contents: list[dict],
project_context: dict | None = None,
document_summaries: list[dict] | None = None,
github_summary: str | None = None,
github_architecture: str | None = None,
language: str = "en",
) -> str:
parts: list[str] = []
Expand All @@ -79,21 +187,39 @@ def build_system_prompt(
if project_context.get("questionnaire_answers"):
parts.append("\n**Questionnaire Answers:**\n")
for key, val in project_context["questionnaire_answers"].items():
parts.append(f"- {key}: {val}\n")
label = QUESTION_LABELS.get(key, key)
if isinstance(val, list):
val = ", ".join(val)
parts.append(f"- **{label}** — {val}\n")

if document_summaries:
parts.append("\n## Uploaded Document Summaries\n")
parts.append("\n## Uploaded Document Context\n")
for doc in document_summaries:
parts.append(f"### {doc['name']}\n{doc['summary']}\n")

if github_summary:
parts.append("\n## GitHub Repository & Task Context\n")
parts.append(github_summary)
parts.append(f"### {doc['name']}\n{_downshift_headings(doc['summary'])}\n")

if github_summary or github_architecture:
parts.append("\n## Codebase & Task Context\n")
if github_summary:
parts.append(_downshift_headings(github_summary))
if github_architecture:
import re

summary_text = github_summary or ""
matches = re.findall(r"^(#{1,5})\s+(\d+)\.", summary_text, re.MULTILINE)
next_num = max((int(n) for _, n in matches), default=0) + 1
prefix = "#" + (matches[0][0] if matches else "##")
parts.append(f"\n{prefix} {next_num}. Architecture\n")
parts.append(github_architecture)

if project_context and project_context.get("reporting_evidence"):
parts.append("\n## Reporting Evidence\n")
for ev in project_context["reporting_evidence"]:
parts.append(f"- **{ev['item_key']}:** {ev['comment']}\n")
parts.append("\n## Reporting Evidence Context\n")
grouped = _group_evidence_by_framework(project_context["reporting_evidence"])
for framework_title, total, items in grouped:
filled = [ev for ev in items if ev.get("comment", "").strip()]
parts.append(f"\n### {framework_title} ({len(filled)} out of {total} questions answered)\n")
for ev in filled:
label = _format_evidence_key(ev["item_key"])
parts.append(f"- **{label}:** {ev['comment']}\n")

context = "\n".join(parts) if parts else "No project context available."
return SYSTEM_INSTRUCTION_TEMPLATE.format(
Expand Down
127 changes: 126 additions & 1 deletion backend/app/api/routes/chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,14 @@
from google.genai.types import Content, Part
from sqlalchemy.orm import Session

from app.agents.norma import build_system_prompt, create_norma_agent
from app.agents.norma import (
QUESTION_LABELS,
_downshift_headings,
_format_evidence_key,
_group_evidence_by_framework,
build_system_prompt,
create_norma_agent,
)
from app.api.dependencies import get_current_user
from app.core.database import SessionLocal, get_db
from app.models.chat import ChatMessage, ChatSession
Expand Down Expand Up @@ -100,16 +107,134 @@ def _assemble_context(project: Project, db: Session, language: str = "en") -> st

integration = db.query(Integration).filter(Integration.project_id == project.id).first()
github_summary = integration.summary if integration else None
github_architecture = integration.architecture_mermaid if integration else None

return build_system_prompt(
framework_contents=framework_contents,
project_context=project_context,
document_summaries=document_summaries if document_summaries else None,
github_summary=github_summary,
github_architecture=github_architecture,
language=language,
)


def _build_section(project: Project, db: Session, section: str) -> str | None:
if section == "overview":
lines = [
"## Current Project Context\n",
f"**Project:** {project.name}\n",
]
if project.description:
lines.append(f"**Description:** {project.description}\n")
if project.intended_purpose:
lines.append(f"**Intended Purpose:** {project.intended_purpose}\n")
if project.intended_users:
lines.append(f"**Intended Users:** {project.intended_users}\n")
if project.deployment_context:
lines.append(f"**Deployment Context:** {project.deployment_context}\n")
lines.append("\n### EU AI Act Risk Classification\n")
lines.append(f"**Risk Classification:** {project.risk_classification or 'Not set'}\n")
if project.questionnaire_answers:
lines.append("**Questionnaire Answers:**\n")
for key, val in project.questionnaire_answers.items():
label = QUESTION_LABELS.get(key, key)
if isinstance(val, list):
val = ", ".join(val)
lines.append(f"- **{label}** — {val}\n")
return "\n".join(lines)

if section == "frameworks":
frameworks = db.query(Framework).all()
if not frameworks:
return None
lines = ["## Compliance Frameworks\n"]
for fw in frameworks:
lines.append(f"### {fw.name}\n{fw.description}\n")
if fw.content:
lines.append(f"\n{fw.content}\n")
return "\n".join(lines)

if section == "documents":
frameworks = db.query(Framework).all()
all_docs = db.query(Document).filter(Document.project_id == project.id).all()
custom_docs = db.query(CustomDocument).filter(CustomDocument.project_id == project.id).all()
lines = ["## Uploaded Document Context\n"]
for fw in frameworks:
fw_docs = [d for d in all_docs if d.definition.framework_id == fw.id]
summarised = [d for d in fw_docs if d.summary]
lines.append(f"\n### {fw.name} ({len(summarised)} out of {len(fw_docs)} documents uploaded)\n")
if summarised:
for doc in summarised:
lines.append(f"- **{doc.definition.name}:** {_downshift_headings(doc.summary)}\n")
else:
lines.append("No documents uploaded for this framework.\n")
custom_summarised = [c for c in custom_docs if c.summary]
lines.append(
f"\n### Additional Documents ({len(custom_summarised)} out of {len(custom_docs)} documents uploaded)\n"
)
if custom_summarised:
for cdoc in custom_summarised:
lines.append(f"- **{cdoc.file_name}:** {_downshift_headings(cdoc.summary)}\n")
else:
lines.append("No additional documents uploaded.\n")
return "\n".join(lines)

if section == "reporting":
evidence = db.query(ReportingEvidence).filter(ReportingEvidence.project_id == project.id).all()
evidence_dicts = [{"item_key": ev.item_key, "comment": ev.comment} for ev in evidence]
lines = ["## Reporting Evidence Context\n"]
grouped = _group_evidence_by_framework(evidence_dicts)
for framework_title, total, items in grouped:
filled = [ev for ev in items if ev.get("comment", "").strip()]
lines.append(f"\n### {framework_title} ({len(filled)} out of {total} questions answered)\n")
for ev in filled:
label = _format_evidence_key(ev["item_key"])
lines.append(f"- **{label}:** {ev['comment']}\n")
return "\n".join(lines)

if section == "github":
integration = db.query(Integration).filter(Integration.project_id == project.id).first()
if not integration:
return None
if not integration.summary and not integration.architecture_mermaid:
return None
parts = ["## Codebase & Task Context\n"]
if integration.summary:
parts.append(_downshift_headings(integration.summary))
if integration.architecture_mermaid:
import re

summary_text = integration.summary or ""
matches = re.findall(r"^(#{1,5})\s+(\d+)\.", summary_text, re.MULTILINE)
next_num = max((int(n) for _, n in matches), default=0) + 1
prefix = "#" + (matches[0][0] if matches else "##")
parts.append(f"\n{prefix} {next_num}. Architecture\n")
parts.append(integration.architecture_mermaid)
return "\n".join(parts)

return None


@router.get("/context")
def get_context(
project_id: uuid.UUID,
section: str = "full",
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
project = db.query(Project).filter(Project.id == project_id, Project.owner_id == current_user.id).first()
if not project:
raise HTTPException(status_code=404, detail="Project not found")

if section == "full":
context = _assemble_context(project, db, language=current_user.language_preference or "en")
else:
context = _build_section(project, db, section)

return {"context": context}


@router.post("/sessions", response_model=ChatSessionResponse, status_code=201)
def create_session(
body: ChatSessionCreate,
Expand Down
Loading
Loading