GraphRAG Implementation (Technical)

# Phase 2: Embedding storage for semantic search
embedding = Column(JSON, nullable=True)
embedding_model = Column(String, nullable=True)
embedding_updated_at = Column(DateTime, nullable=True)

def to_networkx(self) -> nx.DiGraph:
    """Convert SQLite graph to NetworkX for algorithms."""

def ego_graph(self, entity_name: str, radius: int = 2) -> dict:
    """Get k-hop neighborhood around an entity."""

def get_nodes_by_type(self, node_type: str) -> List[Node]:
    """Filter nodes by type (CHARACTER, LOCATION, etc.)."""

def get_edges_by_type(self, relation_type: str) -> List[Edge]:
    """Filter edges by relation type (MOTIVATES, HINDERS, etc.)."""

def get_all_nodes(self) -> List[Node]:
    """Return all nodes in the graph."""

def get_all_edges(self) -> List[Edge]:
    """Return all edges in the graph."""

class EmbeddingProvider(ABC):
    """Base class for embedding providers."""

    @abstractmethod
    async def embed(self, text: str) -> List[float]: ...

    @abstractmethod
    async def embed_batch(self, texts: List[str]) -> List[List[float]]: ...


class OllamaEmbedding(EmbeddingProvider):
    """Ollama-based embeddings with automatic model detection."""

    EMBEDDING_MODELS = [
        "nomic-embed-text",
        "mxbai-embed-large",
        "all-minilm",
    ]

    FALLBACK_MODELS = [
        "llama3.2:3b",
        "mistral:7b",
    ]


class OpenAIEmbedding(EmbeddingProvider):
    """OpenAI text-embedding-3-small/large."""


class CohereEmbedding(EmbeddingProvider):
    """Cohere embed-english-v3.0."""

from backend.services.embedding_service import get_embedding_service

service = get_embedding_service(provider="ollama")
embedding = await service.embed("Character description text")
similarity = service.cosine_similarity(embedding_a, embedding_b)

class EmbeddingIndexService:
    async def index_node(self, node: Node) -> bool:
        """Generate and store embedding for a node."""

    async def reindex_all(self) -> Dict[str, int]:
        """Rebuild all embeddings."""

    async def get_stale_nodes(self) -> List[Node]:
        """Find nodes needing reindexing."""

class KnowledgeRouter:
    async def route_query(self, query: str) -> RoutedContext:
        """
        Classify query → Retrieve from graph → Assemble context.

        Returns token-counted context block with priorities.
        """

class NarrativeEdgeType(Enum):
    # Goal-Obstacle-Conflict
    MOTIVATES = "MOTIVATES"
    HINDERS = "HINDERS"
    CAUSES = "CAUSES"

    # Character Dynamics
    CHALLENGES = "CHALLENGES"
    KNOWS = "KNOWS"
    CONTRADICTS = "CONTRADICTS"

    # Narrative Threading
    FORESHADOWS = "FORESHADOWS"
    CALLBACKS = "CALLBACKS"

    # Basic Relationships
    LOCATED_IN = "LOCATED_IN"
    OWNS = "OWNS"
    PART_OF = "PART_OF"
    HAS_TRAIT = "HAS_TRAIT"
    LOVES = "LOVES"
    HATES = "HATES"
    ALLIES_WITH = "ALLIES_WITH"
    CONFLICTS_WITH = "CONFLICTS_WITH"
    REVEALS = "REVEALS"
    CUSTOM = "CUSTOM"

class NarrativeExtractor:
    def __init__(
        self,
        graph_service: KnowledgeGraphService,
        ollama_url: str = "http://localhost:11434/api/chat",
        model: str = "llama3.2:3b"
    ): ...

    async def extract_from_scene(
        self,
        scene_content: str,
        scene_id: str,
        current_beat: str = "Unknown"
    ) -> Dict[str, Any]:
        """
        Extract entities and relationships from scene text.

        Returns:
        - entities: New characters, locations, objects, events
        - relationships: Narrative edges between entities
        - flaw_challenge: Whether protagonist's flaw was tested
        - beat_alignment: Whether scene matches expected beat
        """

    async def merge_to_graph(self, extraction: Dict) -> Dict[str, Any]:
        """Merge extraction results into the knowledge graph."""

class VerificationTier(Enum):
    FAST = "fast"      # <500ms - inline
    MEDIUM = "medium"  # 2-5s - background
    SLOW = "slow"      # 5-30s - on-demand


class VerificationService:
    async def run_fast_checks(
        self, content: str, context: Dict
    ) -> VerificationResult:
        """
        FAST tier checks:
        - Character alive/dead status
        - Known fact contradictions
        - Required callbacks present
        """

    async def run_medium_checks(
        self, content: str, context: Dict
    ) -> VerificationResult:
        """
        MEDIUM tier checks:
        - Flaw challenge gap detection
        - Beat alignment
        - Timeline consistency
        """

    async def run_slow_checks(
        self, content: str, context: Dict
    ) -> VerificationResult:
        """
        SLOW tier checks:
        - Full LLM semantic analysis
        - Delegates to GraphHealthService if available
        """

class SceneWriterService:
    async def verify_scene(
        self,
        content: str,
        scene_context: Optional[Dict] = None,
        run_background: bool = True
    ) -> Dict[str, Any]:
        """Run verification on generated content."""

    async def generate_and_verify(
        self,
        scene_id: str,
        scaffold: Scaffold,
        ...
    ) -> SceneGenerationResult:
        """Generate scene with automatic verification."""

class GraphAnalyzer:
    def detect_communities(self) -> Dict[str, List[str]]:
        """
        Louvain community detection on character nodes.
        Returns: {"primary_cast": [...], "secondary_cast": [...], ...}
        """

    def find_bridge_characters(self, top_k: int = 5) -> List[Dict]:
        """
        Betweenness centrality to find connecting characters.
        Returns role inference: protagonist/major/supporting/minor
        """

    def calculate_tension(self) -> Dict[str, Any]:
        """
        Tension = weighted sum of:
        - HINDERS edges (weight 2.0)
        - FORESHADOWS edges (weight 1.5)
        - CONTRADICTS edges (weight 3.0)
        - CHALLENGES edges (weight 2.5)

        Returns: score (0-1), level, indicators, recommendation
        """

    def analyze_pacing(self) -> Dict[str, Any]:
        """
        Categorize edges as action/setup/resolution.
        Returns: pacing type, ratios, recommendation
        """

    def get_narrative_summary(self) -> Dict[str, Any]:
        """
        Comprehensive analysis combining all methods.
        """

"graph": {
    "edge_types": {
        "MOTIVATES": True,
        "HINDERS": True,
        "CHALLENGES": True,
        "FORESHADOWS": True,
        "CALLBACKS": True,
        "CONTRADICTS": False,  # Experimental
    },
    "extraction_triggers": {
        "on_manuscript_promote": True,
        "before_foreman_chat": True,
        "periodic_minutes": 0,
    },
    "verification_level": "standard",  # minimal | standard | thorough
    "embedding_provider": "ollama",    # ollama | openai | cohere | none
}

"graph.verification_level": {"type": str, "choices": ["minimal", "standard", "thorough"]},
"graph.embedding_provider": {"type": str, "choices": ["ollama", "openai", "cohere", "none"]},
"graph.extraction_triggers.periodic_minutes": {"type": int, "min": 0, "max": 60},