{
"$type": "site.standard.document",
"bskyPostRef": {
"cid": "bafyreiapbzgfjqb4qpvtqzslsymu3mryqwbmci5t2gklzwgm6tejovj22y",
"uri": "at://did:plc:25rdn5elo5izoxrmtis34zuk/app.bsky.feed.post/3mpu2sciocxc2"
},
"coverImage": {
"$type": "blob",
"ref": {
"$link": "bafkreieifqvfgw4dll2fai7wgh6o7pfqos5lzyv4xd44a65hd3n44v5ldq"
},
"mimeType": "image/webp",
"size": 71112
},
"path": "/hiroki-kameyama/ai-governance-eu-ai-act-compliance-risk-assessment-and-audit-logging-big",
"publishedAt": "2026-07-04T21:55:25.000Z",
"site": "https://dev.to",
"tags": [
"ai",
"mlops",
"llm",
"python",
"Chapter 7 (Multi-Agent)",
"@dataclass"
],
"textContent": "## Introduction\n\nThrough Chapter 7 (Multi-Agent), we have a complete, functioning AI system. The final step is building _organizational_ infrastructure to operate AI safely over time.\n\n\n\n [Before] Technical safety\n Security → Block malicious input\n Evals → Measure quality\n\n [Now] Organizational / regulatory safety\n Governance → Know what AI systems are in use\n Risk mgmt → Classify and assess risks\n Audit logs → Record who did what, when\n EU AI Act → Regulatory compliance\n\n\n## EU AI Act Status (as of June 2026)\n\nThe EU AI Act came into force on August 1, 2024, with full enforcement on **August 2, 2026**. Transparency rules (disclosing when users are interacting with AI, labeling AI-generated content) also take effect on that date.\n\nAI systems are classified into three risk tiers:\n\nRisk Level | Description | Examples\n---|---|---\n**Prohibited** | Not permitted | Social scoring, manipulative AI\n**High Risk** | Strict regulation | Hiring, credit scoring, law enforcement\n**Limited Risk** | Transparency obligations | Chatbots, AI-generated content\n**Minimal Risk** | No regulation | Spam filters, game AI\n\n**Our RAG system's classification:** Limited Risk (chatbot). We are required to disclose to users that they are interacting with AI.\n\n## Directory Structure\n\n\n pgvector-tutorial/\n ├── existing files\n └── governance/\n ├── ai_registry.py # ★ AI system inventory\n ├── risk_assessor.py # ★ Risk assessment\n ├── audit_logger.py # ★ Audit logging\n └── compliant_rag.py # ★ Governance-compliant RAG\n\n\n## 1. AI System Inventory — `governance/ai_registry.py`\n\nMost organizations lack a systematic inventory of their AI systems, making risk classification and compliance planning difficult. Knowing what you have is the essential first step.\n\n\n\n # governance/ai_registry.py\n \"\"\"\n AI system inventory\n\n Centrally manage all AI systems in use across the organization.\n Forms the foundation for technical documentation required by EU AI Act Annex IV.\n \"\"\"\n from datetime import datetime\n from dataclasses import dataclass, asdict\n from enum import Enum\n\n\n class RiskLevel(Enum):\n UNACCEPTABLE = \"prohibited\"\n HIGH = \"high_risk\"\n LIMITED = \"limited_risk\"\n MINIMAL = \"minimal_risk\"\n\n\n class SystemStatus(Enum):\n ACTIVE = \"active\"\n TESTING = \"testing\"\n DEPRECATED = \"deprecated\"\n\n\n @dataclass\n class AISystem:\n system_id: str\n name: str\n description: str\n purpose: str\n risk_level: str\n status: str\n model: str\n data_sources: list[str]\n owner: str\n human_oversight: bool\n eu_ai_act_category: str\n registered_at: str\n last_reviewed: str\n\n\n REGISTRY = {\n \"rag-search-001\": AISystem(\n system_id=\"rag-search-001\",\n name=\"pgvector RAG Search System\",\n description=\"Document search and answer generation using pgvector and Gemini Embedding\",\n purpose=\"Technical document search and Q&A for engineers. Internal use only.\",\n risk_level=RiskLevel.LIMITED.value,\n status=SystemStatus.ACTIVE.value,\n model=\"gemini-2.5-flash + gemini-embedding-001\",\n data_sources=[\"pgvector (internal document DB)\"],\n owner=\"Hiroki Kameyama\",\n human_oversight=True,\n eu_ai_act_category=\"Limited Risk - Chatbot (Article 50 transparency obligations apply)\",\n registered_at=\"2026-06-01\",\n last_reviewed=datetime.now().strftime(\"%Y-%m-%d\"),\n ),\n \"multiagent-001\": AISystem(\n system_id=\"multiagent-001\",\n name=\"Multi-Agent Search System\",\n description=\"Collaborative system with Orchestrator + Search Worker + Quality Check Worker\",\n purpose=\"High-quality answer generation for complex technical questions. Internal use only.\",\n risk_level=RiskLevel.LIMITED.value,\n status=SystemStatus.TESTING.value,\n model=\"gemini-2.5-flash (multiple Agents)\",\n data_sources=[\"pgvector (internal document DB)\"],\n owner=\"Hiroki Kameyama\",\n human_oversight=True,\n eu_ai_act_category=\"Limited Risk - Chatbot (Article 50 transparency obligations apply)\",\n registered_at=\"2026-06-25\",\n last_reviewed=datetime.now().strftime(\"%Y-%m-%d\"),\n ),\n }\n\n\n def get_system(system_id: str) -> AISystem | None:\n return REGISTRY.get(system_id)\n\n\n def list_systems(risk_level: str = None, status: str = None) -> list[AISystem]:\n systems = list(REGISTRY.values())\n if risk_level:\n systems = [s for s in systems if s.risk_level == risk_level]\n if status:\n systems = [s for s in systems if s.status == status]\n return systems\n\n\n def generate_inventory_report() -> dict:\n \"\"\"Generate an inventory report (EU AI Act compliant).\"\"\"\n systems = list(REGISTRY.values())\n return {\n \"generated_at\": datetime.now().isoformat(),\n \"total_systems\": len(systems),\n \"by_risk_level\": {\n level.value: sum(1 for s in systems if s.risk_level == level.value)\n for level in RiskLevel\n },\n \"by_status\": {\n status.value: sum(1 for s in systems if s.status == status.value)\n for status in SystemStatus\n },\n \"systems\": [asdict(s) for s in systems],\n }\n\n\n if __name__ == \"__main__\":\n print(\"=== AI System Inventory ===\\n\")\n for system in list_systems():\n print(f\"[{system.system_id}] {system.name}\")\n print(f\" Purpose: {system.purpose}\")\n print(f\" Risk: {system.risk_level}\")\n print(f\" Status: {system.status}\")\n print(f\" EU AI Act: {system.eu_ai_act_category}\")\n print(f\" Human oversight: {'Yes' if system.human_oversight else 'No'}\")\n print()\n\n report = generate_inventory_report()\n print(f\"Total systems: {report['total_systems']}\")\n print(f\"By risk level: {report['by_risk_level']}\")\n\n\n## 2. Risk Assessment — `governance/risk_assessor.py`\n\n\n # governance/risk_assessor.py\n \"\"\"\n Risk assessment module\n\n Quantitatively assess AI system risk.\n Supports the EU AI Act risk-based approach.\n \"\"\"\n from dataclasses import dataclass\n from datetime import datetime, timedelta\n\n\n @dataclass\n class RiskAssessment:\n system_id: str\n assessed_at: str\n scores: dict[str, float]\n overall_risk: float\n risk_level: str # low / medium / high / critical\n mitigations: list[str]\n next_review: str\n\n\n RISK_CRITERIA = {\n \"data_privacy\": {\n \"name\": \"Data Privacy\",\n \"description\": \"Does it handle personal or sensitive data?\",\n \"weight\": 0.25,\n },\n \"decision_impact\": {\n \"name\": \"Decision Impact\",\n \"description\": \"Does it influence important human decisions?\",\n \"weight\": 0.25,\n },\n \"autonomy\": {\n \"name\": \"Autonomy\",\n \"description\": \"Does it operate autonomously without human intervention?\",\n \"weight\": 0.20,\n },\n \"bias_risk\": {\n \"name\": \"Bias Risk\",\n \"description\": \"Is there risk of discrimination or bias?\",\n \"weight\": 0.15,\n },\n \"explainability\": {\n \"name\": \"Explainability\",\n \"description\": \"Can it explain why it gave an answer? (lower = higher risk)\",\n \"weight\": 0.15,\n },\n }\n\n\n def assess_risk(\n system_id: str,\n data_privacy: float = 0.1,\n decision_impact: float = 0.2,\n autonomy: float = 0.3,\n bias_risk: float = 0.1,\n explainability: float = 0.7, # Higher = safer\n ) -> RiskAssessment:\n \"\"\"\n Run a risk assessment.\n\n Args:\n system_id: ID of the system to assess\n data_privacy: Data privacy risk (0.0–1.0)\n decision_impact: Decision influence level (0.0–1.0)\n autonomy: Autonomy level (0.0–1.0)\n bias_risk: Bias risk (0.0–1.0)\n explainability: Explainability (higher = safer, inverted in scoring)\n \"\"\"\n scores = {\n \"data_privacy\": data_privacy,\n \"decision_impact\": decision_impact,\n \"autonomy\": autonomy,\n \"bias_risk\": bias_risk,\n \"explainability\": 1.0 - explainability, # Invert: high explainability = safer\n }\n\n overall_risk = sum(\n scores[key] * RISK_CRITERIA[key][\"weight\"]\n for key in scores\n )\n\n if overall_risk < 0.2:\n risk_level = \"low\"\n elif overall_risk < 0.4:\n risk_level = \"medium\"\n elif overall_risk < 0.7:\n risk_level = \"high\"\n else:\n risk_level = \"critical\"\n\n mitigations = []\n if scores[\"data_privacy\"] > 0.5:\n mitigations.append(\"Implement PII masking and anonymization\")\n if scores[\"decision_impact\"] > 0.5:\n mitigations.append(\"Require human review for all important decisions\")\n if scores[\"autonomy\"] > 0.5:\n mitigations.append(\"Limit autonomous execution scope, add human approval steps\")\n if scores[\"bias_risk\"] > 0.3:\n mitigations.append(\"Conduct bias audit and ensure diversity in training data\")\n if scores[\"explainability\"] > 0.5:\n mitigations.append(\"Always present the source documents (reference) for answers\")\n if not mitigations:\n mitigations.append(\"Current risk level is within acceptable range. Continue regular reviews.\")\n\n days = {\"low\": 180, \"medium\": 90, \"high\": 30, \"critical\": 7}\n next_review = (datetime.now() + timedelta(days=days[risk_level])).strftime(\"%Y-%m-%d\")\n\n return RiskAssessment(\n system_id=system_id,\n assessed_at=datetime.now().isoformat(),\n scores=scores,\n overall_risk=round(overall_risk, 3),\n risk_level=risk_level,\n mitigations=mitigations,\n next_review=next_review,\n )\n\n\n if __name__ == \"__main__\":\n print(\"=== RAG System Risk Assessment ===\\n\")\n\n assessment = assess_risk(\n system_id=\"rag-search-001\",\n data_privacy=0.1, # No personal data handled\n decision_impact=0.2, # Provides reference info only — humans make final decisions\n autonomy=0.3, # Partially autonomous but human-reviewed\n bias_risk=0.1, # Technical docs only — low bias risk\n explainability=0.8, # Can show reference documents\n )\n\n print(f\"System ID: {assessment.system_id}\")\n print(f\"Overall risk score: {assessment.overall_risk}\")\n print(f\"Risk level: {assessment.risk_level.upper()}\")\n print(f\"\\nPer-dimension scores:\")\n for key, score in assessment.scores.items():\n name = RISK_CRITERIA[key][\"name\"]\n print(f\" {name}: {score:.2f}\")\n print(f\"\\nRecommended mitigations:\")\n for mitigation in assessment.mitigations:\n print(f\" - {mitigation}\")\n print(f\"\\nNext review: {assessment.next_review}\")\n\n\n## 3. Audit Logging — `governance/audit_logger.py`\n\nHigh-risk AI system providers are required to design systems that automatically log relevant events across the system lifecycle.\n\n\n\n # governance/audit_logger.py\n \"\"\"\n Audit logging module\n\n Record all operations of AI systems.\n Complies with EU AI Act Article 12 (record keeping).\n \"\"\"\n import json\n import time\n from datetime import datetime\n from pathlib import Path\n from dataclasses import dataclass, asdict\n from enum import Enum\n\n\n class EventType(Enum):\n QUERY = \"query\"\n SEARCH = \"search\"\n GENERATION = \"generation\"\n SECURITY_BLOCK = \"security_block\"\n QUALITY_CHECK = \"quality_check\"\n ERROR = \"error\"\n HUMAN_REVIEW = \"human_review\"\n\n\n @dataclass\n class AuditEvent:\n event_id: str\n timestamp: str\n event_type: str\n system_id: str\n user_id: str\n input_summary: str # Summary of input (no PII)\n output_summary: str # Summary of output\n metadata: dict\n duration_ms: float\n\n\n class AuditLogger:\n \"\"\"\n Audit logger class.\n Records all AI operations to a JSONL file.\n \"\"\"\n\n def __init__(self, log_file: str = \"governance/audit_log.jsonl\"):\n self.log_file = Path(log_file)\n self.log_file.parent.mkdir(exist_ok=True)\n\n def _generate_event_id(self) -> str:\n return f\"evt_{int(time.time() * 1000)}\"\n\n def log(\n self,\n event_type: EventType,\n system_id: str,\n user_id: str,\n input_summary: str,\n output_summary: str = \"\",\n metadata: dict = None,\n duration_ms: float = 0.0,\n ) -> AuditEvent:\n event = AuditEvent(\n event_id=self._generate_event_id(),\n timestamp=datetime.now().isoformat(),\n event_type=event_type.value,\n system_id=system_id,\n user_id=user_id,\n input_summary=input_summary[:200],\n output_summary=output_summary[:200],\n metadata=metadata or {},\n duration_ms=duration_ms,\n )\n\n with open(self.log_file, \"a\", encoding=\"utf-8\") as f:\n f.write(json.dumps(asdict(event), ensure_ascii=False) + \"\\n\")\n\n return event\n\n def get_recent_events(self, n: int = 10) -> list[AuditEvent]:\n if not self.log_file.exists():\n return []\n\n events = []\n with open(self.log_file, \"r\", encoding=\"utf-8\") as f:\n for line in f:\n if line.strip():\n data = json.loads(line)\n events.append(AuditEvent(**data))\n\n return events[-n:]\n\n def generate_compliance_report(self) -> dict:\n \"\"\"Generate an EU AI Act compliance report.\"\"\"\n if not self.log_file.exists():\n return {\"error\": \"No audit log found\"}\n\n events = []\n with open(self.log_file, \"r\", encoding=\"utf-8\") as f:\n for line in f:\n if line.strip():\n events.append(json.loads(line))\n\n total = len(events)\n by_type = {}\n for event in events:\n et = event[\"event_type\"]\n by_type[et] = by_type.get(et, 0) + 1\n\n security_blocks = by_type.get(EventType.SECURITY_BLOCK.value, 0)\n block_rate = security_blocks / total if total > 0 else 0\n\n return {\n \"report_generated_at\": datetime.now().isoformat(),\n \"total_events\": total,\n \"events_by_type\": by_type,\n \"security_block_rate\": round(block_rate, 4),\n \"compliance_status\": {\n \"audit_logging\": \"✅ Enabled\",\n \"human_oversight\": \"✅ Enabled\",\n \"transparency_disclosure\": \"✅ Implemented\",\n \"data_retention\": \"✅ Retained in JSONL files\",\n },\n }\n\n\n## 4. Governance-Compliant RAG — `governance/compliant_rag.py`\n\nA RAG implementation compliant with EU AI Act Article 50 (transparency obligations).\n\n\n\n # governance/compliant_rag.py\n import sys\n import os\n sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))\n\n import psycopg2\n import time\n from google import genai\n from google.genai import types\n from dotenv import load_dotenv\n from governance.audit_logger import AuditLogger, EventType\n from governance.ai_registry import get_system\n\n load_dotenv()\n\n client = genai.Client(api_key=os.getenv(\"GEMINI_API_KEY\"))\n logger = AuditLogger()\n\n conn = psycopg2.connect(\n host=os.getenv(\"DB_HOST\"),\n port=os.getenv(\"DB_PORT\"),\n dbname=os.getenv(\"DB_NAME\"),\n user=os.getenv(\"DB_USER\"),\n password=os.getenv(\"DB_PASSWORD\"),\n )\n cur = conn.cursor()\n\n # EU AI Act Article 50 — Chatbot transparency disclosure\n AI_DISCLOSURE = \"\"\"\n ⚠️ This system provides answers generated by Artificial Intelligence.\n Responses are based on pgvector database search results but may contain errors.\n For important decisions, please consult a qualified expert.\n \"\"\"\n\n SYSTEM_ID = \"rag-search-001\"\n\n\n def get_embedding(text: str) -> list[float]:\n result = client.models.embed_content(\n model=\"gemini-embedding-001\",\n contents=text,\n config=types.EmbedContentConfig(\n task_type=\"RETRIEVAL_QUERY\",\n output_dimensionality=768,\n ),\n )\n return result.embeddings[0].values\n\n\n def search(query: str, top_k: int = 3) -> list[dict]:\n query_embedding = get_embedding(query)\n cur.execute(\"\"\"\n SELECT title, body,\n 1 - (embedding <=> %s::vector) AS similarity\n FROM documents\n ORDER BY embedding <=> %s::vector\n LIMIT %s;\n \"\"\", (query_embedding, query_embedding, top_k))\n rows = cur.fetchall()\n return [\n {\"title\": r[0], \"body\": r[1], \"similarity\": round(r[2], 4)}\n for r in rows\n ]\n\n\n def compliant_rag_answer(\n question: str,\n user_id: str = \"anonymous\",\n show_disclosure: bool = True,\n ) -> dict:\n \"\"\"\n Governance-compliant RAG pipeline.\n\n EU AI Act compliance:\n - Article 50: Disclose that the system is AI\n - Article 12: Record audit logs\n - Only run systems registered in the inventory\n\n Returns:\n {\n \"answer\": str,\n \"disclosure\": str,\n \"sources\": list,\n \"audit_event_id\": str\n }\n \"\"\"\n start_time = time.time()\n\n system = get_system(SYSTEM_ID)\n if not system:\n return {\"error\": \"System not registered in inventory\"}\n\n query_event = logger.log(\n event_type=EventType.QUERY,\n system_id=SYSTEM_ID,\n user_id=user_id,\n input_summary=question,\n metadata={\"system_name\": system.name},\n )\n\n docs = search(question, top_k=3)\n search_duration = (time.time() - start_time) * 1000\n\n logger.log(\n event_type=EventType.SEARCH,\n system_id=SYSTEM_ID,\n user_id=user_id,\n input_summary=question,\n output_summary=f\"Retrieved {len(docs)} documents\",\n metadata={\"docs_count\": len(docs)},\n duration_ms=search_duration,\n )\n\n if not docs:\n return {\n \"answer\": \"No relevant documents found.\",\n \"disclosure\": AI_DISCLOSURE,\n \"sources\": [],\n \"audit_event_id\": query_event.event_id,\n }\n\n context = \"\\n\\n\".join([f\"[{d['title']}]\\n{d['body']}\" for d in docs])\n prompt = f\"\"\"Answer the question based on the following documents.\n\n # Reference Documents\n {context}\n\n # Question\n {question}\n\n # Answer (concisely, based on the documents)\"\"\"\n\n for attempt in range(3):\n try:\n response = client.models.generate_content(\n model=\"gemini-2.5-flash\",\n contents=prompt,\n )\n answer = response.text\n break\n except Exception as e:\n if (\"503\" in str(e) or \"429\" in str(e)) and attempt < 2:\n time.sleep((attempt + 1) * 10)\n else:\n raise\n\n gen_duration = (time.time() - start_time) * 1000\n\n logger.log(\n event_type=EventType.GENERATION,\n system_id=SYSTEM_ID,\n user_id=user_id,\n input_summary=question,\n output_summary=answer[:100],\n metadata={\"answer_length\": len(answer)},\n duration_ms=gen_duration,\n )\n\n return {\n \"answer\": answer,\n \"disclosure\": AI_DISCLOSURE if show_disclosure else \"\",\n \"sources\": [{\"title\": d[\"title\"], \"similarity\": d[\"similarity\"]} for d in docs],\n \"audit_event_id\": query_event.event_id,\n }\n\n\n if __name__ == \"__main__\":\n print(\"=== Governance-Compliant RAG ===\")\n print(AI_DISCLOSURE)\n\n result = compliant_rag_answer(\n question=\"How do you calculate the F1 score?\",\n user_id=\"user_001\",\n )\n\n print(f\"Answer:\\n{result['answer']}\")\n print(f\"\\nSource documents:\")\n for source in result[\"sources\"]:\n print(f\" - {source['title']} (similarity: {source['similarity']})\")\n print(f\"\\nAudit log event ID: {result['audit_event_id']}\")\n\n\n## 5. EU AI Act Compliance Status\n\nWhat this implementation covers:\n\nEU AI Act Article | Content | Implementation\n---|---|---\nArticle 4 | AI literacy | This tutorial itself\nArticle 12 | Record keeping / audit logs | `audit_logger.py`\nArticle 50 | Chatbot transparency disclosure | Display `AI_DISCLOSURE`\nArticle 9 | Risk management | `risk_assessor.py`\nAnnex IV | Technical documentation | `ai_registry.py`\n\n## 6. The Four Principles of AI Governance\n\n\n ① Know\n → Use ai_registry.py to track every AI system in use\n\n ② Assess\n → Use risk_assessor.py to quantify risk\n\n ③ Monitor\n → Use audit_logger.py + Langfuse for real-time monitoring\n\n ④ Respond\n → Report to authorities within 72 hours in case of incidents\n (for high-risk systems)\n\n\n## FAQ\n\n**Q: Is our RAG system subject to EU AI Act regulation?**\nA: It falls under \"Limited Risk (chatbot)\". We have a transparency obligation (Article 50) to disclose to users that they are interacting with AI. It does not qualify as high-risk, so strict compliance requirements don't apply.\n\n**Q: Does the EU AI Act apply to non-EU companies?**\nA: If you provide services to EU users, it applies regardless of where your company is based. If you serve EU users, compliance is required.\n\n**Q: What happens if classified as high-risk?**\nA: Conformity assessment, EU Declaration of Conformity, CE marking, and registration in the EU database become required.\n\n## Next Steps\n\n * **[Chapter 9: Summary]** — Your next steps as an AI architect\n * **ISO 42001 Certification** — AI Management System international standard, maps directly to EU AI Act requirements\n * **Incident response planning** — Design the 72-hour reporting flow for when problems occur\n * **Third-party risk management** — Include governance for external AI services like the Gemini API\n\n",
"title": "AI Governance — EU AI Act Compliance, Risk Assessment, and Audit Logging"
}