Starphix
Agentic Engineering / Sources
The Lab / Sources / Agentic Engineering / Frontiers/Neurosymbolic-Integration.md

Neurosymbolic Integration

Combining neural networks with symbolic reasoning for robust, interpretable, and verifiable agents.


Neurosymbolic Integration Taxonomy

Neurosymbolic Integration
├── Neural → Symbolic
│   ├── Neural Perception → Symbolic Reasoning
│   ├── Neural Program Synthesis
│   └── Neural Theorem Proving
├── Symbolic → Neural
│   ├── Symbolic Knowledge → Neural Priors
│   ├── Symbolic Constraints → Neural Training
│   └── Symbolic Search → Neural Guidance
├── Hybrid Architectures
│   ├── Neural-Symbolic Machines
│   ├── Neuro-Symbolic Concept Learners
│   ├── Differentiable Reasoning
│   └── Neural-Symbolic Program Synthesis
├── Reasoning Integration
│   ├── Neural Theorem Proving
│   ├── Neural-Symbolic RL
│   ├── Neural-Symbolic Planning
│   └── Neural-Symbolic Verification
└── Verification & Explanation
    ├── Symbolic Verification of Neural Nets
    ├── Neural Network Explanation via Symbols
    ├── Counterfactual Reasoning
    └── Causal Reasoning

Neural → Symbolic

Neural Program Synthesis

class NeuralProgramSynthesizer:
    def __init__(self):
        self.encoder = CodeEncoder()
        self.decoder = ProgramDecoder()
        self.executor = ProgramExecutor()

    async def synthesize(self, spec: Specification) -> Program:
        # 1. Encode specification
        spec_embedding = self.encoder.encode(spec)

        # 2. Generate candidate programs (beam search)
        candidates = await self.decoder.beam_search(
            spec_embedding, beam_width=10, max_length=100
        )

        # 3. Execute and verify
        for program in candidates:
            try:
                result = await self.executor.execute(program, spec.test_cases)
                if result.passed:
                    return program
            except Exception:
                continue

        raise SynthesisFailed("No valid program found")

Neural Theorem Proving

class NeuralTheoremProver:
    def __init__(self):
        self.encoder = TheoremEncoder()
        self.tactic_generator = TacticGenerator()
        self.proof_search = ProofSearch()

    async def prove(self, theorem: Theorem) -> Proof | None:
        # 1. Encode theorem statement
        theorem_emb = self.encoder.encode(theorem.statement)

        # 2. Generate tactic sequence
        tactics = await self.tactic_generator.generate(theorem_emb)

        # 3. Search for proof
        proof = await self.proof_search.search(
            theorem, tactics, max_depth=20
        )

        return proof

Neural Theorem Proving with Lean/Coq

class LeanTheoremProver:
    def __init__(self):
        self.lean = LeanInterface()
        self.tactic_model = TacticModel()

    async def prove(self, theorem: str) -> ProofResult:
        # 1. Parse theorem
        parsed = self.parse_lean(theorem)

        # 2. Generate tactic sequence
        tactics = []
        state = self.lean.get_initial_state()

        for _ in range(max_steps):
            # Neural tactic prediction
            tactic = await self.tactic_model.predict(state)
            tactics.append(tactic)

            # Apply in Lean
            result = self.lean.apply_tactic(tactic)
            if result.proof_complete:
                return ProofResult(proof=tactics, success=True)
            state = result.new_state

        return ProofResult(success=False)

Symbolic → Neural

Symbolic Knowledge → Neural Priors

class KnowledgeDistillation:
    def __init__(self, knowledge_base: KnowledgeBase):
        self.kb = knowledge_base

    def extract_rules(self) -> list[Rule]:
        """Extract logical rules from knowledge base"""
        rules = []
        for fact in self.kb.facts:
            # Convert to Horn clauses
            rules.append(self.fact_to_rule(fact))
        return rules

    def create_prior_network(self, rules: list[Rule]) -> PriorNetwork:
        """Create neural network with symbolic priors"""
        class PriorNetwork(nn.Module):
            def __init__(self, rules):
                super().__init__()
                self.rules = rules
                # Soft logic layer
                self.rule_weights = nn.Parameter(torch.ones(len(rules)))

            def forward(self, x):
                # Apply rules as soft constraints
                for i, rule in enumerate(self.rules):
                    # Soft logic: rule satisfaction as differentiable
                    satisfaction = self.eval_rule(rule, x)
                    x = x + self.rule_weights[i] * satisfaction
                return x

        return PriorNetwork(rules)

class ConstraintSatisfactionLayer(nn.Module):
    """Differentiable constraint satisfaction"""
    def __init__(self, constraints: list[Constraint]):
        super().__init__()
        self.constraints = constraints
        self.penalty_weight = nn.Parameter(torch.tensor(1.0))

    def forward(self, x: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
        # Compute constraint violations
        violations = []
        for constraint in self.constraints:
            violation = constraint.violation(x)
            violations.append(violation)

        violation_tensor = torch.stack(violations)
        penalty = self.penalty_weight * violation_tensor.sum()

        return x, penalty

Hybrid Architectures

Neural-Symbolic Machines

class NeuralSymbolicMachine:
    def __init__(self):
        self.perception = NeuralPerception()
        self.reasoner = SymbolicReasoner()
        self.interface = NeuralSymbolicInterface()

    async def run(self, input: Input) -> Output:
        # 1. Neural perception
        perceptions = await self.perception.process(input)

        # 2. Ground to symbolic
        symbols = self.interface.ground(perceptions)

        # 3. Symbolic reasoning
        conclusions = await self.reasoner.reason(symbols)

        # 4. Execute actions
        actions = self.plan_actions(conclusions)

        return Output(actions=actions, reasoning=conclusions)

class NeuralSymbolicInterface:
    def __init__(self):
        self.grounder = NeuralGrounder()
        self.lifter = SymbolicLifter()

    def ground(self, perceptions: Perceptions) -> SymbolicState:
        """Ground neural perceptions to symbolic representations"""
        entities = self.grounder.ground_entities(perceptions)
        relations = self.grounder.ground_relations(perceptions)
        return SymbolicState(entities=entities, relations=relations)

    def lift(self, symbolic: SymbolicState) -> NeuralRepresentation:
        """Lift symbolic to neural for further processing"""
        return self.lifter.lift(symbolic)

class DifferentiableReasoning:
    def __init__(self, knowledge_base: KnowledgeBase):
        self.kb = knowledge_base
        self.reasoner = DifferentiableReasoner()

    def forward(self, query: Query) -> tuple[Answer, Proof]:
        # Differentiable forward chaining
        facts = self.kb.facts
        rules = self.kb.rules

        # Soft forward chaining
        for _ in range(max_depth):
            new_facts = self.soft_forward_chain(facts, rules)
            facts = facts + new_facts

        # Extract answer with confidence
        answer = self.extract_answer(query, facts)
        proof = self.extract_proof(query, facts)

        return answer, proof

Neural-Symbolic Program Synthesis

class NeuroSymbolicProgramSynthesizer:
    def __init__(self):
        self.encoder = SpecificationEncoder()
        self.sketch_generator = SketchGenerator()
        self.synthesizer = ProgramSynthesizer()
        self.verifier = ProgramVerifier()

    async def synthesize(self, spec: Specification) -> Program:
        # 1. Generate program sketch
        sketch = await self.sketch_generator.generate(spec)

        # 2. Fill holes with neural synthesis
        program = await self.synthesizer.fill_sketch(sketch)

        # 3. Verify with symbolic verification
        verified = await self.verifier.verify(program, spec)
        if not verified:
            # Counterexample-guided refinement
            return await self.refine(program, spec)

        return program

class SketchGenerator:
    def __init__(self):
        self.encoder = SpecificationEncoder()
        self.decoder = SketchDecoder()

    def generate(self, spec: Specification) -> Sketch:
        # Generate program sketch with holes
        spec_emb = self.encoder.encode(spec)
        sketch = self.decoder.decode(spec_emb)
        return sketch

Reasoning Integration

Neural-Symbolic RL

class NeuroSymbolicRL:
    def __init__(self):
        self.policy_net = PolicyNetwork()
        self.value_net = ValueNetwork()
        self.reasoner = SymbolicReasoner()
        self.reward_shaper = RewardShaper()

    async def act(self, state: State) -> Action:
        # 1. Neural policy proposes actions
        neural_actions = self.policy_net(state)

        # 2. Symbolic reasoning filters/validates
        valid_actions = await self.reasoner.filter_actions(state, neural_actions)

        # 3. Select best valid action
        scores = self.value_net(state, valid_actions)
        return valid_actions[scores.argmax()]

    async def learn(self, trajectory: Trajectory):
        # 1. Standard RL update
        await self.rl_update(trajectory)

        # 2. Symbolic reward shaping
        shaped_rewards = self.reward_shaper.shape(trajectory)

        # 3. Symbolic policy distillation
        await self.distill_symbolic_policy(trajectory)

Neural-Symbolic Planning

class NeuroSymbolicPlanner:
    def __init__(self):
        self.neural_planner = NeuralPlanner()
        self.symbolic_planner = SymbolicPlanner(PDDLPlanner())
        self.verifier = PlanVerifier()

    async def plan(self, state: State, goal: Goal) -> Plan:
        # 1. Neural planner proposes high-level plan
        neural_plan = await self.neural_planner.plan(state, goal)

        # 2. Symbolic planner refines
        symbolic_plan = await self.symbolic_planner.refine(neural_plan)

        # 3. Verify
        if not await self.verifier.verify(symbolic_plan):
            # Refinement loop
            return await self.refine_plan(state, goal)

        return symbolic_plan

    async def refine_plan(self, state: State, goal: Goal) -> Plan:
        # Counterexample-guided refinement
        for _ in range(max_refinements):
            try:
                plan = await self.symbolic_planner.plan(state, goal)
                if await self.verifier.verify(plan):
                    return plan
            except VerificationError as e:
                # Add constraint and retry
                self.add_constraint(e.counterexample)
        raise PlanningFailed()

Neural-Symbolic Verification

class NeuroSymbolicVerifier:
    def __init__(self):
        self.neural_verifier = NeuralVerifier()
        self.symbolic_verifier = SymbolicVerifier(Z3Solver())

    async def verify(self, program: Program, spec: Specification) -> VerificationResult:
        # 1. Neural verification (fast, approximate)
        neural_result = await self.neural_verifier.verify(program, spec)

        if neural_result.confidence > 0.95:
            return neural_result

        # 2. Symbolic verification (exact, slower)
        symbolic_result = await self.symbolic_verifier.verify(program, spec)

        return VerificationResult(
            proven=symbolic_result.proven,
            counterexample=symbolic_result.counterexample,
            proof=symbolic_result.proof
        )

class NeuralVerifier:
    def __init__(self):
        self.model = VerificationModel()

    async def verify(self, program: Program, spec: Specification) -> VerificationResult:
        # Neural network predicts verification result
        code_emb = self.encoder.encode(program.code)
        spec_emb = self.encoder.encode(spec)

        result = self.model(code_emb, spec_emb)

        return VerificationResult(
            proven=result.proven > 0.5,
            confidence=result.confidence,
            counterexample=result.counterexample if not result.proven else None
        )

Verification & Explanation

Symbolic Verification of Neural Networks

class NeuralNetworkVerifier:
    def __init__(self):
        self.abstractor = NetworkAbstractor()
        self.solver = Z3Solver()

    async def verify_robustness(self, network: nn.Module, 
                                 input_bounds: Bounds,
                                 property: Property) -> VerificationResult:
        # 1. Abstract network to linear constraints
        constraints = self.abstractor.abstract(network)

        # 2. Add input bounds
        constraints.add_bounds(input_bounds)

        # 3. Add property as constraint
        constraints.add_property(property)

        # 4. Solve
        result = await self.solver.solve(constraints)

        return VerificationResult(
            verified=result == unsat,
            counterexample=self.extract_counterexample() if result == sat else None
        )

    async def verify_lipschitz(self, network: nn.Module, 
                               epsilon: float) -> float:
        """Compute Lipschitz constant"""
        return await self.compute_lipschitz(network, epsilon)

Neural Network Explanation via Symbols

class SymbolicExplainer:
    def __init__(self):
        self.concept_extractor = ConceptExtractor()
        self.rule_extractor = RuleExtractor()
        self.counterfactual = CounterfactualGenerator()

    def explain(self, model: Model, input: Input, prediction: Prediction) -> Explanation:
        # 1. Extract concepts
        concepts = self.concept_extractor.extract(model, input)

        # 2. Extract decision rules
        rules = self.rule_extractor.extract(model, input)

        # 3. Generate counterfactuals
        counterfactuals = self.counterfactual.generate(input, prediction)

        return Explanation(
            concepts=concepts,
            rules=rules,
            counterfactuals=counterfactuals,
            attribution=self.compute_attribution(input)
        )

    def extract_concepts(self, model: Model, input: Input) -> list[Concept]:
        # TCAV (Testing with Concept Activation Vectors)
        concepts = self.extract_concepts_tcav(model, input)
        return concepts

    def extract_rules(self, model: Model, input: Input) -> list[Rule]:
        # Anchors: High-precision rules
        anchors = self.anchor_explainer.explain(input, model.predict)
        return anchors

Counterfactual Reasoning

class CounterfactualReasoner:
    def __init__(self, model: Model, world_model: WorldModel):
        self.model = model
        self.world = world_model

    def generate_counterfactual(self, input: Input, target: Prediction) -> Counterfactual:
        """Find minimal change to input to change prediction"""
        # Gradient-based search
        x = input.clone().requires_grad_(True)
        target_class = target.class_id

        for _ in range(max_iter):
            output = model(input)
            loss = F.cross_entropy(output, target)
            loss.backward()

            # Gradient step
            input.data -= lr * input.grad.sign()
            input.data = clamp(input, input_bounds)

        return Counterfactual(original=input, counterfactual=input.data)

    def causal_reasoning(self, fact: Fact, query: CausalQuery) -> CausalAnswer:
        # Use world model for causal reasoning
        # do-calculus / structural causal models
        effect = self.world_model.intervene(query.intervention)
        return CausalAnswer(effect=effect, confidence=0.9)

Causal Reasoning

class CausalReasoner:
    def __init__(self):
        self.scm = StructuralCausalModel()
        self.estimator = CausalEstimator()

    def identify_effect(self, treatment: Variable, outcome: Variable, 
                        graph: CausalGraph) -> IdentificationResult:
        """Identify causal effect using do-calculus"""
        # Check backdoor criterion
        adjustment_set = self.find_adjustment_set(treatment, outcome, graph)

        if adjustment_set is None:
            return IdentificationResult(identifiable=False)

        return IdentificationResult(
            identifiable=True,
            adjustment_set=adjustment_set,
            formula=self.derive_formula(treatment, outcome, adjustment_set)
        )

    def estimate_effect(self, data: DataFrame, treatment: str, outcome: str,
                       adjustment_set: list[str]) -> CausalEffect:
        # Use multiple estimators
        estimates = {
            'backdoor': self.estimator.backdoor(data, treatment, outcome, adjustment_set),
            'iv': self.estimator.instrumental_variable(data, treatment, outcome),
            'frontdoor': self.estimator.frontdoor(data, treatment, outcome),
            'dml': self.estimator.double_ml(data, treatment, outcome, adjustment_set)
        }

        # Aggregate
        return CausalEffect(
            ate=statistics.mean([e.ate for e in estimates.values()]),
            ci=self.compute_ci([e.ate for e in estimates.values()]),
            estimates=estimates
        )

Integration Patterns

Pattern 1: Neural Perception + Symbolic Reasoning

class PerceptionReasoningAgent:
    def __init__(self):
        self.perception = NeuralPerception()
        self.reasoner = SymbolicReasoner()
        self.interface = NeuroSymbolicInterface()

    async def run(self, observation: Observation) -> Action:
        # Perceive
        perceptions = await self.perception.process(observation)

        # Ground to symbolic
        symbolic_state = self.interface.ground(perceptions)

        # Reason
        conclusions = await self.reasoner.reason(symbolic_state)

        # Plan
        plan = await self.planner.plan(conclusions)

        return plan.first_action()

Pattern 2: Symbolic Specification + Neural Synthesis

class SpecToCode:
    def __init__(self):
        self.spec_parser = SpecParser()
        self.neural_synthesizer = NeuralSynthesizer()
        self.symbolic_verifier = Verifier()

    async def generate(self, spec: Spec) -> Code:
        # 1. Parse spec to logical form
        logic_form = self.spec_parser.parse(spec)

        # 2. Neural synthesis with symbolic constraints
        code = await self.neural_synthesizer.generate(logic_form)

        # 3. Verify
        if not await self.verifier.verify(code, spec):
            return await self.repair(code, spec)

        return code

Pattern 3: Neural Search + Symbolic Verification

class NeuralSymbolicSearch:
    def __init__(self):
        self.neural_search = NeuralSearch()
        self.symbolic_verifier = SymbolicVerifier()

    async def search(self, problem: Problem) -> Solution:
        # Neural search for candidates
        candidates = await self.neural_search.search(problem, beam_width=10)

        # Symbolic verification
        for candidate in candidates:
            if await self.symbolic_verifier.verify(candidate):
                return candidate

        # Refinement loop
        return await self.refine_search(problem, candidates)


Sources: "Neural-Symbolic Learning and Reasoning" (Besold et al.), "Neural-Symbolic Integration" surveys, "Deep Learning for Symbolic Mathematics" (Lample & Charton), "Neural Theorem Proving" (Polu et al.), "Neural-Symbolic Machines" (Liang et al.), "Differentiable Reasoning" (Rocktäschel & Riedel), "Neural Theorem Proving" (Wang et al.), "DeepMath" (Bunel et al.), "HolStep" (Kaliszyk et al.), "Lean" (de Moura et al.), "Coq" (Bertot & Castéran), "Neural Theorem Proving" surveys, "Neural Program Synthesis" (Balog et al.), "RobustFill" (Devlin et al.), "DeepCoder" (Balog et al.), "Neural Program Synthesis" (Ellis et al.), "DreamCoder" (Ellis et al.), "Neurosymbolic AI" surveys, "Differentiable Reasoning" (Evans & Grefenstette), "Neural-Symbolic VQA" (Yi et al.), "NS-CL" (Mao et al.), "Neural-Symbolic RL" (Garnelo et al.), "Neural-Symbolic Planning" (Liu et al.), "Neural-Symbolic Verification" (Zhang et al.).

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown