peter thrill’s diabolical dialectic the module

module.py

“””
dspy.Dialectic v3 — Generic Hegelian dialectical reasoning module.

Works with ANY DSPy signature, like ChainOfThought or ReAct.
Follows the multi-signature orchestration pattern (Pattern 2 from DSPy architecture).

Usage:
# Works with any signature
dialectic = Dialectic(“question -> answer”)
dialectic = Dialectic(“question, context -> answer: int, confidence: float”)
dialectic = Dialectic(MyCustomSignature)

```
# Drop-in replacement for ChainOfThought in most cases
result = dialectic(question="How should we balance X and Y?")
print(result.answer)

# Access dialectical history if needed
print(result.dialectic_history)
```

Architecture:
Internally builds 5 Predict instances from the user’s signature:
- self.generate_thesis: user inputs -> thesis + assumptions
- self.negate: thesis -> antithesis (grounded in specific assumption)
- self.sublate: thesis + antithesis -> synthesis (with predictive bet)
- self.check_stability: synthesis -> stability verdict
- self.extract: final synthesis + user inputs -> user’s typed outputs

```
All 5 are discoverable by optimizers via named_parameters().
The extract step (stolen from ReAct) maps dialectical output back to
the user's typed output fields, preserving type safety.
```

“””

import dspy
from dspy.signatures.signature import ensure_signature
from typing import Optional, Literal

def _build_dialectic_signatures(user_signature):
“”“Build all internal signatures from the user’s signature.

```
This is the ReAct pattern: accept any signature, generate internal
signatures that reference the user's fields, and map back at the end.
"""
input_fields = user_signature.input_fields
output_field_names = list(user_signature.output_fields.keys())
output_desc = ", ".join(f"`{k}`" for k in output_field_names)
task_desc = user_signature.instructions or f"Given the inputs, produce {output_desc}."

# --- 1. THESIS: user's inputs -> thesis position ---
# The thesis is a committed position on what the outputs SHOULD be.
thesis_sig = (
    dspy.Signature(
        {**input_fields},
        f"You are the THESIS generator in a dialectical reasoning process.\n\n"
        f"The underlying task: {task_desc}\n\n"
        f"Your job: take a clear, committed POSITION on what {output_desc} should be. "
        f"Don't hedge. Actually commit to specific answers and explain your reasoning.\n\n"
        f"If context from a prior synthesis is provided, you MUST engage with it. "
        f"Quote the instability reason and explain what you're re-examining."
    )
    .append("prior_context", dspy.InputField(
        prefix="Prior Synthesis Context:",
        desc="Empty on first iteration. On subsequent iterations: the prior synthesis, "
             "its stability prediction, the verdict, and the instability reason. "
             "You MUST engage with this if present."
    ), type_=str)
    .append("prior_engagement", dspy.OutputField(
        prefix="Prior Engagement:",
        desc="null if first iteration (prior_context is empty). "
             "OTHERWISE REQUIRED: quote the instability reason verbatim, then state whether you are "
             "DEEPENING (same ground, higher resolution), PIVOTING (prior synthesis was reductive, "
             "trying different frame), or NARROWING (prior was right but too broad). "
             "This must visibly inform your thesis below."
    ), type_=Optional[str])
    .append("thesis", dspy.OutputField(
        prefix="Thesis:",
        desc=f"A committed position on {output_desc}. This is not a summary of options — "
             f"it's a specific claim about what the answer should be and why."
    ), type_=str)
    .append("thesis_reasoning", dspy.OutputField(
        prefix="Thesis Reasoning:",
        desc="The logic and assumptions this thesis depends on. Be explicit."
    ), type_=str)
    .append("key_assumptions", dspy.OutputField(
        prefix="Key Assumptions:",
        desc="The 2-4 load-bearing assumptions in your reasoning. Format: '1. [assumption] 2. [assumption]'. "
             "These are the specific claims that, if wrong, would collapse the thesis entirely. "
             "The negation step will target one of these — so be honest about what your argument actually depends on."
    ), type_=str)
)

# --- 2. NEGATE: thesis -> antithesis (grounded in specific assumption) ---
negate_sig = dspy.Signature(
    {**input_fields},
    f"You are the NEGATION step in a dialectical reasoning process.\n\n"
    f"The underlying task: {task_desc}\n\n"
    f"Your job: find where the thesis undermines ITSELF. Not an external objection — "
    f"identify the specific assumption that generates a contradiction when followed "
    f"to its logical conclusion.\n\n"
    f"You MUST quote the specific numbered assumption you're targeting. "
    f"If you can't point to one, your negation is not grounded."
).append("thesis", dspy.InputField(
    prefix="Thesis:",
), type_=str).append("thesis_reasoning", dspy.InputField(
    prefix="Thesis Reasoning:",
), type_=str).append("key_assumptions", dspy.InputField(
    prefix="Key Assumptions:",
), type_=str).append("targeted_assumption", dspy.OutputField(
    prefix="Targeted Assumption:",
    desc="QUOTE the specific numbered assumption you are targeting. "
         "Copy it exactly. Format: 'Assumption [N]: \"[exact quote]\"'"
), type_=str).append("self_destruction_chain", dspy.OutputField(
    prefix="Self-Destruction Chain:",
    desc="Trace the logical steps showing how the assumption destroys the thesis from within. "
         "Format as a chain: '[assumption] implies [X]. But [X] requires [Y]. "
         "And [Y] contradicts [the thesis's own claim that Z].' "
         "Every step must follow from the PRIOR step — no jumps. "
         "If you can't build an unbroken chain, the contradiction isn't real."
), type_=str).append("antithesis", dspy.OutputField(
    prefix="Antithesis:",
    desc="The position that emerges when the targeted assumption collapses. "
         "Not 'the opposite view' — what you're forced to conclude when the "
         "thesis's own logic is taken seriously."
), type_=str).append("contradiction", dspy.OutputField(
    prefix="Contradiction:",
    desc="The specific tension: thesis claims [A] but its own assumption [N] "
         "implies [not-A]. State both sides using the thesis's own terms."
), type_=str)

# --- 3. SUBLATE: thesis + antithesis -> synthesis (with bet) ---
sublate_sig = dspy.Signature(
    {**input_fields},
    f"You are the SUBLATION (aufhebung) step in a dialectical reasoning process.\n\n"
    f"The underlying task: {task_desc}\n\n"
    f"Your synthesis must NOT be a compromise or average. It must be a new concept "
    f"that EXPLAINS why the contradiction was necessary, preserving both moments "
    f"while transcending the framing that made them contradictory.\n\n"
    f"You must also make a FALSIFIABLE prediction about your synthesis's stability."
).append("thesis", dspy.InputField(
    prefix="Thesis:",
), type_=str).append("antithesis", dspy.InputField(
    prefix="Antithesis:",
), type_=str).append("contradiction", dspy.InputField(
    prefix="Contradiction:",
), type_=str).append("targeted_assumption", dspy.InputField(
    prefix="Targeted Assumption:",
), type_=str).append("shared_framing", dspy.OutputField(
    prefix="Shared Framing Exposed:",
    desc="The assumption that BOTH thesis and antithesis share — the invisible frame "
         "that made them appear contradictory. This is the thing being transcended. "
         "It should be something neither side questioned. "
         "Format: 'Both thesis and antithesis assume [X]. But [X] is the problem.' "
         "If you can't find a shared assumption, the contradiction may be genuine "
         "rather than dialectical."
), type_=str).append("synthesis", dspy.OutputField(
    prefix="Synthesis:",
    desc=f"A new position that dissolves the contradiction. Must explain WHY the "
         f"contradiction existed. Must imply specific answers for {output_desc}."
), type_=str).append("preserved_from_thesis", dspy.OutputField(
    prefix="Preserved From Thesis:",
    desc="What specific insight from the thesis survives? Quote the part that "
         "remains true under the new framing."
), type_=str).append("preserved_from_antithesis", dspy.OutputField(
    prefix="Preserved From Antithesis:",
    desc="What specific insight from the antithesis survives?"
), type_=str).append("transcended", dspy.OutputField(
    prefix="What Was Transcended:",
    desc="What assumption was revealed as inadequate?"
), type_=str).append("stability_prediction", dspy.OutputField(
    prefix="Stability Prediction:",
    desc="Make a FALSIFIABLE prediction about this synthesis's stability. "
         "Format: 'This synthesis is stable IF [specific condition]. "
         "It would become unstable IF [specific scenario] because [reason].' "
         "The stability check will evaluate this prediction directly — "
         "so make it concrete enough to be proven wrong."
), type_=str).append("vulnerability_type", dspy.OutputField(
    prefix="Vulnerability Type:",
    desc="One of: SELF_UNDERMINING, SCOPE_LIMITATION, DEEPER_ASSUMPTION, "
         "EMPIRICAL. Pick one, explain briefly."
), type_=str)

# --- 4. STABILITY CHECK: evaluates the bet ---
stability_sig = dspy.Signature(
    {**input_fields},
    f"You are evaluating a dialectical synthesis's stability.\n\n"
    f"The underlying task: {task_desc}\n\n"
    f"The synthesis made a prediction about its own stability. "
    f"Your job is to evaluate THAT prediction — does the stability condition hold? "
    f"Does the instability scenario obtain? Do not freelance."
).append("synthesis", dspy.InputField(
    prefix="Synthesis:",
), type_=str).append("stability_prediction", dspy.InputField(
    prefix="Stability Prediction:",
), type_=str).append("vulnerability_type", dspy.InputField(
    prefix="Vulnerability Type:",
), type_=str).append("transcended", dspy.InputField(
    prefix="What Was Transcended:",
), type_=str).append("prediction_evaluation", dspy.OutputField(
    prefix="Prediction Evaluation:",
    desc="Evaluate the synthesis's stability prediction directly. "
         "Does condition [X] hold? Does scenario [Y] obtain? "
         "Cite specific aspects of the synthesis."
), type_=str).append("verdict", dspy.OutputField(
    prefix="Stability Verdict:",
    desc="One of: STABLE, SELF_UNDERMINING, INCOMPLETE, REDUCTIVE, DEEPER_ASSUMPTION."
), type_=str).append("instability_reason", dspy.OutputField(
    prefix="Instability Reason:",
    desc="If not STABLE: the specific tension that remains. Must be concrete enough "
         "for the next iteration to target. If STABLE: 'none'."
), type_=str)

# --- 5. EXTRACT: synthesis -> user's typed outputs ---
# This is the ReAct pattern: map from internal representation
# back to the user's actual typed output fields.
extract_sig = dspy.Signature(
    {**input_fields, **user_signature.output_fields},
    f"Extract the final answer from a dialectical reasoning process.\n\n"
    f"The underlying task: {task_desc}\n\n"
    f"You have been given the final synthesis from a dialectical analysis. "
    f"Your job is to extract the specific values for {output_desc} that the "
    f"synthesis implies. Be faithful to the synthesis — do not add your own reasoning."
).append("dialectic_synthesis", dspy.InputField(
    prefix="Dialectic Synthesis:",
    desc="The final synthesis from dialectical reasoning. Extract answers from this."
), type_=str).append("dialectic_history_summary", dspy.InputField(
    prefix="Dialectic History:",
    desc="Summary of the dialectical process: iterations, key contradictions resolved."
), type_=str)

return thesis_sig, negate_sig, sublate_sig, stability_sig, extract_sig
```

class Dialectic(dspy.Module):
“”“Generic Hegelian dialectical reasoning.

```
Works with any DSPy signature. Drives into contradiction, synthesizes,
iterates until stable, then extracts typed outputs.

Architecture (ReAct pattern — multi-signature orchestration):
    5 internal Predict instances, all discoverable by optimizers:
    - generate_thesis: take a committed position
    - negate: find internal contradiction (grounded in specific assumption)
    - sublate: aufhebung with predictive bet
    - check_stability: evaluate the bet (structured taxonomy)
    - extract: map synthesis -> user's typed outputs

Optimization surface:
    Each Predict has independently tunable demos and instructions.
    An optimizer could learn e.g. better negation strategies or
    sublation demos that avoid compromise.

Args:
    signature: Any DSPy signature (string or class).
    max_iterations: Cap on dialectical turns (default 3).
    check_stability: Whether to dynamically terminate on convergence.
    **config: Passed to all internal Predict instances (e.g. temperature).
"""

def __init__(self, signature, max_iterations=3, check_stability=True, **config):
    super().__init__()
    self.user_signature = ensure_signature(signature)
    self.max_iterations = max_iterations
    self._check_stability = check_stability

    # Build internal signatures from user's signature
    (thesis_sig, negate_sig, sublate_sig,
     stability_sig, extract_sig) = _build_dialectic_signatures(self.user_signature)

    # All stored as instance attrs -> discoverable by named_parameters()
    self.generate_thesis = dspy.Predict(thesis_sig, **config)
    self.negate = dspy.Predict(negate_sig, **config)
    self.sublate = dspy.Predict(sublate_sig, **config)
    self.check_stability = dspy.Predict(stability_sig, **config)
    # Extract uses ChainOfThought bc it needs to reason about
    # mapping synthesis -> typed outputs
    self.extract = dspy.ChainOfThought(extract_sig, **config)

def forward(self, **kwargs):
    # Separate user inputs from any privileged kwargs
    user_inputs = {k: v for k, v in kwargs.items()
                   if k in self.user_signature.input_fields}

    history = []
    prior_context = ""

    for i in range(self.max_iterations):
        # --- thesis ---
        thesis_result = self.generate_thesis(
            **user_inputs,
            prior_context=prior_context,
        )

        # --- negation (must target specific assumption) ---
        negate_result = self.negate(
            **user_inputs,
            thesis=thesis_result.thesis,
            thesis_reasoning=thesis_result.thesis_reasoning,
            key_assumptions=thesis_result.key_assumptions,
        )

        # --- sublation (with predictive bet) ---
        sublate_result = self.sublate(
            **user_inputs,
            thesis=thesis_result.thesis,
            antithesis=negate_result.antithesis,
            contradiction=negate_result.contradiction,
            targeted_assumption=negate_result.targeted_assumption,
        )

        moment = {
            "iteration": i,
            "prior_engagement": thesis_result.prior_engagement,
            "thesis": thesis_result.thesis,
            "key_assumptions": thesis_result.key_assumptions,
            "targeted_assumption": negate_result.targeted_assumption,
            "self_destruction_chain": negate_result.self_destruction_chain,
            "antithesis": negate_result.antithesis,
            "contradiction": negate_result.contradiction,
            "shared_framing": sublate_result.shared_framing,
            "synthesis": sublate_result.synthesis,
            "preserved_from_thesis": sublate_result.preserved_from_thesis,
            "preserved_from_antithesis": sublate_result.preserved_from_antithesis,
            "transcended": sublate_result.transcended,
            "stability_prediction": sublate_result.stability_prediction,
            "vulnerability_type": sublate_result.vulnerability_type,
        }

        # --- stability check (evaluates the bet) ---
        if self._check_stability and i < self.max_iterations - 1:
            stability_result = self.check_stability(
                **user_inputs,
                synthesis=sublate_result.synthesis,
                stability_prediction=sublate_result.stability_prediction,
                vulnerability_type=sublate_result.vulnerability_type,
                transcended=sublate_result.transcended,
            )

            moment["prediction_evaluation"] = stability_result.prediction_evaluation
            moment["verdict"] = stability_result.verdict
            moment["instability_reason"] = stability_result.instability_reason

            if stability_result.verdict.strip().upper() == "STABLE":
                history.append(moment)
                break

            # spiral context for next iteration
            prior_context = (
                f"PRIOR SYNTHESIS: {sublate_result.synthesis}\n\n"
                f"STABILITY PREDICTION: {sublate_result.stability_prediction}\n\n"
                f"VERDICT: {stability_result.verdict}\n\n"
                f"INSTABILITY REASON: {stability_result.instability_reason}\n\n"
                f"PREDICTION EVALUATION: {stability_result.prediction_evaluation}\n\n"
                f"You MUST engage with this instability. Quote the instability reason."
            )
        else:
            moment["verdict"] = "FINAL_ITERATION"
            moment["instability_reason"] = "max iterations reached"

        history.append(moment)

    # --- extract: map synthesis -> user's typed outputs ---
    final = history[-1]

    # Build a summary of the dialectical process
    history_summary_parts = []
    for m in history:
        history_summary_parts.append(
            f"Iteration {m['iteration']}: "
            f"Thesis challenged assumption [{m['targeted_assumption'][:80]}...]. "
            f"Resolved via: {m.get('shared_framing', 'N/A')[:80]}... "
            f"Verdict: {m.get('verdict', 'N/A')}"
        )
    history_summary = "\n".join(history_summary_parts)

    extraction = self.extract(
        **user_inputs,
        dialectic_synthesis=final["synthesis"],
        dialectic_history_summary=history_summary,
    )

    # Build final Prediction with user's output fields + dialectic metadata
    output_kwargs = {}
    for field_name in self.user_signature.output_fields:
        if hasattr(extraction, field_name):
            output_kwargs[field_name] = getattr(extraction, field_name)

    return dspy.Prediction(
        **output_kwargs,
        # metadata (accessible but not part of the typed contract)
        dialectic_history=history,
        dialectic_iterations=len(history),
        dialectic_converged=final.get("verdict", "").strip().upper() == "STABLE",
        dialectic_final_synthesis=final["synthesis"],
    )
```

# —————————————————————————

# Convenience: Dialectic with ChainOfThought internals

# —————————————————————————

class ChainOfDialectic(Dialectic):
“”“Same as Dialectic but uses ChainOfThought for all internal Predicts.

```
More expensive (extra reasoning tokens per step) but may produce
better dialectical reasoning, especially for complex tasks.

The reasoning fields are independently optimizable — an optimizer
could learn different reasoning styles for thesis vs negation vs sublation.
"""

def __init__(self, signature, max_iterations=3, check_stability=True, **config):
    # Initialize Module but NOT Dialectic's __init__
    dspy.Module.__init__(self)
    self.user_signature = ensure_signature(signature)
    self.max_iterations = max_iterations
    self._check_stability = check_stability

    (thesis_sig, negate_sig, sublate_sig,
     stability_sig, extract_sig) = _build_dialectic_signatures(self.user_signature)

    # All ChainOfThought instead of Predict
    self.generate_thesis = dspy.ChainOfThought(thesis_sig, **config)
    self.negate = dspy.ChainOfThought(negate_sig, **config)
    self.sublate = dspy.ChainOfThought(sublate_sig, **config)
    self.check_stability = dspy.ChainOfThought(stability_sig, **config)
    self.extract = dspy.ChainOfThought(extract_sig, **config)
```

# —————————————————————————

# Usage examples

# —————————————————————————

if **name** == “**main**”:
lm = dspy.LM(“openai/gpt-4o-mini”)
dspy.configure(lm=lm)

```
# --- Example 1: Simple signature ---
dialectic = Dialectic("question -> answer")
result = dialectic(question="How should society balance individual privacy with collective security?")

print(f"Answer: {result.answer}")
print(f"Converged: {result.dialectic_converged} in {result.dialectic_iterations} iteration(s)")
print()

# --- Example 2: Complex typed signature ---
class PolicyAnalysis(dspy.Signature):
    """Analyze a policy proposal and assess its viability."""
    proposal: str = dspy.InputField(desc="the policy proposal to analyze")
    context: str = dspy.InputField(desc="relevant political/economic context")
    assessment: str = dspy.OutputField(desc="detailed assessment of viability")
    recommendation: str = dspy.OutputField(desc="one of: ADOPT, MODIFY, REJECT")
    confidence: float = dspy.OutputField(desc="confidence score 0-1")

dialectic = Dialectic(PolicyAnalysis, max_iterations=2)
result = dialectic(
    proposal="Universal basic income at $1000/month for all adults",
    context="Post-pandemic economy with high automation anxiety"
)

print(f"Assessment: {result.assessment}")
print(f"Recommendation: {result.recommendation}")
print(f"Confidence: {result.confidence}")
print()

# --- Example 3: Composable with other DSPy modules ---
class DialecticalRAG(dspy.Module):
    """RAG pipeline that uses dialectical reasoning for answer generation."""
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Predict("question -> passages: list[str]")
        self.answer = Dialectic("question, passages: list[str] -> answer, confidence: float")

    def forward(self, question):
        passages = self.retrieve(question=question).passages
        return self.answer(question=question, passages=passages)

# named_parameters() would find:
# ("retrieve", <Predict>)                    -- the retriever
# ("answer.generate_thesis", <Predict>)      -- dialectic thesis
# ("answer.negate", <Predict>)               -- dialectic negation
# ("answer.sublate", <Predict>)              -- dialectic sublation
# ("answer.check_stability", <Predict>)      -- dialectic stability
# ("answer.extract.predict", <Predict>)      -- dialectic extraction (CoT)
#
# All independently optimizable by BootstrapFewShot, MIPRO, etc.

# --- Example 4: With optimization ---
# from dspy.teleprompt import BootstrapFewShot
#
# def metric(example, prediction):
#     return prediction.answer == example.answer
#
# optimizer = BootstrapFewShot(metric=metric, max_bootstrapped_demos=3)
# optimized = optimizer.compile(
#     DialecticalRAG(),
#     trainset=training_examples,
# )
#
# # The optimizer will independently tune demos for:
# # - the retriever (what good retrieval looks like)
# # - thesis generation (what good initial positions look like)
# # - negation (what good internal contradictions look like)
# # - sublation (what good synthesis looks like)
# # - extraction (what faithful extraction looks like)
#
# optimized.save("dialectical_rag.json")

# --- Print optimizer surface ---
rag = DialecticalRAG()
print("Optimizer surface (named_predictors):")
for name, pred in rag.named_predictors():
    print(f"  {name}: {pred.signature.signature}")
```