aishastr.live

Agentic Architecture Pattern: Knowledge Base

Written by

admin

in


    """
Pattern 66: Knowledge Base Pattern

Agents maintain structured knowledge about the world, rules, facts, and relationships.
Supports querying, inference, and reasoning over stored knowledge.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class FactType(Enum):
    """Type of knowledge fact."""
    ENTITY = "entity"
    RELATION = "relation"
    RULE = "rule"
    CONCEPT = "concept"
    PROCEDURE = "procedure"


class ConfidenceLevel(Enum):
    """Confidence in knowledge."""
    CERTAIN = 1.0
    HIGH = 0.8
    MEDIUM = 0.5
    LOW = 0.3
    UNCERTAIN = 0.1


@dataclass
class Fact:
    """Knowledge fact with metadata."""
    fact_id: str
    fact_type: FactType
    subject: str
    predicate: str
    object: str
    confidence: float = 1.0
    source: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)
    tags: List[str] = field(default_factory=list)


@dataclass
class Rule:
    """Logical rule for inference."""
    rule_id: str
    name: str
    conditions: List[str]  # IF conditions
    conclusion: str  # THEN conclusion
    confidence: float = 1.0
    description: Optional[str] = None


@dataclass
class Query:
    """Knowledge base query."""
    query_id: str
    question: str
    query_type: str  # "fact", "inference", "explanation"
    results: List[Fact] = field(default_factory=list)
    inferred_facts: List[Fact] = field(default_factory=list)
    explanation: Optional[str] = None


class KnowledgeBase:
    """Structured knowledge storage with reasoning."""
    
    def __init__(self, name: str):
        self.name = name
        self.facts: Dict[str, Fact] = {}
        self.rules: Dict[str, Rule] = {}
        self.indexes: Dict[str, Set[str]] = {
            "subject": {},
            "predicate": {},
            "object": {}
        }
    
    def add_fact(self, fact: Fact):
        """Add fact to knowledge base."""
        self.facts[fact.fact_id] = fact
        
        # Update indexes
        for index_type in ["subject", "predicate", "object"]:
            value = getattr(fact, index_type)
            if value not in self.indexes[index_type]:
                self.indexes[index_type][value] = set()
            self.indexes[index_type][value].add(fact.fact_id)
    
    def add_rule(self, rule: Rule):
        """Add inference rule."""
        self.rules[rule.rule_id] = rule
    
    def query_facts(self, subject: Optional[str] = None, 
                   predicate: Optional[str] = None,
                   object: Optional[str] = None) -> List[Fact]:
        """Query facts by subject, predicate, or object."""
        result_ids = None
        
        if subject:
            result_ids = self.indexes["subject"].get(subject, set())
        if predicate:
            pred_ids = self.indexes["predicate"].get(predicate, set())
            result_ids = result_ids & pred_ids if result_ids else pred_ids
        if object:
            obj_ids = self.indexes["object"].get(object, set())
            result_ids = result_ids & obj_ids if result_ids else obj_ids
        
        if result_ids is None:
            return list(self.facts.values())
        
        return [self.facts[fid] for fid in result_ids if fid in self.facts]
    
    def apply_rules(self) -> List[Fact]:
        """Apply inference rules to derive new facts."""
        inferred = []
        
        for rule in self.rules.values():
            # Simple rule matching (can be extended)
            conditions_met = True
            
            for condition in rule.conditions:
                # Parse condition: "subject predicate object"
                parts = condition.split()
                if len(parts) >= 3:
                    matching_facts = self.query_facts(
                        subject=parts[0] if parts[0] != "?" else None,
                        predicate=parts[1] if parts[1] != "?" else None,
                        object=parts[2] if parts[2] != "?" else None
                    )
                    if not matching_facts:
                        conditions_met = False
                        break
            
            if conditions_met:
                # Create inferred fact from conclusion
                parts = rule.conclusion.split()
                if len(parts) >= 3:
                    inferred_fact = Fact(
                        fact_id=f"inferred_{rule.rule_id}_{len(inferred)}",
                        fact_type=FactType.RELATION,
                        subject=parts[0],
                        predicate=parts[1],
                        object=parts[2],
                        confidence=rule.confidence,
                        source=f"rule:{rule.rule_id}"
                    )
                    inferred.append(inferred_fact)
        
        return inferred


class KnowledgeAgent:
    """Agent with knowledge base and reasoning capabilities."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.kb = KnowledgeBase(f"{name}_kb")
        self.queries: List[Query] = []
    
    def learn_fact(self, subject: str, predicate: str, object: str, 
                   fact_type: FactType = FactType.RELATION,
                   confidence: float = 1.0):
        """Learn new fact."""
        fact = Fact(
            fact_id=f"fact_{len(self.kb.facts)}",
            fact_type=fact_type,
            subject=subject,
            predicate=predicate,
            object=object,
            confidence=confidence,
            source=self.name
        )
        self.kb.add_fact(fact)
        print(f"   šŸ“š {self.name} learned: {subject} {predicate} {object}")
    
    def answer_query(self, question: str) -> Query:
        """Answer question using knowledge base and LLM."""
        # First, try direct fact lookup
        facts_text = "\n".join([
            f"- {f.subject} {f.predicate} {f.object} (confidence: {f.confidence:.2f})"
            for f in self.kb.facts.values()
        ])
        
        # Apply inference rules
        inferred = self.kb.apply_rules()
        inferred_text = "\n".join([
            f"- {f.subject} {f.predicate} {f.object} (inferred, confidence: {f.confidence:.2f})"
            for f in inferred
        ]) if inferred else "None"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, an AI agent with a knowledge base. "
                      f"Answer questions based on your knowledge and reasoning."),
            ("human", f"Known facts:\n{facts_text}\n\n"
                     f"Inferred facts:\n{inferred_text}\n\n"
                     f"Question: {question}\n\n"
                     f"Provide a clear answer based on the facts and inferences. "
                     f"If you need to infer something, explain your reasoning.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        query = Query(
            query_id=f"q{len(self.queries)}",
            question=question,
            query_type="inference",
            inferred_facts=inferred,
            explanation=str(response.content)
        )
        
        self.queries.append(query)
        return query


class KnowledgeBasePattern:
    """
    Knowledge Base Pattern - Structured knowledge with reasoning.
    
    Agents maintain facts, rules, and relationships, enabling them to
    answer questions through direct lookup and logical inference.
    """
    
    def __init__(self):
        self.agents: Dict[str, KnowledgeAgent] = {}
    
    def register_agent(self, agent: KnowledgeAgent):
        """Register agent with knowledge base."""
        self.agents[agent.name] = agent
    
    def share_knowledge(self, source_agent: str, target_agent: str, fact: Fact):
        """Share knowledge between agents."""
        if source_agent in self.agents and target_agent in self.agents:
            self.agents[target_agent].kb.add_fact(fact)
    
    def run(self, domain: str) -> Dict[str, Any]:
        """Execute knowledge base pattern."""
        total_facts = sum(len(agent.kb.facts) for agent in self.agents.values())
        total_rules = sum(len(agent.kb.rules) for agent in self.agents.values())
        total_queries = sum(len(agent.queries) for agent in self.agents.values())
        
        return {
            "pattern": "KnowledgeBase",
            "domain": domain,
            "total_agents": len(self.agents),
            "total_facts": total_facts,
            "total_rules": total_rules,
            "total_queries": total_queries
        }


# Example usage
if __name__ == "__main__":
    print("=== Knowledge Base Pattern Demo ===\n")
    
    # Create pattern
    pattern = KnowledgeBasePattern()
    
    # Create agent with knowledge base
    print("1. Creating agent with knowledge base...")
    agent = KnowledgeAgent("Scholar")
    pattern.register_agent(agent)
    
    # Learn facts about programming languages
    print("\n2. Learning facts about programming...")
    agent.learn_fact("Python", "is_a", "programming_language", FactType.CONCEPT)
    agent.learn_fact("Python", "has_feature", "dynamic_typing", FactType.RELATION)
    agent.learn_fact("Python", "used_for", "data_science", FactType.RELATION)
    agent.learn_fact("Python", "used_for", "web_development", FactType.RELATION)
    agent.learn_fact("JavaScript", "is_a", "programming_language", FactType.CONCEPT)
    agent.learn_fact("JavaScript", "used_for", "web_development", FactType.RELATION)
    agent.learn_fact("TypeScript", "extends", "JavaScript", FactType.RELATION)
    agent.learn_fact("TypeScript", "has_feature", "static_typing", FactType.RELATION)
    
    # Add inference rules
    print("\n3. Adding inference rules...")
    rule1 = Rule(
        rule_id="rule1",
        name="Inheritance",
        conditions=["TypeScript extends JavaScript", "JavaScript used_for web_development"],
        conclusion="TypeScript used_for web_development",
        confidence=0.95,
        description="If A extends B, and B is used for X, then A is used for X"
    )
    agent.kb.add_rule(rule1)
    print(f"   āš™ļø Added rule: {rule1.name}")
    
    # Query knowledge base
    print("\n4. Querying knowledge base...")
    
    # Direct queries
    print("\n   Direct fact lookup:")
    python_facts = agent.kb.query_facts(subject="Python")
    for fact in python_facts:
        print(f"   - {fact.subject} {fact.predicate} {fact.object}")
    
    web_dev_facts = agent.kb.query_facts(predicate="used_for", object="web_development")
    print(f"\n   Languages used for web development: {len(web_dev_facts)}")
    for fact in web_dev_facts:
        print(f"   - {fact.subject}")
    
    # Natural language queries
    print("\n5. Answering natural language questions...")
    questions = [
        "What is Python used for?",
        "Which programming languages support web development?",
        "What are the differences between Python and TypeScript?",
        "Can TypeScript be used for web development?"
    ]
    
    for question in questions:
        print(f"\n   Q: {question}")
        query = agent.answer_query(question)
        print(f"   A: {query.explanation[:200]}...")
        if query.inferred_facts:
            print(f"   Inferred: {len(query.inferred_facts)} new facts")
    
    # Knowledge statistics
    print("\n6. Knowledge Base Statistics:")
    summary = pattern.run("Programming Languages")
    print(f"   Total Facts: {summary['total_facts']}")
    print(f"   Total Rules: {summary['total_rules']}")
    print(f"   Total Queries: {summary['total_queries']}")
    
    # Show knowledge graph structure
    print("\n7. Knowledge Graph:")
    subjects = set(f.subject for f in agent.kb.facts.values())
    predicates = set(f.predicate for f in agent.kb.facts.values())
    print(f"   Entities: {len(subjects)}")
    print(f"   Relation types: {len(predicates)}")
    print(f"   Predicates: {', '.join(sorted(predicates))}")
    
    print("\nāœ“ Knowledge base pattern enabled structured reasoning!")

Agentic Architecture Pattern: Knowledge Base

Pattern 66: 66 Knowledge Base – Line-by-Line Explanation

Overview

Pattern 66: Knowledge Base Pattern

Agents maintain structured knowledge about the world, rules, facts, and relationships. Supports querying, inference, and reasoning over stored knowledge.

Code Structure Analysis

File Information

  • Pattern Number: 66
  • Pattern Name: 66 Knowledge Base
  • File Path: patterns\memory_knowledge\66_knowledge_base.py
  • Total Lines: 340

Detailed Code Explanation

Module Docstring

"""
Pattern 66: Knowledge Base Pattern

Agents maintain structured knowledge about the world, rules, facts, and relationships.
Supports querying, inference, and reasoning over stored knowledge.
"""

Explanation: This module docstring describes the purpose and key features of the 66 Knowledge Base pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

FactType (Line 16)

class FactType(Enum):
    ENTITY
    RELATION
    RULE
    CONCEPT
    PROCEDURE

Purpose: Type of knowledge fact.

Values: – ENTITY: Represents entity – RELATION: Represents relation – RULE: Represents rule – CONCEPT: Represents concept – PROCEDURE: Represents procedure

ConfidenceLevel (Line 25)

class ConfidenceLevel(Enum):
    CERTAIN
    HIGH
    MEDIUM
    LOW
    UNCERTAIN

Purpose: Confidence in knowledge.

Values: – CERTAIN: Represents certain – HIGH: Represents high – MEDIUM: Represents medium – LOW: Represents low – UNCERTAIN: Represents uncertain

Classes

FactType (Line 16)

Description: Type of knowledge fact.

ConfidenceLevel (Line 25)

Description: Confidence in knowledge.

Fact (Line 35)

Description: Knowledge fact with metadata.

Rule (Line 49)

Description: Logical rule for inference.

Query (Line 60)

Description: Knowledge base query.

KnowledgeBase (Line 70)

Description: Structured knowledge storage with reasoning.

Methods:

__init__ (Line 73)
def __init__(self, name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

add_fact (Line 83)
def add_fact(self, fact):
    """Add fact to knowledge base."""

Purpose: Add fact to knowledge base.

Parameters: – self: Instance reference – fact: Parameter for fact

add_rule (Line 94)
def add_rule(self, rule):
    """Add inference rule."""

Purpose: Add inference rule.

Parameters: – self: Instance reference – rule: Parameter for rule

query_facts (Line 98)
def query_facts(self, subject, predicate, object):
    """Query facts by subject, predicate, or object."""

Purpose: Query facts by subject, predicate, or object.

Parameters: – self: Instance reference – subject: Parameter for subject – predicate: Parameter for predicate – object: Parameter for object

apply_rules (Line 118)
def apply_rules(self):
    """Apply inference rules to derive new facts."""

Purpose: Apply inference rules to derive new facts.

Parameters: – self: Instance reference

KnowledgeAgent (Line 157)

Description: Agent with knowledge base and reasoning capabilities.

Methods:

__init__ (Line 160)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

learn_fact (Line 166)
def learn_fact(self, subject, predicate, object, fact_type, confidence):
    """Learn new fact."""

Purpose: Learn new fact.

Parameters: – self: Instance reference – subject: Parameter for subject – predicate: Parameter for predicate – object: Parameter for object – fact_type: Parameter for fact_type – confidence: Parameter for confidence

answer_query (Line 182)
def answer_query(self, question):
    """Answer question using knowledge base and LLM."""

Purpose: Answer question using knowledge base and LLM.

Parameters: – self: Instance reference – question: Parameter for question

KnowledgeBasePattern (Line 222)

Description: Knowledge Base Pattern – Structured knowledge with reasoning.

Agents maintain facts, rules, and relationships, enabling them to answer questions through direct lookup and logical inference.

Methods:

__init__ (Line 230)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 233)
def register_agent(self, agent):
    """Register agent with knowledge base."""

Purpose: Register agent with knowledge base.

Parameters: – self: Instance reference – agent: Parameter for agent

share_knowledge (Line 237)
def share_knowledge(self, source_agent, target_agent, fact):
    """Share knowledge between agents."""

Purpose: Share knowledge between agents.

Parameters: – self: Instance reference – source_agent: Parameter for source_agent – target_agent: Parameter for target_agent – fact: Parameter for fact

run (Line 242)
def run(self, domain):
    """Execute knowledge base pattern."""

Purpose: Execute knowledge base pattern.

Parameters: – self: Instance reference – domain: Parameter for domain

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = FactType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts