Agentic Architecture Pattern: Symbolic Neural Hybrid

Written by

in


    """
Pattern 151: Symbolic-Neural Hybrid Pattern

Combines symbolic reasoning with neural networks for robust AI systems.
Integrates rule-based logic with LLM capabilities using LangChain.
"""

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


class ReasoningMode(Enum):
    """Mode of reasoning."""
    SYMBOLIC = "symbolic"
    NEURAL = "neural"
    HYBRID = "hybrid"


@dataclass
class Rule:
    """Symbolic rule for logical reasoning."""
    rule_id: str
    name: str
    conditions: List[str]  # Logical conditions
    conclusion: str
    confidence: float = 1.0


@dataclass
class Fact:
    """Fact in the knowledge base."""
    fact_id: str
    predicate: str
    arguments: List[str]
    confidence: float = 1.0
    source: str = "user"  # user, symbolic, neural, hybrid


@dataclass
class InferenceResult:
    """Result of inference."""
    conclusion: str
    mode: ReasoningMode
    confidence: float
    explanation: str
    supporting_facts: List[Fact] = field(default_factory=list)
    applied_rules: List[Rule] = field(default_factory=list)


class SymbolicReasoner:
    """Symbolic reasoning engine using rules."""
    
    def __init__(self):
        self.rules: Dict[str, Rule] = {}
        self.facts: Dict[str, Fact] = {}
        
    def add_rule(self, rule: Rule):
        """Add a reasoning rule."""
        self.rules[rule.rule_id] = rule
        
    def add_fact(self, fact: Fact):
        """Add a fact to knowledge base."""
        self.facts[fact.fact_id] = fact
        
    def infer(self, query: str) -> List[InferenceResult]:
        """Perform symbolic inference."""
        results = []
        
        # Find applicable rules
        for rule in self.rules.values():
            # Check if all conditions are satisfied
            satisfied = True
            supporting_facts = []
            
            for condition in rule.conditions:
                # Find matching facts
                matching = [f for f in self.facts.values() if condition in f.predicate]
                if matching:
                    supporting_facts.extend(matching)
                else:
                    satisfied = False
                    break
            
            if satisfied:
                # Calculate confidence
                if supporting_facts:
                    confidence = min(f.confidence for f in supporting_facts) * rule.confidence
                else:
                    confidence = rule.confidence
                
                result = InferenceResult(
                    conclusion=rule.conclusion,
                    mode=ReasoningMode.SYMBOLIC,
                    confidence=confidence,
                    explanation=f"Applied rule '{rule.name}' with {len(supporting_facts)} supporting facts",
                    supporting_facts=supporting_facts,
                    applied_rules=[rule]
                )
                results.append(result)
        
        return results


class NeuralReasoner:
    """Neural reasoning using LLM."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def infer(self, query: str, context: Dict[str, Any]) -> InferenceResult:
        """Perform neural inference using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a reasoning AI. Analyze the query and context to draw conclusions."),
            ("human", """Query: {query}

Context:
{context}

Provide:
1. Your conclusion
2. Confidence (0.0 to 1.0)
3. Brief explanation

Format: CONCLUSION|confidence|explanation""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "query": query,
            "context": "\n".join([f"{k}: {v}" for k, v in context.items()])
        })
        
        # Parse response
        parts = str(response.content).split('|')
        conclusion = parts[0].strip() if len(parts) > 0 else "Unknown"
        confidence = float(parts[1].strip()) if len(parts) > 1 else 0.5
        explanation = parts[2].strip() if len(parts) > 2 else "LLM reasoning"
        
        return InferenceResult(
            conclusion=conclusion,
            mode=ReasoningMode.NEURAL,
            confidence=confidence,
            explanation=explanation
        )


class SymbolicNeuralHybrid:
    """
    Hybrid system combining symbolic and neural reasoning.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.symbolic = SymbolicReasoner()
        self.neural = NeuralReasoner(model_name=model_name)
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        
    def reason(
        self,
        query: str,
        mode: ReasoningMode = ReasoningMode.HYBRID
    ) -> List[InferenceResult]:
        """Perform reasoning using specified mode."""
        if mode == ReasoningMode.SYMBOLIC:
            return self.symbolic.infer(query)
        
        elif mode == ReasoningMode.NEURAL:
            context = self._build_context()
            return [self.neural.infer(query, context)]
        
        elif mode == ReasoningMode.HYBRID:
            return self._hybrid_reasoning(query)
        
        return []
    
    def _hybrid_reasoning(self, query: str) -> List[InferenceResult]:
        """Combine symbolic and neural reasoning."""
        # First try symbolic reasoning
        symbolic_results = self.symbolic.infer(query)
        
        # Get neural reasoning
        context = self._build_context()
        context["query"] = query
        neural_result = self.neural.infer(query, context)
        
        # Combine results
        if symbolic_results:
            # Use LLM to synthesize both approaches
            synthesized = self._synthesize_results(query, symbolic_results, neural_result)
            return [synthesized] + symbolic_results + [neural_result]
        else:
            # Only neural result available
            return [neural_result]
    
    def _synthesize_results(
        self,
        query: str,
        symbolic_results: List[InferenceResult],
        neural_result: InferenceResult
    ) -> InferenceResult:
        """Synthesize symbolic and neural results."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a reasoning synthesizer. Combine symbolic and neural reasoning results."),
            ("human", """Query: {query}

Symbolic Reasoning:
{symbolic}

Neural Reasoning:
{neural}

Synthesize a final conclusion that combines both approaches.
Format: CONCLUSION|confidence|explanation""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "query": query,
            "symbolic": "\n".join([f"- {r.conclusion} (confidence: {r.confidence})" for r in symbolic_results]),
            "neural": f"{neural_result.conclusion} (confidence: {neural_result.confidence})"
        })
        
        # Parse response
        parts = str(response.content).split('|')
        conclusion = parts[0].strip() if len(parts) > 0 else "Hybrid conclusion"
        confidence = float(parts[1].strip()) if len(parts) > 1 else 0.7
        explanation = parts[2].strip() if len(parts) > 2 else "Combined symbolic and neural reasoning"
        
        return InferenceResult(
            conclusion=conclusion,
            mode=ReasoningMode.HYBRID,
            confidence=confidence,
            explanation=explanation,
            supporting_facts=symbolic_results[0].supporting_facts if symbolic_results else [],
            applied_rules=symbolic_results[0].applied_rules if symbolic_results else []
        )
    
    def _build_context(self) -> Dict[str, Any]:
        """Build context from knowledge base."""
        context = {}
        
        # Add facts
        if self.symbolic.facts:
            context["facts"] = [
                f"{f.predicate}({', '.join(f.arguments)})"
                for f in self.symbolic.facts.values()
            ]
        
        # Add rules
        if self.symbolic.rules:
            context["rules"] = [
                f"{r.name}: IF {' AND '.join(r.conditions)} THEN {r.conclusion}"
                for r in self.symbolic.rules.values()
            ]
        
        return context


class SymbolicNeuralHybridPattern:
    """Complete hybrid pattern implementation."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.hybrid = SymbolicNeuralHybrid(model_name=model_name)
        
    def run(
        self,
        query: str,
        mode: ReasoningMode = ReasoningMode.HYBRID
    ) -> Dict[str, Any]:
        """Run hybrid reasoning."""
        results = self.hybrid.reason(query, mode)
        
        return {
            "query": query,
            "mode": mode.value,
            "results": [
                {
                    "conclusion": r.conclusion,
                    "confidence": r.confidence,
                    "mode": r.mode.value,
                    "explanation": r.explanation,
                    "supporting_facts_count": len(r.supporting_facts),
                    "applied_rules_count": len(r.applied_rules)
                }
                for r in results
            ]
        }


def main():
    """Example usage of Symbolic-Neural Hybrid Pattern."""
    print("=" * 80)
    print("Symbolic-Neural Hybrid Pattern Demo")
    print("=" * 80)
    
    # Create hybrid pattern
    pattern = SymbolicNeuralHybridPattern()
    hybrid = pattern.hybrid
    
    # Add symbolic rules
    hybrid.symbolic.add_rule(Rule(
        rule_id="rule-1",
        name="Mammal Classification",
        conditions=["has_fur", "warm_blooded", "feeds_milk"],
        conclusion="is_mammal",
        confidence=0.95
    ))
    
    hybrid.symbolic.add_rule(Rule(
        rule_id="rule-2",
        name="Bird Classification",
        conditions=["has_feathers", "has_wings", "lays_eggs"],
        conclusion="is_bird",
        confidence=0.95
    ))
    
    hybrid.symbolic.add_rule(Rule(
        rule_id="rule-3",
        name="Carnivore Classification",
        conditions=["eats_meat", "has_sharp_teeth"],
        conclusion="is_carnivore",
        confidence=0.9
    ))
    
    # Add facts
    hybrid.symbolic.add_fact(Fact(
        fact_id="f1",
        predicate="has_fur",
        arguments=["dog"],
        confidence=1.0
    ))
    
    hybrid.symbolic.add_fact(Fact(
        fact_id="f2",
        predicate="warm_blooded",
        arguments=["dog"],
        confidence=1.0
    ))
    
    hybrid.symbolic.add_fact(Fact(
        fact_id="f3",
        predicate="feeds_milk",
        arguments=["dog"],
        confidence=1.0
    ))
    
    hybrid.symbolic.add_fact(Fact(
        fact_id="f4",
        predicate="eats_meat",
        arguments=["dog"],
        confidence=0.8
    ))
    
    # Test different reasoning modes
    queries = [
        "What type of animal is a dog?",
        "Is a dog a carnivore?",
        "What can we infer about a dog's diet?"
    ]
    
    modes = [ReasoningMode.SYMBOLIC, ReasoningMode.NEURAL, ReasoningMode.HYBRID]
    
    for query in queries:
        print(f"\n{'=' * 80}")
        print(f"šŸ” Query: {query}")
        print('=' * 80)
        
        for mode in modes:
            print(f"\nšŸ“Š Mode: {mode.value.upper()}")
            print('-' * 80)
            
            result = pattern.run(query, mode)
            
            for i, res in enumerate(result["results"], 1):
                print(f"\n  Result {i}:")
                print(f"    Conclusion: {res['conclusion']}")
                print(f"    Confidence: {res['confidence']:.2f}")
                print(f"    Mode: {res['mode']}")
                print(f"    Explanation: {res['explanation']}")
                if res['supporting_facts_count'] > 0:
                    print(f"    Supporting Facts: {res['supporting_facts_count']}")
                if res['applied_rules_count'] > 0:
                    print(f"    Applied Rules: {res['applied_rules_count']}")


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Symbolic Neural Hybrid

Pattern 151: 151 Symbolic Neural Hybrid Pattern – Line-by-Line Explanation

Overview

Pattern 151: Symbolic Neural Hybrid Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration


Code Structure Analysis

File Information

  • Pattern Number: 151
  • Pattern Name: 151 Symbolic Neural Hybrid Pattern
  • File Path: patterns\hybrid_ai\151_symbolic_neural_hybrid_pattern.py
  • Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 151: Symbolic Neural Hybrid Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 151 Symbolic Neural Hybrid Pattern pattern.


Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


Classes

SymbolicNeuralHybridConfig (Line 18)

Description: Configuration for Symbolic Neural Hybrid Pattern.

SymbolicNeuralHybridState (Line 27)

Description: State information for Symbolic Neural Hybrid Pattern.

SymbolicNeuralHybridAgent (Line 36)

Description: Agent implementing Symbolic Neural Hybrid Pattern.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)
def process(self, input_data):
    """Process input according to Symbolic Neural Hybrid Pattern."""

Purpose: Process input according to Symbolic Neural Hybrid Pattern.

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

get_statistics (Line 75)
def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

SymbolicNeuralHybridPattern (Line 85)

Description: Symbolic Neural Hybrid Pattern – Enhanced implementation

This pattern provides comprehensive functionality for symbolic neural hybrid with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)
def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)
def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)
def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)
def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)
def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)
def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.


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 = SymbolicNeuralHybridConfig(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 *