Building Neuro-Symbolic Projects - Neuro-Symbolic AI Tutorial

Your First Neuro-Symbolic Project

This chapter walks you through building a complete neuro-symbolic AI system from scratch. We'll create a Medical Triage System that safely processes patient input and recommends appropriate actions.

What We're Building: A system that uses neural networks to understand patient symptoms and symbolic rules to make safe clinical recommendations.

Project Architecture Overview

Step 1: Set Up Your Project Structure

Create the following folder structure:

medical-triage-system/
├── README.md
├── requirements.txt
├── src/
│   ├── __init__.py
│   ├── extraction.py      # Neural NLU
│   ├── knowledge_graph.py  # KG definition
│   ├── rules.py           # Inference rules
│   ├── triage.py          # Main logic
│   └── utils.py           # Helpers
├── data/
│   ├── symptoms.json      # Symptom definitions
│   ├── red_flags.json     # Critical symptoms
│   └── test_cases.json    # Test scenarios
├── tests/
│   ├── test_extraction.py
│   ├── test_rules.py
│   └── test_integration.py
└── notebooks/
    └── analysis.ipynb     # Analysis & benchmarking

Step 2: Define Your Knowledge Graph

Create `src/knowledge_graph.py` - Define entities, relationships, and rules

import networkx as nx
import json

class MedicalKnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.rules = []
        self._build_graph()

    def _build_graph(self):
        """Build basic medical KG"""
        # Add nodes (entities)
        diseases = ['acute_coronary_syndrome', 'pneumonia', 'anxiety', 'asthma']
        symptoms = ['chest_pain', 'shortness_of_breath', 'sweating', 'cough']

        for disease in diseases:
            self.graph.add_node(disease, type='disease')
        for symptom in symptoms:
            self.graph.add_node(symptom, type='symptom')

        # Add relationships
        edges = [
            ('chest_pain', 'acute_coronary_syndrome', 'manifests_as'),
            ('shortness_of_breath', 'acute_coronary_syndrome', 'manifests_as'),
            ('sweating', 'acute_coronary_syndrome', 'manifests_as'),
            ('cough', 'pneumonia', 'manifests_as'),
            ('shortness_of_breath', 'pneumonia', 'manifests_as'),
            ('chest_pain', 'anxiety', 'manifests_as'),
        ]

        for src, tgt, rel in edges:
            self.graph.add_edge(src, tgt, relation=rel)

    def add_rule(self, name, condition_fn, consequence_fn):
        """Add inference rule"""
        self.rules.append({
            'name': name,
            'condition': condition_fn,
            'consequence': consequence_fn
        })

    def query_symptom_matches(self, symptoms_detected):
        """Find diseases matching detected symptoms"""
        matches = {}
        for disease in [n for n, attr in self.graph.nodes(data=True)
                        if attr.get('type') == 'disease']:
            # Count matching symptoms
            predecessors = set(self.graph.predecessors(disease))
            matched = sum(1 for s in symptoms_detected if s in predecessors)
            if matched > 0:
                matches[disease] = matched / len(list(self.graph.predecessors(disease)))
        return matches

Step 3: Implement Neural Extraction

Create `src/extraction.py` - Use LLM to extract symptoms from patient text

from anthropic import Anthropic

class SymptomExtractor:
    def __init__(self):
        self.client = Anthropic()

    def extract_symptoms(self, patient_text: str) -> dict:
        """
        Use Claude to extract symptoms from patient description
        """
        prompt = f"""
Given the patient's description, extract:
1. List of symptoms mentioned
2. Duration and severity (if mentioned)
3. Associated context

Patient text: "{patient_text}"

Respond in JSON format:
{{
    "symptoms": ["symptom1", "symptom2"],
    "severity": "mild|moderate|severe|unknown",
    "duration_days": 3,
    "context": "additional relevant info"
}}
        """

        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )

        # Parse response
        import json
        response_text = message.content[0].text
        result = json.loads(response_text)
        return result

Step 4: Implement Inference Rules

Create `src/rules.py` - Define clinical decision rules

class TriageRules:
    # Red flag symptoms requiring immediate escalation
    RED_FLAGS = {
        'chest_pain': {'severity_threshold': 'moderate', 'escalate': True},
        'shortness_of_breath': {'severity_threshold': 'moderate', 'escalate': True},
        'severe_headache': {'severity_threshold': 'severe', 'escalate': True},
        'loss_of_consciousness': {'severity_threshold': 'mild', 'escalate': True},
        'uncontrolled_bleeding': {'severity_threshold': 'mild', 'escalate': True},
    }

    @staticmethod
    def check_red_flags(symptoms: list, severity: str) -> bool:
        """
        Check if any red flag symptoms are present
        """
        for symptom in symptoms:
            if symptom in TriageRules.RED_FLAGS:
                threshold = TriageRules.RED_FLAGS[symptom]['severity_threshold']
                if TriageRules._meets_threshold(severity, threshold):
                    return True
        return False

    @staticmethod
    def _meets_threshold(actual: str, required: str) -> bool:
        """Compare severity levels"""
        severity_order = {'mild': 1, 'moderate': 2, 'severe': 3}
        return severity_order.get(actual, 0) >= severity_order.get(required, 0)

    @staticmethod
    def determine_triage_level(symptoms: list, severity: str) -> str:
        """
        Determine triage level based on symptoms and severity
        Return: 'emergency' | 'urgent' | 'standard' | 'minor'
        """
        if TriageRules.check_red_flags(symptoms, severity):
            return 'emergency'
        elif severity == 'severe':
            return 'urgent'
        elif severity == 'moderate':
            return 'standard'
        else:
            return 'minor'

Step 5: Integrate Components

Create `src/triage.py` - Main system logic

from extraction import SymptomExtractor
from knowledge_graph import MedicalKnowledgeGraph
from rules import TriageRules

class MedicalTriageSystem:
    def __init__(self):
        self.extractor = SymptomExtractor()
        self.kg = MedicalKnowledgeGraph()

    def process_patient(self, patient_text: str) -> dict:
        """
        Full pipeline: extraction → KG reasoning → rules → output
        """
        # Step 1: Extract symptoms using neural network
        extracted = self.extractor.extract_symptoms(patient_text)
        symptoms = extracted['symptoms']
        severity = extracted['severity']

        # Step 2: Query knowledge graph
        disease_matches = self.kg.query_symptom_matches(symptoms)

        # Step 3: Apply safety rules
        triage_level = TriageRules.determine_triage_level(symptoms, severity)

        # Step 4: Generate response
        response = {
            'symptoms_detected': symptoms,
            'severity': severity,
            'possible_conditions': disease_matches,
            'triage_level': triage_level,
            'recommendation': self._generate_recommendation(triage_level),
            'explanation': self._explain_decision(symptoms, triage_level)
        }

        return response

    def _generate_recommendation(self, triage_level: str) -> str:
        recommendations = {
            'emergency': 'Go to Emergency Department immediately',
            'urgent': 'See doctor today or visit urgent care',
            'standard': 'Schedule appointment with primary care doctor',
            'minor': 'Monitor symptoms; seek care if worsening'
        }
        return recommendations.get(triage_level, 'Seek medical advice')

    def _explain_decision(self, symptoms: list, level: str) -> str:
        return f"Based on symptoms {symptoms} and severity, " \
               f"triage level is {level}"

# Usage
system = MedicalTriageSystem()
result = system.process_patient("I have severe chest pain and can't breathe")
print(result)

Step 6: Write Tests

Create `tests/test_integration.py` - Validate with real medical scenarios

import pytest
from src.triage import MedicalTriageSystem

class TestTriageSystem:
    @pytest.fixture
    def system(self):
        return MedicalTriageSystem()

    def test_emergency_scenario(self, system):
        """Test case: patient with red flag symptoms"""
        result = system.process_patient(
            "I have severe chest pain, shortness of breath, and I'm sweating"
        )
        assert result['triage_level'] == 'emergency'
        assert 'Go to Emergency Department' in result['recommendation']

    def test_minor_scenario(self, system):
        """Test case: patient with minor symptoms"""
        result = system.process_patient(
            "I have a slight headache and feeling a bit tired"
        )
        assert result['triage_level'] in ['minor', 'standard']

    def test_moderate_scenario(self, system):
        """Test case: patient with moderate symptoms"""
        result = system.process_patient(
            "I have a cough, low fever, and mild chest discomfort"
        )
        assert result['triage_level'] in ['standard', 'urgent']

# Run: pytest tests/test_integration.py

Step 7: Deploy and Monitor

Create a simple Streamlit UI for interaction

import streamlit as st
from src.triage import MedicalTriageSystem

st.title("Medical Triage System")
st.markdown("**Safety Notice:** This is an educational system. Always consult medical professionals.")

system = MedicalTriageSystem()

# Input
patient_input = st.text_area("Describe your symptoms:")

if st.button("Analyze Symptoms"):
    if patient_input:
        result = system.process_patient(patient_input)

        # Display results
        st.subheader("Triage Assessment")

        # Color-code by triage level
        level = result['triage_level']
        color_map = {
            'emergency': '🔴',
            'urgent': '🟠',
            'standard': '🟡',
            'minor': '🟢'
        }

        st.write(f"### {color_map[level]} Level: {level.upper()}")
        st.write(f"**Recommendation:** {result['recommendation']}")
        st.write(f"**Detected Symptoms:** {', '.join(result['symptoms_detected'])}")
        st.write(f"**Explanation:** {result['explanation']}")

Key Design Principles

Principle	Why It Matters	Example
Separation of Concerns	Neural (NLU) and symbolic (rules) are separate; easier to test and update	Extraction module doesn't know about rules
Explainability	Every decision is traceable; critical for healthcare	System shows which symptoms triggered escalation
Safety First	Rules enforce hard constraints that cannot be violated	Red flags always escalate, no exceptions
Testability	Every component can be tested independently and in combination	Unit tests for rules, integration tests for pipeline
Modularity	Easy to swap components (different LLM, different rules)	Can upgrade to Claude 4.0 without changing rules

Common Mistakes to Avoid

❌ Don't	✓ Do
Put all logic in the neural network. This makes the system opaque and unsafe.	Use neural for extraction, rules for decisions. This makes the system transparent and trustworthy.
Write rules in code if they change frequently. Store them in a database or config file instead.	Structure your rules so they can be updated without re-deploying code. Keep rules externalized and versioned.
Skip testing. "It works on my machine" doesn't work in healthcare or high-stakes domains.	Test extensively with real and synthetic medical scenarios. Validate every edge case and failure mode.

What's Next?

Now that you understand how to build a neuro-symbolic system, the next chapters show how to apply this approach to specific domains:

Chapter 5: Healthcare AI - clinical decision support in practice
Chapter 6: Legal AI - contract analysis and case law reasoning
Chapter 7: Insurance AI - fraud detection and risk assessment
Chapter 8: Pharmacy & Drug Discovery - personalized medicine

✓ Key Takeaways:

Neuro-symbolic systems separate neural perception from symbolic reasoning
This separation makes systems safer, more explainable, and easier to test
Start with a clear architecture diagram before coding
Implement rules as configuration, not hard-coded logic
Test extensively with realistic scenarios
In healthcare and other high-stakes domains, safety is non-negotiable

🎯 Ready for Assessment?

10 Questions

~10 mins

70% to Pass

Take Quiz Now →