Knowledge Engineering in the Age of LLMs: Bridging Symbolic AI and Neural Networks

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class KnowledgeEmbedder:
    """
    Transforms textual knowledge into semantic vector representations.
    """

    def __init__(self, model_name: str = "all-mpnet-base-v2"):
        self.encoder = SentenceTransformer(model_name)
        self.knowledge_base = {}
        self.embeddings = None

    def add_knowledge(self, entities: dict[str, str]):
        """
        Add knowledge entities with their descriptions.

        Args:
            entities: Dict mapping entity names to descriptions
        """
        self.knowledge_base.update(entities)
        descriptions = list(self.knowledge_base.values())
        self.embeddings = self.encoder.encode(descriptions)

    def find_related(self, query: str, top_k: int = 5) -> list[tuple[str, float]]:
        """
        Find knowledge entities semantically related to a query.
        """
        query_embedding = self.encoder.encode([query])
        similarities = cosine_similarity(query_embedding, self.embeddings)[0]

        indices = np.argsort(similarities)[::-1][:top_k]
        entities = list(self.knowledge_base.keys())

        return [(entities[i], float(similarities[i])) for i in indices]

    def compute_relation_strength(self, entity_a: str, entity_b: str) -> float:
        """
        Compute semantic relatedness between two entities.
        """
        emb_a = self.encoder.encode([self.knowledge_base[entity_a]])
        emb_b = self.encoder.encode([self.knowledge_base[entity_b]])
        return float(cosine_similarity(emb_a, emb_b)[0][0])

from rdflib import Graph, Namespace, URIRef, Literal
from rdflib.namespace import RDF, RDFS, OWL
from typing import List, Tuple

class KnowledgeGraphBuilder:
    """
    Constructs RDF knowledge graphs from LLM-extracted information.
    """

    def __init__(self, namespace: str = "http://example.org/ontology#"):
        self.graph = Graph()
        self.ns = Namespace(namespace)
        self.graph.bind("ex", self.ns)
        self.graph.bind("owl", OWL)

    def add_class(self, class_name: str, parent: str = None):
        """Define an ontological class."""
        class_uri = self.ns[class_name]
        self.graph.add((class_uri, RDF.type, OWL.Class))
        self.graph.add((class_uri, RDFS.label, Literal(class_name)))

        if parent:
            self.graph.add((class_uri, RDFS.subClassOf, self.ns[parent]))

    def add_entity(self, entity: str, entity_class: str, properties: dict = None):
        """Add an instance to the knowledge graph."""
        entity_uri = self.ns[entity.replace(" ", "_")]
        self.graph.add((entity_uri, RDF.type, self.ns[entity_class]))
        self.graph.add((entity_uri, RDFS.label, Literal(entity)))

        if properties:
            for prop, value in properties.items():
                prop_uri = self.ns[prop]
                if isinstance(value, str) and value.startswith("http"):
                    self.graph.add((entity_uri, prop_uri, URIRef(value)))
                else:
                    self.graph.add((entity_uri, prop_uri, Literal(value)))

    def add_relation(self, subject: str, predicate: str, obj: str):
        """Add a relationship between entities."""
        subj_uri = self.ns[subject.replace(" ", "_")]
        pred_uri = self.ns[predicate]
        obj_uri = self.ns[obj.replace(" ", "_")]
        self.graph.add((subj_uri, pred_uri, obj_uri))

    def query(self, sparql: str) -> List[Tuple]:
        """Execute a SPARQL query against the knowledge graph."""
        return list(self.graph.query(sparql))

    def serialize(self, format: str = "turtle") -> str:
        """Export the graph in the specified format."""
        return self.graph.serialize(format=format)

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class KnowledgeReasoner:
    """
    Implements chain-of-thought reasoning over knowledge bases.
    """

    REASONING_TEMPLATE = """You are a knowledge reasoning system. Given the following
knowledge context and question, reason step-by-step to derive the answer.

Knowledge Context:
{context}

Question: {question}

Let's think through this step-by-step:
1. First, identify the relevant facts from the knowledge context.
2. Then, determine what logical connections exist between these facts.
3. Apply any necessary inference rules.
4. Derive the final conclusion.

Reasoning:"""

    def __init__(self, model_name: str = "mistralai/Mistral-7B-Instruct-v0.2"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    def reason(self, knowledge_context: str, question: str) -> dict:
        """
        Perform chain-of-thought reasoning over knowledge.

        Returns:
            Dict with 'reasoning_chain' and 'conclusion'
        """
        prompt = self.REASONING_TEMPLATE.format(
            context=knowledge_context,
            question=question
        )

        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=0.3,
                do_sample=True,
                pad_token_id=self.tokenizer.eos_token_id
            )

        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        reasoning = response.split("Reasoning:")[-1].strip()

        # Extract conclusion from reasoning chain
        lines = reasoning.split("\n")
        conclusion = lines[-1] if lines else reasoning

        return {
            "reasoning_chain": reasoning,
            "conclusion": conclusion
        }

class SymbolicValidator:
    """
    Validates LLM reasoning against knowledge graph constraints.
    """

    def __init__(self, knowledge_graph: KnowledgeGraphBuilder):
        self.kg = knowledge_graph

    def validate_assertion(self, subject: str, predicate: str, obj: str) -> dict:
        """
        Check if an assertion is consistent with the knowledge graph.
        """
        # Check if entities exist
        subject_query = f"""
        ASK {{ ?s rdfs:label "{subject}" }}
        """
        subject_exists = bool(list(self.kg.query(subject_query)))

        # Check domain/range constraints
        constraint_query = f"""
        SELECT ?domain ?range WHERE {{
            ex:{predicate} rdfs:domain ?domain .
            ex:{predicate} rdfs:range ?range .
        }}
        """
        constraints = list(self.kg.query(constraint_query))

        # Validate type compatibility
        type_query = f"""
        SELECT ?type WHERE {{
            ?s rdfs:label "{subject}" .
            ?s rdf:type ?type .
        }}
        """
        subject_types = [str(t[0]) for t in self.kg.query(type_query)]

        return {
            "subject_exists": subject_exists,
            "constraints": constraints,
            "subject_types": subject_types,
            "is_valid": subject_exists and len(constraints) > 0
        }

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.schema import Document
from typing import List, Optional
import hashlib

class KnowledgeRAGSystem:
    """
    Production-grade RAG system with hybrid retrieval and fact validation.
    """

    def __init__(
        self,
        embedding_model: str = "sentence-transformers/all-mpnet-base-v2",
        chunk_size: int = 512,
        chunk_overlap: int = 50
    ):
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={'device': 'cpu'},
            encode_kwargs={'normalize_embeddings': True}
        )

        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

        self.vector_store = None
        self.knowledge_graph = KnowledgeGraphBuilder()
        self.document_registry = {}

    def _compute_doc_id(self, content: str) -> str:
        """Generate unique document identifier."""
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def ingest_documents(self, documents: List[dict]):
        """
        Ingest documents into both vector store and knowledge graph.

        Args:
            documents: List of dicts with 'content', 'metadata', and optional 'entities'
        """
        all_chunks = []

        for doc in documents:
            doc_id = self._compute_doc_id(doc['content'])
            self.document_registry[doc_id] = doc

            # Split into chunks for vector store
            chunks = self.text_splitter.split_text(doc['content'])
            for i, chunk in enumerate(chunks):
                chunk_doc = Document(
                    page_content=chunk,
                    metadata={
                        **doc.get('metadata', {}),
                        'doc_id': doc_id,
                        'chunk_index': i
                    }
                )
                all_chunks.append(chunk_doc)

            # Extract entities for knowledge graph
            if 'entities' in doc:
                for entity in doc['entities']:
                    self.knowledge_graph.add_entity(
                        entity['name'],
                        entity['type'],
                        entity.get('properties', {})
                    )

                    if 'relations' in entity:
                        for rel in entity['relations']:
                            self.knowledge_graph.add_relation(
                                entity['name'],
                                rel['predicate'],
                                rel['object']
                            )

        # Build vector store
        if self.vector_store is None:
            self.vector_store = FAISS.from_documents(all_chunks, self.embeddings)
        else:
            self.vector_store.add_documents(all_chunks)

        print(f"Ingested {len(documents)} documents ({len(all_chunks)} chunks)")

    def hybrid_retrieve(
        self,
        query: str,
        vector_k: int = 5,
        graph_depth: int = 2
    ) -> dict:
        """
        Perform hybrid retrieval combining vector similarity and graph traversal.
        """
        # Vector similarity search
        vector_results = self.vector_store.similarity_search_with_score(
            query, k=vector_k
        )

        # Extract entities from query for graph search
        # In production, use NER or LLM for entity extraction
        query_terms = query.lower().split()

        graph_results = []
        for term in query_terms:
            sparql = f"""
            SELECT ?entity ?label ?type WHERE {{
                ?entity rdfs:label ?label .
                ?entity rdf:type ?type .
                FILTER(CONTAINS(LCASE(?label), "{term}"))
            }} LIMIT 10
            """
            results = self.knowledge_graph.query(sparql)
            graph_results.extend(results)

        return {
            "vector_results": [
                {"content": doc.page_content, "score": float(score), "metadata": doc.metadata}
                for doc, score in vector_results
            ],
            "graph_results": [
                {"entity": str(r[0]), "label": str(r[1]), "type": str(r[2])}
                for r in graph_results
            ]
        }

    def generate_response(
        self,
        query: str,
        llm,
        validate: bool = True
    ) -> dict:
        """
        Generate a knowledge-grounded response.
        """
        # Retrieve relevant context
        retrieval = self.hybrid_retrieve(query)

        # Assemble context
        vector_context = "\n\n".join([
            f"[Source {i+1}]: {r['content']}"
            for i, r in enumerate(retrieval['vector_results'][:3])
        ])

        graph_context = ""
        if retrieval['graph_results']:
            graph_context = "\n\nRelated Entities:\n" + "\n".join([
                f"- {r['label']} ({r['type'].split('#')[-1]})"
                for r in retrieval['graph_results'][:5]
            ])

        # Generate response
        prompt = f"""Based on the following knowledge context, answer the question accurately.
If the context doesn't contain enough information, acknowledge the limitation.

Knowledge Context:
{vector_context}
{graph_context}

Question: {query}

Answer:"""

        response = llm.generate(prompt)

        result = {
            "query": query,
            "response": response,
            "sources": retrieval['vector_results'],
            "related_entities": retrieval['graph_results']
        }

        # Optional fact validation
        if validate and retrieval['graph_results']:
            result["validation"] = self._validate_response(response, retrieval)

        return result

    def _validate_response(self, response: str, retrieval: dict) -> dict:
        """
        Validate response claims against knowledge graph.
        """
        # In production, extract claims and verify each
        # Simplified validation here
        entity_labels = {r['label'].lower() for r in retrieval['graph_results']}
        response_lower = response.lower()

        mentioned_entities = [e for e in entity_labels if e in response_lower]

        return {
            "entities_grounded": len(mentioned_entities),
            "total_retrieved_entities": len(entity_labels),
            "grounding_ratio": len(mentioned_entities) / max(len(entity_labels), 1)
        }

from dataclasses import dataclass
from typing import List, Callable
import numpy as np

@dataclass
class EvaluationMetric:
    name: str
    compute: Callable
    threshold: float

class KnowledgeSystemEvaluator:
    """
    Comprehensive evaluation framework for knowledge-augmented LLM systems.
    """

    def __init__(self):
        self.metrics = [
            EvaluationMetric(
                name="retrieval_precision",
                compute=self._compute_retrieval_precision,
                threshold=0.7
            ),
            EvaluationMetric(
                name="answer_relevance",
                compute=self._compute_answer_relevance,
                threshold=0.75
            ),
            EvaluationMetric(
                name="factual_consistency",
                compute=self._compute_factual_consistency,
                threshold=0.85
            ),
            EvaluationMetric(
                name="hallucination_rate",
                compute=self._compute_hallucination_rate,
                threshold=0.1  # Lower is better
            )
        ]

    def _compute_retrieval_precision(
        self,
        retrieved_docs: List[str],
        relevant_docs: List[str]
    ) -> float:
        """Precision of retrieved documents."""
        if not retrieved_docs:
            return 0.0
        relevant_retrieved = set(retrieved_docs) & set(relevant_docs)
        return len(relevant_retrieved) / len(retrieved_docs)

    def _compute_answer_relevance(
        self,
        query: str,
        response: str,
        embedder
    ) -> float:
        """Semantic similarity between query and response."""
        query_emb = embedder.encode([query])
        response_emb = embedder.encode([response])
        from sklearn.metrics.pairwise import cosine_similarity
        return float(cosine_similarity(query_emb, response_emb)[0][0])

    def _compute_factual_consistency(
        self,
        response: str,
        source_docs: List[str],
        nli_model
    ) -> float:
        """
        Check if response is entailed by source documents.
        Uses Natural Language Inference.
        """
        consistency_scores = []
        for doc in source_docs:
            # NLI: premise=doc, hypothesis=response
            score = nli_model.predict(premise=doc, hypothesis=response)
            consistency_scores.append(score['entailment'])
        return np.mean(consistency_scores) if consistency_scores else 0.0

    def _compute_hallucination_rate(
        self,
        response: str,
        source_docs: List[str],
        claim_extractor
    ) -> float:
        """
        Detect claims in response not supported by sources.
        """
        claims = claim_extractor.extract(response)
        unsupported = 0

        for claim in claims:
            if not any(claim.lower() in doc.lower() for doc in source_docs):
                unsupported += 1

        return unsupported / max(len(claims), 1)

    def evaluate(self, test_cases: List[dict], system, embedder, nli_model, claim_extractor) -> dict:
        """
        Run comprehensive evaluation on test cases.
        """
        results = {metric.name: [] for metric in self.metrics}

        for case in test_cases:
            response = system.generate_response(case['query'], case.get('llm'))

            # Compute each metric
            results['retrieval_precision'].append(
                self._compute_retrieval_precision(
                    [r['content'] for r in response['sources']],
                    case.get('relevant_docs', [])
                )
            )

            results['answer_relevance'].append(
                self._compute_answer_relevance(
                    case['query'],
                    response['response'],
                    embedder
                )
            )

            # Additional metrics computed similarly...

        # Aggregate results
        summary = {}
        for metric in self.metrics:
            scores = results[metric.name]
            avg_score = np.mean(scores) if scores else 0.0
            summary[metric.name] = {
                'score': avg_score,
                'threshold': metric.threshold,
                'passed': avg_score >= metric.threshold if metric.name != 'hallucination_rate'
                         else avg_score <= metric.threshold
            }

        return summary

from functools import lru_cache
from datetime import datetime, timedelta
import asyncio

class ProductionKnowledgeService:
    """
    Production-ready knowledge service with caching and monitoring.
    """

    def __init__(self, rag_system: KnowledgeRAGSystem, cache_ttl: int = 3600):
        self.rag = rag_system
        self.cache_ttl = cache_ttl
        self.query_cache = {}
        self.metrics = {
            'total_queries': 0,
            'cache_hits': 0,
            'avg_latency_ms': 0,
            'error_count': 0
        }

    def _cache_key(self, query: str) -> str:
        """Generate cache key for query."""
        return hashlib.sha256(query.lower().strip().encode()).hexdigest()

    def _is_cache_valid(self, cached_entry: dict) -> bool:
        """Check if cache entry is still valid."""
        if not cached_entry:
            return False
        cached_time = cached_entry.get('timestamp')
        if not cached_time:
            return False
        return datetime.now() - cached_time < timedelta(seconds=self.cache_ttl)

    async def query(self, query: str, bypass_cache: bool = False) -> dict:
        """
        Handle query with caching and monitoring.
        """
        start_time = datetime.now()
        self.metrics['total_queries'] += 1

        cache_key = self._cache_key(query)

        # Check cache
        if not bypass_cache and cache_key in self.query_cache:
            if self._is_cache_valid(self.query_cache[cache_key]):
                self.metrics['cache_hits'] += 1
                return self.query_cache[cache_key]['response']

        try:
            # Execute query
            response = await asyncio.to_thread(
                self.rag.generate_response,
                query,
                self.llm
            )

            # Update cache
            self.query_cache[cache_key] = {
                'response': response,
                'timestamp': datetime.now()
            }

            # Update latency metrics
            latency = (datetime.now() - start_time).total_seconds() * 1000
            n = self.metrics['total_queries']
            self.metrics['avg_latency_ms'] = (
                (self.metrics['avg_latency_ms'] * (n-1) + latency) / n
            )

            return response

        except Exception as e:
            self.metrics['error_count'] += 1
            raise

    def get_metrics(self) -> dict:
        """Return service metrics for monitoring."""
        return {
            **self.metrics,
            'cache_hit_rate': self.metrics['cache_hits'] / max(self.metrics['total_queries'], 1),
            'error_rate': self.metrics['error_count'] / max(self.metrics['total_queries'], 1)
        }

    async def refresh_knowledge(self, documents: List[dict]):
        """
        Incrementally update knowledge base.
        """
        await asyncio.to_thread(self.rag.ingest_documents, documents)
        # Invalidate affected cache entries
        self.query_cache.clear()