Knowledge Graph

Query: "Who authored the security audit and what did they recommend?"

Vector search returns:
  1. "The security audit identified 3 critical vulnerabilities..."
  2. "Best practices for security audits include..."
  3. "John Chen, Lead Security Architect, conducted the review..."

The answer requires connecting chunk 3 (who) to chunk 1 (what).
Vector search finds both but doesn't know they're related.

John Chen  ──AUTHORED──▶  Security Audit Report
John Chen  ──HAS_ROLE──▶  Lead Security Architect
Security Audit  ──IDENTIFIED──▶  3 Critical Vulnerabilities
Security Audit  ──RECOMMENDED──▶  Multi-Factor Authentication
Security Audit  ──RECOMMENDED──▶  Network Segmentation

# forge/ingestion/graph_extractor.py
class GraphExtractor:
    """Extracts entities and relationships from document chunks."""
 
    EXTRACTION_PROMPT = """Extract all entities and relationships from this text.
 
Entity types: PERSON, ORGANIZATION, CONCEPT, DOCUMENT, TECHNOLOGY, LOCATION, DATE
Relationship types: AUTHORED_BY, REFERENCES, PART_OF, RELATED_TO, LOCATED_IN, OCCURRED_ON
 
Output as JSON:
{{
  "entities": [
    {{"name": "...", "type": "...", "description": "..."}}
  ],
  "relationships": [
    {{"source": "...", "target": "...", "type": "...", "description": "..."}}
  ]
}}
 
Text:
{chunk_text}"""
 
    async def extract(self, chunk: DocumentChunk) -> GraphExtractionResult:
        response = await self.llm.generate(
            self.EXTRACTION_PROMPT.format(chunk_text=chunk.text),
            max_tokens=500,
            temperature=0.0,
        )
        return self._parse_response(response, chunk)

# Entity stored in Qdrant point payload
payload = {
    "text": chunk.text,
    "level": "L2",
    "entities": [
        {"name": "Dr. Sarah Kim", "type": "PERSON"},
        {"name": "RAPTOR", "type": "CONCEPT"},
        {"name": "MIT", "type": "ORGANIZATION"},
    ],
    "relationships": [
        {"source": "Dr. Sarah Kim", "target": "RAPTOR", "type": "AUTHORED"},
    ]
}

# forge/retrieval/graph.py
class GraphStore:
    """Redis-backed adjacency list for knowledge graph traversal."""
 
    async def store_relationship(self, rel: Relationship):
        """Store a relationship in Redis adjacency list."""
        # Forward edge
        await self.redis.sadd(
            f"graph:out:{rel.source_id}",
            json.dumps({
                "target": rel.target_id,
                "type": rel.type,
                "chunk_id": rel.source_chunk_id,
            })
        )
        # Reverse edge
        await self.redis.sadd(
            f"graph:in:{rel.target_id}",
            json.dumps({
                "source": rel.source_id,
                "type": rel.type,
                "chunk_id": rel.source_chunk_id,
            })
        )
 
    async def traverse(
        self,
        start_entity: str,
        max_hops: int = 2,
        relationship_filter: list[str] | None = None,
    ) -> list[GraphPath]:
        """BFS traversal from a starting entity."""
        visited = set()
        queue = [(start_entity, 0, [])]
        paths = []
 
        while queue:
            entity, depth, path = queue.pop(0)
            if entity in visited or depth > max_hops:
                continue
            visited.add(entity)
 
            edges = await self.redis.smembers(f"graph:out:{entity}")
            for edge_json in edges:
                edge = json.loads(edge_json)
                if relationship_filter and edge["type"] not in relationship_filter:
                    continue
                new_path = path + [(entity, edge["type"], edge["target"])]
                paths.append(GraphPath(path=new_path, chunk_ids=[edge["chunk_id"]]))
                queue.append((edge["target"], depth + 1, new_path))
 
        return paths

@tool
async def graph_traverse(
    entity: str,
    max_hops: int = 2,
    relationship_types: list[str] | None = None,
) -> list[GraphResult]:
    """Traverse the knowledge graph from a starting entity.
 
    Args:
        entity: The entity name to start traversal from
        max_hops: Maximum traversal depth (default: 2)
        relationship_types: Optional filter for specific relationship types
    """
    # First, find the entity in Qdrant
    entity_results = await qdrant.search(
        collection="forge_documents",
        query_filter=Filter(must=[
            FieldCondition(
                key="entities[].name",
                match=MatchText(text=entity),
            )
        ]),
        limit=5,
    )
 
    if not entity_results:
        return []
 
    # Traverse from found entity
    paths = await graph_store.traverse(
        start_entity=entity,
        max_hops=max_hops,
        relationship_filter=relationship_types,
    )
 
    # Fetch source chunks for each path
    results = []
    for path in paths:
        chunks = await qdrant.get_batch(path.chunk_ids)
        results.append(GraphResult(
            path=path,
            source_chunks=chunks,
        ))
 
    return results

graph:
  enabled: true
  extraction_model: "llm"
  entity_types:
    - PERSON
    - ORGANIZATION
    - CONCEPT
    - DOCUMENT
    - TECHNOLOGY
    - LOCATION
    - DATE
  relationship_types:
    - AUTHORED_BY
    - REFERENCES
    - PART_OF
    - RELATED_TO
    - LOCATED_IN
    - OCCURRED_ON
  max_entities_per_chunk: 15
  max_relationships_per_chunk: 20
  storage: "hybrid"           # Qdrant payload + Redis adjacency list