Knowledge Graph as the Brain - MIZ OKI 3.0 Architecture
Document: KG_BRAIN_ARCHITECTURE.md Version: 1.0.0 Created: December 9, 2025 Purpose: Define how KG serves as the central intelligence directing all agents, services, and MCP tools
Executive Summary
Paradigm Shift: The Knowledge Graph (KG) is the BRAIN of the system, not just storage. It contains: - Complete business process understanding - Full customer journey mapping (awareness → sale → retention) - Next-best-action intelligence - Orchestration rules for all agents/services/MCP
Boss Agent becomes the EXECUTOR that: - Queries KG for what to do next - Executes KG's recommendations - Reports results back to KG for learning
1. The KG Brain Model
┌─────────────────────────────────────────────────────────────────────────┐
│ KNOWLEDGE GRAPH (THE BRAIN) │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ BUSINESS │ │ CUSTOMER │ │ ORCHESTRATION │ │
│ │ PROCESSES │ │ JOURNEY │ │ RULES │ │
│ │ │ │ │ │ │ │
│ │ • Sales Pipeline│ │ • Awareness │ │ • Which agent? │ │
│ │ • Marketing Ops │ │ • Consideration │ │ • Which cell? │ │
│ │ • Customer Svc │ │ • Decision │ │ • Which service?│ │
│ │ • Operations │ │ • Purchase │ │ • Which MCP? │ │
│ │ • Finance │ │ • Retention │ │ • Next step? │ │
│ │ • Analytics │ │ • Advocacy │ │ • Priority? │ │
│ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │ │
│ └────────────────────┼────────────────────┘ │
│ │ │
│ ┌───────────▼───────────┐ │
│ │ NEXT-BEST-ACTION │ │
│ │ ENGINE │ │
│ └───────────┬───────────┘ │
│ │ │
└────────────────────────────────┼─────────────────────────────────────────┘
│
┌────────────▼────────────┐
│ BOSS AGENT │
│ (THE EXECUTOR) │
│ │
│ "KG, what should I do │
│ next for this user?" │
└────────────┬────────────┘
│
┌────────────┬───────────┼───────────┬────────────┐
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ Cells │ │ MOA │ │ MOE │ │ REWOO │ │ MCP │
│ (1-32) │ │ │ │ │ │ │ │ Tools │
└─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘
2. Customer Journey in the KG
Journey Stages & KG Nodes
CUSTOMER JOURNEY (stored as connected nodes in KG)
═══════════════════════════════════════════════════
[AWARENESS] ──► [CONSIDERATION] ──► [DECISION] ──► [PURCHASE] ──► [RETENTION] ──► [ADVOCACY]
│ │ │ │ │ │
▼ ▼ ▼ ▼ ▼ ▼
Touchpoints: Touchpoints: Touchpoints: Touchpoints: Touchpoints: Touchpoints:
- Ad impression - Website visit - Demo request - Cart add - Onboarding - Review
- Social post - Content DL - Pricing page - Checkout - Support - Referral
- Email open - Webinar - Sales call - Payment - Usage - Social share
- Search query - Comparison - Trial start - Confirmation - Renewal - Case study
KG Node Types for Journey
# Node types in Firestore KG
JOURNEY_NODE_TYPES = {
"user": {
"attributes": ["user_id", "segment", "lifetime_value", "current_stage", "health_score"]
},
"touchpoint": {
"attributes": ["type", "channel", "timestamp", "value", "sentiment"]
},
"stage": {
"attributes": ["name", "entry_criteria", "exit_criteria", "typical_duration"]
},
"action": {
"attributes": ["type", "agent_required", "priority", "expected_outcome"]
},
"business_process": {
"attributes": ["name", "department", "steps", "kpis", "owners"]
}
}
# Edge types (relationships)
JOURNEY_EDGE_TYPES = {
"EXPERIENCED": "user -> touchpoint",
"AT_STAGE": "user -> stage",
"PROGRESSED_TO": "stage -> stage",
"TRIGGERED_BY": "touchpoint -> action",
"REQUIRES": "action -> agent/cell/service",
"PART_OF": "touchpoint -> business_process",
"LEADS_TO": "action -> outcome"
}
3. Business Process Integration
Process Categories & Cell Mapping
BUSINESS PROCESS │ KG KNOWLEDGE │ CELLS/SERVICES
═════════════════════════════════│═══════════════════════════│═════════════════════
│ │
MARKETING │ │
├─ Campaign Management │ Campaign -> Audience │ Cell 16 (Segmentation)
├─ Lead Generation │ Lead -> Score -> Nurture │ Cell 7 (Causal)
├─ Content Marketing │ Content -> Engagement │ Cell 22 (NLP)
├─ Email Marketing │ Email -> Open -> Click │ Cell 31 (Recommendations)
└─ Attribution │ Channel -> Conversion │ Cell 7 (Attribution)
│ │
SALES │ │
├─ Lead Qualification │ Lead -> MQL -> SQL │ Cell 9 (Prediction)
├─ Opportunity Management │ Opp -> Stage -> Close │ Cell 14 (Strategy)
├─ Forecasting │ Pipeline -> Forecast │ Cell 30 (Time Series)
├─ Pricing │ Product -> Price -> Deal │ Cell 12 (Optimization)
└─ Quote Management │ Quote -> Approval │ Cell 13 (Risk)
│ │
CUSTOMER SUCCESS │ │
├─ Onboarding │ User -> Activation │ Cell 26 (Journey)
├─ Health Scoring │ User -> Health -> Churn │ Cell 18 (CLV)
├─ Support Tickets │ Ticket -> Resolution │ Cell 22 (NLP)
├─ Renewals │ Contract -> Renewal │ Cell 9 (Prediction)
└─ Upsell/Cross-sell │ User -> Propensity │ Cell 31 (Recommendations)
│ │
ANALYTICS │ │
├─ ROI Analysis │ Investment -> Return │ Cell 32 (ROI Dashboard)
├─ A/B Testing │ Variant -> Significance │ Cell 28 (A/B Testing)
├─ Anomaly Detection │ Metric -> Anomaly │ Cell 29 (Anomaly)
├─ Forecasting │ Historical -> Future │ Cell 30 (Time Series)
└─ Causal Analysis │ Cause -> Effect │ Cell 7 (Causal)
4. Next-Best-Action Engine
How KG Determines Next Step
class KGBrain:
"""KG as the Brain that directs all operations"""
async def get_next_best_action(self, user_id: str, context: Dict) -> Dict:
"""
Query KG to determine what Boss Agent should do next.
Returns:
{
"action": "engage_lead",
"agent": "cell_31", # Recommendation Engine
"service": "moa", # Use multi-agent if complex
"mcp_tools": ["write_kg_edge", "invoke_cell"],
"priority": "high",
"reason": "User at consideration stage, high intent signals",
"expected_outcome": "Move to decision stage"
}
"""
# 1. Get user's current journey state
journey_state = await self.get_user_journey_state(user_id)
# 2. Get relevant business process rules
process_rules = await self.get_process_rules(journey_state["current_stage"])
# 3. Analyze user signals and intent
intent_signals = await self.analyze_intent_signals(user_id, context)
# 4. Query orchestration rules
orchestration = await self.get_orchestration_rules(
stage=journey_state["current_stage"],
intent=intent_signals["primary_intent"],
segment=journey_state["segment"]
)
# 5. Determine next best action
return {
"action": orchestration["recommended_action"],
"agent": orchestration["primary_agent"],
"service": orchestration["orchestration_service"],
"mcp_tools": orchestration["required_tools"],
"priority": intent_signals["urgency"],
"reason": self.explain_decision(journey_state, process_rules, intent_signals),
"expected_outcome": orchestration["expected_outcome"],
"fallback_actions": orchestration["fallback_actions"]
}
async def record_outcome(self, action_id: str, outcome: Dict):
"""Record action outcome in KG for learning"""
# Update edges: action -> outcome
# Update user journey state
# Trigger learning (Cell 20, 24)
pass
Orchestration Rules in KG
# Stored in Firestore: kg_orchestration_rules collection
ORCHESTRATION_RULES = {
"awareness_to_consideration": {
"trigger": "high_engagement_score",
"primary_agent": "cell_31", # Recommendations
"orchestration_service": None, # Direct cell call
"actions": [
{"type": "send_content", "priority": 1},
{"type": "schedule_nurture", "priority": 2}
],
"mcp_tools": ["write_kg_edge", "invoke_cell"]
},
"consideration_high_intent": {
"trigger": "pricing_page_visit + demo_request",
"primary_agent": "cell_14", # Strategy Planning
"orchestration_service": "moa", # Multi-agent for complex decision
"actions": [
{"type": "prioritize_lead", "priority": 1},
{"type": "assign_sales_rep", "priority": 1},
{"type": "prepare_proposal", "priority": 2}
],
"mcp_tools": ["invoke_cell", "generate_code", "write_kg_node"]
},
"churn_risk_detected": {
"trigger": "health_score < 40 AND days_since_login > 14",
"primary_agent": "cell_26", # Journey Intelligence
"orchestration_service": "rewoo", # Planning for intervention
"actions": [
{"type": "analyze_drop_points", "priority": 1},
{"type": "generate_intervention", "priority": 1},
{"type": "alert_csm", "priority": 2}
],
"mcp_tools": ["read_file", "write_kg_edge", "invoke_cell"]
}
}
5. Boss Agent as Executor
Updated Boss Agent Flow
┌──────────────────────────────────────────────────────────────────┐
│ BOSS AGENT (EXECUTOR) │
│ │
│ 1. RECEIVE REQUEST │
│ └─► User query OR System event OR Scheduled trigger │
│ │
│ 2. CONSULT KG BRAIN │
│ └─► kg_brain.get_next_best_action(user_id, context) │
│ Returns: {agent, service, mcp_tools, priority, reason} │
│ │
│ 3. EXECUTE KG'S DECISION │
│ ├─► If agent only: cell_orchestrator.invoke_cell(agent, task) │
│ ├─► If service: agent_deployer.invoke_orchestrator(service) │
│ └─► MCP tools: mcp_registry.invoke_tool(tool, params) │
│ │
│ 4. REPORT BACK TO KG │
│ └─► kg_brain.record_outcome(action_id, result) │
│ - Updates user journey │
│ - Triggers learning │
│ - Refines orchestration rules │
│ │
│ 5. RESPOND TO USER │
│ └─► Format response with KG context │
└──────────────────────────────────────────────────────────────────┘
Implementation
class BossAgentExecutor:
"""Boss Agent as the Executor of KG Brain decisions"""
def __init__(self):
self.kg_brain = KGBrain()
self.cell_orchestrator = CellOrchestrator()
self.agent_deployer = AgentDeployer()
self.mcp_registry = MCPToolRegistry()
async def process_request(self, request: Dict) -> Dict:
"""Process any request by consulting KG first"""
user_id = request.get("user_id", "anonymous")
query = request.get("message", "")
context = request.get("context", {})
# 1. CONSULT KG BRAIN - What should I do?
kg_decision = await self.kg_brain.get_next_best_action(
user_id=user_id,
context={
"query": query,
"channel": context.get("channel"),
"session_data": context.get("session_data"),
"recent_actions": context.get("recent_actions", [])
}
)
# 2. EXECUTE KG'S DECISION
action_id = str(uuid.uuid4())
result = await self._execute_kg_decision(kg_decision, query, context)
# 3. REPORT BACK TO KG
await self.kg_brain.record_outcome(
action_id=action_id,
outcome={
"decision": kg_decision,
"result": result,
"success": result.get("status") == "success",
"timestamp": datetime.utcnow().isoformat()
}
)
# 4. RESPOND
return {
"response": result.get("response"),
"kg_context": {
"user_stage": kg_decision.get("user_stage"),
"action_taken": kg_decision.get("action"),
"agent_used": kg_decision.get("agent"),
"next_suggested": kg_decision.get("expected_outcome")
},
"metadata": {
"action_id": action_id,
"kg_directed": True
}
}
async def _execute_kg_decision(self, decision: Dict, query: str, context: Dict) -> Dict:
"""Execute whatever the KG Brain decided"""
agent = decision.get("agent")
service = decision.get("orchestration_service")
mcp_tools = decision.get("mcp_tools", [])
# Use orchestration service if specified
if service:
return await self.agent_deployer.invoke_orchestrator(
orchestrator=service,
task=query,
context={
**context,
"kg_decision": decision,
"primary_agent": agent
}
)
# Otherwise, invoke cell directly
if agent and agent.startswith("cell_"):
return await self.cell_orchestrator.invoke_cell(
cell_id=agent,
task=query,
context={
**context,
"kg_decision": decision
}
)
# Fallback to LLM with KG context
return await self._llm_response_with_kg_context(query, decision, context)
6. Required KG Collections (Firestore)
KG_COLLECTIONS = {
# Core Journey Data
"kg_users": "User profiles with journey state",
"kg_touchpoints": "All user touchpoints/interactions",
"kg_stages": "Journey stage definitions",
"kg_stage_transitions": "User stage transitions history",
# Business Process Data
"kg_business_processes": "All business processes",
"kg_process_steps": "Steps within each process",
"kg_process_metrics": "KPIs and metrics for processes",
# Orchestration Intelligence
"kg_orchestration_rules": "Rules for what agent/service to use",
"kg_action_outcomes": "Historical outcomes of actions",
"kg_agent_performance": "Performance metrics per agent/cell",
# Real-time State
"kg_active_sessions": "Currently active user sessions",
"kg_pending_actions": "Actions waiting to be executed",
"kg_intervention_queue": "Interventions queued by KG"
}
7. API Additions Required
New KG Brain Endpoints
# POST /api/v1/kg/brain/next-action
# Get next best action for a user
@app.post("/api/v1/kg/brain/next-action")
async def get_next_action(request: NextActionRequest):
"""Query KG Brain for next best action"""
return await kg_brain.get_next_best_action(
user_id=request.user_id,
context=request.context
)
# POST /api/v1/kg/brain/record-outcome
# Record action outcome for learning
@app.post("/api/v1/kg/brain/record-outcome")
async def record_outcome(request: OutcomeRequest):
"""Record outcome of an action in KG"""
return await kg_brain.record_outcome(
action_id=request.action_id,
outcome=request.outcome
)
# GET /api/v1/kg/journey/{user_id}
# Get complete user journey
@app.get("/api/v1/kg/journey/{user_id}")
async def get_user_journey(user_id: str):
"""Get complete journey state for a user"""
return await kg_brain.get_user_journey_state(user_id)
# POST /api/v1/kg/journey/transition
# Record a stage transition
@app.post("/api/v1/kg/journey/transition")
async def record_transition(request: TransitionRequest):
"""Record a user's journey stage transition"""
return await kg_brain.record_stage_transition(
user_id=request.user_id,
from_stage=request.from_stage,
to_stage=request.to_stage,
trigger=request.trigger
)
# GET /api/v1/kg/process/{process_name}
# Get business process definition
@app.get("/api/v1/kg/process/{process_name}")
async def get_process(process_name: str):
"""Get business process definition from KG"""
return await kg_brain.get_business_process(process_name)
# GET /api/v1/kg/orchestration/rules
# Get orchestration rules
@app.get("/api/v1/kg/orchestration/rules")
async def get_orchestration_rules(stage: str = None, intent: str = None):
"""Get orchestration rules from KG"""
return await kg_brain.get_orchestration_rules(stage=stage, intent=intent)
8. Implementation Roadmap
Phase 1: KG Brain Foundation (v5.2.0)
- [ ] Implement
KGBrainclass with basic next-action logic - [ ] Add journey stage tracking in Firestore
- [ ] Create orchestration rules collection
- [ ] Update Boss Agent to consult KG before acting
Phase 2: Business Process Integration (v5.3.0)
- [ ] Import business process definitions into KG
- [ ] Map processes to cells/services
- [ ] Implement process-aware routing
Phase 3: Learning Loop (v5.4.0)
- [ ] Record all action outcomes
- [ ] Implement outcome analysis
- [ ] Auto-refine orchestration rules based on success rates
Phase 4: Advanced Intelligence (v5.5.0)
- [ ] Predictive next-action (before user asks)
- [ ] Cross-journey optimization
- [ ] Multi-user pattern learning
9. Summary: The New Paradigm
| Component | OLD (Current) | NEW (KG Brain) |
|---|---|---|
| Brain | Boss Agent | Knowledge Graph |
| Boss Agent | Orchestrator | Executor of KG decisions |
| Decision Logic | In Boss Agent code | In KG rules/edges |
| Journey State | Per-request | Persistent in KG |
| Business Process | Implicit | Explicit in KG |
| Next Action | LLM decides | KG rules + LLM |
| Learning | Conversation memory | Action outcome tracking |
Key Principle: The KG knows the business. Boss Agent executes what KG decides. All agents/cells/services are tools the KG orchestrates through Boss Agent.
This document defines the target architecture for KG as the central intelligence of MIZ OKI 3.0.