09: Knowledge Graph and Datalake
From Document Retrieval to Reasoning-Aware Knowledge Fabric
Pillar: #9 of 12 - Supporting Priority
Timeline: 8 Weeks
Dependencies: #04 (RAG Next Gen)
Expected ROI: Multi-hop reasoning, 95% entity resolution, decision provenance tracking
Executive Summary
Move beyond flat document retrieval to a knowledge fabric that understands entities, relationships, and reasoning chains. Current RAG retrieves documents; knowledge graphs retrieve understanding. This pillar implements entity extraction, relationship mapping, and provenance tracking that enables AI to reason across the entire organizational knowledge base.
Key Insight: Documents are fragments; knowledge graphs are the whole picture.
Problem Statement
Current State: Document Islands
# File: maestro/learning/rag.py
# Current: Flat document retrieval
async def retrieve(self, query: str) -> List[Document]:
# Embed query
embedding = await self.embedder.embed(query)
# Find similar documents
results = await self.vector_store.search(
embedding,
limit=5,
)
return results # No entity understanding, no relationships
What's Missing
Query: "What authentication method did we choose for the mobile app?"
Current RAG returns:
- Document about OAuth2 (mentions authentication)
- Document about mobile frameworks (mentions mobile app)
- Document about security policies (mentions authentication)
What we need:
- The DECISION that linked mobile_app → authentication → OAuth2
- WHO made the decision
- WHEN it was made
- WHY it was chosen over alternatives
- WHAT other systems are affected
Knowledge Fragmentation
Current State:
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Requirements │ │ Design │ │ Code │
│ Docs │ │ Docs │ │ Comments │
└──────────────┘ └──────────────┘ └──────────────┘
↓ ↓ ↓
┌─────────────────────────────────────────────────┐
│ Flat Vector Store │
│ (No understanding of relationships) │
└─────────────────────────────────────────────────┘
Desired State:
┌─────────────────────────────────────────────────┐
│ KNOWLEDGE GRAPH │
│ │
│ User ──(creates)──► Requirement │
│ │ │ │
│ │ (specifies) │
│ │ ▼ │
│ └──(makes)──► Decision ──(chooses)──► Tech │
│ │ │
│ (documented in) │
│ ▼ │
│ Document │
└─────────────────────────────────────────────────┘
Solution Architecture
Knowledge Fabric
┌────────────────────────────────────────────┐
│ KNOWLEDGE FABRIC │
├────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────────┐ │
Input ────────────► │ │ Entity │ │ Relationship │ │
(docs, code, │ │ Extractor │ ─► │ Mapper │ │
conversations) │ └──────┬──────┘ └────────┬────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ GRAPH DATABASE │ │
│ │ (Entities + Relations + Properties) │ │
│ └─────────────────────────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ Vector │ │ Provenance │ │
│ │ Store │ │ Tracker │ │
│ └─────────────┘ └─────────────────┘ │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ QUERY ENGINE │ │
│ │ (Graph + Vector + Hybrid) │ │
│ └─────────────────────────────────────┘ │
│ │
└────────────────────────────────────────────┘
Core Components
class EntityExtractor:
"""
Extract entities from documents, code, and conversations.
"""
ENTITY_TYPES = [
"PERSON", "TEAM", "PROJECT", "SERVICE", "API",
"TECHNOLOGY", "REQUIREMENT", "DECISION", "DOCUMENT",
"CODE_MODULE", "DATABASE", "ENDPOINT", "CONFIG",
]
def __init__(self):
self.ner_model = load_model("maestro-ner")
self.code_parser = CodeEntityParser()
self.doc_parser = DocumentEntityParser()
async def extract(self, content: Content) -> List[Entity]:
if content.type == "code":
return await self._extract_from_code(content)
elif content.type == "document":
return await self._extract_from_document(content)
elif content.type == "conversation":
return await self._extract_from_conversation(content)
async def _extract_from_code(self, code: Code) -> List[Entity]:
"""Extract entities from source code."""
entities = []
# Parse AST
ast = self.code_parser.parse(code.content)
# Extract modules
for module in ast.modules:
entities.append(Entity(
type="CODE_MODULE",
name=module.name,
properties={
"path": module.path,
"language": code.language,
"dependencies": module.imports,
},
source=code.path,
))
# Extract functions/classes
for func in ast.functions:
entities.append(Entity(
type="FUNCTION",
name=func.name,
properties={
"signature": func.signature,
"docstring": func.docstring,
"complexity": func.complexity,
},
source=code.path,
))
# Extract API endpoints
for endpoint in self._find_endpoints(ast):
entities.append(Entity(
type="ENDPOINT",
name=endpoint.path,
properties={
"method": endpoint.method,
"handler": endpoint.handler,
"parameters": endpoint.params,
},
source=code.path,
))
return entities
async def _extract_from_document(self, doc: Document) -> List[Entity]:
"""Extract entities from documentation."""
# Use NER model
ner_results = await self.ner_model.extract(doc.content)
entities = []
for result in ner_results:
entities.append(Entity(
type=result.label,
name=result.text,
properties={
"context": result.context,
"confidence": result.confidence,
},
source=doc.path,
))
# Extract decisions (special handling)
decisions = self._extract_decisions(doc.content)
entities.extend(decisions)
return entities
2. Relationship Mapper
class RelationshipMapper:
"""
Map relationships between entities.
"""
RELATION_TYPES = [
"DEPENDS_ON", "IMPLEMENTS", "USES", "CALLS",
"CREATED_BY", "DECIDED", "DOCUMENTS", "BELONGS_TO",
"REPLACES", "EXTENDS", "CONFIGURES", "AUTHENTICATES",
]
async def map_relationships(
self,
entities: List[Entity],
context: Context
) -> List[Relationship]:
relationships = []
# Code-based relationships
if context.type == "code":
relationships.extend(
await self._map_code_relationships(entities, context)
)
# Document-based relationships
if context.type == "document":
relationships.extend(
await self._map_document_relationships(entities, context)
)
# Cross-reference relationships
relationships.extend(
await self._map_cross_references(entities)
)
return relationships
async def _map_code_relationships(
self,
entities: List[Entity],
context: Context
) -> List[Relationship]:
relationships = []
# Import relationships
for entity in entities:
if entity.type == "CODE_MODULE":
for dep in entity.properties.get("dependencies", []):
target = await self._resolve_entity(dep)
if target:
relationships.append(Relationship(
source=entity.id,
target=target.id,
type="DEPENDS_ON",
properties={"import_type": "direct"},
))
# Call relationships
call_graph = await self._build_call_graph(context)
for caller, callees in call_graph.items():
for callee in callees:
relationships.append(Relationship(
source=caller.id,
target=callee.id,
type="CALLS",
properties={"call_count": callee.count},
))
return relationships
async def _map_cross_references(
self,
entities: List[Entity]
) -> List[Relationship]:
"""Find relationships across entity types."""
relationships = []
# Link requirements to implementations
requirements = [e for e in entities if e.type == "REQUIREMENT"]
implementations = [e for e in entities if e.type in ["CODE_MODULE", "FUNCTION"]]
for req in requirements:
# Find implementations that mention this requirement
matches = await self._find_implementations(req, implementations)
for impl in matches:
relationships.append(Relationship(
source=impl.id,
target=req.id,
type="IMPLEMENTS",
properties={"confidence": matches[impl]},
))
return relationships
3. Graph Database Manager
class GraphDatabaseManager:
"""
Manage the knowledge graph in Neo4j.
"""
def __init__(self, uri: str, auth: tuple):
self.driver = GraphDatabase.driver(uri, auth=auth)
async def add_entity(self, entity: Entity):
"""Add or update an entity."""
query = """
MERGE (e:{type} {{id: $id}})
SET e += $properties
SET e.name = $name
SET e.updated_at = datetime()
RETURN e
""".format(type=entity.type)
async with self.driver.session() as session:
await session.run(
query,
id=entity.id,
name=entity.name,
properties=entity.properties,
)
async def add_relationship(self, rel: Relationship):
"""Add a relationship between entities."""
query = """
MATCH (source {{id: $source_id}})
MATCH (target {{id: $target_id}})
MERGE (source)-[r:{type}]->(target)
SET r += $properties
SET r.updated_at = datetime()
RETURN r
""".format(type=rel.type)
async with self.driver.session() as session:
await session.run(
query,
source_id=rel.source,
target_id=rel.target,
properties=rel.properties,
)
async def query_graph(self, cypher: str, params: dict = None) -> List[dict]:
"""Execute a Cypher query."""
async with self.driver.session() as session:
result = await session.run(cypher, params or {})
return [record.data() async for record in result]
async def find_path(
self,
source: str,
target: str,
max_hops: int = 5
) -> List[Path]:
"""Find paths between entities."""
query = """
MATCH path = shortestPath(
(source {id: $source})-[*1..$max_hops]-(target {id: $target})
)
RETURN path
"""
result = await self.query_graph(
query,
{"source": source, "target": target, "max_hops": max_hops}
)
return [self._parse_path(r["path"]) for r in result]
4. Hybrid Query Engine
class HybridQueryEngine:
"""
Query across graph, vector, and structured data.
"""
def __init__(self):
self.graph_db = GraphDatabaseManager()
self.vector_store = VectorStore()
self.llm = LLMClient()
async def query(
self,
query: str,
context: Optional[Context] = None
) -> QueryResult:
# Step 1: Extract entities from query
query_entities = await self._extract_query_entities(query)
# Step 2: Graph-based retrieval
graph_results = await self._graph_retrieve(query_entities)
# Step 3: Vector-based retrieval
vector_results = await self._vector_retrieve(query)
# Step 4: Merge and rank
merged = await self._merge_results(graph_results, vector_results)
# Step 5: Build reasoning chain
reasoning = await self._build_reasoning_chain(merged, query)
return QueryResult(
entities=merged.entities,
relationships=merged.relationships,
documents=merged.documents,
reasoning_chain=reasoning,
provenance=self._collect_provenance(merged),
)
async def _graph_retrieve(
self,
entities: List[Entity]
) -> GraphResult:
"""Multi-hop retrieval from graph."""
results = []
for entity in entities:
# Find entity in graph
matches = await self.graph_db.query_graph(
"""
MATCH (e)
WHERE e.name CONTAINS $name OR e.id = $name
RETURN e
""",
{"name": entity.name}
)
# Expand to related entities
for match in matches:
related = await self.graph_db.query_graph(
"""
MATCH (e {id: $id})-[r*1..2]-(related)
RETURN e, r, related
""",
{"id": match["e"]["id"]}
)
results.extend(related)
return GraphResult(results)
async def _build_reasoning_chain(
self,
results: MergedResults,
query: str
) -> ReasoningChain:
"""Build explicit reasoning chain for the answer."""
steps = []
for entity in results.key_entities:
# Find how this entity relates to the query
path = await self.graph_db.find_path(
source=entity.id,
target=results.primary_entity.id
)
if path:
steps.append(ReasoningStep(
entity=entity,
relationship=path[0].relationship,
explanation=self._explain_step(path),
))
return ReasoningChain(steps=steps, confidence=self._calculate_confidence(steps))
5. Provenance Tracker
class ProvenanceTracker:
"""
Track the origin and lineage of all knowledge.
"""
async def record_provenance(
self,
entity: Entity,
source: Source,
actor: Actor,
timestamp: datetime
):
"""Record where knowledge came from."""
provenance = Provenance(
entity_id=entity.id,
source_type=source.type,
source_path=source.path,
source_version=source.version,
actor_id=actor.id,
actor_type=actor.type, # human, agent, system
timestamp=timestamp,
confidence=self._calculate_confidence(source),
)
await self.graph_db.add_entity(provenance)
await self.graph_db.add_relationship(Relationship(
source=entity.id,
target=provenance.id,
type="HAS_PROVENANCE",
))
async def get_lineage(self, entity_id: str) -> Lineage:
"""Get full lineage of an entity."""
query = """
MATCH (e {id: $id})-[:HAS_PROVENANCE]->(p:Provenance)
OPTIONAL MATCH (e)-[:DERIVED_FROM*]->(ancestor)
RETURN e, p, collect(ancestor) as ancestors
"""
result = await self.graph_db.query_graph(query, {"id": entity_id})
return Lineage(
entity=result["e"],
provenance=result["p"],
ancestors=result["ancestors"],
)
async def validate_knowledge(self, entity_id: str) -> ValidationResult:
"""Check if knowledge is still valid."""
lineage = await self.get_lineage(entity_id)
issues = []
# Check if sources still exist
for source in lineage.sources:
if not await self._source_exists(source):
issues.append(ValidationIssue(
type="SOURCE_MISSING",
source=source,
))
# Check if sources have been updated
for source in lineage.sources:
current_version = await self._get_current_version(source)
if current_version != source.version:
issues.append(ValidationIssue(
type="SOURCE_UPDATED",
source=source,
old_version=source.version,
new_version=current_version,
))
return ValidationResult(
valid=len(issues) == 0,
issues=issues,
)
Integration Points
Files to Modify
RAG Enhancement
# File: maestro/learning/rag.py
class EnhancedRAG:
"""RAG with knowledge graph integration."""
def __init__(self):
self.vector_store = VectorStore()
self.knowledge_graph = KnowledgeGraph()
self.hybrid_engine = HybridQueryEngine()
async def retrieve(
self,
query: str,
context: Context
) -> EnhancedRetrievalResult:
# Hybrid retrieval
result = await self.hybrid_engine.query(query, context)
return EnhancedRetrievalResult(
documents=result.documents,
entities=result.entities,
relationships=result.relationships,
reasoning_chain=result.reasoning_chain,
provenance=result.provenance,
)
Implementation Plan
- [ ] Build entity extractors for code
- [ ] Build entity extractors for documents
- [ ] Train NER model on Maestro data
- [ ] Create entity resolution logic
Weeks 3-4: Relationship Mapping
- [ ] Implement relationship extractors
- [ ] Build call graph analyzer
- [ ] Create cross-reference mapper
- [ ] Set up Neo4j database
Weeks 5-6: Query Engine
- [ ] Build graph query interface
- [ ] Implement hybrid retrieval
- [ ] Create reasoning chain builder
- [ ] Add caching layer
Weeks 7-8: Provenance & Integration
- [ ] Implement provenance tracker
- [ ] Build lineage visualization
- [ ] Integrate with RAG
- [ ] Production deployment
Expected Results
Retrieval Quality
Use Case Improvements
Risks & Mitigations
Success Criteria
- #04 RAG Next Gen: Knowledge graph enhances retrieval
- #08 Observability: Graph metrics in dashboards
- #10 Security: Provenance for audit trails
Status: Ready for Implementation
Key Insight: Documents are fragments; knowledge graphs are the whole picture
Owner: Knowledge Engineering Team
