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